Add TSO state machine

adapter/redis_compat_commands_stream_test.go

@@ -590,7 +590,8 @@ func TestRedis_StreamLegacyDataIsDiscarded(t *testing.T) {
 	}}
 	payload, err := marshalStreamValue(legacy)
 	require.NoError(t, err)
-	seedTS := nowNanos(t)
+	seedTS, err := nodes[0].redisServer.coordinator.Clock().NextFenced()
+	require.NoError(t, err)
 	require.NoError(t, nodes[0].redisServer.store.PutAt(ctx, redisStreamKey(key), payload, seedTS, 0))
 
 	// XLEN on the legacy-only stream must report zero — the legacy blob
@@ -616,7 +617,7 @@ func TestRedis_StreamLegacyDataIsDiscarded(t *testing.T) {
 
 	// Legacy blob is now gone; pick a readTS clearly in the future of
 	// any commit above so MVCC visibility does not hide a still-living blob.
-	readTS := nowNanos(t) + uint64(time.Minute)
+	readTS := nodes[0].redisServer.store.LastCommitTS() + 1
 	_, getErr := nodes[0].redisServer.store.GetAt(ctx, redisStreamKey(key), readTS)
 	require.Error(t, getErr, "legacy blob must be deleted by the first write")
 
@@ -757,21 +758,6 @@ func TestRedis_StreamXAddRecreatesExpiredStreamWithoutStaleEntriesOrTTL(t *testi
 	requireMissingAt(t, server.store, redisTTLKey([]byte(key)), server.readTS())
 }
 
-// nowNanos returns the current UnixNano timestamp as uint64, failing the
-// test if the reading is non-positive. Centralising the bounds check here
-// keeps the int64->uint64 conversion safe and the individual test sites
-// free of gosec waivers.
-func nowNanos(t *testing.T) uint64 {
-	t.Helper()
-	ns := time.Now().UnixNano()
-	require.Positive(t, ns)
-	if ns < 0 {
-		// Unreachable after require.Positive, but lets gosec see the bound.
-		return 0
-	}
-	return uint64(ns)
-}
-
 // TestXAddEnforceMaxWideColumn is a pure-function regression guard: the
 // maxWideColumnItems cap must reject unbounded XADDs on a stream that is
 // already at the ceiling, but must NOT reject when the caller supplied a

docs/design/2026_04_16_partial_centralized_tso.md

@@ -1,11 +1,12 @@
 # Centralized Timestamp Oracle (TSO) Design
 
 - Status: Partial — M1 all-led-group HLC renewal, the M2 reserved-group
-  bootstrap bridge, and the M3-M5 TSO allocator/batch cutover are implemented;
-  the minimal TSO-only FSM and follower redirect/admin exposure remain open
+  bootstrap bridge, the M3-M5 TSO allocator/batch cutover, and the minimal
+  TSO-only FSM are implemented; group-0 timestamp issuance, follower
+  redirect/admin exposure, and shadow validation remain open
 - Author: bootjp
 - Date: 2026-04-16
-- Updated: 2026-07-11
+- Updated: 2026-07-23
 
 ---
 
@@ -44,11 +45,18 @@ Implemented:
    bridge still issues timestamps from the locally led data shard through
    `LocalTSOAllocator`; pinning timestamp issuance to group 0 remains deferred
    until the TSO-leader redirect path exists.
+9. `TSOStateMachine` implements a TSO-only Raft FSM that accepts HLC lease
+   entries, snapshots exactly the physical ceiling as eight big-endian bytes,
+   restores with malformed-input rejection, advances the HLC handoff floor
+   through the committed ceiling, and classifies HLC lease entries as
+   volatile-only for safe cold-start replay. This removes the requirement for
+   group 0 to carry the KV FSM's data store once startup wiring switches to
+   the dedicated FSM.
 
 Remaining:
 
-1. Replace the compatibility bridge FSM with the minimal TSO-only FSM that
-   persists only the HLC ceiling.
+1. Wire group 0 startup to `TSOStateMachine` instead of the compatibility KV
+   FSM and pin production timestamp issuance to the group-0 leader.
 2. Add follower redirect/admin exposure for the dedicated TSO leader.
 3. Add Phase B shadow-read validation before making dedicated TSO the only
    production timestamp path.
@@ -452,6 +460,7 @@ TSO Leader
   ├─ Propose([0x02][ceilingMs]) to TSO Raft group
   │
   └─ TSO FSM.Apply() on all TSO members
+       → HLC.Observe(ceilingMs|maxLogical)
        → HLC.SetPhysicalCeiling(ceilingMs)
          ↳ shared HLC ceiling updated on every node ✅
 ```
@@ -467,8 +476,10 @@ New TSO leader elected via Raft
   ├─ FSM.Restore() or Raft log replay
   │    → physicalCeiling restored to the last committed value
   │
-  └─ HLC.Next() uses max(now, physicalCeiling)
-       → new leader issues timestamps strictly above the old leader's window ✅
+  └─ HLC.NextBatchFenced() rejects the restored exhausted ceiling; group-0
+     timestamp serving stays deferred until it can publish a fresh usable
+     lease window before allocation
+       → new leader does not reuse timestamps from the old leader's window ✅
 ```
 
 ---
@@ -633,7 +644,7 @@ least as large as the maximum shard ceiling.
 | M3 — shipped | Define `TSOAllocator` interface; implement backed by `defaultGroup` | Medium |
 | M4 — shipped | `BatchAllocator` with atomic counter for low-latency timestamp serving | Medium |
 | M5 — shipped for default-group bridge | Coordinator feature-flag cutover via `--tsoEnabled`; shadow validation against a dedicated group remains deferred to M6 | Medium |
-| M6 — partial | Dedicated TSO Raft group (`groupID = 0`) is reserved/bootstrap-capable and warmed by the HLC renewal bridge; TSO-leader-only timestamp issuance and the minimal `TSOStateMachine` remain open | Low |
+| M6 — partial | Dedicated TSO Raft group (`groupID = 0`) is reserved/bootstrap-capable and warmed by the HLC renewal bridge; minimal `TSOStateMachine` is implemented; TSO-leader-only timestamp issuance remains open | Low |
 | M7 | Phase D legacy cleanup + cross-shard SSI read-timestamp validation via TSO | Low |
 
 ---

kv/hlc.go

@@ -9,11 +9,11 @@ import (
 )
 
 // ErrCeilingExpired is returned by HLC.NextFenced() when the
-// Raft-agreed physical ceiling has expired — i.e. the wall clock has
-// caught up to or passed the ceiling, meaning RunHLCLeaseRenewal has
-// not applied a fresh ceiling within `hlcPhysicalWindowMs` of the
-// current wall time. Callers MUST refuse to commit and propagate this
-// to the client.
+// Raft-agreed physical ceiling has expired or the current in-memory
+// logical window has been exhausted — i.e. issuing another fenced
+// timestamp would require a physical millisecond above the committed
+// ceiling. Callers MUST refuse to commit and propagate this to the
+// client.
 //
 // This implements HLC-4 precondition (iii) from
 // docs/design/2026_05_28_implemented_tla_safety_spec.md §5.1: every
@@ -26,10 +26,12 @@ import (
 // to issue ts before the first RunHLCLeaseRenewal cycle.  Strict
 // bootstrap fencing is a follow-up consideration; the current
 // semantics match the spec wherever the prior-leader hazard is real.
-var ErrCeilingExpired = errors.New("hlc: physical ceiling expired (wall_now >= physicalCeiling); refusing to issue persistence timestamp")
+var ErrCeilingExpired = errors.New("hlc: physical/logical ceiling exhausted; refusing to issue persistence timestamp")
 var ErrInvalidHLCBatchSize = errors.New("hlc: invalid batch size")
 
 const hlcLogicalBits = 16
+const hlcTimestampBits = 64
+const hlcPhysicalBits = hlcTimestampBits - hlcLogicalBits
 const hlcLogicalMask uint64 = (1 << hlcLogicalBits) - 1
 const maxHLCBatchSize = 1 << hlcLogicalBits
 
@@ -143,6 +145,8 @@ func (h *HLC) Next() uint64 {
 //   - ceiling == 0 (pre-bootstrap, no prior leader): no fence, identical to Next.
 //   - ceiling > 0 AND wall_now >= ceiling: returns (0, ErrCeilingExpired).
 //   - ceiling > 0 AND wall_now < ceiling: floor wall at ceiling, then proceed.
+//   - ceiling > 0 AND the next timestamp's physical part would exceed ceiling:
+//     returns (0, ErrCeilingExpired) and waits for a fresh committed lease.
 //
 // The TLA+ proof for this lives in tla/hlc/MCHLC_gap.cfg (HLC-4
 // counterexample, depth 5) — see docs/design/2026_05_28_implemented_tla_safety_spec.md §5.1.
@@ -157,6 +161,30 @@ func (h *HLC) NextBatchFenced(n int) (uint64, error) {
 	return h.nextBatchLocked(n, true)
 }
 
+func (h *HLC) fencedNowMillis(fence bool) (int64, int64, error) {
+	nowMs := time.Now().UnixMilli()
+	ceiling := h.physicalCeiling.Load()
+	if ceiling <= 0 {
+		return nowMs, ceiling, nil
+	}
+	if fence && nowMs >= ceiling {
+		h.nextFencedRejections.Add(1)
+		return 0, ceiling, errors.WithStack(ErrCeilingExpired)
+	}
+	if nowMs < ceiling {
+		return ceiling, ceiling, nil
+	}
+	return nowMs, ceiling, nil
+}
+
+func (h *HLC) rejectFencedPhysicalOverflow(fence bool, ceiling, newWall int64) error {
+	if !fence || ceiling <= 0 || newWall <= ceiling {
+		return nil
+	}
+	h.nextFencedRejections.Add(1)
+	return errors.WithStack(ErrCeilingExpired)
+}
+
 func (h *HLC) nextLocked(fence bool) (uint64, error) {
 	for {
 		prev := h.last.Load()
@@ -165,25 +193,9 @@ func (h *HLC) nextLocked(fence bool) (uint64, error) {
 		prevWall := clampUint64ToInt64(wallPart)
 		prevLogical := clampUint64ToUint16(logicalPart)
 
-		nowMs := time.Now().UnixMilli()
-		ceiling := h.physicalCeiling.Load()
-		if ceiling > 0 {
-			if fence && nowMs >= ceiling {
-				// HLC-4 precondition (iii): ceiling has expired.  Fail
-				// closed rather than issue a timestamp that could collide
-				// with a subsequent leader's window after renewal catches
-				// up.  Increment the counter so the monitoring layer can
-				// alert on the rate (see NextFencedRejections).
-				h.nextFencedRejections.Add(1)
-				return 0, errors.WithStack(ErrCeilingExpired)
-			}
-			if nowMs < ceiling {
-				// Physical part: floor at the Raft-agreed ceiling so a
-				// new leader always starts above the previous leader's
-				// issued window.
-				nowMs = ceiling
-			}
-			// Non-fenced path with nowMs >= ceiling: keep nowMs as-is.
+		nowMs, ceiling, err := h.fencedNowMillis(fence)
+		if err != nil {
+			return 0, err
 		}
 
 		newWall := nowMs
@@ -197,6 +209,9 @@ func (h *HLC) nextLocked(fence bool) (uint64, error) {
 				newWall++
 			}
 		}
+		if err := h.rejectFencedPhysicalOverflow(fence, ceiling, newWall); err != nil {
+			return 0, err
+		}
 
 		next := (nonNegativeUint64(newWall) << hlcLogicalBits) | uint64(newLogical)
 		if h.last.CompareAndSwap(prev, next) {
@@ -216,16 +231,9 @@ func (h *HLC) nextBatchLocked(n int, fence bool) (uint64, error) {
 		prevWall := clampUint64ToInt64(prev >> hlcLogicalBits)
 		prevLogical := prev & hlcLogicalMask
 
-		nowMs := time.Now().UnixMilli()
-		ceiling := h.physicalCeiling.Load()
-		if ceiling > 0 {
-			if fence && nowMs >= ceiling {
-				h.nextFencedRejections.Add(1)
-				return 0, errors.WithStack(ErrCeilingExpired)
-			}
-			if nowMs < ceiling {
-				nowMs = ceiling
-			}
+		nowMs, ceiling, err := h.fencedNowMillis(fence)
+		if err != nil {
+			return 0, err
 		}
 
 		newWall := nowMs
@@ -238,6 +246,9 @@ func (h *HLC) nextBatchLocked(n int, fence bool) (uint64, error) {
 				baseLogical = 0
 			}
 		}
+		if err := h.rejectFencedPhysicalOverflow(fence, ceiling, newWall); err != nil {
+			return 0, err
+		}
 		base := (nonNegativeUint64(newWall) << hlcLogicalBits) | baseLogical
 		next := base + batch - 1
 		if h.last.CompareAndSwap(prev, next) {

kv/hlc_test.go

@@ -222,6 +222,19 @@ func TestHLCNextFencedFailsClosedOnExpiredCeiling(t *testing.T) {
 	require.ErrorIs(t, err, ErrCeilingExpired)
 }
 
+func TestHLCNextFencedRejectsExhaustedCeilingWindow(t *testing.T) {
+	t.Parallel()
+
+	h := NewHLC()
+	ceiling := time.Now().Add(time.Hour).UnixMilli()
+	h.SetPhysicalCeiling(ceiling)
+	h.Observe((uint64(ceiling) << hlcLogicalBits) | hlcLogicalMask) //nolint:gosec // ceiling is a positive Unix ms timestamp.
+
+	_, err := h.NextFenced()
+	require.ErrorIs(t, err, ErrCeilingExpired)
+	require.Equal(t, uint64(1), h.NextFencedRejections())
+}
+
 // TestHLCNextFencedIgnoredPreBootstrap verifies the documented soft-fence
 // semantics: ceiling == 0 (no prior leader) is NOT fenced — NextFenced must
 // behave identically to Next so that demo/test bootstrap can issue ts