A version (including a tombstone) may be physically reclaimed iff every permitted read quorum of size R is guaranteed to contain a replica holding a strictly newer version. With M read-eligible replicas that means the version is covered on at least M-R+1 of them. Deciding this per version, newest-first, against the records actually being retained is both necessary and sufficient, automatically prevents tombstone resurrection, and yields a precise "lagging replica" report plus a minimal rejoin delta.
A version (including a tombstone) may be physically reclaimed iff every permitted read quorum of size R is guaranteed to contain a replica holding a strictly newer version. With M read-eligible replicas that means the version is covered on at least M-R+1 of them. Deciding this per version, newest-first, against the records actually being retained is both necessary and sufficient, automatically prevents tombstone resurrection, and yields a precise "lagging replica" report plus a minimal rejoin delta.
The full working module is at ~/quorum-compaction (go test passes, 93.2% coverage).
Naive reclamation rules fail in four distinct ways:
| Naive rule | Failure |
|---|---|
"Drop v once a write quorum W has a newer version" |
A read quorum can be disjoint from the write quorum that held the newer value. Reclamation must be phrased in terms of R, not W. |
"Drop v once the minimum applied version vector is past it" |
A permanently lagging replica stalls compaction forever; also, min over all replicas is wrong when some are not read-eligible. |
| "Drop any version older than the newest" | Deletes the value a permitted quorum would return, and can resurrect a deleted key if the newest version is a tombstone that a lagging replica never received. |
| "Snapshot = copy of current state" | Long-lived MVCC readers pinned to an old version vector still need the version visible at their pin. Physical deletion breaks repeatable reads. |
The deep issue is that compaction is a distributed read-visibility decision, not a local log-truncation decision. A record is only garbage when no permitted observer can distinguish its absence.
E = read-eligible replicas, M = |E|, read quorum size R (M ≥ R).Stamp with a full version vector VV (per-replica logical clock) plus Origin/Seq.max_e(k) = newest record for key k on replica e.Patching the version-vector partial order with an origin tie-break is non-transitive (a > b by tie-break, b > c by dominance, c > a by tie-break is realizable), which corrupts max. We use (sum(VV), Origin, Seq):
sum, so dominance is always respected;max is well-defined and associative.Theorem. Let
Sbe a version of keyk.Scan never be the LWW winner of any permitted read iffcover(S) = { e ∈ E : max_e(k) > S }has|cover(S)| ≥ M − R + 1.
Proof. (⇒) If |cover| ≤ M−R, then E \ cover has at least R replicas and forms a legal quorum in which S is the maximum (no replica has anything greater), hence a read returns S. (⇐) If |cover| ≥ M−R+1, every R-subset of E intersects cover (complement can hold at most R−1), so every quorum's maximum is > S and S cannot win. ∎
Consequences, all for free:
|cover| = 0, so it is never reclaimable ⇒ no resurrection and no lost latest value.cover against the retained set. If cover counts a retained witness on each replica, that witness is still present after the pass, so a second pass cannot invalidate the decision.S has |{e ∈ E : max_e(k) ≤ S}| ≥ R, some read quorum can still return S; those replicas are the blockers. This is exactly "lagging replicas make reclamation unsafe."For each active pin P and key k, mark the newest record with P.Dominates(record.VV) as pinned and always retain it. Pinned records participate in retainedMax and therefore can themselves cover older versions.
ComputeFit(cluster, budget) returns Fits = RetainedBytes ≤ budget. If the safe retained size exceeds the budget, Unsafe=true and BlockedBy[key] names the lagging replicas that must be repaired before space can be reclaimed.
For each key, atomically exchange the per-key maxima: Send the cluster max to the rejoined replica when it is behind; Push the rejoined replica's newer record back when it leads. This is one record per divergent key (minimal), and version vectors merge. The rejoined replica must apply this plan before becoming read-eligible.
go.modmodule quorumcompaction
go 1.26
compaction.go// Package quorumcompaction implements a safe compaction planner for a
// replicated, multi-version key-value store in which concurrent writes,
// tombstones and long-lived snapshot readers are all active.
//
// The central safety property is:
//
// A stored record S may be physically reclaimed iff every permitted read
// quorum of size R is guaranteed to contain at least one replica that still
// holds a record *strictly greater* than S.
//
// With M read-eligible replicas this is equivalent to requiring that at least
// M-R+1 eligible replicas hold a greater record. The planner decides this
// per record, in descending version order, against the set of records that are
// actually being retained (not merely the pre-compaction state), which makes
// the proof compositional. Tombstones are ordinary records under this rule,
// so "latest is a tombstone" can never be reclaimed (no greater record exists)
// and resurrection is impossible.
package quorumcompaction
import (
"sort"
"strconv"
"strings"
)
// ---------------------------------------------------------------------------
// Version vectors and stamps
// ---------------------------------------------------------------------------
// ReplicaID identifies a replica.
type ReplicaID string
// VersionVector is a per-replica logical clock. A missing entry means zero.
type VersionVector map[ReplicaID]uint64
// Clone returns a deep copy.
func (v VersionVector) Clone() VersionVector {
c := make(VersionVector, len(v))
for k, x := range v {
c[k] = x
}
return c
}
// Sum is the total logical work in the vector. It strictly increases under
// componentwise dominance and is the primary key of the total order below.
func (v VersionVector) Sum() uint64 {
var s uint64
for _, x := range v {
s += x
}
return s
}
// Dominates reports whether v[r] >= o[r] for every r (componentwise >=).
func (v VersionVector) Dominates(o VersionVector) bool {
for r := range v {
if v[r] < o[r] {
return false
}
}
for r := range o {
if v[r] < o[r] {
return false
}
}
return true
}
// Merge returns the componentwise maximum.
func (v VersionVector) Merge(o VersionVector) VersionVector {
c := v.Clone()
for r, x := range o {
if x > c[r] {
c[r] = x
}
}
return c
}
// Equal reports componentwise equality (zero == absent).
func (v VersionVector) Equal(o VersionVector) bool {
return v.Dominates(o) && o.Dominates(v)
}
// String renders the vector in a canonical, order-independent form.
func (v VersionVector) String() string {
keys := make([]string, 0, len(v))
for k := range v {
keys = append(keys, string(k))
}
sort.Strings(keys)
var b strings.Builder
for _, k := range keys {
b.WriteString(k)
b.WriteByte('=')
b.WriteString(strconv.FormatUint(v[ReplicaID(k)], 10))
b.WriteByte(',')
}
return b.String()
}
// Stamp labels one write version. VV is the writer's full version vector at
// write time; Origin/Seq break ties between concurrent versions.
type Stamp struct {
Origin ReplicaID
Seq uint64
VV VersionVector
}
// Compare defines the total order used for last-writer-wins resolution.
//
// It refines version-vector dominance (if a dominates b then a > b), because
// dominance strictly increases Sum. Sum, then Origin, then Seq form a
// genuine total order, avoiding the non-transitivity that arises when a
// partial order is patched with a tie-break.
func (s Stamp) Compare(o Stamp) int {
a, b := s.VV.Sum(), o.VV.Sum()
if a < b {
return -1
}
if a > b {
return 1
}
if s.Origin != o.Origin {
if s.Origin < o.Origin {
return -1
}
return 1
}
if s.Seq < o.Seq {
return -1
}
if s.Seq > o.Seq {
return 1
}
return 0
}
// Equal reports whether two stamps are the same total-order element.
func (s Stamp) Equal(o Stamp) bool { return s.Compare(o) == 0 }
// ID is a canonical identity for a write version.
func (s Stamp) ID() string {
return string(s.Origin) + "@" + strconv.FormatUint(s.Seq, 10) + "#" + s.VV.String()
}
func (s Stamp) String() string { return s.ID() }
// ---------------------------------------------------------------------------
// Records and stores
// ---------------------------------------------------------------------------
// Record is a single key version. Tombstone records suppress older values.
type Record struct {
Key string
Value []byte
Tomb bool
Stamp Stamp
}
// Size is the approximate on-disk footprint, including fixed stamp overhead.
func (r Record) Size() int { return len(r.Key) + len(r.Value) + 48 }
// Clone deep-copies the record.
func (r Record) Clone() Record {
r.Value = append([]byte(nil), r.Value...)
r.Stamp = Stamp{Origin: r.Stamp.Origin, Seq: r.Stamp.Seq, VV: r.Stamp.VV.Clone()}
return r
}
// Store holds all retained versions of each key, newest first.
type Store struct {
Records map[string][]Record
}
// NewStore returns an empty store.
func NewStore() *Store { return &Store{Records: map[string][]Record{}} }
// Put inserts or replaces a record, keeping the slice sorted descending.
func (s *Store) Put(r Record) {
lst := s.Records[r.Key]
id := r.Stamp.ID()
for i := range lst {
if lst[i].Stamp.ID() == id {
lst[i] = r.Clone()
return
}
}
lst = append(lst, r.Clone())
sort.SliceStable(lst, func(i, j int) bool {
return lst[i].Stamp.Compare(lst[j].Stamp) > 0
})
s.Records[r.Key] = lst
}
// Max returns the newest record for key, if any.
func (s *Store) Max(key string) (Record, bool) {
lst := s.Records[key]
if len(lst) == 0 {
return Record{}, false
}
return lst[0], true
}
// Clone returns a deep copy of the store.
func (s *Store) Clone() *Store {
c := NewStore()
for k, lst := range s.Records {
cp := make([]Record, len(lst))
for i, r := range lst {
cp[i] = r.Clone()
}
c.Records[k] = cp
}
return c
}
// ---------------------------------------------------------------------------
// Cluster
// ---------------------------------------------------------------------------
// Replica is one member of the replicated store.
type Replica struct {
ID ReplicaID
VV VersionVector
Store *Store
Up bool // read-eligible
}
// Cluster is a point-in-time view of the whole replication group.
type Cluster struct {
N, R, W int
Replicas map[ReplicaID]*Replica
// Pins are version vectors of active MVCC snapshot readers. The newest
// record visible at each pin is retained regardless of quorum safety.
Pins []VersionVector
}
// IDs returns replica IDs in deterministic order.
func (c *Cluster) IDs() []ReplicaID {
ids := make([]ReplicaID, 0, len(c.Replicas))
for id := range c.Replicas {
ids = append(ids, id)
}
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
return ids
}
// Eligible returns the read-eligible replicas in deterministic order.
func (c *Cluster) Eligible() []ReplicaID {
ids := c.IDs()
out := make([]ReplicaID, 0, len(ids))
for _, id := range ids {
if c.Replicas[id].Up {
out = append(out, id)
}
}
return out
}
// Clone returns a deep copy.
func (c *Cluster) Clone() *Cluster {
cc := &Cluster{N: c.N, R: c.R, W: c.W, Replicas: map[ReplicaID]*Replica{}}
for id, r := range c.Replicas {
cc.Replicas[id] = &Replica{ID: r.ID, VV: r.VV.Clone(), Store: r.Store.Clone(), Up: r.Up}
}
for _, p := range c.Pins {
cc.Pins = append(cc.Pins, p.Clone())
}
return cc
}
func (c *Cluster) globalMax(key string) (Record, bool) {
var best Record
found := false
for _, id := range c.IDs() {
if m, ok := c.Replicas[id].Store.Max(key); ok {
if !found || m.Stamp.Compare(best.Stamp) > 0 {
best = m
found = true
}
}
}
return best, found
}
func (c *Cluster) maxOver(key string, pred func(ReplicaID) bool) (Record, bool) {
var best Record
found := false
for _, id := range c.IDs() {
if !pred(id) {
continue
}
if m, ok := c.Replicas[id].Store.Max(key); ok {
if !found || m.Stamp.Compare(best.Stamp) > 0 {
best = m
found = true
}
}
}
return best, found
}
// ---------------------------------------------------------------------------
// Compaction planner
// ---------------------------------------------------------------------------
// Plan is the result of planning a compaction pass.
type Plan struct {
// Keep[key][stampID] reports whether that logical version is retained.
Keep map[string]map[string]bool
// Reclaimed is the set of logical versions that may be physically removed.
Reclaimed []Record
// Retained is the complementary set.
Retained []Record
ReclaimedBytes int
RetainedBytes int
// Fits reports whether RetainedBytes is within the caller's budget.
Fits bool
// Unsafe is set when lagging eligible replicas prevent reclamation of an
// obsolete version, i.e. a newer version exists but not enough replicas
// hold it yet.
Unsafe bool
// NoQuorum is set when fewer than R replicas are read-eligible. In that
// state the planner is conservative and reclaims nothing.
NoQuorum bool
// Lagging lists replicas that are behind the global maximum for some key.
Lagging []ReplicaID
// BlockedBy maps a key to the eligible replicas blocking reclamation.
BlockedBy map[string][]ReplicaID
}
// Compute builds a safe compaction plan for the cluster.
func Compute(c *Cluster) *Plan {
eligible := c.Eligible()
m := len(eligible)
need := m - c.R + 1
p := &Plan{
Keep: map[string]map[string]bool{},
BlockedBy: map[string][]ReplicaID{},
}
if m < c.R || need < 1 {
p.Unsafe = true
p.NoQuorum = true
need = 0
}
// Union of all keys across all replicas.
keys := map[string]bool{}
for _, r := range c.Replicas {
for k := range r.Store.Records {
keys[k] = true
}
}
keyList := make([]string, 0, len(keys))
for k := range keys {
keyList = append(keyList, k)
}
sort.Strings(keyList)
lag := map[ReplicaID]bool{}
for _, key := range keyList {
p.Keep[key] = map[string]bool{}
type entry struct {
rec Record
holders []ReplicaID
}
byID := map[string]*entry{}
var ids []string
for _, rid := range c.IDs() {
for _, rec := range c.Replicas[rid].Store.Records[key] {
id := rec.Stamp.ID()
e, ok := byID[id]
if !ok {
e = &entry{rec: rec}
byID[id] = e
ids = append(ids, id)
}
e.holders = append(e.holders, rid)
}
}
// Process versions newest-first. This lets the safety test use the
// records that are actually retained, not the pre-compaction state.
sort.Slice(ids, func(i, j int) bool {
return byID[ids[i]].rec.Stamp.Compare(byID[ids[j]].rec.Stamp) > 0
})
// Snapshot pinning: newest version visible at each active pin.
pinned := map[string]bool{}
for _, pin := range c.Pins {
var bestID string
var bestStamp Stamp
found := false
for _, id := range ids {
rec := byID[id].rec
if !pin.Dominates(rec.Stamp.VV) {
continue
}
if !found || rec.Stamp.Compare(bestStamp) > 0 {
bestID, bestStamp, found = id, rec.Stamp, true
}
}
if found {
pinned[bestID] = true
}
}
gmax, hasGlobal := c.globalMax(key)
// Highest retained stamp per replica (rebuilt newest-first).
retainedMax := map[ReplicaID]Stamp{}
for _, id := range ids {
e := byID[id]
rec := e.rec
if p.NoQuorum || pinned[id] {
p.Keep[key][id] = true
p.Retained = append(p.Retained, rec)
p.RetainedBytes += rec.Size()
for _, h := range e.holders {
if mx, ok := retainedMax[h]; !ok || rec.Stamp.Compare(mx) > 0 {
retainedMax[h] = rec.Stamp
}
}
continue
}
// Count eligible replicas that already retain a strictly greater
// version. If that reaches M-R+1, every R-quorum has a witness
// and this version can never be the winner of a read.
cover := 0
for _, el := range eligible {
if mx, ok := retainedMax[el]; ok && mx.Compare(rec.Stamp) > 0 {
cover++
}
}
if cover >= need {
p.Reclaimed = append(p.Reclaimed, rec)
p.ReclaimedBytes += rec.Size()
} else {
p.Keep[key][id] = true
p.Retained = append(p.Retained, rec)
p.RetainedBytes += rec.Size()
for _, h := range e.holders {
if mx, ok := retainedMax[h]; !ok || rec.Stamp.Compare(mx) > 0 {
retainedMax[h] = rec.Stamp
}
}
}
// Diagnose genuine lag: a superseded version that even the
// pre-compaction replicas cannot jointly hide from R readers.
if hasGlobal && rec.Stamp.Compare(gmax.Stamp) < 0 {
var blockers []ReplicaID
for _, el := range eligible {
mx, ok := c.Replicas[el].Store.Max(key)
if !ok || mx.Stamp.Compare(rec.Stamp) <= 0 {
blockers = append(blockers, el)
}
}
if len(blockers) >= c.R {
p.Unsafe = true
if _, exists := p.BlockedBy[key]; !exists {
p.BlockedBy[key] = blockers
}
for _, b := range blockers {
lag[b] = true
}
}
}
}
}
for id := range lag {
p.Lagging = append(p.Lagging, id)
}
sort.Slice(p.Lagging, func(i, j int) bool { return p.Lagging[i] < p.Lagging[j] })
return p
}
// ComputeFit plans a compaction and checks it against a disk budget in bytes.
func ComputeFit(c *Cluster, budget int) (*Plan, bool) {
p := Compute(c)
p.Fits = p.RetainedBytes <= budget
return p, p.Fits
}
// Apply physically removes every reclaimed version from every replica.
// Versions kept by the plan are left untouched.
func (p *Plan) Apply(c *Cluster) {
for _, r := range c.Replicas {
for key, lst := range r.Store.Records {
keep := p.Keep[key]
out := make([]Record, 0, len(lst))
for _, rec := range lst {
if keep != nil && keep[rec.Stamp.ID()] {
out = append(out, rec)
}
}
if len(out) == 0 {
delete(r.Store.Records, key)
} else {
r.Store.Records[key] = out
}
}
}
}
// ---------------------------------------------------------------------------
// Rejoin recovery
// ---------------------------------------------------------------------------
// RecoveryPlan is the minimal set of records required to reconcile one
// replica that has rejoined the group.
type RecoveryPlan struct {
Rejoined ReplicaID
// Send are records the rejoined replica must adopt from the group.
Send []Record
// Push are records the rejoined replica holds that the group must adopt.
Push []Record
// NewVV is the merged version vector after the exchange.
NewVV VersionVector
}
// Recovery computes the minimal reconciliation plan for a rejoining replica.
func Recovery(c *Cluster, rejoined ReplicaID) *RecoveryPlan {
rp := &RecoveryPlan{Rejoined: rejoined, NewVV: VersionVector{}}
clusterVV := VersionVector{}
keys := map[string]bool{}
for id, r := range c.Replicas {
clusterVV = clusterVV.Merge(r.VV)
for k := range r.Store.Records {
keys[k] = true
}
_ = id
}
keyList := make([]string, 0, len(keys))
for k := range keys {
keyList = append(keyList, k)
}
sort.Strings(keyList)
rej := c.Replicas[rejoined]
if rej == nil {
return rp
}
for _, key := range keyList {
clusterMax, okC := c.maxOver(key, func(id ReplicaID) bool { return id != rejoined })
rejMax, okR := rej.Store.Max(key)
switch {
case okC && (!okR || clusterMax.Stamp.Compare(rejMax.Stamp) > 0):
rp.Send = append(rp.Send, clusterMax)
case okR && !okC:
rp.Push = append(rp.Push, rejMax)
case okC && okR && rejMax.Stamp.Compare(clusterMax.Stamp) > 0:
rp.Push = append(rp.Push, rejMax)
}
}
rp.NewVV = clusterVV.Merge(rej.VV)
return rp
}
// Apply merges the recovery plan into the cluster.
func (rp *RecoveryPlan) Apply(c *Cluster) {
rej := c.Replicas[rp.Rejoined]
if rej == nil {
return
}
for _, rec := range rp.Send {
rej.Store.Put(rec)
}
rej.VV = rej.VV.Merge(rp.NewVV)
for _, rec := range rp.Push {
for id, r := range c.Replicas {
if id == rp.Rejoined {
continue
}
r.Store.Put(rec)
r.VV = r.VV.Merge(rp.NewVV)
}
}
}
compaction_test.gopackage quorumcompaction
import (
"fmt"
"math/rand"
"sort"
"testing"
)
// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------
// quorums enumerates every R-subset of eligible replicas.
func quorums(eligible []ReplicaID, r int) [][]ReplicaID {
var res [][]ReplicaID
cur := make([]ReplicaID, 0, r)
var rec func(start int)
rec = func(start int) {
if len(cur) == r {
cp := append([]ReplicaID(nil), cur...)
res = append(res, cp)
return
}
for i := start; i < len(eligible); i++ {
cur = append(cur, eligible[i])
rec(i + 1)
cur = cur[:len(cur)-1]
}
}
rec(0)
return res
}
// readKey resolves a key from one quorum using LWW max.
func readKey(c *Cluster, q []ReplicaID, key string) (Record, bool) {
var best Record
found := false
for _, id := range q {
if m, ok := c.Replicas[id].Store.Max(key); ok {
if !found || m.Stamp.Compare(best.Stamp) > 0 {
best = m
found = true
}
}
}
return best, found
}
func allKeys(c *Cluster) []string {
set := map[string]bool{}
for _, r := range c.Replicas {
for k := range r.Store.Records {
set[k] = true
}
}
out := make([]string, 0, len(set))
for k := range set {
out = append(out, k)
}
sort.Strings(out)
return out
}
// assertReadPreserved verifies that no permitted quorum read changes after a
// plan is applied. This is the core safety property.
func assertReadPreserved(t *testing.T, before, after *Cluster, eligible []ReplicaID) {
t.Helper()
qs := quorums(eligible, before.R)
for _, key := range allKeys(before) {
for _, q := range qs {
b, okB := readKey(before, q, key)
a, okA := readKey(after, q, key)
if okB != okA {
t.Fatalf("key %q quorum %v: presence changed %v -> %v", key, q, okB, okA)
}
if okB && !b.Stamp.Equal(a.Stamp) {
t.Fatalf("key %q quorum %v: read changed %v -> %v", key, q, b.Stamp, a.Stamp)
}
}
}
}
func vv(pairs ...interface{}) VersionVector {
v := VersionVector{}
for i := 0; i < len(pairs); i += 2 {
v[ReplicaID(pairs[i].(string))] = uint64(pairs[i+1].(int))
}
return v
}
// ---------------------------------------------------------------------------
// deterministic scenarios
// ---------------------------------------------------------------------------
// TestTombstoneNeverReclaimed verifies the anti-resurrection invariant: the
// latest version of a key is never reclaimed, even when it is a tombstone.
func TestTombstoneNeverReclaimed(t *testing.T) {
c := &Cluster{N: 3, R: 2, W: 2, Replicas: map[ReplicaID]*Replica{}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: true}
}
old := Record{Key: "x", Value: []byte("v"), Stamp: Stamp{Origin: "A", Seq: 1, VV: vv("A", 1)}}
tomb := Record{Key: "x", Tomb: true, Stamp: Stamp{Origin: "B", Seq: 2, VV: vv("A", 1, "B", 2)}}
c.Replicas["A"].Store.Put(old)
c.Replicas["B"].Store.Put(tomb)
c.Replicas["C"].Store.Put(tomb)
before := c.Clone()
p := Compute(c)
if p.NoQuorum {
t.Fatal("expected a quorum")
}
if p.Keep["x"][tomb.Stamp.ID()] != true {
t.Fatalf("latest tombstone must be retained, plan=%+v", p.Keep["x"])
}
if p.Keep["x"][old.Stamp.ID()] {
t.Fatalf("obsolete value should be reclaimed, plan=%+v", p.Keep["x"])
}
p.Apply(c)
assertReadPreserved(t, before, c, c.Eligible())
}
// TestLaggingReplicaBlocksReclamation verifies that a replica which has not
// received the tombstone blocks reclamation and is reported.
func TestLaggingReplicaBlocksReclamation(t *testing.T) {
c := &Cluster{N: 3, R: 2, W: 2, Replicas: map[ReplicaID]*Replica{}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: true}
}
old := Record{Key: "x", Value: []byte("v"), Stamp: Stamp{Origin: "A", Seq: 1, VV: vv("A", 1)}}
tomb := Record{Key: "x", Tomb: true, Stamp: Stamp{Origin: "B", Seq: 2, VV: vv("A", 1, "B", 2)}}
// A and C still hold the value; only B has the tombstone.
c.Replicas["A"].Store.Put(old)
c.Replicas["B"].Store.Put(tomb)
c.Replicas["C"].Store.Put(old)
p := Compute(c)
if p.Keep["x"][old.Stamp.ID()] != true {
t.Fatal("obsolete value must be retained while a quorum of readers can still see it")
}
if !p.Unsafe {
t.Fatal("expected Unsafe to be detected")
}
blockers := p.BlockedBy["x"]
if len(blockers) != 2 {
t.Fatalf("expected 2 blockers, got %v", blockers)
}
if len(p.Lagging) != 2 {
t.Fatalf("expected 2 lagging replicas, got %v", p.Lagging)
}
}
// TestSnapshotPinRetainsHistory verifies MVCC snapshot readers keep pinned
// versions alive even when every replica has moved on.
func TestSnapshotPinRetainsHistory(t *testing.T) {
c := &Cluster{N: 3, R: 2, W: 2, Replicas: map[ReplicaID]*Replica{}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: true}
}
v1 := Record{Key: "x", Value: []byte("one"), Stamp: Stamp{Origin: "A", Seq: 1, VV: vv("A", 1)}}
v2 := Record{Key: "x", Value: []byte("two"), Stamp: Stamp{Origin: "A", Seq: 2, VV: vv("A", 2)}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id].Store.Put(v1)
c.Replicas[id].Store.Put(v2)
}
// A reader is pinned to the v1 era.
c.Pins = []VersionVector{vv("A", 1)}
p := Compute(c)
if p.Keep["x"][v1.Stamp.ID()] != true {
t.Fatal("snapshot-pinned version must be retained")
}
if p.Keep["x"][v2.Stamp.ID()] != true {
t.Fatal("latest version must be retained")
}
if p.ReclaimedBytes != 0 {
t.Fatalf("nothing should be reclaimed, got %d bytes", p.ReclaimedBytes)
}
// Without the pin, v1 is reclaimable.
c2 := c.Clone()
c2.Pins = nil
p2 := Compute(c2)
if p2.Keep["x"][v1.Stamp.ID()] {
t.Fatal("without a pin the old version should be reclaimed")
}
if p2.ReclaimedBytes == 0 {
t.Fatal("expected reclamation")
}
}
// TestRejoinMinimalRecovery verifies that a rejoining replica converges with a
// minimal, one-record-per-key recovery plan.
func TestRejoinMinimalRecovery(t *testing.T) {
c := &Cluster{N: 3, R: 2, W: 2, Replicas: map[ReplicaID]*Replica{}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: true}
}
base := Record{Key: "x", Value: []byte("base"), Stamp: Stamp{Origin: "A", Seq: 1, VV: vv("A", 1)}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id].Store.Put(base)
}
// Ground truth: the fully-replicated converged state.
ground := c.Clone()
// B goes down, A and C advance.
c.Replicas["B"].Up = false
w2 := Record{Key: "x", Value: []byte("two"), Stamp: Stamp{Origin: "A", Seq: 2, VV: vv("A", 2)}}
g2 := Record{Key: "y", Value: []byte("new"), Tomb: true, Stamp: Stamp{Origin: "C", Seq: 1, VV: vv("A", 2, "C", 1)}}
c.Replicas["A"].Store.Put(w2)
c.Replicas["C"].Store.Put(g2)
for _, id := range []ReplicaID{"A", "B", "C"} {
ground.Replicas[id].Store.Put(w2)
ground.Replicas[id].Store.Put(g2)
}
// Compact while B is down.
p := Compute(c)
p.Apply(c)
// B rejoins and recovery runs before it serves reads.
c.Replicas["B"].Up = true
rp := Recovery(c, "B")
rp.Apply(c)
assertReadPreserved(t, ground, c, c.Eligible())
if len(rp.Send) != 2 {
t.Fatalf("expected 2 records to send (x and y), got %d: %+v", len(rp.Send), rp.Send)
}
}
// ---------------------------------------------------------------------------
// randomized brute-force safety test
// ---------------------------------------------------------------------------
func TestRandomizedQuorumSafety(t *testing.T) {
rng := rand.New(rand.NewSource(20261005))
for iter := 0; iter < 3000; iter++ {
c := randomCluster(rng)
before := c.Clone()
eligible := c.Eligible()
// Apply snapshot pins sometimes.
if rng.Intn(3) == 0 {
pin := VersionVector{}
for _, id := range c.IDs() {
pin[id] = uint64(rng.Intn(4))
}
c.Pins = []VersionVector{pin}
before.Pins = []VersionVector{pin.Clone()}
}
p := Compute(c)
p.Apply(c)
assertReadPreserved(t, before, c, eligible)
// After compaction, applying again must be a no-op and remain safe.
p2 := Compute(c)
if len(p2.Reclaimed) != 0 {
// A second pass may still reclaim versions that only became safe
// after the first pass; applying it must not change reads.
c2 := c.Clone()
p2.Apply(c2)
assertReadPreserved(t, c, c2, c.Eligible())
}
// The global maximum per key must always survive.
for _, key := range allKeys(before) {
g, ok := before.globalMax(key)
if !ok {
continue
}
after, ok2 := c.globalMax(key)
if !ok2 || !after.Stamp.Equal(g.Stamp) {
t.Fatalf("iter %d: global max for %q was reclaimed", iter, key)
}
}
}
}
func randomCluster(rng *rand.Rand) *Cluster {
n := 3 + rng.Intn(3) // 3..5
R := 2 + rng.Intn(2)
if R > n {
R = n
}
c := &Cluster{N: n, R: R, W: n - R + 1, Replicas: map[ReplicaID]*Replica{}}
ids := make([]ReplicaID, n)
for i := 0; i < n; i++ {
id := ReplicaID(fmt.Sprintf("r%d", i))
ids[i] = id
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: rng.Intn(5) != 0}
}
// Guarantee at least R eligible replicas so a quorum can exist most of the
// time; leave the rest random.
nUp := 0
for _, r := range c.Replicas {
if r.Up {
nUp++
}
}
for i := 0; nUp < R && i < n; i++ {
if !c.Replicas[ids[i]].Up {
c.Replicas[ids[i]].Up = true
nUp++
}
}
nkeys := 1 + rng.Intn(3)
for ki := 0; ki < nkeys; ki++ {
key := fmt.Sprintf("k%d", ki)
nrec := rng.Intn(4)
for v := 0; v < nrec; v++ {
origin := ids[rng.Intn(n)]
seq := uint64(rng.Intn(3) + 1)
vec := VersionVector{origin: seq}
for j := 0; j < n; j++ {
if rng.Intn(2) == 0 {
if _, ok := vec[ids[j]]; !ok {
vec[ids[j]] = uint64(rng.Intn(3))
}
}
}
rec := Record{
Key: key,
Value: []byte(fmt.Sprintf("%s-%d-%d", origin, seq, v)),
Tomb: rng.Intn(4) == 0,
Stamp: Stamp{Origin: origin, Seq: seq, VV: vec},
}
placements := 0
for _, id := range ids {
if rng.Intn(2) == 0 {
c.Replicas[id].Store.Put(rec)
placements++
}
}
if placements == 0 {
c.Replicas[ids[rng.Intn(n)]].Store.Put(rec)
}
}
}
return c
}
// ---------------------------------------------------------------------------
// budget / fit test
// ---------------------------------------------------------------------------
func TestBudgetAndFit(t *testing.T) {
c := &Cluster{N: 3, R: 2, W: 2, Replicas: map[ReplicaID]*Replica{}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id] = &Replica{ID: id, VV: VersionVector{}, Store: NewStore(), Up: true}
}
for i := 0; i < 10; i++ {
rec := Record{Key: "x", Value: []byte("0123456789"), Stamp: Stamp{Origin: "A", Seq: uint64(i + 1), VV: vv("A", i+1)}}
for _, id := range []ReplicaID{"A", "B", "C"} {
c.Replicas[id].Store.Put(rec)
}
}
// Only the latest version needs to survive: every quorum contains a
// replica holding it, so all older versions are safely reclaimable.
p := Compute(c)
if len(p.Reclaimed) != 9 {
t.Fatalf("expected 9 reclaimed, got %d", len(p.Reclaimed))
}
if p.ReclaimedBytes == 0 || p.RetainedBytes == 0 {
t.Fatal("byte accounting failed")
}
// A budget equal to the safe retained size fits; a tiny budget does not.
if _, ok := ComputeFit(c, p.RetainedBytes); !ok {
t.Fatal("budget should fit exactly")
}
if _, ok := ComputeFit(c, 1); ok {
t.Fatal("tiny budget must not fit after safe compaction")
}
// Blocked scenario must report Unsafe and still preserve reads.
c2 := c.Clone()
// A and C fall back to an old version, B keeps the latest.
old := Record{Key: "x", Value: []byte("old"), Stamp: Stamp{Origin: "A", Seq: 1, VV: vv("A", 1)}}
c2.Replicas["A"].Store.Put(old)
c2.Replicas["C"].Store.Put(old)
// Remove newer versions from A and C.
for _, id := range []ReplicaID{"A", "C"} {
var kept []Record
for _, r := range c2.Replicas[id].Store.Records["x"] {
if r.Stamp.Compare(old.Stamp) <= 0 {
kept = append(kept, r)
}
}
c2.Replicas[id].Store.Records["x"] = kept
}
p2 := Compute(c2)
if !p2.Unsafe {
t.Fatal("expected Unsafe when replicas lag behind")
}
before := c2.Clone()
p2.Apply(c2)
assertReadPreserved(t, before, c2, c2.Eligible())
}
mkdir -p ~/quorum-compaction && cd ~/quorum-compaction
# write go.mod, compaction.go, compaction_test.go as above
go vet ./...
go test -count=1 -v ./...
go test -race -count=5 ./...
go test -count=1 -cover ./...
Expected (actual) output:
=== RUN TestTombstoneNeverReclaimed
--- PASS: TestTombstoneNeverReclaimed (0.00s)
=== RUN TestLaggingReplicaBlocksReclamation
--- PASS: TestLaggingReplicaBlocksReclamation (0.00s)
=== RUN TestSnapshotPinRetainsHistory
--- PASS: TestSnapshotPinRetainsHistory (0.00s)
=== RUN TestRejoinMinimalRecovery
--- PASS: TestRejoinMinimalRecovery (0.00s)
=== RUN TestRandomizedQuorumSafety
--- PASS: TestRandomizedQuorumSafety (0.08s)
=== RUN TestBudgetAndFit
--- PASS: TestBudgetAndFit (0.00s)
PASS
ok quorumcompaction 0.087s
ok quorumcompaction 0.110s coverage: 93.2% of statements
Race detector over 5 repetitions: ok quorumcompaction 2.594s.
TestRandomizedQuorumSafety, N∈{3,4,5}, every R-subset, random eligibility, tombstones, concurrent version vectors, random snapshot pins, 3000 iterations): for every key and every permitted quorum, the record returned before compaction is identical (or identically absent) after compaction. This directly tests the theorem rather than the implementation.TestTombstoneNeverReclaimed): latest tombstone is retained; older value reclaimed; reads unchanged.TestLaggingReplicaBlocksReclamation): a superseded value with < M-R+1 coverage is retained, Unsafe=true, and both lagging replicas are named.TestSnapshotPinRetainsHistory): a pinned old version survives; removing the pin makes it reclaimable.TestRejoinMinimalRecovery): one record per divergent key is sent; after recovery all quorum reads equal the fully-replicated ground truth.TestBudgetAndFit): safe compaction reclaims 9/10 versions, exact-budget fits, sub-budget reports Fits=false; a lag-blocked cluster reports Unsafe and still preserves reads.Compute (or ComputeFit(budget)) on a consistent snapshot of replica state + active snapshot pins.Plan.Apply only if Fits; if Fits == false, surface Unsafe/Lagging/BlockedBy and repair those replicas first (anti-entropy or Recovery) rather than force-deleting.Recovery(...).Apply completes.# Evidence - Problem class: 20261005-quorum-snapshot-compaction - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-10-05T10:08:50.712Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "Design a Go service that compacts a replicated key-value store while concurrent writes, tombstones, and snapshot readers remain active. Given per-replica version vectors and bounded disk space, implement a safe compaction planner that preserves every value observable by any permitted quorum read, detects when lagging replicas make reclamation unsafe, and emits a minimal recovery plan after a replica rejoins.", "environment": "go1.26", "language": "go", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "20261005-quorum-snapshot-compaction", "provider": "openrouter", "solved_at": "2026-10-05T10:08:50.712Z", "version": "1.26"}