diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index 59e9cf5fe..fe57ecaa1 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -72,9 +72,15 @@ 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 } // DeltaCompactorOption configures a DeltaCompactor. @@ -340,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 { @@ -349,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) } @@ -357,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_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index dc7eadebd..247ea507c 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" @@ -559,8 +560,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 { @@ -575,10 +576,20 @@ 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, []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) { @@ -606,14 +617,356 @@ 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) + require.NoError(t, err) + 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. +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) + } +} + +// 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) } +// 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 { + 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 +// 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) { + 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..2f992ea10 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -23,12 +23,13 @@ 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. 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 { - ScanGroupAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) -} - -type ttlInlineLocalLeaderGroups interface { - LocalLeaderGroupIDs() []uint64 + ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) } type ttlInlineMigrationScanPlan struct { @@ -37,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{ { @@ -179,26 +206,13 @@ 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, 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 } @@ -238,38 +252,56 @@ 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 } +// 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 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) - 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 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 whole-prefix scans") + }) +} + 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 { @@ -278,19 +310,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.ScanGroupAt(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/adapter/redis_ttl_inline_migrator_plan_test.go b/adapter/redis_ttl_inline_migrator_plan_test.go index d4ad5f43a..a83281fe8 100644 --- a/adapter/redis_ttl_inline_migrator_plan_test.go +++ b/adapter/redis_ttl_inline_migrator_plan_test.go @@ -15,8 +15,8 @@ type groupScanningStore struct { store.MVCCStore } -func (groupScanningStore) ScanGroupAt(context.Context, uint64, []byte, []byte, int, uint64) ([]*store.KVPair, error) { - return nil, nil +func (groupScanningStore) ScanGroupOwnedAt(context.Context, uint64, []byte, []byte, int, uint64) ([]*store.KVPair, []byte, error) { + return nil, nil, nil } type ttlPlanLeaderGroupsCoordinator struct { @@ -25,7 +25,7 @@ type ttlPlanLeaderGroupsCoordinator struct { groups []uint64 } -func (c ttlPlanLeaderGroupsCoordinator) LocalLeaderGroupIDs() []uint64 { return c.groups } +func (c ttlPlanLeaderGroupsCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { return c.groups, true } // In production the compactor holds WithKeyVizLabel(startupGatedCoordinator{ // sharded}); kv.CoordinatorForwarder stands in for the startup gate, which diff --git a/kv/coordinator.go b/kv/coordinator.go index f14fb86cf..460415ecc 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/coordinator_forwarder.go b/kv/coordinator_forwarder.go index 0406a2fda..9cb06d03a 100644 --- a/kv/coordinator_forwarder.go +++ b/kv/coordinator_forwarder.go @@ -28,6 +28,7 @@ var ( _ AllGroupsLeaseTimestampCoordinator = CoordinatorForwarder{} _ GroupRoutableCoordinator = CoordinatorForwarder{} _ GroupLeaderRoutableCoordinator = CoordinatorForwarder{} + _ LocalLeaderGroupsCoordinator = CoordinatorForwarder{} _ RaftMembershipCoordinator = CoordinatorForwarder{} _ TimestampAllocatorProvider = CoordinatorForwarder{} _ ConfiguredTimestampAllocatorProvider = CoordinatorForwarder{} @@ -256,11 +257,10 @@ func (f CoordinatorForwarder) EngineGroupIDForKey(key []byte) uint64 { // LocalLeaderGroupIDs lists the data groups this process leads. The Redis // delta compactor's TTL-inline migration plans one scan per group from it. -func (f CoordinatorForwarder) LocalLeaderGroupIDs() []uint64 { - if groups, ok := f.Inner.(interface{ LocalLeaderGroupIDs() []uint64 }); ok { - return groups.LocalLeaderGroupIDs() - } - return nil +// An Inner without the capability stays unknown rather than leading no group, +// so the migration falls back to whole-prefix scans instead of planning none. +func (f CoordinatorForwarder) LocalLeaderGroupIDs() ([]uint64, bool) { + return LocalLeaderGroupIDsThrough(f.Inner) } func (f CoordinatorForwarder) RaftMembers(ctx context.Context) ([]RaftMember, error) { 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/shard_store.go b/kv/shard_store.go index c52b4cfbf..0ce37ce54 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -904,6 +904,111 @@ 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 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, 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. 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: +// 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 + } + g, ok := s.groupForID(groupID) + if !ok || g == nil || g.Store == nil { + return []*store.KVPair{}, 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 + } + var resume []byte + if page.full { + resume = page.advanceKey + } + 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, nil +} + +// 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, nil + } + if route.GroupID == groupID { + return true, nil + } + if !s.legacyRouteInGroup(groupID, key) { + return false, 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 @@ -3084,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) } @@ -3103,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 @@ -3128,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 { @@ -3143,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 @@ -3172,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 diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index a119bca0a..c01dc0c9e 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" @@ -1512,6 +1513,351 @@ 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: 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 := []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("!redis|str|") + end := prefixScanEnd(prefix) + + kvs, resume, err := st.ScanGroupOwnedAt(ctx, 1, prefix, end, 10, 10) + require.NoError(t, err) + require.Nil(t, resume) + require.Equal(t, []*store.KVPair{{Key: owned, Value: []byte("owned")}}, kvs) + + 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) +} + +// 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 + + 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, 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() + + families := []struct { + name string + prefix []byte + key []byte + }{ + // 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"))}, + } + 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) + } + }) + } + } +} + +// Whether the logical owner holds the key can be answered here only by that +// 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) + 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)}) + } + prefix := []byte(store.HashMetaPrefix) + + kvs, _, err := st.ScanGroupOwnedAt(ctx, 1, prefix, prefixScanEnd(prefix), 10, 10) + require.NoError(t, err) + require.Equal(t, want, kvs) + require.Zero(t, counting.pointReads()) +} + func TestShardStoreGetGroupAt_UsesExplicitGroup(t *testing.T) { t.Parallel() 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 10da667eb..6efe3f584 100644 --- a/main.go +++ b/main.go @@ -2711,6 +2711,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) diff --git a/main_coordinator_capabilities_test.go b/main_coordinator_capabilities_test.go new file mode 100644 index 000000000..8d114231b --- /dev/null +++ b/main_coordinator_capabilities_test.go @@ -0,0 +1,335 @@ +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" +) + +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 newStartupGatedCoordinator(inner, gate) +} + +// startupWrapperSurfaces are the values capability consumers type-assert in +// the server: main probes the startup gate itself, and every adapter probes +// WithKeyVizLabel(startupGatedCoordinator). +var startupWrapperSurfaces = []startupWrapperSurface{ + {name: "startupGatedCoordinator", wrap: startupGateOver}, + { + 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") + }, +} + +// TestStartupWrappersForwardShardedCoordinatorCapabilities calls every +// ShardedCoordinator method not listed in shardedMethodsNotForwarded through +// both wrappers, against a spy, and requires the call to reach the spy and +// return its result. TestAdapterCoordinatorWrappersForwardShardedCapabilities +// checks only that the methods exist. A method can exist and still not +// forward, when the wrapper probes its inner coordinator through an interface +// that no longer matches and answers with its own fallback. That is how +// LocalLeaderGroupIDs stopped the Redis TTL-inline migration while unit tests +// on an unwrapped coordinator stayed green. 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 shardedMethodsNotForwarded { + _, ok := shardedType.MethodByName(name) + require.Truef(t, ok, "ShardedCoordinator.%s no longer exists; drop it from shardedMethodsNotForwarded", 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 + } + if _, notForwarded := shardedMethodsNotForwarded[name]; notForwarded { + continue + } + spyMethod := spyMethods.MethodByName(name) + require.Truef(t, spyMethod.IsValid() && spyMethod.Type() == sig, + "shardedCapabilitySpy.%s must mirror ShardedCoordinator.%s %s", name, name, sig) + t.Run(name, func(t *testing.T) { + t.Parallel() + for _, s := range startupWrapperSurfaces { + 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 +} diff --git a/main_coordinator_chain_test.go b/main_coordinator_chain_test.go index fe796b002..d482813c0 100644 --- a/main_coordinator_chain_test.go +++ b/main_coordinator_chain_test.go @@ -19,7 +19,7 @@ type localLeaderGroupsCoordinator struct { groups []uint64 } -func (c localLeaderGroupsCoordinator) LocalLeaderGroupIDs() []uint64 { return c.groups } +func (c localLeaderGroupsCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { return c.groups, true } // Every adapter receives WithKeyVizLabel(startupGatedCoordinator{sharded}). The // Redis delta compactor's TTL-inline migration plans one scan per group this @@ -32,9 +32,9 @@ func TestAdapterCoordinatorChainForwardsLocalLeaderGroups(t *testing.T) { inner := localLeaderGroupsCoordinator{groups: []uint64{1, 3}} chain := kv.WithKeyVizLabel(newStartupGatedCoordinator(inner, &startupPublicKVGate{}), keyviz.LabelRedis) - groups, ok := chain.(interface{ LocalLeaderGroupIDs() []uint64 }) - require.True(t, ok, "the adapter coordinator chain must expose LocalLeaderGroupIDs") - require.Equal(t, []uint64{1, 3}, groups.LocalLeaderGroupIDs()) + groups, known := kv.LocalLeaderGroupIDsThrough(chain) + require.True(t, known, "the adapter coordinator chain must report LocalLeaderGroupIDs") + require.Equal(t, []uint64{1, 3}, groups) } // shardedMethodsNotForwarded lists the *kv.ShardedCoordinator methods the diff --git a/main_redis_ttl_migration_wiring_test.go b/main_redis_ttl_migration_wiring_test.go new file mode 100644 index 000000000..6536f27e5 --- /dev/null +++ b/main_redis_ttl_migration_wiring_test.go @@ -0,0 +1,113 @@ +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 ScanGroupOwnedAt is what lets the +// TTL-inline migrator plan one scan per locally led group instead of a +// whole-prefix 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) 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() + 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) { + 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} + ids, known := coord.LocalLeaderGroupIDs() + require.True(t, known) + require.Equal(t, localLeaders, ids) + + gate := &startupPublicKVGate{} + gate.markReady() + gated := newStartupGatedCoordinator(coord, gate) + + 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()} + 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 whole-prefix scan when the groups are known") +}