diff --git a/internal/filesystem/fuseadapter/adapter.go b/internal/filesystem/fuseadapter/adapter.go index eb275874f..0dff214f2 100644 --- a/internal/filesystem/fuseadapter/adapter.go +++ b/internal/filesystem/fuseadapter/adapter.go @@ -106,7 +106,7 @@ var errnoMappings = []struct { {filesystem.ErrInvalid, syscall.EINVAL}, {filesystem.ErrUnsupported, syscall.EOPNOTSUPP}, {filesystem.ErrStaleHome, syscall.EAGAIN}, - {filesystem.ErrContentionLimit, syscall.EBUSY}, + {filesystem.ErrContentionLimit, syscall.EAGAIN}, {filesystem.ErrOperationQuota, syscall.ENOSPC}, {store.ErrWriteConflict, syscall.EAGAIN}, } diff --git a/internal/filesystem/fuseadapter/adapter_test.go b/internal/filesystem/fuseadapter/adapter_test.go index 3e1a94f00..6550c92e0 100644 --- a/internal/filesystem/fuseadapter/adapter_test.go +++ b/internal/filesystem/fuseadapter/adapter_test.go @@ -32,7 +32,7 @@ func TestErrnoMapsFilesystemErrors(t *testing.T) { {name: "invalid", err: filesystem.ErrInvalid, want: syscall.EINVAL}, {name: "unsupported", err: filesystem.ErrUnsupported, want: syscall.EOPNOTSUPP}, {name: "stale home", err: filesystem.ErrStaleHome, want: syscall.EAGAIN}, - {name: "contention limit", err: filesystem.ErrContentionLimit, want: syscall.EBUSY}, + {name: "contention limit", err: filesystem.ErrContentionLimit, want: syscall.EAGAIN}, {name: "operation journal quota", err: filesystem.ErrOperationQuota, want: syscall.ENOSPC}, {name: "write conflict", err: cerrors.Wrap(store.ErrWriteConflict, "txn"), want: syscall.EAGAIN}, { diff --git a/internal/filesystem/read_ts_fence_test.go b/internal/filesystem/read_ts_fence_test.go index 87a4b7700..3264c85c8 100644 --- a/internal/filesystem/read_ts_fence_test.go +++ b/internal/filesystem/read_ts_fence_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/kv/kvtest" "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) @@ -15,6 +16,7 @@ import ( type allGroupsTimestampFenceCoordinator struct { inner *testCoordinator fenceTS uint64 + fenceFn func() uint64 fenceErr error calls int plainCall int @@ -22,6 +24,9 @@ type allGroupsTimestampFenceCoordinator struct { func (c *allGroupsTimestampFenceCoordinator) LeaseReadAllGroupsTimestamp(context.Context) (uint64, error) { c.calls++ + if c.fenceFn != nil { + return c.fenceFn(), c.fenceErr + } return c.fenceTS, c.fenceErr } @@ -70,7 +75,26 @@ func TestServiceReadTSRaisesSnapshotToAllGroupsFenceTimestamp(t *testing.T) { coord.fenceTS = 5 ts, err = svc.readTS(ctx) require.NoError(t, err) - require.EqualValues(t, 7, ts, "local store ahead of the fence keeps its own watermark") + require.EqualValues(t, 5, ts, "leader fence remains authoritative when local state is ahead") +} + +func TestServiceReadTSUsesLocalWatermarkForLegacyForwarder(t *testing.T) { + ctx := context.Background() + st := store.NewMVCCStore() + require.NoError(t, st.PutAt(ctx, []byte("key"), []byte("value"), 7, 0)) + seed, err := NewService(st, &testCoordinator{st: st}, WithChunkSize(testChunkSize)) + require.NoError(t, err) + require.NoError(t, seed.InitializeRoot(ctx, testRootMode, 1000, 1000)) + coord := kv.CoordinatorForwarder{Inner: &kvtest.NopCoordinator{}} + svc, err := NewService(st, coord, WithChunkSize(testChunkSize)) + require.NoError(t, err) + + ts, err := svc.readTS(ctx) + require.NoError(t, err) + require.Equal(t, st.LastCommitTS(), ts, "legacy fence has no leader-side commit watermark") + root, err := svc.GetAttr(ctx, RootInode) + require.NoError(t, err) + require.Equal(t, RootInode, root.Inode) } func TestServiceReadTSPropagatesAllGroupsFenceTimestampError(t *testing.T) { @@ -90,24 +114,18 @@ func TestServiceFollowerReadsObserveOwnForwardedWrite(t *testing.T) { backing := store.NewMVCCStore() local := &laggingStore{MVCCStore: backing} inner := &testCoordinator{st: backing} - coord := &allGroupsTimestampFenceCoordinator{inner: inner} + coord := &allGroupsTimestampFenceCoordinator{inner: inner, fenceFn: backing.LastCommitTS} svc, err := NewService(local, coord, WithChunkSize(testChunkSize), WithIDAllocator(sequenceIDAllocator(2))) require.NoError(t, err) - fence := func() { coord.fenceTS = backing.LastCommitTS() } - - fence() require.NoError(t, svc.InitializeRoot(ctx, testRootMode, 1000, 1000)) - fence() file, err := svc.Create(ctx, RootInode, []byte("file"), CreateOptions{Mode: testFileMode}) require.NoError(t, err) - fence() _, err = svc.Write(ctx, file.Inode, 0, 0, []byte("abcd")) require.NoError(t, err) // The write committed on the leader (backing store) but this node's // applied watermark still trails it by one commit. local.lag = 1 - fence() stat, err := svc.GetAttr(ctx, file.Inode) require.NoError(t, err) diff --git a/internal/filesystem/service.go b/internal/filesystem/service.go index 4d39615d3..747abd83e 100644 --- a/internal/filesystem/service.go +++ b/internal/filesystem/service.go @@ -2774,15 +2774,22 @@ func (s *Service) openHandleExpiresNsec(now time.Time) int64 { return now.Add(s.openLeaseTTL).UnixNano() } -// readTS returns the snapshot timestamp for a read: the local applied -// watermark, raised to the fence's leader-side commit timestamp when the -// fence reports one. Lease and linearizable fences return Raft indexes, not -// commit timestamps, so only the all-groups timestamp fence can raise it. +// readTS returns the leader-fenced snapshot timestamp when available. A +// follower's local applied watermark can run ahead of the leader that serves +// its routed reads. Taking the maximum would let a read at that timestamp +// miss an in-flight write with the same commit timestamp, so OCC could accept +// a stale read-modify-write. Older fences report only Raft indexes and still +// use the local applied watermark. func (s *Service) readTS(ctx context.Context) (uint64, error) { fenceTS, err := s.fenceReadSnapshot(ctx) if err != nil { return 0, err } + if reader, ok := s.dispatch.(allGroupsLeaseTimestampReader); ok { + if support, wrapped := reader.(interface{ SupportsAllGroupsLeaseTimestamp() bool }); !wrapped || support.SupportsAllGroupsLeaseTimestamp() { + return fenceTS, nil + } + } return max(s.store.LastCommitTS(), fenceTS), nil } diff --git a/kv/coordinator.go b/kv/coordinator.go index d72f5694a..4c4567020 100644 --- a/kv/coordinator.go +++ b/kv/coordinator.go @@ -1013,7 +1013,7 @@ func (c *Coordinate) RecoverHLCLease(ctx context.Context) error { } c.hlcRecoveryMu.Lock() defer c.hlcRecoveryMu.Unlock() - if !hlcCeilingExpired(c.clock) { + if !hlcCeilingExhausted(c.clock) { return nil } if !c.IsLeaderAcceptingWrites() { @@ -1024,7 +1024,7 @@ func (c *Coordinate) RecoverHLCLease(ctx context.Context) error { } rctx, cancel := hlcRecoveryContext(ctx) defer cancel() - ceilingMs := time.Now().UnixMilli() + hlcPhysicalWindowMs + ceilingMs := hlcRecoveryCeiling(c.clock) return errors.Wrap(c.ProposeHLCLease(rctx, ceilingMs), "recover hlc lease") } diff --git a/kv/coordinator_forwarder.go b/kv/coordinator_forwarder.go index 6555dd848..672a35a34 100644 --- a/kv/coordinator_forwarder.go +++ b/kv/coordinator_forwarder.go @@ -200,6 +200,17 @@ func (f CoordinatorForwarder) LeaseReadAllGroupsTimestamp(ctx context.Context) ( return LeaseReadAllGroupsTimestampThrough(f.Inner, ctx) } +// SupportsAllGroupsLeaseTimestamp reports whether the inner coordinator can +// return a leader-side commit watermark. The forwarder's timestamp method also +// fences legacy coordinators, but returns zero for those fences. +func (f CoordinatorForwarder) SupportsAllGroupsLeaseTimestamp() bool { + if support, ok := f.Inner.(interface{ SupportsAllGroupsLeaseTimestamp() bool }); ok { + return support.SupportsAllGroupsLeaseTimestamp() + } + _, ok := f.Inner.(AllGroupsLeaseTimestampCoordinator) + return ok +} + // IsLeaderForGroup / RaftLeaderForGroup / LeaseReadForGroup carry the // group-keyed read-fence path. Without them every fence target falls back to // key resolution, which re-introduces the legacy wide-column group collapse diff --git a/kv/coordinator_forwarder_test.go b/kv/coordinator_forwarder_test.go index 9fadc21e4..1c96c0d1a 100644 --- a/kv/coordinator_forwarder_test.go +++ b/kv/coordinator_forwarder_test.go @@ -1,6 +1,7 @@ package kv import ( + "context" "testing" "github.com/stretchr/testify/require" @@ -54,3 +55,23 @@ func TestCoordinatorForwarderGroupLeadershipDelegatesToGroupRouter(t *testing.T) require.False(t, f.IsLeaderForGroup(2)) require.Equal(t, "n2:50051", f.RaftLeaderForGroup(2)) } + +func TestCoordinatorForwarderTimestampSupportFollowsInner(t *testing.T) { + t.Parallel() + + legacy := CoordinatorForwarder{Inner: &nopCoordinator{}} + require.False(t, legacy.SupportsAllGroupsLeaseTimestamp()) + require.False(t, (CoordinatorForwarder{Inner: legacy}).SupportsAllGroupsLeaseTimestamp()) + + withTimestamp := CoordinatorForwarder{Inner: timestampCoordinator{Coordinator: &nopCoordinator{}}} + require.True(t, withTimestamp.SupportsAllGroupsLeaseTimestamp()) + require.True(t, (CoordinatorForwarder{Inner: withTimestamp}).SupportsAllGroupsLeaseTimestamp()) +} + +type timestampCoordinator struct { + Coordinator +} + +func (timestampCoordinator) LeaseReadAllGroupsTimestamp(context.Context) (uint64, error) { + return 1, nil +} diff --git a/kv/lease_warmup_test.go b/kv/lease_warmup_test.go index afa9d8976..b6225e6b2 100644 --- a/kv/lease_warmup_test.go +++ b/kv/lease_warmup_test.go @@ -366,6 +366,44 @@ func TestShardedCoordinator_RecoverHLCLease_ProposesToEveryLedGroup(t *testing.T require.NotZero(t, got) } +func TestShardedCoordinator_RecoverHLCLease_LogicalCeilingExhausted(t *testing.T) { + t.Parallel() + clock := NewHLC() + ceiling := time.Now().Add(time.Minute).UnixMilli() + clock.SetPhysicalCeiling(ceiling) + clock.Observe((uint64(ceiling) << hlcLogicalBits) | hlcLogicalMask) + eng := newShardedLeaseEngine(100) + eng.proposeApply = applyHLCLeaseEntryToClock(t, clock) + follower := newShardedLeaseEngine(200) + follower.state.Store(raftengine.StateFollower) + coord := mustShardedLeaseCoord(t, eng, follower) + coord.clock = clock + + _, err := coord.nextTxnTSAfter(context.Background(), 0) + require.NoError(t, err) + require.Equal(t, int32(1), eng.proposeCalls.Load()) + _, err = clock.NextFenced() + require.NoError(t, err) + require.Greater(t, clock.PhysicalCeiling(), ceiling) +} + +func TestCoordinate_RecoverHLCLease_LogicalCeilingExhausted(t *testing.T) { + t.Parallel() + clock := NewHLC() + ceiling := time.Now().Add(time.Minute).UnixMilli() + clock.SetPhysicalCeiling(ceiling) + clock.Observe((uint64(ceiling) << hlcLogicalBits) | hlcLogicalMask) + eng := &fakeLeaseEngine{applied: 11, leaseDur: time.Hour} + eng.proposeApply = applyHLCLeaseEntryToClock(t, clock) + coord := NewCoordinatorWithEngine(nil, eng, WithHLC(clock)) + + require.NoError(t, coord.RecoverHLCLease(context.Background())) + require.Equal(t, int32(1), eng.proposeCalls.Load()) + require.Greater(t, clock.PhysicalCeiling(), ceiling) + _, err := clock.NextFenced() + require.NoError(t, err) +} + func TestShardedCoordinator_RecoverHLCLease_SlowTargetDoesNotBlockSuccessfulPeer(t *testing.T) { t.Parallel() clock := NewHLC() diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index 8fb93ac58..0970fb892 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -3847,7 +3847,7 @@ func (c *ShardedCoordinator) RecoverHLCLease(ctx context.Context) error { } c.hlcRecoveryMu.Lock() defer c.hlcRecoveryMu.Unlock() - if !hlcCeilingExpired(c.clock) { + if !hlcCeilingExhausted(c.clock) { return nil } if c.hlcRenewalBlocked != nil && c.hlcRenewalBlocked() { @@ -3859,7 +3859,7 @@ func (c *ShardedCoordinator) RecoverHLCLease(ctx context.Context) error { } rctx, cancel := hlcRecoveryContext(ctx) defer cancel() - return c.recoverHLCLeaseTargets(rctx, targets, time.Now().UnixMilli()+hlcPhysicalWindowMs) + return c.recoverHLCLeaseTargets(rctx, targets, hlcRecoveryCeiling(c.clock)) } func (c *ShardedCoordinator) recoverHLCLeaseTargets(ctx context.Context, targets []hlcLeaseRecoveryTarget, ceilingMs int64) error { @@ -3880,7 +3880,7 @@ func (c *ShardedCoordinator) recoverHLCLeaseTargets(ctx context.Context, targets for err := range errCh { recoveryErr = combineHLCRecoveryErrors(recoveryErr, err) } - if !hlcCeilingExpired(c.clock) { + if !hlcCeilingExhausted(c.clock) { return nil } return hlcRecoveryFailedError(recoveryErr) diff --git a/kv/tso.go b/kv/tso.go index 214b954e5..e2828de80 100644 --- a/kv/tso.go +++ b/kv/tso.go @@ -747,12 +747,32 @@ func hlcRecoveryContext(ctx context.Context) (context.Context, context.CancelFun return context.WithTimeout(ctx, dispatchLeaderRetryBudget) } -func hlcCeilingExpired(clock *HLC) bool { +// hlcCeilingExhausted covers both wall-clock expiry and a logical counter +// that cannot advance beyond the currently committed physical ceiling. +func hlcCeilingExhausted(clock *HLC) bool { if clock == nil { return false } ceiling := clock.PhysicalCeiling() - return ceiling > 0 && time.Now().UnixMilli() >= ceiling + if ceiling <= 0 { + return false + } + if time.Now().UnixMilli() >= ceiling { + return true + } + last := clock.Current() + physical := last >> hlcLogicalBits + return physical > uint64(ceiling) || + (physical == uint64(ceiling) && last&hlcLogicalMask == hlcLogicalMask) +} + +// hlcRecoveryCeiling advances beyond both the wall clock and the clock's +// committed or observed physical time. A renewal based only on wall time can +// be lower than an already committed ceiling after clock skew or logical +// counter exhaustion, leaving timestamp issuance blocked. +func hlcRecoveryCeiling(clock *HLC) int64 { + base := max(time.Now().UnixMilli(), clock.PhysicalCeiling(), clampUint64ToInt64(clock.Current()>>hlcLogicalBits)) + return base + hlcPhysicalWindowMs } type tsoCoordinator interface {