crdt-state-merge-engine

Answer

Architecture

The CRDT merge engine is organized into four modules:

1. Vector Clock (clock.go) — Tracks causal history as a map of replicaID → counter. Provides thread-safe operations: - Increment(replicaID) — advances the clock and returns a unique Tag - Merge(other) — element-wise maximum (commutative, associative, idempotent) - Compare(other) / CompareExact(other) — causal ordering with proper happened-before, concurrent, and equal detection

// Vector clock comparison (core of causal consistency)
// Returns -1 (happens-before), 0 (equal/concurrent), 1 (happens-after)
for k := range allKeys {
    v := vc.clocks[k]
    o := other.clocks[k]
    if v > o { vcLeq = false }   // vc[i] > other[i] → vc NOT <= other
    if o > v { otherLeq = false } // other[i] > vc[i] → other NOT <= vc
}

2. AWSet — Add-Wins Set (awset.go) — Based on the Observed-Remove Set (CvRDT). Each addition generates a unique dot (value, tag). A remove only removes dots visible to the caller. Concurrent adds survive because the remover doesn't know their tags:

// AWSet.Add: each add creates a unique dot
func (s *AWSet) Add(value string, tag Tag) {
    s.adds[value][tag.String()] = struct{}{}
}

// AWSet.Remove: only removes observed dots
func (s *AWSet) Remove(value string) {
    for tagStr := range s.adds[value] {
        s.removes[value][tagStr] = struct{}{}
    }
}

// AWSet.Contains: value present iff ∃ dot in adds \ removes
func (s *AWSet) Contains(value string) bool {
    for tagStr := range s.adds[value] {
        if _, removed := s.removes[value][tagStr]; !removed {
            return true
        }
    }
    return false
}

3. RWSet — Remove-Wins Set (rwset.go) — Each operation carries a vector clock snapshot. A value is present only if some add dot's clock strictly dominates all remove dots' clocks. When an add and remove are concurrent, remove wins:

// RWSet.Contains: value present iff ∃ add dot with clock > all remove clocks
func (s *RWSet) Contains(value string) bool {
    for _, a := range addDots {
        dominatesAll := true
        for _, r := range remDots {
            if a.Clock.Compare(r.Clock) <= 0 { // add NOT strictly after remove
                dominatesAll = false
                break
            }
        }
        if dominatesAll { return true }
    }
    return false
}

4. MergeEngine + Replica (store.go) — The distributed KV store replicas with the merge engine: - Each Replica has an ID, a VectorClock, and a map[string]Value - Value is a tagged union over AWSet and RWSet - MergeEngine.Merge(a, b) produces the least upper bound of two replica states, preserving all causal relationships

func (e *MergeEngine) Merge(a, b *Replica) MergeResult {
    // 1. Vector clock: element-wise max
    mergedClock := a.Clock.Merge(b.Clock)

    // 2. For each key: CRDT-specific merge
    for key := range allKeys {
        merged, conflict := e.mergeValues(key, va, vb)
        mergedData[key] = merged
    }
    return MergeResult{MergedClock: mergedClock, MergedData: mergedData}
}

func (e *MergeEngine) mergeValues(key string, a, b Value) (Value, bool) {
    if a.Type != b.Type { return b.Clone(), true } // type conflict

    switch a.Type {
    case TypeAWSet:
        merged.AWSet.Merge(a.AWSet).Merge(b.AWSet) // union of dot stores
    case TypeRWSet:
        merged.RWSet.Merge(a.RWSet).Merge(b.RWSet) // union of dot stores
    }
}