Fix scheduled strict FUSE Jepsen failures

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},
 }

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},
 		{

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,13 +16,17 @@ import (
 type allGroupsTimestampFenceCoordinator struct {
 	inner     *testCoordinator
 	fenceTS   uint64
+	fenceFn   func() uint64
 	fenceErr  error
 	calls     int
 	plainCall int
 }
 
 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)

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
 }
 

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")
 }
 

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

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
+}

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()

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)

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 {