mvcc-key-value-store

Answer

File: mvcc-kv-store/mvcc.go — The core MVCC key-value store implementation.

package mvcc

import (
    "sync"
    "time"
)

// Version represents a single version of a key-value pair committed by a transaction.
type Version struct {
    Timestamp uint64
    Value     string
    Deleted   bool
}

// MVCCStore is a concurrent key-value store using multi-version concurrency control.
type MVCCStore struct {
    mu          sync.RWMutex
    versions    map[string][]*Version // key → versions sorted by timestamp ascending
    nextWriteTS uint64                // next transaction timestamp
    activeSnaps map[uint64]int        // snapshot timestamp → reference count
    closed      chan struct{}
    wg          sync.WaitGroup
}

func New() *MVCCStore {
    s := &MVCCStore{
        versions:    make(map[string][]*Version),
        nextWriteTS: 2, // start at 2 so Snapshot ts = nextWriteTS-1 = 1
        activeSnaps: make(map[uint64]int),
        closed:      make(chan struct{}),
    }
    s.wg.Add(1)
    go s.gcLoop()
    return s
}

func (s *MVCCStore) Close() {
    close(s.closed)
    s.wg.Wait()
}

// Snapshot creates a read snapshot observing all writes committed at or before
// its timestamp. The snapshot ts = last allocated write TS, so any concurrent
// commit gets a strictly greater TS and is invisible — guaranteeing snapshot isolation.
func (s *MVCCStore) Snapshot() *Snapshot {
    s.mu.Lock()
    ts := s.nextWriteTS - 1
    s.activeSnaps[ts]++
    s.mu.Unlock()
    return &Snapshot{
        store:     s,
        timestamp: ts,
        writes:    make(map[string]*Version),
    }
}

type Snapshot struct {
    store     *MVCCStore
    timestamp uint64
    writes    map[string]*Version // local write buffer for read-your-writes
}

// Get checks local writes first (read-your-writes), then scans global versions
// newest-first for a version with timestamp <= snapshot timestamp.
func (snap *Snapshot) Get(key string) (string, bool) {
    if v, ok := snap.writes[key]; ok {
        if v.Deleted { return "", false }
        return v.Value, true
    }
    snap.store.mu.RLock()
    defer snap.store.mu.RUnlock()
    versions, ok := snap.store.versions[key]
    if !ok || len(versions) == 0 { return "", false }
    for i := len(versions) - 1; i >= 0; i-- {
        v := versions[i]
        if v.Timestamp <= snap.timestamp {
            if v.Deleted { return "", false }
            return v.Value, true
        }
    }
    return "", false
}

func (snap *Snapshot) Set(key, value string) {
    snap.writes[key] = &Version{Value: value, Deleted: false}
}

func (snap *Snapshot) Delete(key string) {
    snap.writes[key] = &Version{Deleted: true}
}

// Commit assigns a single fresh timestamp to all buffered writes atomically.
func (snap *Snapshot) Commit() {
    if len(snap.writes) == 0 { snap.Release(); return }
    snap.store.mu.Lock()
    ts := snap.store.nextWriteTS
    snap.store.nextWriteTS++
    for key, v := range snap.writes {
        v.Timestamp = ts
        snap.store.versions[key] = append(snap.store.versions[key], v)
    }
    snap.store.mu.Unlock()
    snap.Release()
}

func (snap *Snapshot) Release() {
    snap.store.mu.Lock()
    snap.store.activeSnaps[snap.timestamp]--
    if snap.store.activeSnaps[snap.timestamp] <= 0 {
        delete(snap.store.activeSnaps, snap.timestamp)
    }
    snap.store.mu.Unlock()
}

// --- Background GC ---

func (s *MVCCStore) gcLoop() {
    defer s.wg.Done()
    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-s.closed: return
        case <-ticker.C: s.gc()
        }
    }
}

func (s *MVCCStore) gc() {
    s.mu.Lock()
    defer s.mu.Unlock()
    // Find minimum active snapshot timestamp
    var minTS *uint64
    for ts := range s.activeSnaps {
        if minTS == nil || ts < *minTS { t := ts; minTS = &t }
    }
    if minTS == nil { return }

    for key, versions := range s.versions {
        if len(versions) <= 1 { continue }
        keep := make([]*Version, 0, len(versions))
        newest := versions[len(versions)-1]
        for _, v := range versions {
            if v == newest || v.Timestamp >= *minTS {
                keep = append(keep, v)
            }
        }
        if len(keep) < len(versions) { s.versions[key] = keep }
    }
}

Key design decisions:

1. Timestamp scheme: nextWriteTS starts at 2. Snapshot() uses nextWriteTS - 1 as its read timestamp (the last allocated write TS). Commit() uses nextWriteTS then increments it. This guarantees every new commit gets a timestamp strictly greater than any existing snapshot's timestamp.

2. Read-your-writes: Each Snapshot carries a local writes map that shadows the global store. Get checks the local buffer first.

3. Snapshot isolation: Since snapshot timestamps reflect the last allocated write TS, any commit that begins after Snapshot() returns will get a strictly larger TS, making it invisible to the snapshot.

4. GC: The background goroutine runs every 100ms, computing the minimum active snapshot timestamp. For each key it drops versions whose timestamp is below that threshold, while always keeping the newest version as a baseline for future snapshots.