kv(composed1): M2 — versioned-snapshot ring + kvFSM RouteHistory wiring (#894)

Author: bootjp

distribution/engine.go

@@ -36,8 +36,27 @@ type Engine struct {
 	catalogVersion   uint64
 	ts               atomic.Uint64
 	hotspotThreshold uint64
+	// history is the M2 versioned-snapshot ring for Composed-1
+	// (docs/design/2026_05_29_proposed_composed1_cross_group_commit_guard.md
+	// §M2).  Keyed by catalogVersion; populated on every successful
+	// ApplySnapshot and seeded by NewEngineWithDefaultRoute so a
+	// transaction that observed catalogVersion = 0 (the engine's
+	// pre-bootstrap snapshot) can still resolve its read-set owner.
+	// Bounded by historyDepth via a FIFO eviction list
+	// (historyOrder).  At M2 the ring is plumbing only — M3 will
+	// read from it via verifyComposed1.
+	history      map[uint64]RouteHistorySnapshot
+	historyOrder []uint64
+	historyDepth int
 }
 
+// DefaultRouteHistoryDepth is the size of Engine's versioned-snapshot
+// ring used by the Composed-1 M2 plumbing.  32 is conservative
+// against current single-leader catalog churn (operator-frequency,
+// not data-plane) per the design doc §9 Q2; raise if a future
+// control plane generates more than ~tens of versions per second.
+const DefaultRouteHistoryDepth = 32
+
 const defaultGroupID uint64 = 1
 const minRouteCountForOrderValidation = 2
 
@@ -54,18 +73,28 @@ func NewEngine() *Engine {
 }
 
 // NewEngineWithDefaultRoute creates an Engine and registers a default route
-// covering the full keyspace with a default group ID.
+// covering the full keyspace with a default group ID.  The default route is
+// also recorded in the M2 history ring as the version-0 snapshot so
+// transactions that observed catalogVersion = 0 can resolve their read-set
+// owner through SnapshotAt(0).
 func NewEngineWithDefaultRoute() *Engine {
 	engine := NewEngine()
 	engine.UpdateRoute([]byte(""), nil, defaultGroupID)
+	engine.recordHistorySnapshotLocked()
 	return engine
 }
 
 // NewEngineWithThreshold creates an Engine and sets a threshold for hotspot
 // detection. A non-zero threshold enables automatic range splitting when the
 // number of accesses to a range exceeds the threshold.
 func NewEngineWithThreshold(threshold uint64) *Engine {
-	return &Engine{routes: make([]Route, 0), hotspotThreshold: threshold}
+	return &Engine{
+		routes:           make([]Route, 0),
+		hotspotThreshold: threshold,
+		history:          make(map[uint64]RouteHistorySnapshot, DefaultRouteHistoryDepth),
+		historyOrder:     make([]uint64, 0, DefaultRouteHistoryDepth),
+		historyDepth:     DefaultRouteHistoryDepth,
+	}
 }
 
 // Version returns current route catalog version applied to the engine.
@@ -106,9 +135,115 @@ func (e *Engine) ApplySnapshot(snapshot CatalogSnapshot) error {
 
 	e.routes = routes
 	e.catalogVersion = snapshot.Version
+	e.recordHistorySnapshotLocked()
 	return nil
 }
 
+// RouteHistorySnapshot is a point-in-time view of the route catalog at
+// a specific version.  Returned by Engine.SnapshotAt for the M3
+// Composed-1 commit-time gate.  Carries an immutable copy of the
+// catalog's routes at the recorded version so a caller can resolve
+// ownership without holding the Engine lock.
+type RouteHistorySnapshot struct {
+	version uint64
+	routes  []Route
+}
+
+// Version returns the catalog version this snapshot was recorded at.
+func (s RouteHistorySnapshot) Version() uint64 { return s.version }
+
+// OwnerOf returns the Raft group ID that owned key at this snapshot's
+// version.  Returns (0, false) when no route covers key (the
+// pre-bootstrap state or an explicitly-uncovered range).  Mirrors
+// Engine.GetRoute's right-half-open interval semantics but against
+// the historical snapshot, not the live engine state.
+//
+// Routes are sorted by Start (recordHistorySnapshotLocked clones from
+// e.routes, which Engine.UpdateRoute / routesFromCatalog keep sorted),
+// so the scan can break the moment key < r.Start — every later route
+// has a strictly greater Start and cannot cover key either.  This
+// matters because M3 puts OwnerOf on every txn commit's apply path
+// (claude review on PR #894 — break-vs-continue lifts the worst-case
+// scan from O(N) to "first non-covering gap" without changing the
+// resolution semantics).
+func (s RouteHistorySnapshot) OwnerOf(key []byte) (uint64, bool) {
+	for _, r := range s.routes {
+		if bytes.Compare(key, r.Start) < 0 {
+			break
+		}
+		if r.End != nil && bytes.Compare(key, r.End) >= 0 {
+			continue
+		}
+		return r.GroupID, true
+	}
+	return 0, false
+}
+
+// SnapshotAt returns the route catalog snapshot recorded at version v.
+// Returns (zero, false) when v is not in the ring — either because v
+// is in the future (> catalogVersion), or because the FIFO ring has
+// evicted v (it was older than the historyDepth-most-recent
+// versions).  The M3 Composed-1 gate (design doc §4.3) treats the
+// not-found case as a hard error and triggers a coordinator retry,
+// so retention depth is a liveness knob, not a safety knob.
+func (e *Engine) SnapshotAt(v uint64) (RouteHistorySnapshot, bool) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+	snap, ok := e.history[v]
+	return snap, ok
+}
+
+// HistoryDepth returns the configured ring depth for diagnostics.
+func (e *Engine) HistoryDepth() int {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+	return e.historyDepth
+}
+
+// recordHistorySnapshotLocked pushes the current (catalogVersion,
+// routes) pair into the ring.  The `Locked` suffix is the Go
+// convention for "caller MUST hold the receiver's lock" — checked
+// by reviewers via name, not by the runtime (claude review on PR
+// #894 — fragile lock contract).  Invoked from ApplySnapshot under
+// the write lock and from NewEngineWithDefaultRoute before the
+// Engine is shared with any concurrent reader.  Idempotent on
+// re-record at the same version.
+func (e *Engine) recordHistorySnapshotLocked() {
+	if e.history == nil {
+		// Engines constructed via the bare struct literal (e.g.
+		// internal test seams) — no history ring configured.  Skip
+		// the record so the M2 plumbing stays optional for those
+		// paths; the M3 gate will observe SnapshotAt → (zero,
+		// false) and trigger the soft-fail-as-retry path.
+		return
+	}
+	if _, exists := e.history[e.catalogVersion]; exists {
+		return
+	}
+	if len(e.historyOrder) >= e.historyDepth {
+		evict := e.historyOrder[0]
+		// Copy the retained tail into fresh storage rather than
+		// reslicing.  `historyOrder[1:]` only advances the slice
+		// header — the head of the original backing array stays
+		// alive and grows unboundedly across evictions.  At depth=32
+		// this is small, but the FIFO eviction is the only place
+		// the array grows, and the compaction is free (single
+		// allocation, single copy of <=historyDepth entries —
+		// claude review on PR #894 — backing-array leak).
+		retained := make([]uint64, len(e.historyOrder)-1, e.historyDepth)
+		copy(retained, e.historyOrder[1:])
+		e.historyOrder = retained
+		delete(e.history, evict)
+	}
+	cloned := make([]Route, len(e.routes))
+	copy(cloned, e.routes)
+	e.history[e.catalogVersion] = RouteHistorySnapshot{
+		version: e.catalogVersion,
+		routes:  cloned,
+	}
+	e.historyOrder = append(e.historyOrder, e.catalogVersion)
+}
+
 func staleSnapshotVersionErr(snapshotVersion, currentVersion uint64) error {
 	return errors.Wrapf(
 		ErrEngineSnapshotVersionStale,

distribution/engine_test.go

@@ -477,3 +477,177 @@ func TestEngineApplySnapshot_Concurrent(t *testing.T) {
 		t.Fatalf("expected route group %d, got %d", maxVersion, route.GroupID)
 	}
 }
+
+// TestEngineSnapshotAt_RecordsApplySnapshot is the M2 round-trip
+// witness for the Composed-1 versioned-snapshot ring (design doc
+// §M2): every successful ApplySnapshot records the (version, routes)
+// pair so SnapshotAt(v) can resolve OwnerOf(k) for any in-flight
+// transaction that observed v at BeginTxn.
+func TestEngineSnapshotAt_RecordsApplySnapshot(t *testing.T) {
+	e := NewEngine()
+	if err := e.ApplySnapshot(CatalogSnapshot{
+		Version: 1,
+		Routes: []RouteDescriptor{
+			{RouteID: 10, Start: []byte(""), End: []byte("m"), GroupID: 7, State: RouteStateActive},
+			{RouteID: 11, Start: []byte("m"), End: nil, GroupID: 9, State: RouteStateActive},
+		},
+	}); err != nil {
+		t.Fatalf("apply snapshot v1: %v", err)
+	}
+
+	snap, ok := e.SnapshotAt(1)
+	if !ok {
+		t.Fatal("expected SnapshotAt(1) to return the v1 snapshot")
+	}
+	if snap.Version() != 1 {
+		t.Fatalf("expected snapshot version 1, got %d", snap.Version())
+	}
+	if owner, found := snap.OwnerOf([]byte("a")); !found || owner != 7 {
+		t.Fatalf("expected key 'a' owner=7 in v1 snapshot; got owner=%d found=%v", owner, found)
+	}
+	if owner, found := snap.OwnerOf([]byte("z")); !found || owner != 9 {
+		t.Fatalf("expected key 'z' owner=9 in v1 snapshot; got owner=%d found=%v", owner, found)
+	}
+}
+
+// TestEngineSnapshotAt_PreservesHistoryAcrossVersions verifies the
+// M3-critical property: after ApplySnapshot has moved the catalog
+// forward, a SnapshotAt for the PRIOR version still returns the old
+// routes.  Without this, the M3 verifyComposed1 gate could not
+// resolve a txn whose observedVer is behind the current catalog —
+// exactly the case the design doc §3 G1c trace requires.
+func TestEngineSnapshotAt_PreservesHistoryAcrossVersions(t *testing.T) {
+	e := NewEngine()
+	if err := e.ApplySnapshot(CatalogSnapshot{
+		Version: 1,
+		Routes: []RouteDescriptor{
+			{RouteID: 10, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive},
+		},
+	}); err != nil {
+		t.Fatalf("apply v1: %v", err)
+	}
+	if err := e.ApplySnapshot(CatalogSnapshot{
+		Version: 2,
+		Routes: []RouteDescriptor{
+			{RouteID: 11, Start: []byte(""), End: nil, GroupID: 2, State: RouteStateActive},
+		},
+	}); err != nil {
+		t.Fatalf("apply v2: %v", err)
+	}
+
+	snapV1, ok := e.SnapshotAt(1)
+	if !ok {
+		t.Fatal("expected v1 still in history after v2 applied")
+	}
+	if owner, _ := snapV1.OwnerOf([]byte("k")); owner != 1 {
+		t.Fatalf("v1 snapshot must still show group=1 owner; got %d", owner)
+	}
+	snapV2, ok := e.SnapshotAt(2)
+	if !ok {
+		t.Fatal("expected v2 in history")
+	}
+	if owner, _ := snapV2.OwnerOf([]byte("k")); owner != 2 {
+		t.Fatalf("v2 snapshot must show group=2 owner; got %d", owner)
+	}
+}
+
+// TestEngineSnapshotAt_FIFOEviction verifies that the ring respects
+// historyDepth: once more than depth versions have been applied, the
+// oldest is evicted and SnapshotAt returns (zero, false) for it.
+// The M3 gate (design doc §4.3) treats the not-found case as a hard
+// retryable error, so retention depth is a liveness knob — this
+// test pins the eviction order so a future depth change does not
+// silently break the M3 contract.
+func TestEngineSnapshotAt_FIFOEviction(t *testing.T) {
+	t.Parallel()
+	e := NewEngine()
+	// Force a tiny depth so the test is bounded and explicit.  The
+	// direct field write is safe because `e` is local to this test
+	// goroutine and the depth is set before any ApplySnapshot fires;
+	// once the Engine is published to concurrent readers, the depth
+	// would have to flow through a constructor option (claude review
+	// on PR #894 — fragile-but-test-local lock contract).
+	e.historyDepth = 3
+
+	for v := uint64(1); v <= 5; v++ {
+		if err := e.ApplySnapshot(CatalogSnapshot{
+			Version: v,
+			Routes: []RouteDescriptor{
+				{RouteID: v, Start: []byte(""), End: nil, GroupID: v, State: RouteStateActive},
+			},
+		}); err != nil {
+			t.Fatalf("apply v%d: %v", v, err)
+		}
+	}
+
+	if _, ok := e.SnapshotAt(1); ok {
+		t.Fatal("v1 must be evicted (oldest) under depth=3 after v2..v5 applied")
+	}
+	if _, ok := e.SnapshotAt(2); ok {
+		t.Fatal("v2 must be evicted (second-oldest) under depth=3 after v3..v5 applied")
+	}
+	for v := uint64(3); v <= 5; v++ {
+		snap, ok := e.SnapshotAt(v)
+		if !ok {
+			t.Fatalf("expected v%d retained (depth=3 keeps the 3 most recent)", v)
+		}
+		if owner, _ := snap.OwnerOf([]byte("k")); owner != v {
+			t.Fatalf("v%d snapshot must show group=%d owner; got %d", v, v, owner)
+		}
+	}
+}
+
+// TestEngineSnapshotAt_UnknownVersionReturnsNotFound documents the
+// M3-relevant contract: a version that has never been applied (e.g.
+// in the future, or a typo) returns (zero, false).  The M3 gate
+// uses this signal to emit ErrComposed1VersionGCd and trigger a
+// coordinator retry.
+func TestEngineSnapshotAt_UnknownVersionReturnsNotFound(t *testing.T) {
+	e := NewEngineWithDefaultRoute()
+	if _, ok := e.SnapshotAt(42); ok {
+		t.Fatal("expected SnapshotAt for an unknown version to return false")
+	}
+}
+
+// TestEngineSnapshotAt_SeedsVersionZeroForDefaultRoute verifies that
+// the NewEngineWithDefaultRoute path records the version-0 default
+// route snapshot.  Without this, every txn that observed v=0 (the
+// common case before any ApplySnapshot lands) would fall through
+// to the M3 not-found path and trigger a spurious retry on its first
+// commit.
+func TestEngineSnapshotAt_SeedsVersionZeroForDefaultRoute(t *testing.T) {
+	e := NewEngineWithDefaultRoute()
+	snap, ok := e.SnapshotAt(0)
+	if !ok {
+		t.Fatal("expected NewEngineWithDefaultRoute to seed a v0 history entry")
+	}
+	if snap.Version() != 0 {
+		t.Fatalf("expected seed snapshot version 0, got %d", snap.Version())
+	}
+	// Default route covers the full keyspace ⇒ every key resolves to
+	// defaultGroupID at v0.
+	owner, ok := snap.OwnerOf([]byte("anything"))
+	if !ok || owner != defaultGroupID {
+		t.Fatalf("expected v0 snapshot to resolve every key to defaultGroupID=%d; got owner=%d found=%v", defaultGroupID, owner, ok)
+	}
+}
+
+// TestEngineSnapshotAt_BareEngineHasNoHistory documents that an
+// Engine constructed via the bare struct literal (e.g. via internal
+// test seams) has a nil history map and SnapshotAt always returns
+// (zero, false).  recordHistorySnapshot is nil-safe so ApplySnapshot
+// still works on such an engine, but the M3 gate will treat every
+// SnapshotAt as "not in ring" → soft-fail-as-retry, which is the
+// correct posture for unconfigured engines.
+func TestEngineSnapshotAt_BareEngineHasNoHistory(t *testing.T) {
+	e := &Engine{routes: make([]Route, 0)} // bare struct literal — no history
+	if err := e.ApplySnapshot(CatalogSnapshot{
+		Version: 1,
+		Routes:  []RouteDescriptor{{RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}},
+	}); err != nil {
+		t.Fatalf("apply on bare engine should succeed: %v", err)
+	}
+	if _, ok := e.SnapshotAt(1); ok {
+		t.Fatal("bare engine has no history ring; SnapshotAt should always be false")
+	}
+}