From ba366222f2131fb7ff319bd2f1079c09318f06bb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 09:04:36 +0000 Subject: [PATCH 01/15] main: reproduce the TTL migration losing local leader groups The Redis delta compactor's TTL-inline migrator does no work in the production server. main.go hands it a startupGatedCoordinator, and NewDeltaCompactor wraps that in WithKeyVizLabel. The label wrapper always exposes LocalLeaderGroupIDs and answers nil when its inner value lacks the method, and the startup gate never forwarded it. The migrator then takes its per-group branch with zero groups, plans no scans, and never reaches the full-keyspace fallback. Pin the real chain: a ShardedCoordinator leading groups 1 and 2 and following group 3, wrapped exactly as main.go wraps it. The test fails here because LocalLeaderGroupIDs comes back nil through the chain and SyncOnce issues no ScanGroupAt at all. The fix follows in the next commit. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- main_redis_ttl_migration_wiring_test.go | 107 ++++++++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 main_redis_ttl_migration_wiring_test.go diff --git a/main_redis_ttl_migration_wiring_test.go b/main_redis_ttl_migration_wiring_test.go new file mode 100644 index 000000000..7488f5721 --- /dev/null +++ b/main_redis_ttl_migration_wiring_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "context" + "sync" + "testing" + + "github.com/bootjp/elastickv/adapter" + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/keyviz" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +// groupScanRecordingStore stands in for the *kv.ShardStore the Redis delta +// compactor holds in production. Exposing ScanGroupAt is what lets the +// TTL-inline migrator plan one scan per locally led group instead of a +// full-keyspace scan. +type groupScanRecordingStore struct { + store.MVCCStore + + mu sync.Mutex + groupScans map[uint64][]string + fullScans []string +} + +func (s *groupScanRecordingStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { + s.mu.Lock() + s.fullScans = append(s.fullScans, string(start)) + s.mu.Unlock() + return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) +} + +func (s *groupScanRecordingStore) ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { + s.mu.Lock() + if s.groupScans == nil { + s.groupScans = make(map[uint64][]string) + } + s.groupScans[groupID] = append(s.groupScans[groupID], string(start)) + s.mu.Unlock() + return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) +} + +func (s *groupScanRecordingStore) recorded() (map[uint64][]string, []string) { + s.mu.Lock() + defer s.mu.Unlock() + groupScans := make(map[uint64][]string, len(s.groupScans)) + for groupID, starts := range s.groupScans { + groupScans[groupID] = append([]string(nil), starts...) + } + return groupScans, append([]string(nil), s.fullScans...) +} + +// main.go hands the Redis delta compactor a startupGatedCoordinator, and +// NewDeltaCompactor wraps that in WithKeyVizLabel. The TTL-inline migrator asks +// that outermost value which data groups this node leads and scans only those. +// The label wrapper always answers, so when the startup gate in between did not +// forward LocalLeaderGroupIDs the migrator got an empty list, planned no scans, +// and migrated nothing in production. Adapter tests hand the compactor an +// unwrapped coordinator and could not see it, so the real chain is pinned here. +func TestRedisTTLInlineMigrationScansLocalLeaderGroupsThroughStartupGate(t *testing.T) { + t.Parallel() + + leader := &mainTSOEngine{state: raftengine.StateLeader} + follower := &mainTSOEngine{state: raftengine.StateFollower} + coord := kv.NewShardedCoordinator(distribution.NewEngine(), map[uint64]*kv.ShardGroup{ + 1: {Engine: leader}, + 2: {Engine: leader}, + 3: {Engine: follower}, + }, 1, kv.NewHLC(), nil) + localLeaders := []uint64{1, 2} + require.Equal(t, localLeaders, coord.LocalLeaderGroupIDs()) + + gate := &startupPublicKVGate{} + gate.markReady() + gated := startupGatedCoordinator{inner: coord, gate: gate} + + labeled, ok := kv.WithKeyVizLabel(gated, keyviz.LabelRedis).(interface{ LocalLeaderGroupIDs() []uint64 }) + require.True(t, ok, "the Redis compactor's coordinator must expose LocalLeaderGroupIDs") + require.Equal(t, localLeaders, labeled.LocalLeaderGroupIDs(), + "the local leader groups must survive the startup gate and the keyviz label") + + st := &groupScanRecordingStore{MVCCStore: store.NewMVCCStore()} + compactor := adapter.NewDeltaCompactor(st, gated) + require.NoError(t, compactor.SyncOnce(context.Background())) + + groupScans, fullScans := st.recorded() + scannedGroups := make([]uint64, 0, len(groupScans)) + for groupID := range groupScans { + scannedGroups = append(scannedGroups, groupID) + } + require.ElementsMatch(t, localLeaders, scannedGroups, + "the migrator must scan every group this node leads and no group it follows") + require.Contains(t, groupScans[1], store.HashMetaPrefix) + require.Equal(t, groupScans[1], groupScans[2], + "every TTL-inline handler must plan one scan per locally led group") + seen := make(map[string]struct{}, len(groupScans[1])) + for _, start := range groupScans[1] { + _, dup := seen[start] + require.Falsef(t, dup, "handler prefix %q was scanned twice in one group", start) + seen[start] = struct{}{} + } + require.NotContains(t, fullScans, store.HashMetaPrefix, + "the migrator must not fall back to a full-keyspace scan when the groups are known") +} From 47f852edbf62afd2b5d8c016fd9741d2add51b70 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 09:13:35 +0000 Subject: [PATCH 02/15] kv, main, adapter: keep local leader groups through the coordinator wrappers The TTL-inline migrator asks its coordinator which data groups this node leads so it scans only those. In the server that coordinator is WithKeyVizLabel(startupGatedCoordinator{ShardedCoordinator}). The label wrapper always exposed LocalLeaderGroupIDs and answered nil when its inner value lacked the method, and the startup gate never forwarded it. The migrator read nil as "this node leads no group", planned no scans, and never reached its full-keyspace fallback. The migration has done nothing in production since a36dac7f. Forward LocalLeaderGroupIDs through the startup gate, and make the answer carry whether it is known. kv.LocalLeaderGroupsCoordinator now returns (ids, known); ShardedCoordinator always knows, and decorators forward through kv.LocalLeaderGroupIDsThrough, which reports a coordinator without the capability as unknown. A decorator that drops the forward can no longer turn "unknown" into "no groups": the migrator falls back to the full-keyspace scan it already had, filtered per key by IsLeaderForKey, and logs once that it did. A known empty list still plans nothing, since this node then leads no group. The regression test from the previous commit now passes; only its two probe lines change for the new signature. Adapter and kv tests pin the unknown and known-empty cases; both fail against the old "nil means no groups" behaviour. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor.go | 2 + adapter/redis_delta_compactor_test.go | 83 ++++++++++++++++++++++++- adapter/redis_ttl_inline_migrator.go | 42 ++++++++----- kv/coordinator.go | 22 +++++++ kv/keyviz_label.go | 11 ++-- kv/keyviz_label_test.go | 52 ++++++++++++++++ kv/sharded_coordinator.go | 9 ++- main.go | 9 +++ main_redis_ttl_migration_wiring_test.go | 10 +-- 9 files changed, 211 insertions(+), 29 deletions(-) diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index 59e9cf5fe..670c40355 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -75,6 +75,8 @@ type DeltaCompactor struct { stateMu sync.Mutex backoffUntil time.Time handlerCursor int + + ttlInlineGroupsUnknownOnce sync.Once } // DeltaCompactorOption configures a DeltaCompactor. diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index dc7eadebd..62a6a27db 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -559,8 +559,8 @@ type localLeaderGroupCompactionCoordinator struct { groupIDs []uint64 } -func (c *localLeaderGroupCompactionCoordinator) LocalLeaderGroupIDs() []uint64 { - return append([]uint64(nil), c.groupIDs...) +func (c *localLeaderGroupCompactionCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { + return append([]uint64(nil), c.groupIDs...), true } type recordingTTLInlineGroupScanStore struct { @@ -614,6 +614,85 @@ func TestDeltaCompactor_TTLInlineUsesLocalLeaderGroupScan(t *testing.T) { require.Equal(t, redisExpireAtMillis(expireAt), meta.ExpireAt) } +// capabilityHidingCompactionCoordinator forwards only the kv.Coordinator +// methods, the way a decorator that forgets an optional forward does. +type capabilityHidingCompactionCoordinator struct { + kv.Coordinator +} + +// The coordinator the compactor holds is wrapped in WithKeyVizLabel, which +// always exposes LocalLeaderGroupIDs. When a decorator underneath hides the +// capability the answer must stay unknown and the migrator must fall back to a +// full-keyspace scan; answering "no groups" there planned nothing and silently +// stopped the migration in production. A known empty list is different: this +// node leads no group, so there is nothing for it to migrate. +func TestDeltaCompactor_TTLInlineScanPlanWhenLocalLeaderGroupsUnknownOrEmpty(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + wrap func(kv.Coordinator) kv.Coordinator + groupIDs []uint64 + wantFullScan bool + wantMigrated bool + }{ + { + name: "unknown behind a decorator", + wrap: func(c kv.Coordinator) kv.Coordinator { + return capabilityHidingCompactionCoordinator{Coordinator: c} + }, + groupIDs: []uint64{2}, + wantFullScan: true, + wantMigrated: true, + }, + { + name: "known with no local leader group", + wrap: func(c kv.Coordinator) kv.Coordinator { return c }, + groupIDs: []uint64{}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := &recordingTTLInlineGroupScanStore{MVCCStore: store.NewMVCCStore()} + inner := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: tc.groupIDs, + } + c := NewDeltaCompactor(st, tc.wrap(inner), WithDeltaCompactorMaxDeltaCount(2)) + + key := []byte("ttl:migrate:group-plan") + expireAt := time.Now().Add(time.Hour) + legacyMeta := make([]byte, redisSimpleMetaLegacySizeBytes) + binary.BigEndian.PutUint64(legacyMeta, 1) + require.NoError(t, st.PutAt(ctx, store.HashMetaKey(key), legacyMeta, 1, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey(key), encodeRedisTTL(expireAt), 2, 0)) + + require.NoError(t, c.SyncOnce(ctx)) + + require.Empty(t, st.groupScans) + if tc.wantFullScan { + require.Contains(t, st.fullStarts, store.HashMetaPrefix) + } else { + require.NotContains(t, st.fullStarts, store.HashMetaPrefix) + } + raw, err := st.GetAt(ctx, store.HashMetaKey(key), st.LastCommitTS()) + require.NoError(t, err) + meta, err := store.UnmarshalHashMeta(raw) + require.NoError(t, err) + wantExpireAt := uint64(0) + if tc.wantMigrated { + wantExpireAt = redisExpireAtMillis(expireAt) + } + require.Equal(t, wantExpireAt, meta.ExpireAt) + }) + } +} + func TestDeltaCompactor_TTLInlineMigratesEmptyCollectionKeys(t *testing.T) { t.Parallel() diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index ea7e7950f..d8b711416 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -27,10 +27,6 @@ type ttlInlineGroupScanner interface { ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) } -type ttlInlineLocalLeaderGroups interface { - LocalLeaderGroupIDs() []uint64 -} - type ttlInlineMigrationScanPlan struct { cursorName string groupID uint64 @@ -242,24 +238,40 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( return nil } +// ttlInlineMigrationScanPlans scans only the groups this node leads when the +// store can scan one group and the coordinator knows which groups those are. A +// known empty list means this node leads no group and plans nothing. When the +// groups are unknown the migrator falls back to one full-keyspace scan, +// filtered per key by IsLeaderForKey, rather than migrating nothing. func (c *DeltaCompactor) ttlInlineMigrationScanPlans(h ttlInlineMigrationHandler) []ttlInlineMigrationScanPlan { scanner, hasGroupScanner := c.st.(ttlInlineGroupScanner) - groups, hasLocalGroups := c.coord.(ttlInlineLocalLeaderGroups) - if hasGroupScanner && hasLocalGroups { - groupIDs := groups.LocalLeaderGroupIDs() - plans := make([]ttlInlineMigrationScanPlan, 0, len(groupIDs)) - for _, groupID := range groupIDs { - plans = append(plans, ttlInlineMigrationScanPlan{ - cursorName: ttlInlineMigrationGroupCursorName(h.typeName, groupID), - groupID: groupID, - scanner: scanner, - }) + if hasGroupScanner { + if groupIDs, known := kv.LocalLeaderGroupIDsThrough(c.coord); known { + plans := make([]ttlInlineMigrationScanPlan, 0, len(groupIDs)) + for _, groupID := range groupIDs { + plans = append(plans, ttlInlineMigrationScanPlan{ + cursorName: ttlInlineMigrationGroupCursorName(h.typeName, groupID), + groupID: groupID, + scanner: scanner, + }) + } + return plans } - return plans + c.warnTTLInlineLocalLeaderGroupsUnknown() } return []ttlInlineMigrationScanPlan{{cursorName: h.typeName}} } +// warnTTLInlineLocalLeaderGroupsUnknown logs once when the store can scan one +// group but the coordinator cannot name the groups this node leads. In the +// server that means a coordinator decorator dropped LocalLeaderGroupIDs, and +// every node is back to full-keyspace migration scans. +func (c *DeltaCompactor) warnTTLInlineLocalLeaderGroupsUnknown() { + c.ttlInlineGroupsUnknownOnce.Do(func() { + c.logger.Warn("ttl inline migrator: coordinator does not report local leader groups; falling back to full-keyspace scans") + }) +} + func ttlInlineMigrationGroupCursorName(typeName string, groupID uint64) string { return typeName + ":group:" + strconv.FormatUint(groupID, 10) } diff --git a/kv/coordinator.go b/kv/coordinator.go index 45581efaf..cc87a3d78 100644 --- a/kv/coordinator.go +++ b/kv/coordinator.go @@ -468,6 +468,28 @@ type GroupLeaderRoutableCoordinator interface { RaftLeaderForGroup(groupID uint64) string } +// LocalLeaderGroupsCoordinator is the optional capability implemented by +// coordinators that know which data groups this process currently leads. +// Background maintenance uses it to scan only those groups instead of the whole +// keyspace. known is false when the coordinator cannot tell -- a decorator whose +// inner coordinator lacks the capability -- and callers must treat that as +// unknown, not as leading no group: a decorator answering "no groups" there +// silently stopped the Redis TTL-inline migration. +type LocalLeaderGroupsCoordinator interface { + LocalLeaderGroupIDs() (ids []uint64, known bool) +} + +// LocalLeaderGroupIDsThrough returns the data groups c leads, with known false +// when c does not expose the capability. Decorators forward through it so an +// inner coordinator without the capability stays unknown. +func LocalLeaderGroupIDsThrough(c Coordinator) ([]uint64, bool) { + lg, ok := c.(LocalLeaderGroupsCoordinator) + if !ok { + return nil, false + } + return lg.LocalLeaderGroupIDs() +} + // LeaseReadGroupTargets collapses targets to one per Raft group, preferring the // group id a target already carries over re-deriving it from the key bytes. A // target with GroupID 0 falls back to key resolution, which is what plain diff --git a/kv/keyviz_label.go b/kv/keyviz_label.go index 2c8c73b87..d50360a37 100644 --- a/kv/keyviz_label.go +++ b/kv/keyviz_label.go @@ -219,9 +219,10 @@ func (c keyVizLabeledCoordinator) EngineGroupIDForKey(key []byte) uint64 { return 0 } -func (c keyVizLabeledCoordinator) LocalLeaderGroupIDs() []uint64 { - if lg, ok := c.inner.(interface{ LocalLeaderGroupIDs() []uint64 }); ok { - return lg.LocalLeaderGroupIDs() - } - return nil +// LocalLeaderGroupIDs forwards which data groups the inner coordinator leads. +// An inner coordinator without the capability stays unknown instead of +// answering "no groups": the TTL-inline migrator then falls back to a +// full-keyspace scan rather than planning nothing. +func (c keyVizLabeledCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { + return LocalLeaderGroupIDsThrough(c.inner) } diff --git a/kv/keyviz_label_test.go b/kv/keyviz_label_test.go index 8a62a1ecd..cd04c0faa 100644 --- a/kv/keyviz_label_test.go +++ b/kv/keyviz_label_test.go @@ -3,6 +3,9 @@ package kv import ( "context" "testing" + + "github.com/bootjp/elastickv/keyviz" + "github.com/stretchr/testify/require" ) // The Redis adapter wraps its coordinator in WithKeyVizLabel, so this wrapper is @@ -22,3 +25,52 @@ func TestKeyVizLabeledCoordinatorForwardsGroupRouting(t *testing.T) { t.Fatal("keyVizLabeledCoordinator must forward LeaseReadForGroup") } } + +type localLeaderGroupsStub struct { + Coordinator + ids []uint64 +} + +func (s localLeaderGroupsStub) LocalLeaderGroupIDs() ([]uint64, bool) { return s.ids, true } + +// The Redis TTL-inline migrator asks the keyviz-labeled coordinator which data +// groups this node leads. The wrapper always has the method, so it must report +// an inner coordinator without the capability as unknown, not as leading no +// group: that answer made the migrator plan no scans at all. A known empty +// answer must stay known, because it means this node leads nothing. +func TestKeyVizLabeledCoordinatorLocalLeaderGroupIDs(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + inner Coordinator + wantIDs []uint64 + wantKnown bool + }{ + { + name: "forwards known groups", + inner: localLeaderGroupsStub{ids: []uint64{2, 5}}, + wantIDs: []uint64{2, 5}, + wantKnown: true, + }, + { + name: "keeps a known empty answer", + inner: localLeaderGroupsStub{ids: []uint64{}}, + wantIDs: []uint64{}, + wantKnown: true, + }, + { + name: "inner without the capability is unknown", + inner: struct{ Coordinator }{}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ids, known := LocalLeaderGroupIDsThrough(WithKeyVizLabel(tc.inner, keyviz.LabelRedis)) + require.Equal(t, tc.wantKnown, known) + require.Equal(t, tc.wantIDs, ids) + }) + } +} diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index 86701e33a..b19917e6e 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -2488,11 +2488,14 @@ func (c *ShardedCoordinator) timestampBridgeCandidateGroupIDs() []uint64 { return ids } +var _ LocalLeaderGroupsCoordinator = (*ShardedCoordinator)(nil) + // LocalLeaderGroupIDs returns the configured data shard groups this process // currently leads. Background maintenance uses this to avoid issuing // whole-keyspace scans from every node and proxying those scans back to the -// same hot leaders. -func (c *ShardedCoordinator) LocalLeaderGroupIDs() []uint64 { +// same hot leaders. The coordinator owns the groups, so the answer is always +// known; an empty list means this process leads none of them. +func (c *ShardedCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { ids := c.timestampBridgeCandidateGroupIDs() out := make([]uint64, 0, len(ids)) for _, gid := range ids { @@ -2501,7 +2504,7 @@ func (c *ShardedCoordinator) LocalLeaderGroupIDs() []uint64 { out = append(out, gid) } } - return out + return out, true } func (c *ShardedCoordinator) invalidateTimestampWindow() { diff --git a/main.go b/main.go index 45e453847..f99530715 100644 --- a/main.go +++ b/main.go @@ -2718,6 +2718,7 @@ var _ kv.LeaseReadableCoordinator = (*startupGatedCoordinator)(nil) var _ kv.AllGroupsLeaseReadableCoordinator = (*startupGatedCoordinator)(nil) var _ kv.AllGroupsLeaseTimestampCoordinator = (*startupGatedCoordinator)(nil) var _ kv.GroupRoutableCoordinator = (*startupGatedCoordinator)(nil) +var _ kv.LocalLeaderGroupsCoordinator = (*startupGatedCoordinator)(nil) var _ kv.RaftMembershipCoordinator = (*startupGatedCoordinator)(nil) var _ kv.TimestampAllocatorProvider = (*startupGatedCoordinator)(nil) var _ kv.ConfiguredTimestampAllocatorProvider = (*startupGatedCoordinator)(nil) @@ -2951,6 +2952,14 @@ func (c startupGatedCoordinator) EngineGroupIDForKey(key []byte) uint64 { return 0 } +// LocalLeaderGroupIDs forwards the data groups this node leads. The Redis delta +// compactor reaches it through WithKeyVizLabel to scope the TTL-inline migration +// to locally led groups; without this forward every node falls back to +// full-keyspace migration scans. +func (c startupGatedCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { + return kv.LocalLeaderGroupIDsThrough(c.inner) +} + func (c startupGatedCoordinator) RaftMembers(ctx context.Context) ([]kv.RaftMember, error) { provider, ok := c.inner.(kv.RaftMembershipCoordinator) if !ok { diff --git a/main_redis_ttl_migration_wiring_test.go b/main_redis_ttl_migration_wiring_test.go index 7488f5721..8794638fd 100644 --- a/main_redis_ttl_migration_wiring_test.go +++ b/main_redis_ttl_migration_wiring_test.go @@ -71,15 +71,17 @@ func TestRedisTTLInlineMigrationScansLocalLeaderGroupsThroughStartupGate(t *test 3: {Engine: follower}, }, 1, kv.NewHLC(), nil) localLeaders := []uint64{1, 2} - require.Equal(t, localLeaders, coord.LocalLeaderGroupIDs()) + ids, known := coord.LocalLeaderGroupIDs() + require.True(t, known) + require.Equal(t, localLeaders, ids) gate := &startupPublicKVGate{} gate.markReady() gated := startupGatedCoordinator{inner: coord, gate: gate} - labeled, ok := kv.WithKeyVizLabel(gated, keyviz.LabelRedis).(interface{ LocalLeaderGroupIDs() []uint64 }) - require.True(t, ok, "the Redis compactor's coordinator must expose LocalLeaderGroupIDs") - require.Equal(t, localLeaders, labeled.LocalLeaderGroupIDs(), + ids, known = kv.LocalLeaderGroupIDsThrough(kv.WithKeyVizLabel(gated, keyviz.LabelRedis)) + require.True(t, known, "the Redis compactor's coordinator must know its local leader groups") + require.Equal(t, localLeaders, ids, "the local leader groups must survive the startup gate and the keyviz label") st := &groupScanRecordingStore{MVCCStore: store.NewMVCCStore()} From 44725dd79d14acba1584f35576155ab1f9dc6228 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 09:17:31 +0000 Subject: [PATCH 03/15] main: guard every ShardedCoordinator capability through the startup wrappers Capabilities are optional interfaces found by type assertion, so a wrapper that does not forward one fails silently: the probe misses and the consumer takes a fallback, or the outer wrapper answers on the inner's behalf. That has now cost the Redis read fence its group routing, Internal.Forward its route-floor gate, and the Redis TTL-inline migration all of its work, each time with green unit tests that used an unwrapped coordinator. Classify every exported ShardedCoordinator method by the surface its consumers probe: the adapters' WithKeyVizLabel(startupGatedCoordinator), the startupGatedCoordinator main probes itself (internal gRPC server options, S3 blob peer discovery), or the raw coordinator main drives directly (lifecycle, HLC lease renewal, autosplit, the legacy TSO bridge). A new method fails the test until it is classified. Each forwarded capability is called through its surface against a spy and must reach the spy and return its result and error. The test fails on one gap: the startup gate does not forward LinearizableReadForKey, so WithKeyVizLabel serves it from the default group's LinearizableRead and a key in another group would be fenced against the wrong group. Nothing in the server calls it through the adapters today, so the gap is latent. The next commit closes it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- main_coordinator_capabilities_test.go | 424 ++++++++++++++++++++++++++ 1 file changed, 424 insertions(+) create mode 100644 main_coordinator_capabilities_test.go diff --git a/main_coordinator_capabilities_test.go b/main_coordinator_capabilities_test.go new file mode 100644 index 000000000..77b8f78f1 --- /dev/null +++ b/main_coordinator_capabilities_test.go @@ -0,0 +1,424 @@ +package main + +import ( + "context" + "reflect" + "strings" + "sync" + "testing" + + "github.com/bootjp/elastickv/keyviz" + "github.com/bootjp/elastickv/kv" + pb "github.com/bootjp/elastickv/proto" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +// coordinatorSurface names the value a capability's consumers type-assert +// against in the server. main.go wraps the ShardedCoordinator in +// startupGatedCoordinator and hands that to the adapters, and every adapter +// wraps it again in WithKeyVizLabel. +type coordinatorSurface int + +const ( + // adapterSurface capabilities are probed by adapters on + // WithKeyVizLabel(startupGatedCoordinator), so both wrappers must forward + // them. + adapterSurface coordinatorSurface = iota + 1 + // gateSurface capabilities are probed by main on the startupGatedCoordinator + // itself; no adapter probes them. If one starts to, move the capability to + // adapterSurface and forward it on WithKeyVizLabel with a fallback that + // matches what the consumer does when the capability is absent. + gateSurface + // rawOnly methods are called by main on the ShardedCoordinator it built, + // never through a wrapper. + rawOnly +) + +// shardedCoordinatorSurfaces classifies every exported ShardedCoordinator +// method except the With* builders. Adding a method to ShardedCoordinator fails +// TestStartupWrappersForwardShardedCoordinatorCapabilities until it is listed +// here. +var shardedCoordinatorSurfaces = map[string]coordinatorSurface{ + // kv.Coordinator. + "Dispatch": adapterSurface, + "IsLeader": adapterSurface, + "VerifyLeader": adapterSurface, + "LinearizableRead": adapterSurface, + "RaftLeader": adapterSurface, + "IsLeaderForKey": adapterSurface, + "VerifyLeaderForKey": adapterSurface, + "RaftLeaderForKey": adapterSurface, + "Clock": adapterSurface, + // Key-routed read fence (kv.LeaderRoutedStore), exposed by WithKeyVizLabel. + "LinearizableReadForKey": adapterSurface, + // Lease reads. + "LeaseRead": adapterSurface, + "LeaseReadForKey": adapterSurface, + "LeaseReadAllGroups": adapterSurface, + "LeaseReadAllGroupsTimestamp": adapterSurface, + // Group-keyed Redis read fence and per-group lease dedup. + "IsLeaderForGroup": adapterSurface, + "RaftLeaderForGroup": adapterSurface, + "LeaseReadForGroup": adapterSurface, + "EngineGroupIDForKey": adapterSurface, + // Scope of the Redis TTL-inline migration. + "LocalLeaderGroupIDs": adapterSurface, + // Timestamp allocation and applied-read vouchers. + "Next": adapterSurface, + "NextAfter": adapterSurface, + "TimestampAllocator": adapterSurface, + "RecoverHLCLease": adapterSurface, + "VouchAppliedReadTimestamp": adapterSurface, + "RevokeAppliedReadTimestamp": adapterSurface, + "SupportsAppliedReadTimestampVoucher": adapterSurface, + + // internalServerOptions installs the route-floor gate and the + // forwarded-write observer from the startup gate for Internal.Forward. + "EnsureMutationsWriteAllowed": gateSurface, + "ObserveForwardedRequests": gateSurface, + // newS3BlobCluster discovers peers through the startup gate. + "RaftMembers": gateSurface, + "RaftMembersForKey": gateSurface, + + // Lifecycle and HLC lease renewal, driven by main on in.coordinate. + "Close": rawOnly, + "RunHLCLeaseRenewal": rawOnly, + "SetHLCLeaseRenewalBlocker": rawOnly, + "ProposeHLCLease": rawOnly, + // Installed on the read tracker from the typed coordinator. + "ObserveTimestampFloor": rawOnly, + // Autosplit is configured with the typed *kv.ShardedCoordinator. + "GroupLeadership": rawOnly, + "LeadershipForKey": rawOnly, + // The legacy TSO bridge wraps the typed coordinator (configureLegacyCoordinatorTSO). + "IsTimestampLeader": rawOnly, +} + +var errSpyForwarded = errors.New("forwarded from the wrapped coordinator") + +const ( + spyIndex uint64 = 4242 + spyLeader = "spy-leader" +) + +// shardedCapabilitySpy stands in for the ShardedCoordinator under the startup +// wrappers. Every method records its name and returns a distinctive value or +// errSpyForwarded, so a forwarded call can be told apart from a wrapper's own +// fallback. +type shardedCapabilitySpy struct { + clock *kv.HLC + + mu sync.Mutex + calls map[string]int +} + +var _ kv.Coordinator = (*shardedCapabilitySpy)(nil) + +func newShardedCapabilitySpy() *shardedCapabilitySpy { + return &shardedCapabilitySpy{clock: kv.NewHLC(), calls: make(map[string]int)} +} + +func (s *shardedCapabilitySpy) record(name string) { + s.mu.Lock() + s.calls[name]++ + s.mu.Unlock() +} + +func (s *shardedCapabilitySpy) count(name string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls[name] +} + +func (s *shardedCapabilitySpy) Dispatch(context.Context, *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + s.record("Dispatch") + return &kv.CoordinateResponse{CommitIndex: spyIndex}, errSpyForwarded +} + +func (s *shardedCapabilitySpy) IsLeader() bool { + s.record("IsLeader") + return true +} + +func (s *shardedCapabilitySpy) VerifyLeader(context.Context) error { + s.record("VerifyLeader") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) LinearizableRead(context.Context) (uint64, error) { + s.record("LinearizableRead") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) RaftLeader() string { + s.record("RaftLeader") + return spyLeader +} + +func (s *shardedCapabilitySpy) IsLeaderForKey([]byte) bool { + s.record("IsLeaderForKey") + return true +} + +func (s *shardedCapabilitySpy) VerifyLeaderForKey(context.Context, []byte) error { + s.record("VerifyLeaderForKey") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) RaftLeaderForKey([]byte) string { + s.record("RaftLeaderForKey") + return spyLeader +} + +func (s *shardedCapabilitySpy) Clock() *kv.HLC { + s.record("Clock") + return s.clock +} + +func (s *shardedCapabilitySpy) LinearizableReadForKey(context.Context, []byte) (uint64, error) { + s.record("LinearizableReadForKey") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) LeaseRead(context.Context) (uint64, error) { + s.record("LeaseRead") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) LeaseReadForKey(context.Context, []byte) (uint64, error) { + s.record("LeaseReadForKey") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) LeaseReadAllGroups(context.Context) error { + s.record("LeaseReadAllGroups") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) LeaseReadAllGroupsTimestamp(context.Context) (uint64, error) { + s.record("LeaseReadAllGroupsTimestamp") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) IsLeaderForGroup(uint64) bool { + s.record("IsLeaderForGroup") + return true +} + +func (s *shardedCapabilitySpy) RaftLeaderForGroup(uint64) string { + s.record("RaftLeaderForGroup") + return spyLeader +} + +func (s *shardedCapabilitySpy) LeaseReadForGroup(context.Context, uint64) (uint64, error) { + s.record("LeaseReadForGroup") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) EngineGroupIDForKey([]byte) uint64 { + s.record("EngineGroupIDForKey") + return spyIndex +} + +func (s *shardedCapabilitySpy) LocalLeaderGroupIDs() ([]uint64, bool) { + s.record("LocalLeaderGroupIDs") + return []uint64{spyIndex}, true +} + +func (s *shardedCapabilitySpy) Next(context.Context) (uint64, error) { + s.record("Next") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) NextAfter(context.Context, uint64) (uint64, error) { + s.record("NextAfter") + return spyIndex, errSpyForwarded +} + +func (s *shardedCapabilitySpy) TimestampAllocator() kv.TimestampAllocator { + s.record("TimestampAllocator") + return s +} + +func (s *shardedCapabilitySpy) RecoverHLCLease(context.Context) error { + s.record("RecoverHLCLease") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) VouchAppliedReadTimestamp(uint64, kv.AppliedReadTimestampVoucherRef) error { + s.record("VouchAppliedReadTimestamp") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) RevokeAppliedReadTimestamp(uint64, kv.AppliedReadTimestampVoucherRef) { + s.record("RevokeAppliedReadTimestamp") +} + +func (s *shardedCapabilitySpy) SupportsAppliedReadTimestampVoucher() bool { + s.record("SupportsAppliedReadTimestampVoucher") + return true +} + +func (s *shardedCapabilitySpy) EnsureMutationsWriteAllowed([]*pb.Mutation, uint64) error { + s.record("EnsureMutationsWriteAllowed") + return errSpyForwarded +} + +func (s *shardedCapabilitySpy) ObserveForwardedRequests([]*pb.Request) { + s.record("ObserveForwardedRequests") +} + +func (s *shardedCapabilitySpy) RaftMembers(context.Context) ([]kv.RaftMember, error) { + s.record("RaftMembers") + return []kv.RaftMember{{NodeID: spyLeader}}, errSpyForwarded +} + +func (s *shardedCapabilitySpy) RaftMembersForKey(context.Context, []byte) ([]kv.RaftMember, error) { + s.record("RaftMembersForKey") + return []kv.RaftMember{{NodeID: spyLeader}}, errSpyForwarded +} + +type startupWrapperSurface struct { + name string + wrap func(kv.Coordinator) kv.Coordinator +} + +func startupGateOver(inner kv.Coordinator) kv.Coordinator { + gate := &startupPublicKVGate{} + gate.markReady() + return startupGatedCoordinator{inner: inner, gate: gate} +} + +var ( + startupGateSurface = startupWrapperSurface{ + name: "startupGatedCoordinator", + wrap: startupGateOver, + } + adapterCoordinatorSurface = startupWrapperSurface{ + name: "WithKeyVizLabel(startupGatedCoordinator)", + wrap: func(inner kv.Coordinator) kv.Coordinator { + return kv.WithKeyVizLabel(startupGateOver(inner), keyviz.LabelRedis) + }, + } +) + +// capabilityChecks replaces the default "call it, expect the same-named method +// on the wrapped coordinator" check for capabilities a wrapper serves +// differently on purpose. +var capabilityChecks = map[string]func(*testing.T, kv.Coordinator, *shardedCapabilitySpy){ + // The startup gate hands out itself as the allocator so it can refuse + // timestamps until startup rotation completes. What must survive is that + // the allocator consumers resolve through the wrappers still allocates + // from the wrapped coordinator. + "TimestampAllocator": func(t *testing.T, surface kv.Coordinator, spy *shardedCapabilitySpy) { + t.Helper() + alloc, ok := kv.TimestampAllocatorThrough(surface) + require.True(t, ok) + ts, err := alloc.Next(context.Background()) + require.ErrorIs(t, err, errSpyForwarded) + require.Equal(t, spyIndex, ts) + require.Positive(t, spy.count("Next"), "the resolved allocator must allocate from the wrapped coordinator") + }, +} + +// main.go hands the adapters startupGatedCoordinator{ShardedCoordinator}, and +// each adapter wraps it in WithKeyVizLabel. Capabilities are optional +// interfaces found by type assertion, so a wrapper that does not forward one +// fails silently: the probe misses and the consumer quietly takes a fallback, +// or the outer wrapper answers on the inner's behalf. That has already cost the +// Redis read fence its group routing, Internal.Forward its route-floor gate, +// and the Redis TTL-inline migration all of its work, each time with green unit +// tests that used an unwrapped coordinator. This test requires every +// ShardedCoordinator capability to be classified by the surface its consumers +// probe, and every forwarded one to reach the wrapped coordinator and return +// its result through that surface. The wrappers' own timestamp-allocator +// surfaces (PhaseD*, ValidateDurableTimestamp, ConfiguredTimestampAllocator) +// are not ShardedCoordinator methods; the TSO routing tests cover them. +func TestStartupWrappersForwardShardedCoordinatorCapabilities(t *testing.T) { + t.Parallel() + + shardedType := reflect.TypeOf((*kv.ShardedCoordinator)(nil)) + for name := range shardedCoordinatorSurfaces { + _, ok := shardedType.MethodByName(name) + require.Truef(t, ok, "ShardedCoordinator.%s no longer exists; drop it from shardedCoordinatorSurfaces", name) + } + sharded := reflect.ValueOf((*kv.ShardedCoordinator)(nil)) + spyMethods := reflect.ValueOf(newShardedCapabilitySpy()) + for i := range shardedType.NumMethod() { + name := shardedType.Method(i).Name + sig := sharded.Method(i).Type() + if isShardedCoordinatorBuilder(name, sig, shardedType) { + continue + } + surface, ok := shardedCoordinatorSurfaces[name] + require.Truef(t, ok, "ShardedCoordinator.%s is not classified in shardedCoordinatorSurfaces: "+ + "forward it on the wrappers whose consumers probe it, or record why main keeps it on the raw coordinator", name) + if surface == rawOnly { + continue + } + spyMethod := spyMethods.MethodByName(name) + require.Truef(t, spyMethod.IsValid() && spyMethod.Type() == sig, + "shardedCapabilitySpy.%s must mirror ShardedCoordinator.%s %s", name, name, sig) + surfaces := []startupWrapperSurface{startupGateSurface} + if surface == adapterSurface { + surfaces = append(surfaces, adapterCoordinatorSurface) + } + t.Run(name, func(t *testing.T) { + t.Parallel() + for _, s := range surfaces { + t.Run(s.name, func(t *testing.T) { + t.Parallel() + requireForwardsCapability(t, s, name, sig) + }) + } + }) + } +} + +func isShardedCoordinatorBuilder(name string, sig reflect.Type, shardedType reflect.Type) bool { + return strings.HasPrefix(name, "With") && sig.NumOut() == 1 && sig.Out(0) == shardedType +} + +func requireForwardsCapability(t *testing.T, s startupWrapperSurface, name string, sig reflect.Type) { + t.Helper() + spy := newShardedCapabilitySpy() + surface := s.wrap(spy) + method := reflect.ValueOf(surface).MethodByName(name) + require.Truef(t, method.IsValid(), "%s does not expose %s", s.name, name) + require.Equalf(t, sig, method.Type(), "%s.%s must keep the ShardedCoordinator signature", s.name, name) + if check, ok := capabilityChecks[name]; ok { + check(t, surface, spy) + return + } + args := capabilityCallArgs(sig) + got := method.Call(args) + require.Positivef(t, spy.count(name), "%s.%s did not reach the wrapped coordinator", s.name, name) + want := reflect.ValueOf(spy).MethodByName(name).Call(args) + errType := reflect.TypeFor[error]() + for i := range want { + if sig.Out(i) == errType { + gotErr, _ := got[i].Interface().(error) + require.ErrorIsf(t, gotErr, errSpyForwarded, "%s.%s must return the wrapped coordinator's error", s.name, name) + continue + } + require.Equalf(t, want[i].Interface(), got[i].Interface(), "%s.%s must return the wrapped coordinator's result", s.name, name) + } +} + +func capabilityCallArgs(sig reflect.Type) []reflect.Value { + ctxType := reflect.TypeFor[context.Context]() + keyType := reflect.TypeFor[[]byte]() + args := make([]reflect.Value, sig.NumIn()) + for i := range args { + switch in := sig.In(i); in { + case ctxType: + args[i] = reflect.ValueOf(context.Background()) + case keyType: + args[i] = reflect.ValueOf([]byte("capability-key")) + default: + args[i] = reflect.Zero(in) + } + } + return args +} From 35f698ee9d10f488674c504129078665b07e38bd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 09:18:43 +0000 Subject: [PATCH 04/15] main: forward LinearizableReadForKey through the startup gate WithKeyVizLabel exposes LinearizableReadForKey to the adapters and serves it from the plain LinearizableRead when its inner coordinator lacks it. The startup gate in between did not forward it, so through the adapters' coordinator a key owned by a non-default group would get a read fence on the default group. Nothing in the server calls it that way today, so the gap was latent; the capability guard from the previous commit flagged it. Forward it, falling back to LinearizableRead for an inner coordinator without a key-routed fence -- one that owns a single group -- as WithKeyVizLabel already does. The guard now passes, and a unit test pins the fallback branch. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- main.go | 16 ++++++++++++++++ main_coordinator_capabilities_test.go | 27 +++++++++++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/main.go b/main.go index f99530715..fa35fb88b 100644 --- a/main.go +++ b/main.go @@ -2748,6 +2748,22 @@ func (c startupGatedCoordinator) LinearizableRead(ctx context.Context) (uint64, return c.inner.LinearizableRead(ctx) //nolint:wrapcheck // Pass through coordinator errors unchanged. } +// LinearizableReadForKey forwards the key-routed read fence. WithKeyVizLabel +// exposes it to the adapters and serves it from the plain LinearizableRead when +// its inner coordinator lacks it, so without this forward a key owned by a +// non-default group would be fenced against the default group. An inner +// coordinator with no key-routed fence has only the one group to fence. +func (c startupGatedCoordinator) LinearizableReadForKey(ctx context.Context, key []byte) (uint64, error) { + if lr, ok := c.inner.(interface { + LinearizableReadForKey(context.Context, []byte) (uint64, error) + }); ok { + idx, err := lr.LinearizableReadForKey(ctx, key) + return idx, errors.WithStack(err) + } + idx, err := c.inner.LinearizableRead(ctx) + return idx, errors.WithStack(err) +} + func (c startupGatedCoordinator) RaftLeader() string { return c.inner.RaftLeader() } diff --git a/main_coordinator_capabilities_test.go b/main_coordinator_capabilities_test.go index 77b8f78f1..66b1a8096 100644 --- a/main_coordinator_capabilities_test.go +++ b/main_coordinator_capabilities_test.go @@ -422,3 +422,30 @@ func capabilityCallArgs(sig reflect.Type) []reflect.Value { } return args } + +// plainLinearizableReadCoordinator has a LinearizableRead but no key-routed +// fence. The embedded interface is nil on purpose: only LinearizableRead is +// exercised here. +type plainLinearizableReadCoordinator struct { + kv.Coordinator + + reads int +} + +func (c *plainLinearizableReadCoordinator) LinearizableRead(context.Context) (uint64, error) { + c.reads++ + return spyIndex, nil +} + +// An inner coordinator without a key-routed fence owns a single group, so the +// startup gate serves LinearizableReadForKey from its plain LinearizableRead, +// the same fallback WithKeyVizLabel uses. +func TestStartupGatedCoordinatorLinearizableReadForKeyFallsBackToLinearizableRead(t *testing.T) { + t.Parallel() + + inner := &plainLinearizableReadCoordinator{} + idx, err := startupGatedCoordinator{inner: inner}.LinearizableReadForKey(context.Background(), []byte("k")) + require.NoError(t, err) + require.Equal(t, spyIndex, idx) + require.Equal(t, 1, inner.reads) +} From b0d42e8847613b5893d873045be6d3d8b5124cd6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 18:05:07 +0000 Subject: [PATCH 05/15] adapter: reproduce TTL migration refusing every shard but the first Review on #1278 found that the per-group TTL-inline scans only work on a single-route store. The migrator scans each locally led group with ShardStore.ScanGroupAt over a bare anchor prefix such as "!redis|str|". ScanGroupAt resolves the requested range through the route catalog, and a bare prefix resolves to the first route only, so every other group is refused with ErrExplicitGroupRouteOwnerMismatch. A node leading several shards migrates at most the first route's anchors and fails every tick on the rest. The existing tests hand the compactor a store whose ScanGroupAt ignores routing, so they could not see it. Run the migrator over a real ShardStore split at "m" across two groups and require legacy string and hash anchors on both sides to migrate. It fails here on group 2's scans for all eight anchor prefixes. The fix follows in the next commit. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 74 +++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 62a6a27db..1e6ef5ec7 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/bootjp/elastickv/distribution" "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" "github.com/bootjp/elastickv/store" @@ -614,6 +615,79 @@ func TestDeltaCompactor_TTLInlineUsesLocalLeaderGroupScan(t *testing.T) { require.Equal(t, redisExpireAtMillis(expireAt), meta.ExpireAt) } +// newSplitRedisShardStore returns a ShardStore whose catalog sends user keys +// below "m" to group 1 and the rest to group 2, the way a multi-shard +// deployment splits the Redis keyspace. +func newSplitRedisShardStore() *kv.ShardStore { + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + return kv.NewShardStore(engine, map[uint64]*kv.ShardGroup{ + 1: {Store: store.NewMVCCStore()}, + 2: {Store: store.NewMVCCStore()}, + }) +} + +// A node scans each data group it leads for legacy TTL anchors. The bare anchor +// prefixes ("!redis|str|", "!hs|meta|", ...) resolve to the first route only, so +// a route-checked group scan refused every other group: anchors outside the +// first route were never migrated and every tick failed. Run the migrator over a +// real split ShardStore and require the anchors of both groups to migrate. +func TestDeltaCompactor_TTLInlineMigratesEveryLocalLeaderShard(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := newSplitRedisShardStore() + coord := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: []uint64{1, 2}, + } + c := NewDeltaCompactor(st, coord, WithDeltaCompactorMaxDeltaCount(2)) + + expireAt := time.Now().Add(time.Hour) + // One key on each side of the split per anchor type. + stringKeys := [][]byte{[]byte("apple"), []byte("zebra")} + hashKeys := [][]byte{[]byte("aardvark"), []byte("yak")} + legacyMeta := make([]byte, redisSimpleMetaLegacySizeBytes) + binary.BigEndian.PutUint64(legacyMeta, 1) + ts := uint64(0) + seed := func(key, value []byte) { + ts++ + require.NoError(t, st.PutAt(ctx, key, value, ts, 0)) + } + for _, key := range stringKeys { + seed(redisStrKey(key), []byte("legacy-value")) + seed(redisTTLKey(key), encodeRedisTTL(expireAt)) + } + for _, key := range hashKeys { + seed(store.HashMetaKey(key), legacyMeta) + seed(redisTTLKey(key), encodeRedisTTL(expireAt)) + } + + require.NoError(t, c.SyncOnce(ctx)) + + readTS := st.LastCommitTS() + for _, key := range stringKeys { + raw, err := st.GetAt(ctx, redisStrKey(key), readTS) + require.NoError(t, err) + require.Truef(t, isNewRedisStrFormat(raw), "string %q was not migrated", key) + value, ttl, err := decodeRedisStr(raw) + require.NoError(t, err) + require.Equal(t, []byte("legacy-value"), value) + require.NotNil(t, ttl) + require.Equal(t, redisExpireAtMillis(expireAt), redisExpireAtMillis(*ttl)) + } + for _, key := range hashKeys { + raw, err := st.GetAt(ctx, store.HashMetaKey(key), readTS) + require.NoError(t, err) + meta, err := store.UnmarshalHashMeta(raw) + require.NoError(t, err) + require.Equalf(t, redisExpireAtMillis(expireAt), meta.ExpireAt, "hash %q was not migrated", key) + } +} + // capabilityHidingCompactionCoordinator forwards only the kv.Coordinator // methods, the way a decorator that forgets an optional forward does. type capabilityHidingCompactionCoordinator struct { From 301aaa060c77bc71bd301467cc2a022ec1133bba Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 18:12:05 +0000 Subject: [PATCH 06/15] kv, adapter: scan each local leader group's own share of the TTL anchors The TTL-inline migrator scanned each locally led group with ShardStore.ScanGroupAt over a bare anchor prefix. ScanGroupAt resolves the requested range through the route catalog first, and a bare prefix such as "!redis|str|" resolves to the first route only, so every other group was refused with ErrExplicitGroupRouteOwnerMismatch. On a store with more than one route, a node migrated at most the first route's anchors and failed every tick on the rest. Add ShardStore.ScanGroupOwnedAt. It scans the raw range on the group's own store, fenced like the other explicit-group reads, and keeps only keys whose route the group currently owns, judged by the same routeAndGroupForKey that routes writes. It keeps paging past skipped keys, so fewer than limit results still means the range is exhausted. A group's store can still hold copies of ranges it no longer owns, and ScanGroupAt returned those to the first route's group; acting on one could rewrite a key from a stale value, so skipping them matters. Each call reads about one page, the same per-tick cost a36dac7f intended, instead of one walk per owned route. The migrator uses it for its per-group plans. The split-store regression test from the previous commit now passes. kv tests cover both anchor encodings (user key directly after the prefix, and length prefixed), a stale copy, and paging past a page of skipped keys. Also correct the fallback wording: when the groups are unknown, the migrator runs one ScanAt over the whole prefix, which on a split catalog reaches only the first route for the string, HLL, TTL and list anchors. That path is not new and is not reached in the server. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 6 +- adapter/redis_ttl_inline_migrator.go | 15 +++-- kv/keyviz_label.go | 2 +- kv/shard_store.go | 51 +++++++++++++++ kv/shard_store_test.go | 85 +++++++++++++++++++++++++ main.go | 2 +- main_redis_ttl_migration_wiring_test.go | 8 +-- 7 files changed, 154 insertions(+), 15 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 1e6ef5ec7..8d1113f3f 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -576,7 +576,7 @@ func (s *recordingTTLInlineGroupScanStore) ScanAt(ctx context.Context, start []b return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) } -func (s *recordingTTLInlineGroupScanStore) ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { +func (s *recordingTTLInlineGroupScanStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { s.groupScans = append(s.groupScans, groupID) s.groupStarts = append(s.groupStarts, string(start)) return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) @@ -607,7 +607,7 @@ func TestDeltaCompactor_TTLInlineUsesLocalLeaderGroupScan(t *testing.T) { require.Contains(t, st.groupScans, uint64(2)) require.Contains(t, st.groupStarts, store.HashMetaPrefix) require.NotContains(t, st.fullStarts, store.HashMetaPrefix, - "sharded ttl migration should scan candidate metadata with ScanGroupAt") + "sharded ttl migration should scan candidate metadata with ScanGroupOwnedAt") raw, err := st.GetAt(ctx, store.HashMetaKey(key), st.LastCommitTS()) require.NoError(t, err) meta, err := store.UnmarshalHashMeta(raw) @@ -697,7 +697,7 @@ type capabilityHidingCompactionCoordinator struct { // The coordinator the compactor holds is wrapped in WithKeyVizLabel, which // always exposes LocalLeaderGroupIDs. When a decorator underneath hides the // capability the answer must stay unknown and the migrator must fall back to a -// full-keyspace scan; answering "no groups" there planned nothing and silently +// whole-prefix scan; answering "no groups" there planned nothing and silently // stopped the migration in production. A known empty list is different: this // node leads no group, so there is nothing for it to migrate. func TestDeltaCompactor_TTLInlineScanPlanWhenLocalLeaderGroupsUnknownOrEmpty(t *testing.T) { diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index d8b711416..e3e997a6d 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -23,8 +23,11 @@ type ttlInlineMigrationHandler struct { buildElems func(context.Context, *store.KVPair, uint64) ([]*kv.Elem[kv.OP], error) } +// ttlInlineGroupScanner is *kv.ShardStore in the server. It uses +// ScanGroupOwnedAt because ScanGroupAt resolves a bare anchor prefix to the +// first route only and refuses every other group. type ttlInlineGroupScanner interface { - ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) + ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) } type ttlInlineMigrationScanPlan struct { @@ -241,8 +244,8 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( // ttlInlineMigrationScanPlans scans only the groups this node leads when the // store can scan one group and the coordinator knows which groups those are. A // known empty list means this node leads no group and plans nothing. When the -// groups are unknown the migrator falls back to one full-keyspace scan, -// filtered per key by IsLeaderForKey, rather than migrating nothing. +// groups are unknown the migrator falls back to one ScanAt over the whole +// prefix, filtered per key by IsLeaderForKey, rather than migrating nothing. func (c *DeltaCompactor) ttlInlineMigrationScanPlans(h ttlInlineMigrationHandler) []ttlInlineMigrationScanPlan { scanner, hasGroupScanner := c.st.(ttlInlineGroupScanner) if hasGroupScanner { @@ -265,10 +268,10 @@ func (c *DeltaCompactor) ttlInlineMigrationScanPlans(h ttlInlineMigrationHandler // warnTTLInlineLocalLeaderGroupsUnknown logs once when the store can scan one // group but the coordinator cannot name the groups this node leads. In the // server that means a coordinator decorator dropped LocalLeaderGroupIDs, and -// every node is back to full-keyspace migration scans. +// every node is back to whole-prefix migration scans. func (c *DeltaCompactor) warnTTLInlineLocalLeaderGroupsUnknown() { c.ttlInlineGroupsUnknownOnce.Do(func() { - c.logger.Warn("ttl inline migrator: coordinator does not report local leader groups; falling back to full-keyspace scans") + c.logger.Warn("ttl inline migrator: coordinator does not report local leader groups; falling back to whole-prefix scans") }) } @@ -295,7 +298,7 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( err error ) if plan.scanner != nil { - kvs, err = plan.scanner.ScanGroupAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) + kvs, err = plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) } else { kvs, err = c.st.ScanAt(ctx, start, end, ttlInlineMigrationTickScanLimit, readTS) } diff --git a/kv/keyviz_label.go b/kv/keyviz_label.go index d50360a37..71d3a8a96 100644 --- a/kv/keyviz_label.go +++ b/kv/keyviz_label.go @@ -222,7 +222,7 @@ func (c keyVizLabeledCoordinator) EngineGroupIDForKey(key []byte) uint64 { // LocalLeaderGroupIDs forwards which data groups the inner coordinator leads. // An inner coordinator without the capability stays unknown instead of // answering "no groups": the TTL-inline migrator then falls back to a -// full-keyspace scan rather than planning nothing. +// whole-prefix scan rather than planning nothing. func (c keyVizLabeledCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { return LocalLeaderGroupIDsThrough(c.inner) } diff --git a/kv/shard_store.go b/kv/shard_store.go index c52b4cfbf..525ed83b9 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -904,6 +904,57 @@ func (s *ShardStore) ScanGroupAt(ctx context.Context, groupID uint64, start []by return s.scanExplicitGroupAtWithReadFence(ctx, groupID, start, end, limit, ts, false, 0, nil, nil) } +// ScanGroupOwnedAt scans [start, end) on groupID's own store and returns, in key +// order, up to limit keys whose route groupID currently owns. ScanGroupAt first +// resolves the requested range through the route catalog, and a bare internal +// prefix such as "!redis|str|" resolves to the first route only, so it refuses +// every other group. Maintenance that walks a whole prefix one group at a time +// uses this instead. The group's store can still hold keys of ranges it no +// longer owns; those are skipped rather than returned. Fewer than limit results +// means the range is exhausted. It is meant for the group's own leader: on any +// other node the read is forwarded, and the leader serves it as ScanGroupAt, +// which may refuse such a range. +func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { + if limit <= 0 { + return []*store.KVPair{}, nil + } + route := distribution.Route{GroupID: groupID} + out := make([]*store.KVPair, 0, limit) + cursor := start + for { + if err := ctx.Err(); err != nil { + return nil, errors.WithStack(err) + } + page, err := s.scanRouteAtDirectionWithReadFence(ctx, route, cursor, end, limit, ts, false, true, 0, nil, nil) + if err != nil { + return nil, err + } + out = s.appendGroupOwnedKVs(out, page, groupID, limit) + if len(out) >= limit || len(page) < limit { + return out, nil + } + cursor = nextScanCursor(lastKVKey(page)) + if end != nil && bytes.Compare(cursor, end) >= 0 { + return out, nil + } + } +} + +func (s *ShardStore) appendGroupOwnedKVs(out []*store.KVPair, kvs []*store.KVPair, groupID uint64, limit int) []*store.KVPair { + for _, kvp := range kvs { + if len(out) >= limit { + break + } + if kvp == nil { + continue + } + if route, _, ok := s.routeAndGroupForKey(kvp.Key); ok && route.GroupID == groupID { + out = append(out, kvp) + } + } + return out +} + func (s *ShardStore) scanExplicitGroupAtWithReadFence(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64, reverse bool, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error) { if limit <= 0 { return []*store.KVPair{}, nil diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index a119bca0a..9707099d4 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -1512,6 +1512,91 @@ func TestShardStoreScanGroupAt_DeduplicatesRouteMappedSameGroupSplits(t *testing require.Equal(t, []*store.KVPair{{Key: key, Value: []byte("msg-2")}}, kvs) } +func newSplitRedisShardStoreForGroupScan() (*ShardStore, map[uint64]*ShardGroup) { + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + groups := map[uint64]*ShardGroup{ + 1: {Store: store.NewMVCCStore()}, + 2: {Store: store.NewMVCCStore()}, + } + return NewShardStore(engine, groups), groups +} + +// The Redis TTL-inline migrator walks each anchor prefix one locally led group at +// a time. A bare prefix resolves to the first route only, so ScanGroupAt refuses +// every other group; ScanGroupOwnedAt must return each group's share, both for +// anchors keyed directly by the user key and for length-prefixed ones. +func TestShardStoreScanGroupOwnedAt_ReturnsEachGroupsShareOfAPrefix(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + prefix []byte + key func(userKey string) []byte + }{ + { + name: "string anchors", + prefix: []byte("!redis|str|"), + key: func(userKey string) []byte { return []byte("!redis|str|" + userKey) }, + }, + { + name: "hash meta anchors", + prefix: []byte(store.HashMetaPrefix), + key: func(userKey string) []byte { return store.HashMetaKey([]byte(userKey)) }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st, groups := newSplitRedisShardStoreForGroupScan() + owned := map[uint64][]byte{1: tc.key("apple"), 2: tc.key("zebra")} + for gid, key := range owned { + require.NoError(t, groups[gid].Store.PutAt(ctx, key, []byte("v"), 5, 0)) + } + end := prefixScanEnd(tc.prefix) + + _, err := st.ScanGroupAt(ctx, 2, tc.prefix, end, 10, 10) + require.ErrorIs(t, err, ErrExplicitGroupRouteOwnerMismatch) + for gid, key := range owned { + kvs, err := st.ScanGroupOwnedAt(ctx, gid, tc.prefix, end, 10, 10) + require.NoError(t, err) + require.Equalf(t, []*store.KVPair{{Key: key, Value: []byte("v")}}, kvs, "group %d", gid) + } + }) + } +} + +// A group's store can still hold keys of a range that has moved to another +// group. ScanGroupOwnedAt must skip them, and must keep paging past them instead +// of reporting a short page as the end of the range: hash meta keys are length +// prefixed, so a stale two-byte key sorts before an owned three-byte one. +func TestShardStoreScanGroupOwnedAt_SkipsKeysOfRangesTheGroupNoLongerOwns(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st, groups := newSplitRedisShardStoreForGroupScan() + stale := store.HashMetaKey([]byte("zz")) + owned := store.HashMetaKey([]byte("abc")) + require.Negative(t, bytes.Compare(stale, owned), "the stale key must sort first") + require.NoError(t, groups[1].Store.PutAt(ctx, stale, []byte("stale"), 5, 0)) + require.NoError(t, groups[1].Store.PutAt(ctx, owned, []byte("owned"), 5, 0)) + require.NoError(t, groups[2].Store.PutAt(ctx, stale, []byte("current"), 6, 0)) + prefix := []byte(store.HashMetaPrefix) + end := prefixScanEnd(prefix) + + kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, end, 1, 10) + require.NoError(t, err) + require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, kvs, + "a page of skipped keys is not the end of the range") + + kvs, err = st.ScanGroupOwnedAt(ctx, 2, prefix, end, 10, 10) + require.NoError(t, err) + require.Equal(t, []*store.KVPair{{Key: stale, Value: []byte("current")}}, kvs) +} + func TestShardStoreGetGroupAt_UsesExplicitGroup(t *testing.T) { t.Parallel() diff --git a/main.go b/main.go index fa35fb88b..3e78456b7 100644 --- a/main.go +++ b/main.go @@ -2971,7 +2971,7 @@ func (c startupGatedCoordinator) EngineGroupIDForKey(key []byte) uint64 { // LocalLeaderGroupIDs forwards the data groups this node leads. The Redis delta // compactor reaches it through WithKeyVizLabel to scope the TTL-inline migration // to locally led groups; without this forward every node falls back to -// full-keyspace migration scans. +// whole-prefix migration scans. func (c startupGatedCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { return kv.LocalLeaderGroupIDsThrough(c.inner) } diff --git a/main_redis_ttl_migration_wiring_test.go b/main_redis_ttl_migration_wiring_test.go index 8794638fd..208e0c329 100644 --- a/main_redis_ttl_migration_wiring_test.go +++ b/main_redis_ttl_migration_wiring_test.go @@ -15,9 +15,9 @@ import ( ) // groupScanRecordingStore stands in for the *kv.ShardStore the Redis delta -// compactor holds in production. Exposing ScanGroupAt is what lets the +// compactor holds in production. Exposing ScanGroupOwnedAt is what lets the // TTL-inline migrator plan one scan per locally led group instead of a -// full-keyspace scan. +// whole-prefix scan. type groupScanRecordingStore struct { store.MVCCStore @@ -33,7 +33,7 @@ func (s *groupScanRecordingStore) ScanAt(ctx context.Context, start []byte, end return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) } -func (s *groupScanRecordingStore) ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { +func (s *groupScanRecordingStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { s.mu.Lock() if s.groupScans == nil { s.groupScans = make(map[uint64][]string) @@ -105,5 +105,5 @@ func TestRedisTTLInlineMigrationScansLocalLeaderGroupsThroughStartupGate(t *test seen[start] = struct{}{} } require.NotContains(t, fullScans, store.HashMetaPrefix, - "the migrator must not fall back to a full-keyspace scan when the groups are known") + "the migrator must not fall back to a whole-prefix scan when the groups are known") } From d795cd3b7f9d52b10ec66f764a187674469d37b5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 18:31:08 +0000 Subject: [PATCH 07/15] kv, adapter: reproduce the group-owned scan's staged, bound and legacy gaps Review of 301aaa06 found three gaps in ShardStore.ScanGroupOwnedAt: - It scans with a bare group route, which carries no staged-visibility flag, so a route being imported is read from its live rows only. An older live copy could feed a rewrite while a newer staged copy waits for promotion. - It reads physical pages until it has limit owned rows. A group holding a large backlog of rows of ranges it no longer owns turns one call into a walk over the whole backlog, which overruns the tick budget and never advances the cursor. - It keeps a key only when its logical owner is the scanned group. Hash, set and zset rows written before wide-column routing still sit at the route of their raw key, and reads fall back to them there, so dropping them leaves those anchors unmigrated. Pin each one: the staged route's key must be deferred, one call must read at most one page, a legacy-route row must be served with the value a point read sees, a tick must stop after one page of stale rows and resume on the next, and the migrator must rewrite a legacy-route hash meta into its user key's group. All five fail here; the fix follows in the next commit. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 89 ++++++++++++++++++ kv/shard_store_test.go | 125 ++++++++++++++++++++++++++ 2 files changed, 214 insertions(+) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 8d1113f3f..08256d1ac 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -688,6 +688,95 @@ func TestDeltaCompactor_TTLInlineMigratesEveryLocalLeaderShard(t *testing.T) { } } +// A group can still hold a full page of copies of ranges it no longer owns. A +// tick must read one page per plan and move the cursor past it, so the group's +// own anchors behind such a page are reached on the next tick; reading on until +// a page of owned rows turned a large stale backlog into a call that never +// finished within the tick budget and never advanced. +func TestDeltaCompactor_TTLInlineAdvancesPastAFullPageOfStaleRows(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 2) + engine.UpdateRoute([]byte("m"), nil, 1) + group1 := store.NewMVCCStore() + st := kv.NewShardStore(engine, map[uint64]*kv.ShardGroup{ + 1: {Store: group1}, + 2: {Store: store.NewMVCCStore()}, + }) + coord := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: []uint64{1}, + } + c := NewDeltaCompactor(st, coord, WithDeltaCompactorMaxDeltaCount(2)) + + // A full page of stale copies of group 2's keys sorts before group 1's key. + for i := range ttlInlineMigrationTickScanLimit { + key := []byte("a" + strconv.Itoa(1000+i)) + require.NoError(t, group1.PutAt(ctx, redisStrKey(key), []byte("stale"), 1, 0)) + } + owned := []byte("zebra") + expireAt := time.Now().Add(time.Hour) + require.NoError(t, st.PutAt(ctx, redisStrKey(owned), []byte("legacy-value"), 2, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey(owned), encodeRedisTTL(expireAt), 3, 0)) + migrated := func() bool { + raw, err := st.GetAt(ctx, redisStrKey(owned), st.LastCommitTS()) + require.NoError(t, err) + return isNewRedisStrFormat(raw) + } + + require.NoError(t, c.SyncOnce(ctx)) + require.False(t, migrated(), "a tick reads one page per plan") + require.NoError(t, c.SyncOnce(ctx)) + require.True(t, migrated(), "the next tick resumes after the stale page") +} + +// Hash, set and zset metadata written before wide-column routing can still sit +// at the route of its raw key while its user key belongs to another group, and +// reads fall back to that row. The group holding it must hand it to the +// migrator, which rewrites it into the user key's group when this node leads +// that group. +func TestDeltaCompactor_TTLInlineMigratesWideColumnRowsAtTheirLegacyRoute(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + group1 := store.NewMVCCStore() + st := kv.NewShardStore(engine, map[uint64]*kv.ShardGroup{ + 1: {Store: group1}, + 2: {Store: store.NewMVCCStore()}, + }) + coord := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: []uint64{1, 2}, + } + c := NewDeltaCompactor(st, coord, WithDeltaCompactorMaxDeltaCount(2)) + + key := []byte("zebra") + expireAt := time.Now().Add(time.Hour) + legacyMeta := make([]byte, redisSimpleMetaLegacySizeBytes) + binary.BigEndian.PutUint64(legacyMeta, 1) + // The raw meta key starts with "!", so its legacy route is group 1. + require.NoError(t, group1.PutAt(ctx, store.HashMetaKey(key), legacyMeta, 1, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey(key), encodeRedisTTL(expireAt), 2, 0)) + + require.NoError(t, c.SyncOnce(ctx)) + + raw, err := st.GetAt(ctx, store.HashMetaKey(key), st.LastCommitTS()) + require.NoError(t, err) + meta, err := store.UnmarshalHashMeta(raw) + require.NoError(t, err) + require.Equal(t, int64(1), meta.Len) + require.Equal(t, redisExpireAtMillis(expireAt), meta.ExpireAt) +} + // capabilityHidingCompactionCoordinator forwards only the kv.Coordinator // methods, the way a decorator that forgets an optional forward does. type capabilityHidingCompactionCoordinator struct { diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 9707099d4..0d57fc858 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "strconv" "sync" "testing" "time" @@ -1597,6 +1598,130 @@ func TestShardStoreScanGroupOwnedAt_SkipsKeysOfRangesTheGroupNoLongerOwns(t *tes require.Equal(t, []*store.KVPair{{Key: stale, Value: []byte("current")}}, kvs) } +// rowCountingMVCCStore counts the rows ScanAt hands back, which is the physical +// work a ShardStore scan does on a group without a Raft engine. +type rowCountingMVCCStore struct { + store.MVCCStore + + mu sync.Mutex + rows int +} + +func (s *rowCountingMVCCStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { + kvs, err := s.MVCCStore.ScanAt(ctx, start, end, limit, ts) + s.mu.Lock() + s.rows += len(kvs) + s.mu.Unlock() + return kvs, err +} + +func (s *rowCountingMVCCStore) rowsRead() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.rows +} + +// A route under staged visibility serves reads from its staged copy first, and +// a group-local scan with a bare group route reads live rows only. Acting on an +// older live copy while a newer staged copy is being imported would rewrite the +// key from stale data, so those keys must be left alone until promotion. +func TestShardStoreScanGroupOwnedAt_DefersKeysOfStagedRoutes(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{ + { + RouteID: 1, Start: []byte(""), End: []byte("m"), GroupID: 1, + State: distribution.RouteStateActive, StagedVisibilityActive: true, MigrationJobID: 9, + }, + {RouteID: 2, Start: []byte("m"), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + }, + })) + group := &ShardGroup{Store: store.NewMVCCStore()} + st := NewShardStore(engine, map[uint64]*ShardGroup{1: group}) + staged := []byte("!redis|str|apple") + live := []byte("!redis|str|zebra") + require.NoError(t, group.Store.PutAt(ctx, staged, []byte("v"), 5, 0)) + require.NoError(t, group.Store.PutAt(ctx, live, []byte("v"), 5, 0)) + prefix := []byte("!redis|str|") + + kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + require.NoError(t, err) + require.Equal(t, []*store.KVPair{{Key: live, Value: []byte("v")}}, kvs) +} + +// Maintenance scans run under a per-tick budget. A group can hold many rows of +// ranges it no longer owns, so one call must read a single physical page no +// matter how few of its rows are kept; reading on until limit kept rows turned +// a large stale backlog into a call that never finished within the budget. +func TestShardStoreScanGroupOwnedAt_ReadsOnePhysicalPagePerCall(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 2) + engine.UpdateRoute([]byte("m"), nil, 1) + counting := &rowCountingMVCCStore{MVCCStore: store.NewMVCCStore()} + st := NewShardStore(engine, map[uint64]*ShardGroup{ + 1: {Store: counting}, + 2: {Store: store.NewMVCCStore()}, + }) + // Stale copies of group 2's keys sort before group 1's only key. + for i := range 6 { + key := []byte("!redis|str|a" + strconv.Itoa(i)) + require.NoError(t, counting.PutAt(ctx, key, []byte("stale"), 5, 0)) + } + owned := []byte("!redis|str|zebra") + require.NoError(t, counting.PutAt(ctx, owned, []byte("owned"), 5, 0)) + prefix := []byte("!redis|str|") + const limit = 2 + + _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), limit, 10) + require.NoError(t, err) + require.LessOrEqual(t, counting.rowsRead(), limit) +} + +// Hash, set and zset rows written before wide-column routing still live at the +// route of their raw key, and reads fall back to that route when the logical +// owner has nothing. The group holding such a row serves it, so its scan must +// return it with the value a point read would see, not drop it because the +// logical owner is another group. +func TestShardStoreScanGroupOwnedAt_ServesWideColumnRowsAtTheirLegacyRoute(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + ownerHasNewer bool + want []byte + }{ + {name: "legacy row only", want: []byte("legacy")}, + {name: "logical owner has a newer row", ownerHasNewer: true, want: []byte("current")}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st, groups := newSplitRedisShardStoreForGroupScan() + // The raw key starts with "!", so its legacy route is group 1; + // the user key "zebra" belongs to group 2. + key := store.HashMetaKey([]byte("zebra")) + require.NoError(t, groups[1].Store.PutAt(ctx, key, []byte("legacy"), 5, 0)) + if tc.ownerHasNewer { + require.NoError(t, groups[2].Store.PutAt(ctx, key, []byte("current"), 6, 0)) + } + prefix := []byte(store.HashMetaPrefix) + + kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + require.NoError(t, err) + require.Equal(t, []*store.KVPair{{Key: key, Value: tc.want}}, kvs) + }) + } +} + func TestShardStoreGetGroupAt_UsesExplicitGroup(t *testing.T) { t.Parallel() From 50494e865a18e94e3b4014817b8ed57e9cdd11bd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 18:37:04 +0000 Subject: [PATCH 08/15] kv, adapter: bound the group-owned scan, defer staged routes, serve legacy rows Close the three gaps the previous commit reproduced in ShardStore.ScanGroupOwnedAt: - One physical page per call. It now reads a single page and returns the last key that page read when it was full, nil once the range is exhausted. The migrator moves its cursor to that key instead of to the last row it kept, so a page holding no row the group serves still advances the cursor and a tick stays within one page per plan. - Staged routes are deferred. A key whose route is under staged visibility is skipped: the page reads live rows only, and an older live copy must not feed a rewrite while a newer staged copy waits for promotion. Such keys are picked up on a later pass. - Legacy-route rows are served. Hash, set and zset rows still stored at the route of their raw key are kept when that route is the scanned group. The page has already re-read them through the point-read path, so they carry the value a reader sees, not a shadowed copy. The migrator still writes one only when this node leads the user key's group, as the HLC rule requires. The earlier stale-copy test used a hash key, which this commit now correctly serves at its legacy route; it uses string keys, which have no legacy route, instead. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 14 +++- adapter/redis_ttl_inline_migrator.go | 31 ++++---- kv/shard_store.go | 97 ++++++++++++++----------- kv/shard_store_test.go | 48 +++++++----- main_redis_ttl_migration_wiring_test.go | 8 +- 5 files changed, 121 insertions(+), 77 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 08256d1ac..7f8910cbf 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -576,10 +576,20 @@ func (s *recordingTTLInlineGroupScanStore) ScanAt(ctx context.Context, start []b return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) } -func (s *recordingTTLInlineGroupScanStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { +func (s *recordingTTLInlineGroupScanStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { s.groupScans = append(s.groupScans, groupID) s.groupStarts = append(s.groupStarts, string(start)) - return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) + return scanPageWithResume(ctx, s.MVCCStore, start, end, limit, ts) +} + +// scanPageWithResume reads one page the way ShardStore.ScanGroupOwnedAt does +// and reports its last key when the page was full. +func scanPageWithResume(ctx context.Context, st store.MVCCStore, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { + kvs, err := st.ScanAt(ctx, start, end, limit, ts) + if err != nil || len(kvs) < limit { + return kvs, nil, err + } + return kvs, kvs[len(kvs)-1].Key, nil } func TestDeltaCompactor_TTLInlineUsesLocalLeaderGroupScan(t *testing.T) { diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index e3e997a6d..34eb2e26c 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -25,9 +25,11 @@ type ttlInlineMigrationHandler struct { // ttlInlineGroupScanner is *kv.ShardStore in the server. It uses // ScanGroupOwnedAt because ScanGroupAt resolves a bare anchor prefix to the -// first route only and refuses every other group. +// first route only and refuses every other group. ScanGroupOwnedAt reads one +// page per call and returns where it stopped, so the cursor advances even when +// the page held no row the group serves. type ttlInlineGroupScanner interface { - ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) + ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) } type ttlInlineMigrationScanPlan struct { @@ -197,7 +199,7 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( plan ttlInlineMigrationScanPlan, readTS uint64, ) error { - kvs, truncated, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) + kvs, cursorKey, truncated, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) if err != nil { return err } @@ -237,7 +239,7 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( if err := flushBatch(); err != nil { return err } - c.advanceCursor(plan.cursorName, lastKVKey(kvs), truncated) + c.advanceCursor(plan.cursorName, cursorKey, truncated) return nil } @@ -279,12 +281,14 @@ func ttlInlineMigrationGroupCursorName(typeName string, groupID uint64) string { return typeName + ":group:" + strconv.FormatUint(groupID, 10) } +// scanTTLInlineMigrationRaw returns the scanned rows, the key the plan's cursor +// moves to, and whether the scan stopped before the end of the prefix. func (c *DeltaCompactor) scanTTLInlineMigrationRaw( ctx context.Context, h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, -) ([]*store.KVPair, bool, error) { +) ([]*store.KVPair, []byte, bool, error) { c.cursorMu.Lock() start := h.prefix if cur := c.cursors[plan.cursorName]; len(cur) > 0 { @@ -293,19 +297,18 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( c.cursorMu.Unlock() end := store.PrefixScanEnd(h.prefix) - var ( - kvs []*store.KVPair - err error - ) if plan.scanner != nil { - kvs, err = plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) - } else { - kvs, err = c.st.ScanAt(ctx, start, end, ttlInlineMigrationTickScanLimit, readTS) + kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) + if err != nil { + return nil, nil, false, errors.WithStack(err) + } + return kvs, resume, resume != nil, nil } + kvs, err := c.st.ScanAt(ctx, start, end, ttlInlineMigrationTickScanLimit, readTS) if err != nil { - return nil, false, errors.WithStack(err) + return nil, nil, false, errors.WithStack(err) } - return kvs, len(kvs) == ttlInlineMigrationTickScanLimit, nil + return kvs, lastKVKey(kvs), len(kvs) == ttlInlineMigrationTickScanLimit, nil } func (c *DeltaCompactor) migrateStringTTLInlineElems(ctx context.Context, pair *store.KVPair, readTS uint64) ([]*kv.Elem[kv.OP], error) { diff --git a/kv/shard_store.go b/kv/shard_store.go index 525ed83b9..2d78740ce 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -904,55 +904,68 @@ func (s *ShardStore) ScanGroupAt(ctx context.Context, groupID uint64, start []by return s.scanExplicitGroupAtWithReadFence(ctx, groupID, start, end, limit, ts, false, 0, nil, nil) } -// ScanGroupOwnedAt scans [start, end) on groupID's own store and returns, in key -// order, up to limit keys whose route groupID currently owns. ScanGroupAt first -// resolves the requested range through the route catalog, and a bare internal -// prefix such as "!redis|str|" resolves to the first route only, so it refuses -// every other group. Maintenance that walks a whole prefix one group at a time -// uses this instead. The group's store can still hold keys of ranges it no -// longer owns; those are skipped rather than returned. Fewer than limit results -// means the range is exhausted. It is meant for the group's own leader: on any -// other node the read is forwarded, and the leader serves it as ScanGroupAt, -// which may refuse such a range. -func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { +// ScanGroupOwnedAt reads one physical page of up to limit rows of [start, end) +// from groupID's own store. It returns, in key order, the rows of that page the +// group serves, and the last key the page read when the page was full (nil once +// the range is exhausted); the caller resumes just after that key. ScanGroupAt +// first resolves the requested range through the route catalog, and a bare +// internal prefix such as "!redis|str|" resolves to the first route only, so it +// refuses every other group. Maintenance that walks a whole prefix one group at +// a time uses this instead, at a cost of one page per call however few of its +// rows are kept. +// +// A group serves the keys whose route it owns, plus Redis hash, set and zset +// rows still stored at the route of their raw key, which reads fall back to; the +// page has already re-read those through the point-read path, so they carry the +// value a reader sees. Rows of ranges the group no longer owns are skipped. Keys +// of a route under staged visibility are skipped too: the page reads live rows +// only, and an older live copy must not be acted on while a newer staged copy +// waits for promotion. It is meant for the group's own leader: on any other node +// the read is forwarded, and the leader serves it as ScanGroupAt, which may +// refuse such a range. +func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { if limit <= 0 { - return []*store.KVPair{}, nil + return []*store.KVPair{}, nil, nil } - route := distribution.Route{GroupID: groupID} - out := make([]*store.KVPair, 0, limit) - cursor := start - for { - if err := ctx.Err(); err != nil { - return nil, errors.WithStack(err) - } - page, err := s.scanRouteAtDirectionWithReadFence(ctx, route, cursor, end, limit, ts, false, true, 0, nil, nil) - if err != nil { - return nil, err - } - out = s.appendGroupOwnedKVs(out, page, groupID, limit) - if len(out) >= limit || len(page) < limit { - return out, nil - } - cursor = nextScanCursor(lastKVKey(page)) - if end != nil && bytes.Compare(cursor, end) >= 0 { - return out, nil + g, ok := s.groupForID(groupID) + if !ok || g == nil || g.Store == nil { + return []*store.KVPair{}, nil, nil + } + page, err := s.scanRouteAtForwardPage(ctx, distribution.Route{GroupID: groupID}, g, start, end, limit, ts, true, 0, nil, nil) + if err != nil { + return nil, nil, err + } + var resume []byte + if page.full { + resume = page.advanceKey + } + canonicalized := redisWideColumnCanonicalizableScan(start) + out := make([]*store.KVPair, 0, len(page.kvs)) + for _, kvp := range page.kvs { + if kvp != nil && s.groupServesKey(groupID, kvp.Key, canonicalized) { + out = append(out, kvp) } } + return out, resume, nil } -func (s *ShardStore) appendGroupOwnedKVs(out []*store.KVPair, kvs []*store.KVPair, groupID uint64, limit int) []*store.KVPair { - for _, kvp := range kvs { - if len(out) >= limit { - break - } - if kvp == nil { - continue - } - if route, _, ok := s.routeAndGroupForKey(kvp.Key); ok && route.GroupID == groupID { - out = append(out, kvp) - } +// groupServesKey reports whether groupID serves key: it owns the key's route, +// or key is a wide-column row stored at its legacy raw-key route in groupID +// and the scan re-read it through the point-read path. +func (s *ShardStore) groupServesKey(groupID uint64, key []byte, canonicalized bool) bool { + route, _, ok := s.routeAndGroupForKey(key) + if !ok || routeHasStagedVisibility(route) { + return false } - return out + if route.GroupID == groupID { + return true + } + legacyKey := redisWideColumnLegacyPointRouteKey(key) + if !canonicalized || legacyKey == nil { + return false + } + legacy, ok := s.engine.GetRoute(legacyKey) + return ok && legacy.GroupID == groupID && !routeHasStagedVisibility(legacy) } func (s *ShardStore) scanExplicitGroupAtWithReadFence(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64, reverse bool, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error) { diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 0d57fc858..1645f71e3 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -1562,7 +1562,7 @@ func TestShardStoreScanGroupOwnedAt_ReturnsEachGroupsShareOfAPrefix(t *testing.T _, err := st.ScanGroupAt(ctx, 2, tc.prefix, end, 10, 10) require.ErrorIs(t, err, ErrExplicitGroupRouteOwnerMismatch) for gid, key := range owned { - kvs, err := st.ScanGroupOwnedAt(ctx, gid, tc.prefix, end, 10, 10) + kvs, _, err := st.ScanGroupOwnedAt(ctx, gid, tc.prefix, end, 10, 10) require.NoError(t, err) require.Equalf(t, []*store.KVPair{{Key: key, Value: []byte("v")}}, kvs, "group %d", gid) } @@ -1571,30 +1571,30 @@ func TestShardStoreScanGroupOwnedAt_ReturnsEachGroupsShareOfAPrefix(t *testing.T } // A group's store can still hold keys of a range that has moved to another -// group. ScanGroupOwnedAt must skip them, and must keep paging past them instead -// of reporting a short page as the end of the range: hash meta keys are length -// prefixed, so a stale two-byte key sorts before an owned three-byte one. +// group. ScanGroupOwnedAt must skip them: acting on such a copy could rewrite +// the key from a stale value. String keys have no legacy route, so a copy in +// the wrong group is just stale. func TestShardStoreScanGroupOwnedAt_SkipsKeysOfRangesTheGroupNoLongerOwns(t *testing.T) { t.Parallel() ctx := context.Background() st, groups := newSplitRedisShardStoreForGroupScan() - stale := store.HashMetaKey([]byte("zz")) - owned := store.HashMetaKey([]byte("abc")) - require.Negative(t, bytes.Compare(stale, owned), "the stale key must sort first") + stale := []byte("!redis|str|zebra") + owned := []byte("!redis|str|apple") require.NoError(t, groups[1].Store.PutAt(ctx, stale, []byte("stale"), 5, 0)) require.NoError(t, groups[1].Store.PutAt(ctx, owned, []byte("owned"), 5, 0)) require.NoError(t, groups[2].Store.PutAt(ctx, stale, []byte("current"), 6, 0)) - prefix := []byte(store.HashMetaPrefix) + prefix := []byte("!redis|str|") end := prefixScanEnd(prefix) - kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, end, 1, 10) + kvs, resume, err := st.ScanGroupOwnedAt(ctx, 1, prefix, end, 10, 10) require.NoError(t, err) - require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, kvs, - "a page of skipped keys is not the end of the range") + require.Nil(t, resume) + require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, kvs) - kvs, err = st.ScanGroupOwnedAt(ctx, 2, prefix, end, 10, 10) + kvs, resume, err = st.ScanGroupOwnedAt(ctx, 2, prefix, end, 10, 10) require.NoError(t, err) + require.Nil(t, resume) require.Equal(t, []*store.KVPair{{Key: stale, Value: []byte("current")}}, kvs) } @@ -1648,7 +1648,7 @@ func TestShardStoreScanGroupOwnedAt_DefersKeysOfStagedRoutes(t *testing.T) { require.NoError(t, group.Store.PutAt(ctx, live, []byte("v"), 5, 0)) prefix := []byte("!redis|str|") - kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) require.NoError(t, err) require.Equal(t, []*store.KVPair{{Key: live, Value: []byte("v")}}, kvs) } @@ -1679,9 +1679,23 @@ func TestShardStoreScanGroupOwnedAt_ReadsOnePhysicalPagePerCall(t *testing.T) { prefix := []byte("!redis|str|") const limit = 2 - _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), limit, 10) - require.NoError(t, err) - require.LessOrEqual(t, counting.rowsRead(), limit) + end := prefixScanEnd(prefix) + + var served []*store.KVPair + start := prefix + for calls := 1; ; calls++ { + before := counting.rowsRead() + kvs, resume, err := st.ScanGroupOwnedAt(ctx, 1, start, end, limit, 10) + require.NoError(t, err) + require.LessOrEqualf(t, counting.rowsRead()-before, limit, "call %d read more than one page", calls) + served = append(served, kvs...) + if resume == nil { + break + } + require.Lessf(t, calls, 10, "the scan must make progress") + start = nextScanCursor(resume) + } + require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, served) } // Hash, set and zset rows written before wide-column routing still live at the @@ -1715,7 +1729,7 @@ func TestShardStoreScanGroupOwnedAt_ServesWideColumnRowsAtTheirLegacyRoute(t *te } prefix := []byte(store.HashMetaPrefix) - kvs, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) require.NoError(t, err) require.Equal(t, []*store.KVPair{{Key: key, Value: tc.want}}, kvs) }) diff --git a/main_redis_ttl_migration_wiring_test.go b/main_redis_ttl_migration_wiring_test.go index 208e0c329..b99e93438 100644 --- a/main_redis_ttl_migration_wiring_test.go +++ b/main_redis_ttl_migration_wiring_test.go @@ -33,14 +33,18 @@ func (s *groupScanRecordingStore) ScanAt(ctx context.Context, start []byte, end return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) } -func (s *groupScanRecordingStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { +func (s *groupScanRecordingStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { s.mu.Lock() if s.groupScans == nil { s.groupScans = make(map[uint64][]string) } s.groupScans[groupID] = append(s.groupScans[groupID], string(start)) s.mu.Unlock() - return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) + kvs, err := s.MVCCStore.ScanAt(ctx, start, end, limit, ts) + if err != nil || len(kvs) < limit { + return kvs, nil, err + } + return kvs, kvs[len(kvs)-1].Key, nil } func (s *groupScanRecordingStore) recorded() (map[uint64][]string, []string) { From 8fec3649eb656b0496d2c01ab9a9158eed326029 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:02:01 +0000 Subject: [PATCH 09/15] kv, adapter: reproduce legacy-route rows served against the owner's state ScanGroupOwnedAt serves a hash, set or zset row stored at the route of its raw key whenever the scan canonicalized it, and never a stream row: - A legacy stream meta row whose user key belongs to another group is dropped, so the stream TTL-inline handler never migrates it, even when this node leads both groups (Codex review). - A wide-column key the logical owner also holds is served by both groups. The migrator then rewrites it from both plans in one tick, and the second dispatch fails its OCC check. Point reads fall back to the legacy route only when the logical owner holds no version of the key, so that is when the legacy group should serve it. These tests fail until the scan follows that rule. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 49 +++++++++++ kv/shard_store_test.go | 113 +++++++++++++++++++------- 2 files changed, 133 insertions(+), 29 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 7f8910cbf..66cf94097 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -787,6 +787,55 @@ func TestDeltaCompactor_TTLInlineMigratesWideColumnRowsAtTheirLegacyRoute(t *tes require.Equal(t, redisExpireAtMillis(expireAt), meta.ExpireAt) } +// Stream metadata written before streams were routed by user key can sit at +// the route of its raw key while the user key belongs to another group, and +// point reads fall back to it while that group holds nothing. The migrator must +// rewrite it into the user key's group with its TTL inline, as it does for +// hash, set and zset metadata. +func TestDeltaCompactor_TTLInlineMigratesStreamMetaAtItsLegacyRoute(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + group1 := store.NewMVCCStore() + st := kv.NewShardStore(engine, map[uint64]*kv.ShardGroup{ + 1: {Store: group1}, + 2: {Store: store.NewMVCCStore()}, + }) + coord := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: []uint64{1, 2}, + } + c := NewDeltaCompactor(st, coord, WithDeltaCompactorMaxDeltaCount(2)) + + key := []byte("zebra") + expireAt := time.Now().Add(time.Hour) + legacyMeta := make([]byte, redisWideMetaLegacySizeBytes) + binary.BigEndian.PutUint64(legacyMeta[0:8], 3) + binary.BigEndian.PutUint64(legacyMeta[8:16], 100) + binary.BigEndian.PutUint64(legacyMeta[16:24], 7) + // The raw meta key starts with "!", so its legacy route is group 1. + require.NoError(t, group1.PutAt(ctx, store.StreamMetaKey(key), legacyMeta, 1, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey(key), encodeRedisTTL(expireAt), 2, 0)) + + require.NoError(t, c.SyncOnce(ctx)) + + raw, err := st.GetAt(ctx, store.StreamMetaKey(key), st.LastCommitTS()) + require.NoError(t, err) + meta, err := store.UnmarshalStreamMeta(raw) + require.NoError(t, err) + require.Equal(t, store.StreamMeta{ + Length: 3, + LastMs: 100, + LastSeq: 7, + ExpireAt: redisExpireAtMillis(expireAt), + }, meta) +} + // capabilityHidingCompactionCoordinator forwards only the kv.Coordinator // methods, the way a decorator that forgets an optional forward does. type capabilityHidingCompactionCoordinator struct { diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 1645f71e3..0ecb65ff9 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -1698,42 +1698,97 @@ func TestShardStoreScanGroupOwnedAt_ReadsOnePhysicalPagePerCall(t *testing.T) { require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, served) } -// Hash, set and zset rows written before wide-column routing still live at the -// route of their raw key, and reads fall back to that route when the logical -// owner has nothing. The group holding such a row serves it, so its scan must -// return it with the value a point read would see, not drop it because the -// logical owner is another group. -func TestShardStoreScanGroupOwnedAt_ServesWideColumnRowsAtTheirLegacyRoute(t *testing.T) { +// Hash, set, zset and stream rows written before their families were routed by +// user key still live at the route of their raw key, and point reads fall back +// to that route only when the logical owner holds no version of the key. The +// group holding such a row must serve it exactly then: when the logical owner +// has its own row, or a tombstone, that is what a reader sees, and the legacy +// copy must not be acted on as well. +func TestShardStoreScanGroupOwnedAt_ServesLegacyRouteRowsTheOwnerDoesNotHold(t *testing.T) { t.Parallel() - cases := []struct { - name string - ownerHasNewer bool - want []byte + families := []struct { + name string + prefix []byte + key []byte }{ - {name: "legacy row only", want: []byte("legacy")}, - {name: "logical owner has a newer row", ownerHasNewer: true, want: []byte("current")}, + // The raw keys start with "!", so their legacy route is group 1; the + // user key "zebra" belongs to group 2. + {name: "hash", prefix: []byte(store.HashMetaPrefix), key: store.HashMetaKey([]byte("zebra"))}, + {name: "stream", prefix: []byte(store.StreamMetaPrefix), key: store.StreamMetaKey([]byte("zebra"))}, } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() + owners := []struct { + name string + write func(ctx context.Context, st store.MVCCStore, key []byte) error + wantGroup uint64 + want []byte + }{ + {name: "owner holds nothing", wantGroup: 1, want: []byte("legacy")}, + { + name: "owner holds a newer row", + write: func(ctx context.Context, st store.MVCCStore, key []byte) error { + return st.PutAt(ctx, key, []byte("current"), 6, 0) + }, + wantGroup: 2, + want: []byte("current"), + }, + { + name: "owner deleted the key", + write: func(ctx context.Context, st store.MVCCStore, key []byte) error { + return st.DeleteAt(ctx, key, 6) + }, + }, + } + for _, family := range families { + for _, owner := range owners { + t.Run(family.name+"/"+owner.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st, groups := newSplitRedisShardStoreForGroupScan() + require.NoError(t, groups[1].Store.PutAt(ctx, family.key, []byte("legacy"), 5, 0)) + if owner.write != nil { + require.NoError(t, owner.write(ctx, groups[2].Store, family.key)) + } + end := prefixScanEnd(family.prefix) + + for _, gid := range []uint64{1, 2} { + kvs, _, err := st.ScanGroupOwnedAt(ctx, gid, family.prefix, end, 10, 10) + require.NoError(t, err) + want := []*store.KVPair{} + if gid == owner.wantGroup { + want = []*store.KVPair{{Key: family.key, Value: owner.want}} + } + require.Equalf(t, want, kvs, "group %d", gid) + } + }) + } + } +} - ctx := context.Background() - st, groups := newSplitRedisShardStoreForGroupScan() - // The raw key starts with "!", so its legacy route is group 1; - // the user key "zebra" belongs to group 2. - key := store.HashMetaKey([]byte("zebra")) - require.NoError(t, groups[1].Store.PutAt(ctx, key, []byte("legacy"), 5, 0)) - if tc.ownerHasNewer { - require.NoError(t, groups[2].Store.PutAt(ctx, key, []byte("current"), 6, 0)) - } - prefix := []byte(store.HashMetaPrefix) +// Whether the logical owner holds the key can be answered here only by that +// group's leader. When another node leads it, the legacy row is left to that +// node: only it may rewrite the key, and guessing "nothing there" could act on +// a copy the owner has superseded. +func TestShardStoreScanGroupOwnedAt_SkipsLegacyRouteRowsOfAnOwnerLedElsewhere(t *testing.T) { + t.Parallel() - kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) - require.NoError(t, err) - require.Equal(t, []*store.KVPair{{Key: key, Value: tc.want}}, kvs) - }) + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + groups := map[uint64]*ShardGroup{ + 1: {Store: store.NewMVCCStore()}, + 2: {Store: store.NewMVCCStore(), Engine: &followerProxyEngine{}}, } + st := NewShardStore(engine, groups) + key := store.StreamMetaKey([]byte("zebra")) + require.NoError(t, groups[1].Store.PutAt(ctx, key, []byte("legacy"), 5, 0)) + prefix := []byte(store.StreamMetaPrefix) + + kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + require.NoError(t, err) + require.Empty(t, kvs) } func TestShardStoreGetGroupAt_UsesExplicitGroup(t *testing.T) { From 49af1bf8bd4ab2ee7a5473a341e8af35f45ff3bb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:04:46 +0000 Subject: [PATCH 10/15] kv: serve legacy-route rows only while the logical owner holds none ScanGroupOwnedAt now serves a row stored at the route of its raw key only when this node leads the key's logical owner and that group holds no version of the key at the read timestamp: exactly when a point read falls back to the legacy route, so the row is the value a reader sees. - Stream metadata is covered: the legacy route candidate is now legacyPointRouteKey, the one point reads use, which includes streams. Streams are not canonicalized by the scan, and they need not be: when the owner holds nothing, the raw row is the canonical value. - A wide-column key the owner also holds is served only by the owner, so the migrator rewrites it once per tick instead of from two plans whose second dispatch failed its OCC check. - When another node leads the owner, or this node does not host it, the row is left out; only the owner's leader may rewrite the key. Cost: one local version probe per legacy-route row, through the same lease-fenced check point reads already use. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- kv/shard_store.go | 83 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 61 insertions(+), 22 deletions(-) diff --git a/kv/shard_store.go b/kv/shard_store.go index 2d78740ce..1c402553d 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -914,15 +914,16 @@ func (s *ShardStore) ScanGroupAt(ctx context.Context, groupID uint64, start []by // a time uses this instead, at a cost of one page per call however few of its // rows are kept. // -// A group serves the keys whose route it owns, plus Redis hash, set and zset -// rows still stored at the route of their raw key, which reads fall back to; the -// page has already re-read those through the point-read path, so they carry the -// value a reader sees. Rows of ranges the group no longer owns are skipped. Keys -// of a route under staged visibility are skipped too: the page reads live rows -// only, and an older live copy must not be acted on while a newer staged copy -// waits for promotion. It is meant for the group's own leader: on any other node -// the read is forwarded, and the leader serves it as ScanGroupAt, which may -// refuse such a range. +// A group serves the keys whose route it owns, plus Redis hash, set, zset and +// stream rows still stored at the route of their raw key while the logical owner +// holds no version of the key: point reads fall back to such a row exactly then, +// so it is the value a reader sees. Only the owner's leader can tell, so on any +// other node those rows are left out. Rows of ranges the group no longer owns +// are skipped. Keys of a route under staged visibility are skipped too: the page +// reads live rows only, and an older live copy must not be acted on while a +// newer staged copy waits for promotion. It is meant for the group's own leader: +// on any other node the read is forwarded, and the leader serves it as +// ScanGroupAt, which may refuse such a range. func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { if limit <= 0 { return []*store.KVPair{}, nil, nil @@ -939,35 +940,73 @@ func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start if page.full { resume = page.advanceKey } - canonicalized := redisWideColumnCanonicalizableScan(start) - out := make([]*store.KVPair, 0, len(page.kvs)) - for _, kvp := range page.kvs { - if kvp != nil && s.groupServesKey(groupID, kvp.Key, canonicalized) { + out, err := s.rowsGroupServesAt(ctx, groupID, page.kvs, ts) + if err != nil { + return nil, nil, err + } + return out, resume, nil +} + +func (s *ShardStore) rowsGroupServesAt(ctx context.Context, groupID uint64, kvs []*store.KVPair, ts uint64) ([]*store.KVPair, error) { + out := make([]*store.KVPair, 0, len(kvs)) + for _, kvp := range kvs { + if kvp == nil { + continue + } + serves, err := s.groupServesKeyAt(ctx, groupID, kvp.Key, ts) + if err != nil { + return nil, err + } + if serves { out = append(out, kvp) } } - return out, resume, nil + return out, nil } -// groupServesKey reports whether groupID serves key: it owns the key's route, -// or key is a wide-column row stored at its legacy raw-key route in groupID -// and the scan re-read it through the point-read path. -func (s *ShardStore) groupServesKey(groupID uint64, key []byte, canonicalized bool) bool { +// groupServesKeyAt reports whether groupID serves key at ts: it owns the key's +// route, or key is stored at its legacy raw-key route in groupID and the +// logical owner holds no version of key at ts. Then the group's own row is the +// value a point read returns. +func (s *ShardStore) groupServesKeyAt(ctx context.Context, groupID uint64, key []byte, ts uint64) (bool, error) { route, _, ok := s.routeAndGroupForKey(key) if !ok || routeHasStagedVisibility(route) { - return false + return false, nil } if route.GroupID == groupID { - return true + return true, nil + } + if !s.legacyRouteInGroup(groupID, key) { + return false, nil } - legacyKey := redisWideColumnLegacyPointRouteKey(key) - if !canonicalized || legacyKey == nil { + return s.ownerHoldsNoVersionAt(ctx, route, key, ts) +} + +// legacyRouteInGroup reports whether key's raw-key route, which point reads +// fall back to, belongs to groupID and is not under staged visibility. +func (s *ShardStore) legacyRouteInGroup(groupID uint64, key []byte) bool { + legacyKey := legacyPointRouteKey(key) + if legacyKey == nil { return false } legacy, ok := s.engine.GetRoute(legacyKey) return ok && legacy.GroupID == groupID && !routeHasStagedVisibility(legacy) } +// ownerHoldsNoVersionAt reports whether owner, the route of key's logical +// owner, holds no version of key at ts. Only the owner's leader on this node +// can answer; anywhere else it reports false. +func (s *ShardStore) ownerHoldsNoVersionAt(ctx context.Context, owner distribution.Route, key []byte, ts uint64) (bool, error) { + if g, ok := s.groupForID(owner.GroupID); !ok || g == nil || g.Store == nil { + return false, nil + } + exists, answered, err := s.routeHasVersionAtOrBefore(ctx, owner, key, ts) + if err != nil { + return false, err + } + return answered && !exists, nil +} + func (s *ShardStore) scanExplicitGroupAtWithReadFence(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64, reverse bool, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error) { if limit <= 0 { return []*store.KVPair{}, nil From 8ca392bfea4de262afc8290758828bee6c1b4e69 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:06:37 +0000 Subject: [PATCH 11/15] adapter: reproduce TTL-inline ticks starving later scan plans Every TTL-inline scan plan (one per handler and locally led group) shares the compactor's per-tick budget, and every tick starts from the first handler's first group. When the earlier plans use up the budget, the later handlers and groups are never reached (Codex review). With three local leader groups, 24 budget-limited ticks scan the first plan 24 times and no other. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor_test.go | 52 +++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 66cf94097..247ea507c 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -625,6 +625,58 @@ func TestDeltaCompactor_TTLInlineUsesLocalLeaderGroupScan(t *testing.T) { require.Equal(t, redisExpireAtMillis(expireAt), meta.ExpireAt) } +// budgetedTTLInlineScanStore ends the tick during its first group scan, the +// way a tick whose first scan uses up the per-tick budget ends, and records +// which group and prefix each tick reached. +type budgetedTTLInlineScanStore struct { + store.MVCCStore + + endTick context.CancelFunc + scanned map[string]int +} + +func (s *budgetedTTLInlineScanStore) ScanGroupOwnedAt(_ context.Context, groupID uint64, start []byte, _ []byte, _ int, _ uint64) ([]*store.KVPair, []byte, error) { + s.scanned[string(start)+"#"+strconv.FormatUint(groupID, 10)]++ + s.endTick() + return nil, nil, nil +} + +// Every TTL-inline scan plan shares one per-tick budget. When the budget runs +// out partway through the plans, the next tick must start where this one +// stopped; starting every tick from the first handler's first group starved +// the later handlers and groups for as long as the earlier ones used the whole +// budget. +func TestDeltaCompactor_TTLInlineRotatesPlansAcrossTicksThatRunOutOfBudget(t *testing.T) { + t.Parallel() + + st := &budgetedTTLInlineScanStore{MVCCStore: store.NewMVCCStore(), scanned: map[string]int{}} + groupIDs := []uint64{1, 2, 3} + coord := &localLeaderGroupCompactionCoordinator{ + shardOnlyLeaderCompactionCoordinator: &shardOnlyLeaderCompactionCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + }, + groupIDs: groupIDs, + } + c := NewDeltaCompactor(st, coord) + handlers := c.ttlInlineMigrationHandlers() + plans := len(handlers) * len(groupIDs) + + for range plans { + ctx, cancel := context.WithCancel(context.Background()) + st.endTick = cancel + require.ErrorIs(t, c.migrateTTLInlineOnce(ctx, 10), context.Canceled) + cancel() + } + + want := make(map[string]int, plans) + for _, h := range handlers { + for _, groupID := range groupIDs { + want[string(h.prefix)+"#"+strconv.FormatUint(groupID, 10)] = 1 + } + } + require.Equal(t, want, st.scanned, "each tick must reach the plan after the one the previous tick reached") +} + // newSplitRedisShardStore returns a ShardStore whose catalog sends user keys // below "m" to group 1 and the rest to group 2, the way a multi-shard // deployment splits the Redis keyspace. From 5c3fc97364e3b61e3b72826f4dad220a3e1aacbc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:07:59 +0000 Subject: [PATCH 12/15] adapter: rotate TTL-inline scan plans across ticks Each tick now flattens every handler's scan plans (one per handler and locally led group) into one list and starts at the plan after the last one the previous tick ran. A tick that runs out of budget hands the rest of the rotation to the next tick instead of starting over at the first handler's first group, so no handler or group is starved while the earlier plans keep using up the budget. The background delta handlers already rotate this way. Their two cursor helpers now take the cursor they rotate, and both loops share them. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- adapter/redis_delta_compactor.go | 29 ++++++++++++------- adapter/redis_ttl_inline_migrator.go | 43 ++++++++++++++++++---------- 2 files changed, 46 insertions(+), 26 deletions(-) diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index 670c40355..fe57ecaa1 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -72,9 +72,13 @@ type DeltaCompactor struct { // quickly without waiting for the next scheduled pass. urgentCh chan urgentCompactionRequest - stateMu sync.Mutex - backoffUntil time.Time - handlerCursor int + stateMu sync.Mutex + backoffUntil time.Time + // handlerCursor and ttlInlinePlanCursor are where the next tick's + // background handlers and TTL-inline scan plans start, so a tick that runs + // out of budget hands the rest of the rotation to the next one. + handlerCursor int + ttlInlinePlanCursor int ttlInlineGroupsUnknownOnce sync.Once } @@ -342,7 +346,7 @@ func (c *DeltaCompactor) compactBackgroundHandlers(ctx context.Context, readTS u return nil } handlers := c.allHandlers() - start := c.backgroundHandlerStart(len(handlers)) + start := c.rotationStart(&c.handlerCursor, len(handlers)) var combined error for offset := range handlers { if ctx.Err() != nil { @@ -351,7 +355,7 @@ func (c *DeltaCompactor) compactBackgroundHandlers(ctx context.Context, readTS u idx := (start + offset) % len(handlers) h := handlers[idx] err := c.compactHandlerSafely(ctx, h, readTS) - c.advanceBackgroundHandlerCursor(idx, len(handlers)) + c.advanceRotation(&c.handlerCursor, idx, len(handlers)) if err != nil && !errors.Is(err, context.Canceled) { combined = errors.CombineErrors(combined, err) } @@ -359,25 +363,28 @@ func (c *DeltaCompactor) compactBackgroundHandlers(ctx context.Context, readTS u return errors.WithStack(combined) } -func (c *DeltaCompactor) backgroundHandlerStart(total int) int { +// rotationStart returns where a rotation over total items starts this tick. +// cursor is one of the compactor's rotation cursors, guarded by stateMu. +func (c *DeltaCompactor) rotationStart(cursor *int, total int) int { if total <= 0 { return 0 } c.stateMu.Lock() defer c.stateMu.Unlock() - if c.handlerCursor < 0 || c.handlerCursor >= total { - c.handlerCursor = 0 + if *cursor < 0 || *cursor >= total { + *cursor = 0 } - return c.handlerCursor + return *cursor } -func (c *DeltaCompactor) advanceBackgroundHandlerCursor(current, total int) { +// advanceRotation moves cursor past current, the item a tick just ran. +func (c *DeltaCompactor) advanceRotation(cursor *int, current, total int) { if total <= 0 { return } c.stateMu.Lock() defer c.stateMu.Unlock() - c.handlerCursor = (current + 1) % total + *cursor = (current + 1) % total } func (c *DeltaCompactor) compactHandlerSafely(ctx context.Context, h collectionDeltaHandler, readTS uint64) (err error) { diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 34eb2e26c..2f992ea10 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -38,19 +38,45 @@ type ttlInlineMigrationScanPlan struct { scanner ttlInlineGroupScanner } +// ttlInlineMigrationTickPlan is one handler's scan plan within a tick. +type ttlInlineMigrationTickPlan struct { + handler ttlInlineMigrationHandler + plan ttlInlineMigrationScanPlan +} + +// migrateTTLInlineOnce runs every handler's scan plans under the tick's +// budget. The plans share that budget, so each tick starts at the plan after +// the last one the previous tick ran, as compactBackgroundHandlers rotates its +// handlers; starting from the first plan every tick starved the later handlers +// and groups whenever the earlier plans used up the budget. func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64) error { + plans := c.ttlInlineMigrationTickPlans() + start := c.rotationStart(&c.ttlInlinePlanCursor, len(plans)) var combined error - for _, h := range c.ttlInlineMigrationHandlers() { + for offset := range plans { if ctx.Err() != nil { return errors.WithStack(errors.CombineErrors(combined, ctx.Err())) } - if err := c.migrateTTLInlineHandler(ctx, h, readTS); err != nil { + idx := (start + offset) % len(plans) + err := c.migrateTTLInlineScanPlan(ctx, plans[idx].handler, plans[idx].plan, readTS) + c.advanceRotation(&c.ttlInlinePlanCursor, idx, len(plans)) + if err != nil { combined = errors.CombineErrors(combined, err) } } return errors.WithStack(combined) } +func (c *DeltaCompactor) ttlInlineMigrationTickPlans() []ttlInlineMigrationTickPlan { + var plans []ttlInlineMigrationTickPlan + for _, h := range c.ttlInlineMigrationHandlers() { + for _, plan := range c.ttlInlineMigrationScanPlans(h) { + plans = append(plans, ttlInlineMigrationTickPlan{handler: h, plan: plan}) + } + } + return plans +} + func (c *DeltaCompactor) ttlInlineMigrationHandlers() []ttlInlineMigrationHandler { return []ttlInlineMigrationHandler{ { @@ -180,19 +206,6 @@ func (c *DeltaCompactor) simpleTTLInlineMigrationHandler( } } -func (c *DeltaCompactor) migrateTTLInlineHandler(ctx context.Context, h ttlInlineMigrationHandler, readTS uint64) error { - var combined error - for _, plan := range c.ttlInlineMigrationScanPlans(h) { - if ctx.Err() != nil { - return errors.WithStack(errors.CombineErrors(combined, ctx.Err())) - } - if err := c.migrateTTLInlineScanPlan(ctx, h, plan, readTS); err != nil { - combined = errors.CombineErrors(combined, err) - } - } - return errors.WithStack(combined) -} - func (c *DeltaCompactor) migrateTTLInlineScanPlan( ctx context.Context, h ttlInlineMigrationHandler, From 832b048fae1f4ffdc497da97feafa21ca79e2280 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:18:27 +0000 Subject: [PATCH 13/15] kv: reproduce the group-owned scan re-reading every wide-column row ScanGroupOwnedAt reads its page through scanRouteAtForwardPage, which re-reads every hash, set and zset row through the point-read path. Under the owner-holds-nothing rule every served row already carries the value a point read returns, so those reads are redundant, and some do harm: - one extra point read per owned wide-column row; - a proxied read to the owner's leader for every legacy-route row the scan then drops; - a page that fails outright with "leader not found" when a legacy row's owner has no known leader, so the plan's cursor cannot advance. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- kv/shard_store_test.go | 89 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 78 insertions(+), 11 deletions(-) diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 0ecb65ff9..c01dc0c9e 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -1767,28 +1767,95 @@ func TestShardStoreScanGroupOwnedAt_ServesLegacyRouteRowsTheOwnerDoesNotHold(t * } // Whether the logical owner holds the key can be answered here only by that -// group's leader. When another node leads it, the legacy row is left to that -// node: only it may rewrite the key, and guessing "nothing there" could act on -// a copy the owner has superseded. +// group's leader. When another node leads it, or no leader is known, the +// legacy row is left to that group's leader: only it may rewrite the key, and +// guessing "nothing there" could act on a copy the owner has superseded. The +// rest of the page must still be served. func TestShardStoreScanGroupOwnedAt_SkipsLegacyRouteRowsOfAnOwnerLedElsewhere(t *testing.T) { t.Parallel() + families := []struct { + name string + prefix []byte + key func(userKey string) []byte + }{ + {name: "hash", prefix: []byte(store.HashMetaPrefix), key: func(u string) []byte { return store.HashMetaKey([]byte(u)) }}, + {name: "stream", prefix: []byte(store.StreamMetaPrefix), key: func(u string) []byte { return store.StreamMetaKey([]byte(u)) }}, + } + for _, family := range families { + t.Run(family.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), []byte("m"), 1) + engine.UpdateRoute([]byte("m"), nil, 2) + groups := map[uint64]*ShardGroup{ + 1: {Store: store.NewMVCCStore()}, + 2: {Store: store.NewMVCCStore(), Engine: &followerProxyEngine{}}, + } + st := NewShardStore(engine, groups) + owned := family.key("apple") + legacy := family.key("zebra") + require.NoError(t, groups[1].Store.PutAt(ctx, owned, []byte("owned"), 5, 0)) + require.NoError(t, groups[1].Store.PutAt(ctx, legacy, []byte("legacy"), 5, 0)) + + kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, family.prefix, prefixScanEnd(family.prefix), 10, 10) + require.NoError(t, err) + require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, kvs) + }) + } +} + +// pointReadCountingMVCCStore counts point reads, which a ShardStore scan uses +// to re-read wide-column rows through the point-read path. +type pointReadCountingMVCCStore struct { + store.MVCCStore + + mu sync.Mutex + reads int +} + +func (s *pointReadCountingMVCCStore) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error) { + s.mu.Lock() + s.reads++ + s.mu.Unlock() + return s.MVCCStore.GetAt(ctx, key, ts) +} + +func (s *pointReadCountingMVCCStore) pointReads() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.reads +} + +// Every row ScanGroupOwnedAt serves already carries the value a point read +// would return, so the page must not re-read its wide-column rows one point +// read at a time. +func TestShardStoreScanGroupOwnedAt_DoesNotReReadServedRows(t *testing.T) { + t.Parallel() + ctx := context.Background() engine := distribution.NewEngine() engine.UpdateRoute([]byte(""), []byte("m"), 1) engine.UpdateRoute([]byte("m"), nil, 2) - groups := map[uint64]*ShardGroup{ - 1: {Store: store.NewMVCCStore()}, - 2: {Store: store.NewMVCCStore(), Engine: &followerProxyEngine{}}, + counting := &pointReadCountingMVCCStore{MVCCStore: store.NewMVCCStore()} + st := NewShardStore(engine, map[uint64]*ShardGroup{ + 1: {Store: counting}, + 2: {Store: store.NewMVCCStore()}, + }) + want := make([]*store.KVPair, 0, 3) + for _, userKey := range []string{"apple", "banana", "cherry"} { + key := store.HashMetaKey([]byte(userKey)) + require.NoError(t, counting.PutAt(ctx, key, []byte(userKey), 5, 0)) + want = append(want, &store.KVPair{Key: key, Value: []byte(userKey)}) } - st := NewShardStore(engine, groups) - key := store.StreamMetaKey([]byte("zebra")) - require.NoError(t, groups[1].Store.PutAt(ctx, key, []byte("legacy"), 5, 0)) - prefix := []byte(store.StreamMetaPrefix) + prefix := []byte(store.HashMetaPrefix) kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) require.NoError(t, err) - require.Empty(t, kvs) + require.Equal(t, want, kvs) + require.Zero(t, counting.pointReads()) } func TestShardStoreGetGroupAt_UsesExplicitGroup(t *testing.T) { From a73a7c0a5960e1582c5d967a0f999f42dc69f224 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 23:20:17 +0000 Subject: [PATCH 14/15] kv: skip the point-read pass for rows the group-owned scan serves The three forward page paths (local, leader, proxy) each re-read the page's hash, set and zset rows through the point-read path. That pass now runs once, in scanRouteAtForwardPage, after whichever path read the page, so its callers behave as before. ScanGroupOwnedAt reads the page through the new scanRouteAtForwardStoredPage, without the pass: every row it serves is owned or held by no one at the owner, so it already carries the value a point read returns. This drops one point read per owned wide-column row, a proxied read per legacy-route row the scan then drops, and the page failure when such a row's owner has no known leader. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- kv/shard_store.go | 57 ++++++++++++++++++++++++++++++----------------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/kv/shard_store.go b/kv/shard_store.go index 1c402553d..0ce37ce54 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -918,7 +918,9 @@ func (s *ShardStore) ScanGroupAt(ctx context.Context, groupID uint64, start []by // stream rows still stored at the route of their raw key while the logical owner // holds no version of the key: point reads fall back to such a row exactly then, // so it is the value a reader sees. Only the owner's leader can tell, so on any -// other node those rows are left out. Rows of ranges the group no longer owns +// other node those rows are left out. Every served row is thus the value a point +// read returns, and the page skips the point-read pass other scans give +// wide-column rows. Rows of ranges the group no longer owns // are skipped. Keys of a route under staged visibility are skipped too: the page // reads live rows only, and an older live copy must not be acted on while a // newer staged copy waits for promotion. It is meant for the group's own leader: @@ -932,7 +934,7 @@ func (s *ShardStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start if !ok || g == nil || g.Store == nil { return []*store.KVPair{}, nil, nil } - page, err := s.scanRouteAtForwardPage(ctx, distribution.Route{GroupID: groupID}, g, start, end, limit, ts, true, 0, nil, nil) + page, err := s.scanRouteAtForwardStoredPage(ctx, distribution.Route{GroupID: groupID}, g, start, end, limit, ts, true, 0, nil, nil) if err != nil { return nil, nil, err } @@ -3187,13 +3189,40 @@ func (s *ShardStore) scanRouteAtForwardPage( readRouteVersion uint64, routeStart []byte, routeEnd []byte, +) (scanRoutePage, error) { + page, err := s.scanRouteAtForwardStoredPage(ctx, route, g, start, end, limit, ts, explicitGroup, readRouteVersion, routeStart, routeEnd) + if err != nil { + return scanRoutePage{}, err + } + page.kvs, err = s.canonicalizeRedisWideColumnScanResults(ctx, page.kvs, start, ts, readRouteVersion) + if err != nil { + return scanRoutePage{}, err + } + return page, nil +} + +// scanRouteAtForwardStoredPage reads one forward page like +// scanRouteAtForwardPage but returns wide-column rows as the group stores +// them, without re-reading each through the point-read path. +func (s *ShardStore) scanRouteAtForwardStoredPage( + ctx context.Context, + route distribution.Route, + g *ShardGroup, + start []byte, + end []byte, + limit int, + ts uint64, + explicitGroup bool, + readRouteVersion uint64, + routeStart []byte, + routeEnd []byte, ) (scanRoutePage, error) { engine := engineForGroup(g) if engine == nil { - return s.scanRouteAtForwardLocalPage(ctx, g, route, start, end, limit, ts, readRouteVersion) + return s.scanRouteAtForwardLocalPage(ctx, g, route, start, end, limit, ts) } if isLinearizableRaftLeader(ctx, engine) { - return s.scanRouteAtForwardLeaderPage(ctx, g, route, start, end, limit, ts, readRouteVersion) + return s.scanRouteAtForwardLeaderPage(ctx, g, route, start, end, limit, ts) } return s.scanRouteAtForwardProxyPage(ctx, route, g, start, end, limit, ts, explicitGroup, readRouteVersion, routeStart, routeEnd) } @@ -3206,18 +3235,13 @@ func (s *ShardStore) scanRouteAtForwardLocalPage( end []byte, limit int, ts uint64, - readRouteVersion uint64, ) (scanRoutePage, error) { raw, err := s.scanRouteLocal(ctx, g, route, start, end, limit, ts, false) if err != nil { return scanRoutePage{}, errors.WithStack(err) } - kvs, err := s.canonicalizeRedisWideColumnScanResults(ctx, filterScanInternalKVs(raw), start, ts, readRouteVersion) - if err != nil { - return scanRoutePage{}, err - } return scanRoutePage{ - kvs: kvs, + kvs: filterScanInternalKVs(raw), advanceKey: lastKVKey(raw), full: len(raw) >= limit, }, nil @@ -3231,7 +3255,6 @@ func (s *ShardStore) scanRouteAtForwardLeaderPage( end []byte, limit int, ts uint64, - readRouteVersion uint64, ) (scanRoutePage, error) { raw, err := s.scanRouteLocal(ctx, g, route, start, end, limit, ts, false) if err != nil { @@ -3246,12 +3269,8 @@ func (s *ShardStore) scanRouteAtForwardLeaderPage( if err != nil { return scanRoutePage{}, err } - kvs, err = s.canonicalizeRedisWideColumnScanResults(ctx, filterScanInternalKVs(kvs), start, ts, readRouteVersion) - if err != nil { - return scanRoutePage{}, err - } return scanRoutePage{ - kvs: kvs, + kvs: filterScanInternalKVs(kvs), advanceKey: lastKVKey(raw), full: len(raw) >= limit, }, nil @@ -3275,12 +3294,8 @@ func (s *ShardStore) scanRouteAtForwardProxyPage( if err != nil { return scanRoutePage{}, err } - kvs, err := s.canonicalizeRedisWideColumnScanResults(ctx, filterScanInternalKVs(raw), start, ts, readRouteVersion) - if err != nil { - return scanRoutePage{}, err - } return scanRoutePage{ - kvs: kvs, + kvs: filterScanInternalKVs(raw), advanceKey: lastKVKey(raw), full: len(raw) >= limit, }, nil From 5ac93191a909a4eec0095b70c7c9859b2b9a358c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 06:58:36 +0000 Subject: [PATCH 15/15] backup: allow pebble's sstable reader through the import boundary TestBackupLinksNoLiveClusterCode fails on main since two merges met: 3f097df6 (#1276) taught internal/backup to read snapshot SST ingest bundles with github.com/cockroachdb/pebble/v2/sstable, and 56780818 (#1277) forbade every github.com/cockroachdb/pebble package there. Allow exactly the sstable package. The bundles are files in pebble's sstable format, and the reader is a file-format dependency, not the live store the boundary keeps out; the rest of pebble stays forbidden. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019P5kzueGgfe3F5iSe8TMtC --- internal/backup/import_boundary_test.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/internal/backup/import_boundary_test.go b/internal/backup/import_boundary_test.go index 06e3f41c6..19c88ff2d 100644 --- a/internal/backup/import_boundary_test.go +++ b/internal/backup/import_boundary_test.go @@ -24,6 +24,13 @@ var backupForbiddenDeps = []string{ "github.com/cockroachdb/pebble", } +// backupAllowedDeps are exceptions inside a forbidden prefix. Snapshot SST +// ingest bundles (snapshot_sst.go) are files in pebble's sstable format, and +// reading them takes pebble's sstable reader, not a live store. +var backupAllowedDeps = []string{ + "github.com/cockroachdb/pebble/v2/sstable", +} + // TestBackupLinksNoLiveClusterCode walks the non-test import closure of // internal/backup through first-party packages and fails on any // forbidden package, naming the import chain that reaches it. @@ -59,6 +66,11 @@ func TestBackupLinksNoLiveClusterCode(t *testing.T) { } func forbiddenBackupDep(pkgPath string) string { + for _, allowed := range backupAllowedDeps { + if pkgPath == allowed { + return "" + } + } for _, forbidden := range backupForbiddenDeps { if pkgPath == forbidden || strings.HasPrefix(pkgPath, forbidden+"/") { return forbidden