From 518a3800b18e56237bf1621b1cbd8b6670bc5ce2 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 04:37:23 +0900 Subject: [PATCH 1/4] Add TSO state machine --- .../2026_04_16_partial_centralized_tso.md | 19 +++- kv/tso_fsm.go | 96 ++++++++++++++++ kv/tso_fsm_test.go | 103 ++++++++++++++++++ 3 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 kv/tso_fsm.go create mode 100644 kv/tso_fsm_test.go diff --git a/docs/design/2026_04_16_partial_centralized_tso.md b/docs/design/2026_04_16_partial_centralized_tso.md index 631ae4dbf..f476ab051 100644 --- a/docs/design/2026_04_16_partial_centralized_tso.md +++ b/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,17 @@ 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, 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. @@ -633,7 +640,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 | --- diff --git a/kv/tso_fsm.go b/kv/tso_fsm.go new file mode 100644 index 000000000..72df98e29 --- /dev/null +++ b/kv/tso_fsm.go @@ -0,0 +1,96 @@ +package kv + +import ( + "encoding/binary" + "io" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/cockroachdb/errors" +) + +const tsoSnapshotLen = 8 + +var _ raftengine.StateMachine = (*TSOStateMachine)(nil) +var _ raftengine.Snapshot = (*tsoSnapshot)(nil) +var _ raftengine.VolatileEntryClassifier = (*TSOStateMachine)(nil) + +// TSOStateMachine is the minimal FSM for the dedicated timestamp-oracle Raft +// group. It tracks only the Raft-agreed HLC physical ceiling; no KV state, +// route catalog, or sidecar state is attached to group 0. +type TSOStateMachine struct { + hlc *HLC +} + +// NewTSOStateMachine constructs the dedicated TSO FSM over the shared HLC. +func NewTSOStateMachine(hlc *HLC) *TSOStateMachine { + return &TSOStateMachine{hlc: hlc} +} + +func (f *TSOStateMachine) Apply(data []byte) any { + if len(data) == 0 || data[0] != raftEncodeHLCLease { + return nil + } + return f.applyHLCLease(data[1:]) +} + +func (f *TSOStateMachine) applyHLCLease(data []byte) any { + if len(data) != hlcLeasePayloadLen { + return errors.Newf("tso fsm: hlc lease: expected %d bytes, got %d", hlcLeasePayloadLen, len(data)) //nolint:wrapcheck // creating new error, nothing to wrap + } + ceilingMs := int64(binary.BigEndian.Uint64(data)) //nolint:gosec // value is a Unix ms timestamp encoded as uint64; fits in int64 for valid deployments. + if f.hlc != nil && ceilingMs > 0 { + f.hlc.SetPhysicalCeiling(ceilingMs) + } + return nil +} + +func (f *TSOStateMachine) Snapshot() (raftengine.Snapshot, error) { + return &tsoSnapshot{ceilingMs: hlcCeilingFromHLC(f.hlc)}, nil +} + +func (f *TSOStateMachine) Restore(r io.Reader) error { + var buf [tsoSnapshotLen]byte + if _, err := io.ReadFull(r, buf[:]); err != nil { + return errors.Wrap(err, "tso fsm: restore snapshot") + } + var extra [1]byte + n, err := r.Read(extra[:]) + if err != nil && !errors.Is(err, io.EOF) { + return errors.Wrap(err, "tso fsm: restore snapshot") + } + if n != 0 { + return errors.New("tso fsm: restore snapshot: trailing bytes") //nolint:wrapcheck // creating new error, nothing to wrap + } + ceilingMs := int64(binary.BigEndian.Uint64(buf[:])) //nolint:gosec // value was written from an int64 Unix ms ceiling. + if f.hlc != nil && ceilingMs > 0 { + f.hlc.SetPhysicalCeiling(ceilingMs) + } + return nil +} + +// IsVolatileOnlyPayload classifies HLC lease entries for the cold-start replay +// gate. Re-applying them is monotonic and reconstructs the in-memory ceiling. +func (f *TSOStateMachine) IsVolatileOnlyPayload(payload []byte) bool { + return len(payload) > 0 && payload[0] == raftEncodeHLCLease +} + +type tsoSnapshot struct { + ceilingMs int64 +} + +func (s *tsoSnapshot) WriteTo(w io.Writer) (int64, error) { + var buf [tsoSnapshotLen]byte + binary.BigEndian.PutUint64(buf[:], uint64(s.ceilingMs)) //nolint:gosec // ceilingMs is a Unix ms timestamp encoded as uint64. + n, err := w.Write(buf[:]) + if err != nil { + return int64(n), errors.WithStack(err) + } + if n != tsoSnapshotLen { + return int64(n), io.ErrShortWrite + } + return tsoSnapshotLen, nil +} + +func (s *tsoSnapshot) Close() error { + return nil +} diff --git a/kv/tso_fsm_test.go b/kv/tso_fsm_test.go new file mode 100644 index 000000000..174cef3fa --- /dev/null +++ b/kv/tso_fsm_test.go @@ -0,0 +1,103 @@ +package kv + +import ( + "bytes" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestTSOStateMachineApplyHLCLeaseUpdatesCeiling(t *testing.T) { + t.Parallel() + + const ceilingMs = int64(9_999_999_999_999) + clock := NewHLC() + fsm := NewTSOStateMachine(clock) + + require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) + require.Equal(t, ceilingMs, clock.PhysicalCeiling()) +} + +func TestTSOStateMachineApplyIgnoresNonLeasePayload(t *testing.T) { + t.Parallel() + + clock := NewHLC() + fsm := NewTSOStateMachine(clock) + + require.Nil(t, fsm.Apply([]byte{raftEncodeSingle})) + require.Zero(t, clock.PhysicalCeiling()) +} + +func TestTSOStateMachineRejectsMalformedHLCLease(t *testing.T) { + t.Parallel() + + fsm := NewTSOStateMachine(NewHLC()) + + requireApplyError(t, fsm.Apply([]byte{raftEncodeHLCLease})) + requireApplyError(t, fsm.Apply(append([]byte{raftEncodeHLCLease}, make([]byte, hlcLeasePayloadLen+1)...))) +} + +func TestTSOStateMachineSnapshotRestoreRoundTrip(t *testing.T) { + t.Parallel() + + const ceilingMs = int64(1_900_000_000_000) + sourceClock := NewHLC() + sourceClock.SetPhysicalCeiling(ceilingMs) + source := NewTSOStateMachine(sourceClock) + + snap, err := source.Snapshot() + require.NoError(t, err) + defer func() { require.NoError(t, snap.Close()) }() + + var buf bytes.Buffer + n, err := snap.WriteTo(&buf) + require.NoError(t, err) + require.EqualValues(t, 8, n) + require.Len(t, buf.Bytes(), 8) + + restoredClock := NewHLC() + restored := NewTSOStateMachine(restoredClock) + require.NoError(t, restored.Restore(bytes.NewReader(buf.Bytes()))) + require.Equal(t, ceilingMs, restoredClock.PhysicalCeiling()) +} + +func TestTSOStateMachineSnapshotWithNilHLCWritesZero(t *testing.T) { + t.Parallel() + + fsm := NewTSOStateMachine(nil) + snap, err := fsm.Snapshot() + require.NoError(t, err) + defer func() { require.NoError(t, snap.Close()) }() + + var buf bytes.Buffer + _, err = snap.WriteTo(&buf) + require.NoError(t, err) + require.Equal(t, make([]byte, 8), buf.Bytes()) +} + +func TestTSOStateMachineRestoreRejectsMalformedSnapshot(t *testing.T) { + t.Parallel() + + fsm := NewTSOStateMachine(NewHLC()) + + require.Error(t, fsm.Restore(bytes.NewReader(nil))) + require.Error(t, fsm.Restore(bytes.NewReader(make([]byte, 9)))) +} + +func TestTSOStateMachineClassifiesHLCLeaseAsVolatileOnly(t *testing.T) { + t.Parallel() + + fsm := NewTSOStateMachine(NewHLC()) + + require.True(t, fsm.IsVolatileOnlyPayload(marshalHLCLeaseRenew(1))) + require.False(t, fsm.IsVolatileOnlyPayload([]byte{raftEncodeSingle})) + require.False(t, fsm.IsVolatileOnlyPayload(nil)) +} + +func requireApplyError(t *testing.T, got any) { + t.Helper() + + err, ok := got.(error) + require.True(t, ok) + require.Error(t, err) +} From ea1e04c89713f88c320ba24e801772b418191108 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 18:00:49 +0900 Subject: [PATCH 2/4] tso: persist fsm ceiling floor --- kv/tso_fsm.go | 43 ++++++++++++++++++++++++++++++++++------- kv/tso_fsm_test.go | 48 ++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 82 insertions(+), 9 deletions(-) diff --git a/kv/tso_fsm.go b/kv/tso_fsm.go index 72df98e29..86845a4ba 100644 --- a/kv/tso_fsm.go +++ b/kv/tso_fsm.go @@ -3,6 +3,7 @@ package kv import ( "encoding/binary" "io" + "sync/atomic" "github.com/bootjp/elastickv/internal/raftengine" "github.com/cockroachdb/errors" @@ -18,7 +19,8 @@ var _ raftengine.VolatileEntryClassifier = (*TSOStateMachine)(nil) // group. It tracks only the Raft-agreed HLC physical ceiling; no KV state, // route catalog, or sidecar state is attached to group 0. type TSOStateMachine struct { - hlc *HLC + hlc *HLC + ceilingMs atomic.Int64 } // NewTSOStateMachine constructs the dedicated TSO FSM over the shared HLC. @@ -38,14 +40,12 @@ func (f *TSOStateMachine) applyHLCLease(data []byte) any { return errors.Newf("tso fsm: hlc lease: expected %d bytes, got %d", hlcLeasePayloadLen, len(data)) //nolint:wrapcheck // creating new error, nothing to wrap } ceilingMs := int64(binary.BigEndian.Uint64(data)) //nolint:gosec // value is a Unix ms timestamp encoded as uint64; fits in int64 for valid deployments. - if f.hlc != nil && ceilingMs > 0 { - f.hlc.SetPhysicalCeiling(ceilingMs) - } + f.applyTSOCeiling(ceilingMs) return nil } func (f *TSOStateMachine) Snapshot() (raftengine.Snapshot, error) { - return &tsoSnapshot{ceilingMs: hlcCeilingFromHLC(f.hlc)}, nil + return &tsoSnapshot{ceilingMs: f.committedCeiling()}, nil } func (f *TSOStateMachine) Restore(r io.Reader) error { @@ -62,10 +62,39 @@ func (f *TSOStateMachine) Restore(r io.Reader) error { return errors.New("tso fsm: restore snapshot: trailing bytes") //nolint:wrapcheck // creating new error, nothing to wrap } ceilingMs := int64(binary.BigEndian.Uint64(buf[:])) //nolint:gosec // value was written from an int64 Unix ms ceiling. - if f.hlc != nil && ceilingMs > 0 { + f.applyTSOCeiling(ceilingMs) + return nil +} + +func (f *TSOStateMachine) applyTSOCeiling(ceilingMs int64) { + if ceilingMs <= 0 { + return + } + f.advanceCommittedCeiling(ceilingMs) + if f.hlc != nil { f.hlc.SetPhysicalCeiling(ceilingMs) + f.hlc.Observe(tsoCeilingMaxTimestamp(ceilingMs)) } - return nil +} + +func (f *TSOStateMachine) advanceCommittedCeiling(ceilingMs int64) { + for { + prev := f.ceilingMs.Load() + if ceilingMs <= prev { + return + } + if f.ceilingMs.CompareAndSwap(prev, ceilingMs) { + return + } + } +} + +func (f *TSOStateMachine) committedCeiling() int64 { + return f.ceilingMs.Load() +} + +func tsoCeilingMaxTimestamp(ceilingMs int64) uint64 { + return (uint64(ceilingMs) << hlcLogicalBits) | hlcLogicalMask //nolint:gosec // ceilingMs is validated positive before conversion. } // IsVolatileOnlyPayload classifies HLC lease entries for the cold-start replay diff --git a/kv/tso_fsm_test.go b/kv/tso_fsm_test.go index 174cef3fa..23cdb18da 100644 --- a/kv/tso_fsm_test.go +++ b/kv/tso_fsm_test.go @@ -3,6 +3,7 @@ package kv import ( "bytes" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -16,6 +17,21 @@ func TestTSOStateMachineApplyHLCLeaseUpdatesCeiling(t *testing.T) { require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) require.Equal(t, ceilingMs, clock.PhysicalCeiling()) + require.Equal(t, tsoCeilingMaxTimestamp(ceilingMs), clock.Current()) + require.Equal(t, ceilingMs, fsm.committedCeiling()) +} + +func TestTSOStateMachineApplyHLCLeaseAdvancesAllocationFloor(t *testing.T) { + t.Parallel() + + ceilingMs := time.Now().Add(time.Hour).UnixMilli() + clock := NewHLC() + fsm := NewTSOStateMachine(clock) + + require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) + base, err := clock.NextBatchFenced(1) + require.NoError(t, err) + require.Equal(t, (uint64(ceilingMs+1) << hlcLogicalBits), base) //nolint:gosec // ceilingMs is a positive Unix ms timestamp. } func TestTSOStateMachineApplyIgnoresNonLeasePayload(t *testing.T) { @@ -40,10 +56,10 @@ func TestTSOStateMachineRejectsMalformedHLCLease(t *testing.T) { func TestTSOStateMachineSnapshotRestoreRoundTrip(t *testing.T) { t.Parallel() - const ceilingMs = int64(1_900_000_000_000) + ceilingMs := time.Now().Add(time.Hour).UnixMilli() sourceClock := NewHLC() - sourceClock.SetPhysicalCeiling(ceilingMs) source := NewTSOStateMachine(sourceClock) + require.Nil(t, source.Apply(marshalHLCLeaseRenew(ceilingMs))) snap, err := source.Snapshot() require.NoError(t, err) @@ -59,6 +75,34 @@ func TestTSOStateMachineSnapshotRestoreRoundTrip(t *testing.T) { restored := NewTSOStateMachine(restoredClock) require.NoError(t, restored.Restore(bytes.NewReader(buf.Bytes()))) require.Equal(t, ceilingMs, restoredClock.PhysicalCeiling()) + require.Equal(t, tsoCeilingMaxTimestamp(ceilingMs), restoredClock.Current()) + require.Equal(t, ceilingMs, restored.committedCeiling()) +} + +func TestTSOStateMachineSnapshotUsesCommittedCeiling(t *testing.T) { + t.Parallel() + + committedCeilingMs := time.Now().Add(time.Hour).UnixMilli() + unrelatedCeilingMs := committedCeilingMs + int64(time.Hour/time.Millisecond) + sourceClock := NewHLC() + source := NewTSOStateMachine(sourceClock) + + require.Nil(t, source.Apply(marshalHLCLeaseRenew(committedCeilingMs))) + sourceClock.SetPhysicalCeiling(unrelatedCeilingMs) + + snap, err := source.Snapshot() + require.NoError(t, err) + defer func() { require.NoError(t, snap.Close()) }() + + var buf bytes.Buffer + _, err = snap.WriteTo(&buf) + require.NoError(t, err) + + restoredClock := NewHLC() + restored := NewTSOStateMachine(restoredClock) + require.NoError(t, restored.Restore(bytes.NewReader(buf.Bytes()))) + require.Equal(t, committedCeilingMs, restoredClock.PhysicalCeiling()) + require.Equal(t, tsoCeilingMaxTimestamp(committedCeilingMs), restoredClock.Current()) } func TestTSOStateMachineSnapshotWithNilHLCWritesZero(t *testing.T) { From fe2dc4c7b8ea358f613da28dde8da9e709ca1b3f Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 18:50:03 +0900 Subject: [PATCH 3/4] tso: reject exhausted ceiling windows --- adapter/redis_compat_commands_stream_test.go | 20 +---- .../2026_04_16_partial_centralized_tso.md | 16 ++-- kv/hlc.go | 77 +++++++++++-------- kv/hlc_test.go | 13 ++++ kv/tso_fsm.go | 2 +- kv/tso_fsm_test.go | 8 +- kv/tso_test.go | 11 +++ 7 files changed, 85 insertions(+), 62 deletions(-) diff --git a/adapter/redis_compat_commands_stream_test.go b/adapter/redis_compat_commands_stream_test.go index 10c38d965..4191ec6a3 100644 --- a/adapter/redis_compat_commands_stream_test.go +++ b/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 diff --git a/docs/design/2026_04_16_partial_centralized_tso.md b/docs/design/2026_04_16_partial_centralized_tso.md index f476ab051..58ea65acb 100644 --- a/docs/design/2026_04_16_partial_centralized_tso.md +++ b/docs/design/2026_04_16_partial_centralized_tso.md @@ -47,10 +47,11 @@ Implemented: 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, 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. + 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: @@ -459,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 ✅ ``` @@ -474,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 ✅ ``` --- diff --git a/kv/hlc.go b/kv/hlc.go index 84daece50..279a23d6b 100644 --- a/kv/hlc.go +++ b/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 @@ -143,6 +143,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 +159,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 +191,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 +207,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 +229,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 +244,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) { diff --git a/kv/hlc_test.go b/kv/hlc_test.go index 0d3868221..ae35267d9 100644 --- a/kv/hlc_test.go +++ b/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 diff --git a/kv/tso_fsm.go b/kv/tso_fsm.go index 86845a4ba..13863c070 100644 --- a/kv/tso_fsm.go +++ b/kv/tso_fsm.go @@ -72,8 +72,8 @@ func (f *TSOStateMachine) applyTSOCeiling(ceilingMs int64) { } f.advanceCommittedCeiling(ceilingMs) if f.hlc != nil { - f.hlc.SetPhysicalCeiling(ceilingMs) f.hlc.Observe(tsoCeilingMaxTimestamp(ceilingMs)) + f.hlc.SetPhysicalCeiling(ceilingMs) } } diff --git a/kv/tso_fsm_test.go b/kv/tso_fsm_test.go index 23cdb18da..91d4b136e 100644 --- a/kv/tso_fsm_test.go +++ b/kv/tso_fsm_test.go @@ -21,7 +21,7 @@ func TestTSOStateMachineApplyHLCLeaseUpdatesCeiling(t *testing.T) { require.Equal(t, ceilingMs, fsm.committedCeiling()) } -func TestTSOStateMachineApplyHLCLeaseAdvancesAllocationFloor(t *testing.T) { +func TestTSOStateMachineApplyHLCLeaseExhaustsCommittedCeiling(t *testing.T) { t.Parallel() ceilingMs := time.Now().Add(time.Hour).UnixMilli() @@ -29,9 +29,9 @@ func TestTSOStateMachineApplyHLCLeaseAdvancesAllocationFloor(t *testing.T) { fsm := NewTSOStateMachine(clock) require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) - base, err := clock.NextBatchFenced(1) - require.NoError(t, err) - require.Equal(t, (uint64(ceilingMs+1) << hlcLogicalBits), base) //nolint:gosec // ceilingMs is a positive Unix ms timestamp. + _, err := clock.NextBatchFenced(1) + require.ErrorIs(t, err, ErrCeilingExpired) + require.Equal(t, uint64(1), clock.NextFencedRejections()) } func TestTSOStateMachineApplyIgnoresNonLeasePayload(t *testing.T) { diff --git a/kv/tso_test.go b/kv/tso_test.go index ef4e58510..aeb4c29b1 100644 --- a/kv/tso_test.go +++ b/kv/tso_test.go @@ -36,6 +36,17 @@ func TestHLCNextBatchFencedRejectsExpiredCeiling(t *testing.T) { require.ErrorIs(t, err, ErrCeilingExpired) } +func TestHLCNextBatchFencedRejectsExhaustedCeilingWindow(t *testing.T) { + h := NewHLC() + ceiling := time.Now().Add(testTSOFutureCeiling).UnixMilli() + h.SetPhysicalCeiling(ceiling) + h.Observe((uint64(ceiling) << hlcLogicalBits) | (hlcLogicalMask - 1)) //nolint:gosec // ceiling is a positive Unix ms timestamp. + + _, err := h.NextBatchFenced(2) + require.ErrorIs(t, err, ErrCeilingExpired) + require.Equal(t, uint64(1), h.NextFencedRejections()) +} + func TestHLCNextBatchFencedBumpsWallOnLogicalOverflow(t *testing.T) { h := NewHLC() futureWall := uint64(time.Now().Add(testTSOFutureCeiling).UnixMilli()) //nolint:gosec // Unix ms is non-negative. From 4ee268c443ae0ce48e2b092fde3d013adc107fe5 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 19:05:17 +0900 Subject: [PATCH 4/4] tso: keep renewed lease windows usable --- kv/hlc.go | 4 ++- kv/tso_fsm.go | 45 +++++++++++++++++++------- kv/tso_fsm_test.go | 79 ++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 109 insertions(+), 19 deletions(-) diff --git a/kv/hlc.go b/kv/hlc.go index 279a23d6b..e7e421122 100644 --- a/kv/hlc.go +++ b/kv/hlc.go @@ -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 diff --git a/kv/tso_fsm.go b/kv/tso_fsm.go index 13863c070..e721bb740 100644 --- a/kv/tso_fsm.go +++ b/kv/tso_fsm.go @@ -10,6 +10,7 @@ import ( ) const tsoSnapshotLen = 8 +const maxHLCPhysicalMillis = int64((uint64(1) << hlcPhysicalBits) - 1) var _ raftengine.StateMachine = (*TSOStateMachine)(nil) var _ raftengine.Snapshot = (*tsoSnapshot)(nil) @@ -39,8 +40,11 @@ func (f *TSOStateMachine) applyHLCLease(data []byte) any { if len(data) != hlcLeasePayloadLen { return errors.Newf("tso fsm: hlc lease: expected %d bytes, got %d", hlcLeasePayloadLen, len(data)) //nolint:wrapcheck // creating new error, nothing to wrap } - ceilingMs := int64(binary.BigEndian.Uint64(data)) //nolint:gosec // value is a Unix ms timestamp encoded as uint64; fits in int64 for valid deployments. - f.applyTSOCeiling(ceilingMs) + ceilingMs, err := decodeTSOCeiling(binary.BigEndian.Uint64(data)) + if err != nil { + return err + } + f.applyTSOCeiling(ceilingMs, false) return nil } @@ -61,30 +65,49 @@ func (f *TSOStateMachine) Restore(r io.Reader) error { if n != 0 { return errors.New("tso fsm: restore snapshot: trailing bytes") //nolint:wrapcheck // creating new error, nothing to wrap } - ceilingMs := int64(binary.BigEndian.Uint64(buf[:])) //nolint:gosec // value was written from an int64 Unix ms ceiling. - f.applyTSOCeiling(ceilingMs) + ceilingMs, err := decodeTSOCeiling(binary.BigEndian.Uint64(buf[:])) + if err != nil { + return errors.Wrap(err, "tso fsm: restore snapshot") + } + f.applyTSOCeiling(ceilingMs, true) return nil } -func (f *TSOStateMachine) applyTSOCeiling(ceilingMs int64) { +func decodeTSOCeiling(raw uint64) (int64, error) { + if raw > uint64(maxHLCPhysicalMillis) { + return 0, errors.Newf("tso fsm: hlc lease: physical ceiling %d exceeds max %d", raw, maxHLCPhysicalMillis) //nolint:wrapcheck // creating new error, nothing to wrap + } + return int64(raw), nil //nolint:gosec // raw is bounded by maxHLCPhysicalMillis. +} + +func (f *TSOStateMachine) applyTSOCeiling(ceilingMs int64, restore bool) { if ceilingMs <= 0 { return } - f.advanceCommittedCeiling(ceilingMs) + prevCeiling, advanced := f.advanceCommittedCeiling(ceilingMs) + if !advanced && !restore { + return + } if f.hlc != nil { - f.hlc.Observe(tsoCeilingMaxTimestamp(ceilingMs)) + floorCeiling := prevCeiling + if restore { + floorCeiling = ceilingMs + } + if floorCeiling > 0 { + f.hlc.Observe(tsoCeilingMaxTimestamp(floorCeiling)) + } f.hlc.SetPhysicalCeiling(ceilingMs) } } -func (f *TSOStateMachine) advanceCommittedCeiling(ceilingMs int64) { +func (f *TSOStateMachine) advanceCommittedCeiling(ceilingMs int64) (int64, bool) { for { prev := f.ceilingMs.Load() if ceilingMs <= prev { - return + return prev, false } if f.ceilingMs.CompareAndSwap(prev, ceilingMs) { - return + return prev, true } } } @@ -115,7 +138,7 @@ func (s *tsoSnapshot) WriteTo(w io.Writer) (int64, error) { return int64(n), errors.WithStack(err) } if n != tsoSnapshotLen { - return int64(n), io.ErrShortWrite + return int64(n), errors.WithStack(io.ErrShortWrite) } return tsoSnapshotLen, nil } diff --git a/kv/tso_fsm_test.go b/kv/tso_fsm_test.go index 91d4b136e..5c3d65151 100644 --- a/kv/tso_fsm_test.go +++ b/kv/tso_fsm_test.go @@ -2,6 +2,8 @@ package kv import ( "bytes" + "encoding/binary" + "io" "testing" "time" @@ -17,21 +19,44 @@ func TestTSOStateMachineApplyHLCLeaseUpdatesCeiling(t *testing.T) { require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) require.Equal(t, ceilingMs, clock.PhysicalCeiling()) - require.Equal(t, tsoCeilingMaxTimestamp(ceilingMs), clock.Current()) + require.Zero(t, clock.Current()) require.Equal(t, ceilingMs, fsm.committedCeiling()) } -func TestTSOStateMachineApplyHLCLeaseExhaustsCommittedCeiling(t *testing.T) { +func TestTSOStateMachineApplyHLCLeaseKeepsRenewedWindowAllocatable(t *testing.T) { t.Parallel() - ceilingMs := time.Now().Add(time.Hour).UnixMilli() + firstCeilingMs := time.Now().Add(time.Hour).UnixMilli() + secondCeilingMs := firstCeilingMs + 100 clock := NewHLC() fsm := NewTSOStateMachine(clock) - require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(ceilingMs))) - _, err := clock.NextBatchFenced(1) - require.ErrorIs(t, err, ErrCeilingExpired) - require.Equal(t, uint64(1), clock.NextFencedRejections()) + require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(firstCeilingMs))) + first, err := clock.NextBatchFenced(1) + require.NoError(t, err) + require.EqualValues(t, firstCeilingMs, first>>hlcLogicalBits) + + require.Nil(t, fsm.Apply(marshalHLCLeaseRenew(secondCeilingMs))) + require.Equal(t, tsoCeilingMaxTimestamp(firstCeilingMs), clock.Current()) + second, err := clock.NextBatchFenced(1) + require.NoError(t, err) + require.EqualValues(t, secondCeilingMs, second>>hlcLogicalBits) + require.Equal(t, uint64(0), clock.NextFencedRejections()) +} + +func TestTSOStateMachineApplyRejectsOutOfRangeHLCLease(t *testing.T) { + t.Parallel() + + clock := NewHLC() + clock.Observe(123) + clock.SetPhysicalCeiling(456) + fsm := NewTSOStateMachine(clock) + + resp := fsm.Apply(marshalRawHLCLeaseRenew(uint64(maxHLCPhysicalMillis) + 1)) + requireApplyError(t, resp) + require.Equal(t, uint64(123), clock.Current()) + require.Equal(t, int64(456), clock.PhysicalCeiling()) + require.Zero(t, fsm.committedCeiling()) } func TestTSOStateMachineApplyIgnoresNonLeasePayload(t *testing.T) { @@ -77,6 +102,8 @@ func TestTSOStateMachineSnapshotRestoreRoundTrip(t *testing.T) { require.Equal(t, ceilingMs, restoredClock.PhysicalCeiling()) require.Equal(t, tsoCeilingMaxTimestamp(ceilingMs), restoredClock.Current()) require.Equal(t, ceilingMs, restored.committedCeiling()) + _, err = restoredClock.NextBatchFenced(1) + require.ErrorIs(t, err, ErrCeilingExpired) } func TestTSOStateMachineSnapshotUsesCommittedCeiling(t *testing.T) { @@ -119,6 +146,22 @@ func TestTSOStateMachineSnapshotWithNilHLCWritesZero(t *testing.T) { require.Equal(t, make([]byte, 8), buf.Bytes()) } +func TestTSOStateMachineRestoreRejectsOutOfRangeSnapshot(t *testing.T) { + t.Parallel() + + clock := NewHLC() + clock.Observe(123) + clock.SetPhysicalCeiling(456) + fsm := NewTSOStateMachine(clock) + + var buf [tsoSnapshotLen]byte + binary.BigEndian.PutUint64(buf[:], uint64(maxHLCPhysicalMillis)+1) + require.Error(t, fsm.Restore(bytes.NewReader(buf[:]))) + require.Equal(t, uint64(123), clock.Current()) + require.Equal(t, int64(456), clock.PhysicalCeiling()) + require.Zero(t, fsm.committedCeiling()) +} + func TestTSOStateMachineRestoreRejectsMalformedSnapshot(t *testing.T) { t.Parallel() @@ -128,6 +171,15 @@ func TestTSOStateMachineRestoreRejectsMalformedSnapshot(t *testing.T) { require.Error(t, fsm.Restore(bytes.NewReader(make([]byte, 9)))) } +func TestTSOStateMachineSnapshotWriteWrapsShortWrite(t *testing.T) { + t.Parallel() + + snap := &tsoSnapshot{ceilingMs: 1} + n, err := snap.WriteTo(shortTSOWriter{}) + require.EqualValues(t, tsoSnapshotLen-1, n) + require.ErrorIs(t, err, io.ErrShortWrite) +} + func TestTSOStateMachineClassifiesHLCLeaseAsVolatileOnly(t *testing.T) { t.Parallel() @@ -145,3 +197,16 @@ func requireApplyError(t *testing.T, got any) { require.True(t, ok) require.Error(t, err) } + +func marshalRawHLCLeaseRenew(raw uint64) []byte { + payload := []byte{raftEncodeHLCLease} + var buf [hlcLeasePayloadLen]byte + binary.BigEndian.PutUint64(buf[:], raw) + return append(payload, buf[:]...) +} + +type shortTSOWriter struct{} + +func (shortTSOWriter) Write(p []byte) (int, error) { + return len(p) - 1, nil +}