From d9b16518e1d97a72c914c224a9fcb4407b8462dd Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 11:36:54 +0000 Subject: [PATCH 1/9] monitoring: export Redis TTL-inline migration passes Add a passes counter by anchor type, data group and outcome, and the last clean pass time per type and group, for the delta compactor to report into. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- monitoring/redis_ttl_inline.go | 66 +++++++++++++++++++++++++++++ monitoring/redis_ttl_inline_test.go | 42 ++++++++++++++++++ monitoring/registry.go | 49 ++++++++++++--------- 3 files changed, 138 insertions(+), 19 deletions(-) create mode 100644 monitoring/redis_ttl_inline.go create mode 100644 monitoring/redis_ttl_inline_test.go diff --git a/monitoring/redis_ttl_inline.go b/monitoring/redis_ttl_inline.go new file mode 100644 index 000000000..0e8e2c42d --- /dev/null +++ b/monitoring/redis_ttl_inline.go @@ -0,0 +1,66 @@ +package monitoring + +import ( + "strconv" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + redisTTLInlinePassClean = "clean" + redisTTLInlinePassNotClean = "not_clean" +) + +// RedisTTLInlineMigrationObserver records the passes the Redis TTL-inline +// migrator completes. A pass covers one anchor type (typeName, such as +// "ttl-inline-string") in one data group; group 0 is the whole-prefix scan the +// migrator falls back to when the coordinator does not report its groups. +type RedisTTLInlineMigrationObserver interface { + ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) +} + +// RedisTTLInlineMigrationMetrics exports the TTL-inline migrator's passes. +// Labels are the anchor type and data group only: eight types times the data +// groups a node leads. +type RedisTTLInlineMigrationMetrics struct { + passes *prometheus.CounterVec + lastCleanPass *prometheus.GaugeVec +} + +func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisTTLInlineMigrationMetrics { + m := &RedisTTLInlineMigrationMetrics{ + passes: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_redis_ttl_inline_migration_passes_total", + Help: "Completed Redis TTL-inline migration passes by anchor type, data group, and outcome. A clean pass read the whole prefix and found every anchor already in the inline format.", + }, + []string{"type", "group", "outcome"}, + ), + lastCleanPass: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds", + Help: "Unix time at which this node last completed a clean Redis TTL-inline migration pass, by anchor type and data group.", + }, + []string{"type", "group"}, + ), + } + if registerer != nil { + registerer.MustRegister(m.passes, m.lastCleanPass) + } + return m +} + +// ObserveRedisTTLInlineMigrationPass counts a completed pass and, when it was +// clean, stamps the type and group's last clean pass time. +func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) { + if m == nil { + return + } + group := strconv.FormatUint(groupID, 10) + outcome := redisTTLInlinePassNotClean + if clean { + outcome = redisTTLInlinePassClean + m.lastCleanPass.WithLabelValues(typeName, group).SetToCurrentTime() + } + m.passes.WithLabelValues(typeName, group, outcome).Inc() +} diff --git a/monitoring/redis_ttl_inline_test.go b/monitoring/redis_ttl_inline_test.go new file mode 100644 index 000000000..5075f7aaa --- /dev/null +++ b/monitoring/redis_ttl_inline_test.go @@ -0,0 +1,42 @@ +package monitoring + +import ( + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +func TestRedisTTLInlineMigrationMetricsExportPasses(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + metrics := newRedisTTLInlineMigrationMetrics(reg) + + before := float64(time.Now().Unix()) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, false) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, true) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, true) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, false) + + require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "not_clean"))) + require.Equal(t, 2.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "clean"))) + require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-hash", "3", "not_clean"))) + require.GreaterOrEqual(t, testutil.ToFloat64(metrics.lastCleanPass.WithLabelValues("ttl-inline-string", "2")), before) + require.Equal(t, 1, testutil.CollectAndCount(metrics.lastCleanPass), + "only a clean pass stamps the last clean pass time") + + families, err := reg.Gather() + require.NoError(t, err) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_passes_total")) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds")) +} + +func TestRedisTTLInlineMigrationObserverIsNilSafe(t *testing.T) { + t.Parallel() + var metrics *RedisTTLInlineMigrationMetrics + require.NotPanics(t, func() { metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, true) }) + var r *Registry + require.Nil(t, r.RedisTTLInlineMigrationObserver()) +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 159bbe822..2ae7698dd 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -14,25 +14,26 @@ type Registry struct { registerer prometheus.Registerer gatherer prometheus.Gatherer - dynamo *DynamoDBMetrics - redis *RedisMetrics - raft *RaftMetrics - lua *LuaMetrics - hotPath *HotPathMetrics - pebble *PebbleMetrics - writeConflict *WriteConflictMetrics - sqs *SQSMetrics - sqsObserver *SQSObserver - fs *FileSystemMetrics - s3 *S3Metrics - hlc *HLCMetrics - hlcObserver *HLCObserver - coldStart *ColdStartMetrics - coldStartObs *ColdStartObserver - tso *TSOMetrics - tsoObserver *TSOObserver - encryption *EncryptionMetrics - encryptionObs *EncryptionStateObserver + dynamo *DynamoDBMetrics + redis *RedisMetrics + raft *RaftMetrics + lua *LuaMetrics + hotPath *HotPathMetrics + pebble *PebbleMetrics + writeConflict *WriteConflictMetrics + sqs *SQSMetrics + sqsObserver *SQSObserver + fs *FileSystemMetrics + s3 *S3Metrics + hlc *HLCMetrics + hlcObserver *HLCObserver + coldStart *ColdStartMetrics + coldStartObs *ColdStartObserver + tso *TSOMetrics + tsoObserver *TSOObserver + encryption *EncryptionMetrics + encryptionObs *EncryptionStateObserver + redisTTLInline *RedisTTLInlineMigrationMetrics } // NewRegistry builds a registry with constant labels that identify the local node. @@ -67,6 +68,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry { r.tsoObserver = newTSOObserver(r.tso) r.encryption = newEncryptionMetrics(registerer) r.encryptionObs = newEncryptionStateObserver(r.encryption) + r.redisTTLInline = newRedisTTLInlineMigrationMetrics(registerer) return r } @@ -288,6 +290,15 @@ func (r *Registry) HLCObserver() *HLCObserver { return r.hlcObserver } +// RedisTTLInlineMigrationObserver returns the observer the Redis delta +// compactor reports its completed TTL-inline migration passes to. +func (r *Registry) RedisTTLInlineMigrationObserver() RedisTTLInlineMigrationObserver { + if r == nil || r.redisTTLInline == nil { + return nil + } + return r.redisTTLInline +} + // TSOObserver returns the shared runtime TSO observer. The kv package consumes // it through its narrow observer interface, keeping Prometheus ownership here. func (r *Registry) TSOObserver() *TSOObserver { From e5c69daf206059df9ad6c695b1f2b435da87cf12 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 11:36:54 +0000 Subject: [PATCH 2/9] adapter, main: report completed TTL-inline migration passes Each scan plan (anchor type, data group) now tallies a pass from the first page of its prefix to the page that reaches the end: anchors scanned, migrated, skipped as not led here, and failed to build. When the pass ends the migrator logs "ttl inline migrator: pass complete" with stable keys and reports it to the observer main wires from the metrics registry. A pass is clean when it started at the top of the prefix in this leadership term and migrated, skipped and failed nothing. A plan that drops out of the tick's plans (its group's leadership moved) loses its open pass but keeps its cursor, so the pass it resumes is not full and cannot be clean. A page retried after a failed batch counts once, plus the keys its committed batches rewrote. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor.go | 28 +- adapter/redis_delta_compactor_test.go | 290 +++++++++++++++++++++ adapter/redis_ttl_inline_migration_pass.go | 164 ++++++++++++ adapter/redis_ttl_inline_migrator.go | 82 ++++-- main.go | 10 +- main_background_env_test.go | 4 +- 6 files changed, 555 insertions(+), 23 deletions(-) create mode 100644 adapter/redis_ttl_inline_migration_pass.go diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index fe57ecaa1..06120e42c 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -10,6 +10,7 @@ import ( "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/monitoring" "github.com/bootjp/elastickv/store" "github.com/cockroachdb/errors" ) @@ -65,6 +66,12 @@ type DeltaCompactor struct { // called from a test while Run is active). cursorMu sync.Mutex cursors map[string][]byte + // ttlInlinePlans tracks each TTL-inline scan plan's pass, keyed like + // cursors by the plan's cursor name. Guarded by cursorMu so a pass moves + // together with its plan's cursor. + ttlInlinePlans map[string]*ttlInlineMigrationPlanState + // ttlInlineObserver, when set, receives every completed TTL-inline pass. + ttlInlineObserver monitoring.RedisTTLInlineMigrationObserver // urgentCh receives requests to immediately compact a single key whose // delta count has exceeded MaxDeltaScanLimit, making reads fail. The Run @@ -134,6 +141,14 @@ func WithDeltaCompactorLogger(l *slog.Logger) DeltaCompactorOption { } } +// WithDeltaCompactorTTLInlineMigrationObserver reports every pass the +// TTL-inline migration completes over an anchor type and data group. +func WithDeltaCompactorTTLInlineMigrationObserver(o monitoring.RedisTTLInlineMigrationObserver) DeltaCompactorOption { + return func(c *DeltaCompactor) { + c.ttlInlineObserver = o + } +} + // NewDeltaCompactor creates a DeltaCompactor that operates on st using coord. func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCompactorOption) *DeltaCompactor { c := &DeltaCompactor{ @@ -145,6 +160,7 @@ func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCo timeout: defaultDeltaCompactorTimeout, timeoutBackoff: defaultDeltaCompactorTimeoutBackoff, cursors: make(map[string][]byte), + ttlInlinePlans: make(map[string]*ttlInlineMigrationPlanState), urgentCh: make(chan urgentCompactionRequest, urgentCompactionChanSize), } for _, opt := range opts { @@ -490,11 +506,19 @@ func (c *DeltaCompactor) scanDeltaRaw(ctx context.Context, h collectionDeltaHand func (c *DeltaCompactor) advanceCursor(typeName string, lastScannedKey []byte, truncated bool) { c.cursorMu.Lock() defer c.cursorMu.Unlock() + c.advanceCursorLocked(typeName, lastScannedKey, truncated) +} + +// advanceCursorLocked is advanceCursor for callers holding cursorMu. It +// reports whether the cursor was reset, i.e. the scan reached the end of the +// prefix and the next one starts from the beginning. +func (c *DeltaCompactor) advanceCursorLocked(typeName string, lastScannedKey []byte, truncated bool) (reset bool) { if truncated && len(lastScannedKey) > 0 { c.cursors[typeName] = bytes.Clone(lastScannedKey) - } else { - delete(c.cursors, typeName) + return false } + delete(c.cursors, typeName) + return true } // compactHandler scans the delta prefix for h, groups entries by userKey, diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 247ea507c..d43eb471e 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -4,6 +4,9 @@ import ( "bytes" "context" "encoding/binary" + "encoding/json" + "fmt" + "log/slog" "strconv" "sync" "testing" @@ -13,6 +16,7 @@ import ( "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" "github.com/stretchr/testify/require" ) @@ -2279,3 +2283,289 @@ func TestListInlineMetaCompaction(t *testing.T) { require.NoError(t, err) require.Empty(t, remaining, "all list delta keys must be removed after inline compaction") } + +// ttlInlinePassTestCoordinator leads the data groups in groupIDs and, when +// leadsKeys is set, every key. Its failDispatch-th non-empty dispatch fails. +type ttlInlinePassTestCoordinator struct { + *localAdapterCoordinator + + mu sync.Mutex + groupIDs []uint64 + leadsKeys bool + dispatches int + failDispatch int +} + +func (c *ttlInlinePassTestCoordinator) IsLeader() bool { return false } + +func (c *ttlInlinePassTestCoordinator) IsLeaderForKey([]byte) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.leadsKeys +} + +func (c *ttlInlinePassTestCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return append([]uint64(nil), c.groupIDs...), true +} + +func (c *ttlInlinePassTestCoordinator) setGroupIDs(ids []uint64) { + c.mu.Lock() + defer c.mu.Unlock() + c.groupIDs = ids +} + +func (c *ttlInlinePassTestCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + if req != nil && len(req.Elems) > 0 { + c.mu.Lock() + c.dispatches++ + fail := c.dispatches == c.failDispatch + c.mu.Unlock() + if fail { + return nil, errors.New("injected dispatch failure") + } + } + return c.localAdapterCoordinator.Dispatch(ctx, req) +} + +type ttlInlinePassObservation struct { + typeName string + groupID uint64 + clean bool +} + +type recordingTTLInlinePassObserver struct { + mu sync.Mutex + passes []ttlInlinePassObservation +} + +func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) { + o.mu.Lock() + defer o.mu.Unlock() + o.passes = append(o.passes, ttlInlinePassObservation{typeName: typeName, groupID: groupID, clean: clean}) +} + +func (o *recordingTTLInlinePassObserver) forType(typeName string) []ttlInlinePassObservation { + o.mu.Lock() + defer o.mu.Unlock() + var out []ttlInlinePassObservation + for _, p := range o.passes { + if p.typeName == typeName { + out = append(out, p) + } + } + return out +} + +type lockedLogBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + n, err := b.buf.Write(p) + return n, errors.WithStack(err) +} + +// ttlInlinePassLine is the part of a "pass complete" log line the tests check. +type ttlInlinePassLine struct { + Level string `json:"level"` + Type string `json:"type"` + GroupID uint64 `json:"group_id"` + Clean bool `json:"clean"` + FullPass bool `json:"full_pass"` + Pages int `json:"pages"` + KeysScanned int `json:"keys_scanned"` + KeysMigrated int `json:"keys_migrated"` + KeysSkipped int `json:"keys_skipped"` + KeysFailed int `json:"keys_failed"` +} + +func (b *lockedLogBuffer) ttlInlinePasses(t *testing.T, typeName string) []ttlInlinePassLine { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + var out []ttlInlinePassLine + for _, line := range bytes.Split(b.buf.Bytes(), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec struct { + Msg string `json:"msg"` + ttlInlinePassLine + } + require.NoError(t, json.Unmarshal(line, &rec)) + if rec.Msg == "ttl inline migrator: pass complete" && rec.Type == typeName { + out = append(out, rec.ttlInlinePassLine) + } + } + return out +} + +func seedTTLInlinePassStrings(t *testing.T, st store.MVCCStore, n uint64, legacy bool) { + t.Helper() + ctx := context.Background() + for i := uint64(0); i < n; i++ { + key := []byte(fmt.Sprintf("ttl:pass:%04d", i)) + value := []byte("value") + if !legacy { + value = encodeRedisStr(value, nil) + } + require.NoError(t, st.PutAt(ctx, redisStrKey(key), value, i+1, 0)) + } +} + +// The migrator reports each pass a scan plan completes over its prefix, so an +// operator can tell from a clean pass on every anchor type and group that the +// legacy TTL backlog has drained. A pass runs from the first page of the +// prefix to the first page that is not truncated, its counters restart with +// the next pass, and a pass that resumed a cursor from an earlier leadership +// term, migrated keys, or left anchors unchecked is not clean. +func TestDeltaCompactor_TTLInlineReportsCompletedPasses(t *testing.T) { + t.Parallel() + + const typeName = "ttl-inline-string" + group1 := []uint64{1} + type pass struct { + level string + clean bool + full bool + pages int + keysScanned int + keysMigrated int + keysSkipped int + keysFailed int + } + clean := func(level string, pages, scanned int) pass { + return pass{level: level, clean: true, full: true, pages: pages, keysScanned: scanned} + } + tests := []struct { + name string + seed func(*testing.T, store.MVCCStore) + notLeader bool + failDispatch int + // ticks lists the groups this node leads at each tick. + ticks [][]uint64 + want []pass + }{ + { + name: "clean pass", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, false) }, + ticks: [][]uint64{group1}, + want: []pass{clean("INFO", 1, 3)}, + }, + { + name: "pass that migrated keys, then clean passes", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, true) }, + ticks: [][]uint64{group1, group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 1, keysScanned: 3, keysMigrated: 3}, + clean("INFO", 1, 3), + clean("DEBUG", 1, 3), + }, + }, + { + name: "pass spanning several ticks", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationTickScanLimit+88, true) + }, + ticks: [][]uint64{group1, group1, group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 2, keysScanned: ttlInlineMigrationTickScanLimit + 88, keysMigrated: ttlInlineMigrationTickScanLimit + 88}, + clean("INFO", 2, ttlInlineMigrationTickScanLimit+88), + }, + }, + { + name: "leadership moves away mid-pass and back", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationTickScanLimit+88, false) + }, + ticks: [][]uint64{group1, {}, group1, group1, group1}, + want: []pass{ + {level: "INFO", full: false, pages: 1, keysScanned: 88}, + clean("INFO", 2, ttlInlineMigrationTickScanLimit+88), + }, + }, + { + name: "anchors this node does not lead", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, false) }, + notLeader: true, + ticks: [][]uint64{group1}, + want: []pass{{level: "INFO", full: true, pages: 1, keysScanned: 3, keysSkipped: 3}}, + }, + { + name: "anchor whose rewrite cannot be built", + seed: func(t *testing.T, st store.MVCCStore) { + ctx := context.Background() + require.NoError(t, st.PutAt(ctx, redisStrKey([]byte("ttl:pass:bad")), []byte("value"), 1, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey([]byte("ttl:pass:bad")), []byte{1}, 2, 0)) + }, + ticks: [][]uint64{group1}, + want: []pass{{level: "INFO", full: true, pages: 1, keysScanned: 1, keysFailed: 1}}, + }, + { + // The first batch of 32 commits and the second fails. The next + // tick reads the page again and rewrites the 33rd key; the pass + // counts the page once and both batches' keys. + name: "page retried after a failed batch", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationBatchKeyLimit+1, true) + }, + failDispatch: 2, + ticks: [][]uint64{group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 1, keysScanned: ttlInlineMigrationBatchKeyLimit + 1, keysMigrated: ttlInlineMigrationBatchKeyLimit + 1}, + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := &recordingTTLInlineGroupScanStore{MVCCStore: store.NewMVCCStore()} + tc.seed(t, st) + coord := &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + leadsKeys: !tc.notLeader, + failDispatch: tc.failDispatch, + } + logs := &lockedLogBuffer{} + observer := &recordingTTLInlinePassObserver{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorLogger(slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + WithDeltaCompactorTTLInlineMigrationObserver(observer), + ) + + for _, groups := range tc.ticks { + coord.setGroupIDs(groups) + err := c.SyncOnce(ctx) + if tc.failDispatch == 0 { + require.NoError(t, err) + } + } + + lines := logs.ttlInlinePasses(t, typeName) + got := make([]pass, 0, len(lines)) + for _, l := range lines { + require.Equal(t, uint64(1), l.GroupID) + got = append(got, pass{ + level: l.Level, clean: l.Clean, full: l.FullPass, pages: l.Pages, + keysScanned: l.KeysScanned, keysMigrated: l.KeysMigrated, + keysSkipped: l.KeysSkipped, keysFailed: l.KeysFailed, + }) + } + require.Equal(t, tc.want, got) + + wantObserved := make([]ttlInlinePassObservation, 0, len(tc.want)) + for _, p := range tc.want { + wantObserved = append(wantObserved, ttlInlinePassObservation{typeName: typeName, groupID: 1, clean: p.clean}) + } + require.Equal(t, wantObserved, observer.forType(typeName)) + }) + } +} diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go new file mode 100644 index 000000000..8050afe4a --- /dev/null +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -0,0 +1,164 @@ +package adapter + +import ( + "context" + "log/slog" + "time" +) + +// ttlInlineMigrationTally counts what the TTL-inline migrator did with the +// anchors it read. +type ttlInlineMigrationTally struct { + // keysScanned counts the anchors read, i.e. the rows whose user key the + // handler could extract. + keysScanned int + // keysMigrated counts the anchors rewritten by a dispatched batch. + keysMigrated int + // keysSkipped counts the anchors left alone because this node did not + // lead their user key's route. + keysSkipped int + // keysFailed counts the anchors whose rewrite could not be built. + keysFailed int +} + +func (t *ttlInlineMigrationTally) add(o ttlInlineMigrationTally) { + t.keysScanned += o.keysScanned + t.keysMigrated += o.keysMigrated + t.keysSkipped += o.keysSkipped + t.keysFailed += o.keysFailed +} + +// ttlInlineMigrationPass is one pass of a scan plan over its prefix: the pages +// from the start of the prefix through the page that reaches its end. +type ttlInlineMigrationPass struct { + ttlInlineMigrationTally + // full reports that the pass started at the beginning of the prefix in + // this leadership term. A pass that resumes a cursor left by an earlier + // term is not full: this term never read the anchors before the cursor. + full bool + pages int + startedAt time.Time +} + +// clean reports that the pass read every anchor of its plan and found each one +// already in the inline format: it read the whole prefix, rewrote nothing, and +// left nothing unchecked. +func (p ttlInlineMigrationPass) clean() bool { + return p.full && p.keysMigrated == 0 && p.keysSkipped == 0 && p.keysFailed == 0 +} + +// ttlInlineMigrationPlanState is a scan plan's pass bookkeeping. +type ttlInlineMigrationPlanState struct { + // pass is the open pass, nil between passes. + pass *ttlInlineMigrationPass + // lastClean reports that the plan's previous pass was clean. It keeps + // repeated clean passes out of the info log. + lastClean bool +} + +// openTTLInlineMigrationPassLocked returns the open pass of the plan named +// name, starting one if needed. fromStart reports that the page being recorded +// starts at the beginning of the prefix; such a page always opens a new pass +// unless the open pass has not completed a page yet. Callers hold cursorMu. +func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, fromStart bool) *ttlInlineMigrationPass { + state := c.ttlInlinePlans[name] + if state == nil { + state = &ttlInlineMigrationPlanState{} + c.ttlInlinePlans[name] = state + } + if state.pass == nil || (fromStart && state.pass.pages > 0) { + state.pass = &ttlInlineMigrationPass{full: fromStart, startedAt: time.Now()} + } + return state.pass +} + +// noteTTLInlineMigrationPageFailed adds the keys a failed page already +// rewrote to its plan's pass. The page itself does not count: its cursor stays, +// and the next tick reads it again. +func (c *DeltaCompactor) noteTTLInlineMigrationPageFailed(name string, fromStart bool, migrated int) { + if migrated == 0 { + return + } + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + c.openTTLInlineMigrationPassLocked(name, fromStart).keysMigrated += migrated +} + +// finishTTLInlineMigrationPage adds a migrated page to its plan's pass and +// moves the plan's cursor past it. When the page reached the end of the prefix +// it closes the pass and returns it with done set, and whether the plan's +// previous pass was clean. +func (c *DeltaCompactor) finishTTLInlineMigrationPage( + name string, + page ttlInlineMigrationPage, + tally ttlInlineMigrationTally, +) (pass ttlInlineMigrationPass, prevClean bool, done bool) { + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + open := c.openTTLInlineMigrationPassLocked(name, page.fromStart) + open.add(tally) + open.pages++ + if !c.advanceCursorLocked(name, page.cursorKey, page.truncated) { + return ttlInlineMigrationPass{}, false, false + } + state := c.ttlInlinePlans[name] + pass, prevClean = *open, state.lastClean + state.pass = nil + state.lastClean = pass.clean() + return pass, prevClean, true +} + +// pruneTTLInlineMigrationPlanStates drops the bookkeeping of plans this tick +// no longer has, such as a group whose leadership moved to another node. The +// plan's cursor stays, so migration resumes where it stopped if the group +// comes back; the pass it resumes is not full and so cannot be clean. +func (c *DeltaCompactor) pruneTTLInlineMigrationPlanStates(plans []ttlInlineMigrationTickPlan) { + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + if len(c.ttlInlinePlans) == 0 { + return + } + live := make(map[string]struct{}, len(plans)) + for _, p := range plans { + live[p.plan.cursorName] = struct{}{} + } + for name := range c.ttlInlinePlans { + if _, ok := live[name]; !ok { + delete(c.ttlInlinePlans, name) + } + } +} + +// reportTTLInlineMigrationPass logs a completed pass and hands it to the +// observer. A pass that is not clean, or the first clean pass after one that +// was not, is logged at info; repeated clean passes are logged at debug so an +// idle migrator does not log every tick. group_id 0 is the whole-prefix scan +// used when the coordinator does not report its local leader groups. +func (c *DeltaCompactor) reportTTLInlineMigrationPass( + ctx context.Context, + h ttlInlineMigrationHandler, + plan ttlInlineMigrationScanPlan, + pass ttlInlineMigrationPass, + prevClean bool, +) { + clean := pass.clean() + level := slog.LevelInfo + if clean && prevClean { + level = slog.LevelDebug + } + c.logger.Log(ctx, level, "ttl inline migrator: pass complete", + "type", h.typeName, + "group_id", plan.groupID, + "clean", clean, + "full_pass", pass.full, + "pages", pass.pages, + "keys_scanned", pass.keysScanned, + "keys_migrated", pass.keysMigrated, + "keys_skipped", pass.keysSkipped, + "keys_failed", pass.keysFailed, + "duration", time.Since(pass.startedAt), + ) + if c.ttlInlineObserver != nil { + c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, clean) + } +} diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 2f992ea10..728fdd4eb 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -51,6 +51,7 @@ type ttlInlineMigrationTickPlan struct { // and groups whenever the earlier plans used up the budget. func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64) error { plans := c.ttlInlineMigrationTickPlans() + c.pruneTTLInlineMigrationPlanStates(plans) start := c.rotationStart(&c.ttlInlinePlanCursor, len(plans)) var combined error for offset := range plans { @@ -212,48 +213,83 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( plan ttlInlineMigrationScanPlan, readTS uint64, ) error { - kvs, cursorKey, truncated, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) + page, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) if err != nil { return err } + tally, err := c.migrateTTLInlinePage(ctx, h, page.kvs, readTS) + if err != nil { + // The next tick reads this page again, so only the keys its earlier + // batches already rewrote count toward the pass. + c.noteTTLInlineMigrationPageFailed(plan.cursorName, page.fromStart, tally.keysMigrated) + return err + } + if pass, prevClean, done := c.finishTTLInlineMigrationPage(plan.cursorName, page, tally); done { + c.reportTTLInlineMigrationPass(ctx, h, plan, pass, prevClean) + } + return nil +} + +// migrateTTLInlinePage rewrites the page's legacy anchors in batches of at most +// ttlInlineMigrationBatchKeyLimit keys and tallies what it did with each +// anchor. On a dispatch error the tally's keysMigrated still counts the keys +// of the batches dispatched before it. +func (c *DeltaCompactor) migrateTTLInlinePage( + ctx context.Context, + h ttlInlineMigrationHandler, + kvs []*store.KVPair, + readTS uint64, +) (ttlInlineMigrationTally, error) { var ( + tally ttlInlineMigrationTally elems []*kv.Elem[kv.OP] batchKeys int + rewrites int flushBatch = func() error { if err := c.dispatchCompaction(ctx, readTS, elems); err != nil { c.logger.WarnContext(ctx, "ttl inline migrator: batch dispatch failed", "type", h.typeName, "error", err) return err } + tally.keysMigrated += rewrites elems = nil batchKeys = 0 + rewrites = 0 return nil } ) for _, pair := range kvs { userKey := h.extractUserKey(pair.Key) - if userKey == nil || !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { + if userKey == nil { + continue + } + tally.keysScanned++ + if !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { + tally.keysSkipped++ continue } built, buildErr := h.buildElems(ctx, pair, readTS) if buildErr != nil { + tally.keysFailed++ c.logger.WarnContext(ctx, "ttl inline migrator: failed to build elems", "type", h.typeName, "key", string(userKey), "error", buildErr) continue } + if len(built) > 0 { + rewrites++ + } elems = append(elems, built...) batchKeys++ if batchKeys >= ttlInlineMigrationBatchKeyLimit { if err := flushBatch(); err != nil { - return err + return tally, err } } } - if err := flushBatch(); err != nil { - return err - } - c.advanceCursor(plan.cursorName, cursorKey, truncated) - return nil + // flushBatch adds the last batch's keys to tally, so call it before + // returning tally. + err := flushBatch() + return tally, err } // ttlInlineMigrationScanPlans scans only the groups this node leads when the @@ -294,34 +330,48 @@ 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. +// ttlInlineMigrationPage is one page a scan plan read in a tick. +type ttlInlineMigrationPage struct { + kvs []*store.KVPair + // cursorKey is the key the plan's cursor moves to. + cursorKey []byte + // truncated reports that the scan stopped before the end of the prefix. + truncated bool + // fromStart reports that the page starts at the beginning of the prefix, + // so it is the first page of a pass. + fromStart bool +} + func (c *DeltaCompactor) scanTTLInlineMigrationRaw( ctx context.Context, h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, -) ([]*store.KVPair, []byte, bool, error) { +) (ttlInlineMigrationPage, error) { c.cursorMu.Lock() start := h.prefix - if cur := c.cursors[plan.cursorName]; len(cur) > 0 { + cur := c.cursors[plan.cursorName] + if len(cur) > 0 { start = append(bytes.Clone(cur), cursorNextByte) } c.cursorMu.Unlock() + page := ttlInlineMigrationPage{fromStart: len(cur) == 0} end := store.PrefixScanEnd(h.prefix) if plan.scanner != nil { kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) if err != nil { - return nil, nil, false, errors.WithStack(err) + return ttlInlineMigrationPage{}, errors.WithStack(err) } - return kvs, resume, resume != nil, nil + page.kvs, page.cursorKey, page.truncated = kvs, resume, resume != nil + return page, nil } kvs, err := c.st.ScanAt(ctx, start, end, ttlInlineMigrationTickScanLimit, readTS) if err != nil { - return nil, nil, false, errors.WithStack(err) + return ttlInlineMigrationPage{}, errors.WithStack(err) } - return kvs, lastKVKey(kvs), len(kvs) == ttlInlineMigrationTickScanLimit, nil + page.kvs, page.cursorKey, page.truncated = kvs, lastKVKey(kvs), len(kvs) == ttlInlineMigrationTickScanLimit + return page, nil } func (c *DeltaCompactor) migrateStringTTLInlineElems(ctx context.Context, pair *store.KVPair, readTS uint64) ([]*kv.Elem[kv.OP], error) { diff --git a/main.go b/main.go index 6efe3f584..e1e23364b 100644 --- a/main.go +++ b/main.go @@ -408,9 +408,13 @@ func startFSMCompactorIfEnabled(ctx context.Context, eg *errgroup.Group, runtime slog.Info("fsm compactor disabled", "env", fsmCompactorEnabledEnvVar) } -func newRedisDeltaCompactorIfEnabled(shardStore *kv.ShardStore, coordinate kv.Coordinator) *adapter.DeltaCompactor { +func newRedisDeltaCompactorIfEnabled(shardStore *kv.ShardStore, coordinate kv.Coordinator, metricsRegistry *monitoring.Registry) *adapter.DeltaCompactor { if enabledEnv(redisDeltaCompactorEnabledEnvVar) { - return adapter.NewDeltaCompactor(shardStore, coordinate) + var opts []adapter.DeltaCompactorOption + if obs := metricsRegistry.RedisTTLInlineMigrationObserver(); obs != nil { + opts = append(opts, adapter.WithDeltaCompactorTTLInlineMigrationObserver(obs)) + } + return adapter.NewDeltaCompactor(shardStore, coordinate, opts...) } slog.Info("redis delta compactor disabled", "env", redisDeltaCompactorEnabledEnvVar) return nil @@ -3482,7 +3486,7 @@ func prepareRedisServer(ctx context.Context, lc *net.ListenConfig, redisAddr str if err != nil { return nil, nil, nil, errors.Wrapf(err, "failed to listen on %s", redisAddr) } - deltaCompactor := newRedisDeltaCompactorIfEnabled(shardStore, coordinate) + deltaCompactor := newRedisDeltaCompactorIfEnabled(shardStore, coordinate, metricsRegistry) redisOpts := []adapter.RedisServerOption{ adapter.WithRedisActiveTimestampTracker(readTracker), adapter.WithRedisRequestObserver(metricsRegistry.RedisObserver()), diff --git a/main_background_env_test.go b/main_background_env_test.go index 17792ca0d..4f6461017 100644 --- a/main_background_env_test.go +++ b/main_background_env_test.go @@ -28,14 +28,14 @@ func TestEnabledEnv(t *testing.T) { func TestRedisDeltaCompactorEnv(t *testing.T) { t.Run("default enabled", func(t *testing.T) { t.Setenv(redisDeltaCompactorEnabledEnvVar, "") - if newRedisDeltaCompactorIfEnabled(nil, nil) == nil { + if newRedisDeltaCompactorIfEnabled(nil, nil, nil) == nil { t.Fatal("newRedisDeltaCompactorIfEnabled() = nil, want compactor") } }) t.Run("false disables", func(t *testing.T) { t.Setenv(redisDeltaCompactorEnabledEnvVar, "false") - if newRedisDeltaCompactorIfEnabled(nil, nil) != nil { + if newRedisDeltaCompactorIfEnabled(nil, nil, nil) != nil { t.Fatal("newRedisDeltaCompactorIfEnabled() returned compactor, want nil") } }) From 41548be6835c48f7bcb84db908c1f7d0f5707fac Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 11:36:54 +0000 Subject: [PATCH 3/9] docs: gate TTL-inline Phase 3 on clean migration passes Document the pass report and the condition to check before disabling the legacy !redis|ttl| read fallback, and what a clean pass does not cover. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- ...2026_04_17_implemented_ttl_inline_value.md | 64 ++++++++++++++++++- 1 file changed, 63 insertions(+), 1 deletion(-) diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index b081b9052..80651213d 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -315,6 +315,68 @@ from `ttlAt()` after the anchor-specific probes miss. The default remains fallback-enabled so rolling upgrades and not-yet-migrated stores preserve behavior. +"All nodes have migrated" is decided from the migrator's pass reports, not +from elapsed time. + +**What a pass is.** The migrator scans one plan per anchor type and per data +group the node leads. There are eight anchor types: `ttl-inline-string`, +`-hll`, `-indexed-collection`, `-list`, `-hash`, `-set`, `-zset` and +`-stream`. A plan reads at most 512 rows a tick, so a pass over a large prefix +spans several ticks. A pass runs from the first page of the prefix to the first +page that reaches its end. It closes on that page, and the next pass starts +again at the top with fresh counters. + +**What each pass reports.** The leader running the plan logs one line per +completed pass, `ttl inline migrator: pass complete`, with these keys: + +| Key | Meaning | +|---|---| +| `type` | Anchor type. | +| `group_id` | Data group. `0` is the whole-prefix fallback scan, used only when the coordinator does not report its local leader groups. | +| `full_pass` | The pass started at the top of the prefix in this leadership term. A plan that loses its group mid-pass drops its counters. If the group comes back, the resumed pass is not full. | +| `pages` | Pages the pass read. | +| `keys_scanned` | Anchors the pass read. | +| `keys_migrated` | Anchors rewritten by committed batches. A page retried after a failed batch still counts the keys its earlier batches rewrote. | +| `keys_skipped` | Anchors left alone because this node did not lead their user key's route. | +| `keys_failed` | Anchors whose rewrite could not be built. | +| `clean` | `full_pass` is true and the other three counts are all zero. | +| `duration` | How long the pass took. | + +A pass that is not clean, and the first clean pass after one that was not, are +logged at info. Repeated clean passes are logged at debug. + +The same outcome is exported per node with labels `type` and `group`: +- `elastickv_redis_ttl_inline_migration_passes_total{outcome="clean"|"not_clean"}`; +- `elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds`. + +**Condition to check before disabling the fallback.** + +1. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is + enabled on every node. A group whose leader runs no compactor gets no pass. +2. Every anchor type in every data group has completed a clean pass. + Leadership moves, so check across nodes: + `max by (type, group) (elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds)` + must have a series for all eight types and every group. The gauge exists + only in builds that also migrate every locally led group + (bootjp/elastickv#1278), so an older build cannot produce it. It is + process-local and gone after a restart: a missing series means no clean + pass is known, not that one failed. +3. No route catalog change has moved a range between groups since those + passes. A pass covers only the rows its group served at its read + timestamps. After such a change, repeat step 2. + +**What a clean pass does not prove.** +- Live writers only produce the inline format. A clean pass therefore shows + that the anchors its group's scan serves were all inline when read. +- The scan skips some legacy-route hash, set, zset and stream rows without + counting them: those whose user key belongs to a group another node leads. + They are not migrated, and they do not keep a pass from being clean. The gate + for them is milestone M4 of the cross-leader design (bootjp/elastickv#1280), + which Phase 3 must also pass. +- The scan also defers, without counting, rows whose route is under staged + visibility. Step 3 excludes them: staging only happens while a route moves. +- Legacy bare string keys are outside the anchor prefixes, as noted above. + ### Phase 4 — Remove TTL buffer The `TTLBuffer`, `runTTLFlusher`, and `flushTTLBuffer` have been removed. TTL @@ -394,5 +456,5 @@ adapter/ | 0 | Add dual-read fallback; new writes use new encoding | Low — reads always fall back | | 1 | Add inline TTL to list/hash/set/zset/stream metadata and HLL payloads | Medium — delta compaction must preserve TTL | | 2 | Background migration of existing keys | Medium — shipped as bounded compactor pass | -| 3 | Remove legacy `!redis\|ttl\|` read path | Medium — safe only after prefix-scanned anchors and legacy bare strings are cleared | +| 3 | Remove legacy `!redis\|ttl\|` read path | Medium — safe only after a clean migration pass for every anchor type and group (§5, Phase 3), the cross-leader legacy-route gate, and legacy bare strings are cleared | | 4 | Remove TTL buffer | Low — shipped | From 4e2a40699f1bf14ee08cb79953e2ced9c73942f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 13:33:06 +0000 Subject: [PATCH 4/9] adapter: drop TTL-inline pages another tick finished; time passes from their first scan Codex review on #1283 found two defects in the pass tally: - Two ticks that read the same cursor both added the same page to the pass. On a middle page the pass counted it twice; on the last page the second tick opened and reported a spurious non-full pass. A page is now dropped when the plan's cursor moved after its scan began. - A pass started when its first page was finished, so its duration left that page out. It now starts when the first page's scan begins. Both carry a regression test that failed before the fix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor_test.go | 107 +++++++++++++++++++++ adapter/redis_ttl_inline_migration_pass.go | 26 +++-- adapter/redis_ttl_inline_migrator.go | 12 ++- 3 files changed, 132 insertions(+), 13 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index d43eb471e..440b4a47d 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -2569,3 +2569,110 @@ func TestDeltaCompactor_TTLInlineReportsCompletedPasses(t *testing.T) { }) } } + +// hookedTTLInlineScanStore runs onScan before each group scan of prefix. +type hookedTTLInlineScanStore struct { + store.MVCCStore + + prefix []byte + onScan func() +} + +func (s *hookedTTLInlineScanStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { + if bytes.HasPrefix(start, s.prefix) && s.onScan != nil { + s.onScan() + } + return scanPageWithResume(ctx, s.MVCCStore, start, end, limit, ts) +} + +func newHookedTTLInlinePassCompactor(t *testing.T, st *hookedTTLInlineScanStore) (*DeltaCompactor, *lockedLogBuffer) { + t.Helper() + coord := &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + groupIDs: []uint64{1}, + leadsKeys: true, + } + logs := &lockedLogBuffer{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorLogger(slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + ) + return c, logs +} + +// Two ticks that run at once read the same cursor and migrate the same page. +// Only the first to finish may add that page to the pass; the other's page is +// stale. Counting both inflated the pass's tallies, and on the last page opened +// a second, spurious pass. +func TestDeltaCompactor_TTLInlinePassIgnoresAPageAnotherTickFinished(t *testing.T) { + t.Parallel() + + ctx := context.Background() + const keys = 2*ttlInlineMigrationTickScanLimit + 88 + st := &hookedTTLInlineScanStore{MVCCStore: store.NewMVCCStore(), prefix: []byte(redisStrPrefix)} + seedTTLInlinePassStrings(t, st, keys, false) + c, logs := newHookedTTLInlinePassCompactor(t, st) + + // The scan of the second string page runs a whole tick in between, as a + // concurrent SyncOnce would: both ticks read the cursor before page two. + scans, nested := 0, false + st.onScan = func() { + scans++ + if scans == 2 && !nested { + nested = true + require.NoError(t, c.SyncOnce(ctx)) + } + } + for range 3 { + require.NoError(t, c.SyncOnce(ctx)) + } + + passes := logs.ttlInlinePasses(t, "ttl-inline-string") + require.Len(t, passes, 1) + require.Equal(t, 3, passes[0].Pages) + require.Equal(t, keys, passes[0].KeysScanned) + require.True(t, passes[0].Clean) +} + +// A pass's duration includes its first page: the pass starts when that page's +// scan starts, not when the page has been migrated. +func TestDeltaCompactor_TTLInlinePassDurationIncludesFirstPage(t *testing.T) { + t.Parallel() + + const scanDelay = 50 * time.Millisecond + st := &hookedTTLInlineScanStore{ + MVCCStore: store.NewMVCCStore(), + prefix: []byte(redisStrPrefix), + onScan: func() { time.Sleep(scanDelay) }, + } + seedTTLInlinePassStrings(t, st, 3, false) + c, logs := newHookedTTLInlinePassCompactor(t, st) + + require.NoError(t, c.SyncOnce(context.Background())) + + durations := logs.ttlInlinePassDurations(t, "ttl-inline-string") + require.Len(t, durations, 1) + require.GreaterOrEqual(t, durations[0], scanDelay) +} + +func (b *lockedLogBuffer) ttlInlinePassDurations(t *testing.T, typeName string) []time.Duration { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + var out []time.Duration + for _, line := range bytes.Split(b.buf.Bytes(), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec struct { + Msg string `json:"msg"` + Type string `json:"type"` + Duration time.Duration `json:"duration"` + } + require.NoError(t, json.Unmarshal(line, &rec)) + if rec.Msg == "ttl inline migrator: pass complete" && rec.Type == typeName { + out = append(out, rec.Duration) + } + } + return out +} diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go index 8050afe4a..4c0f9b7c2 100644 --- a/adapter/redis_ttl_inline_migration_pass.go +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -1,6 +1,7 @@ package adapter import ( + "bytes" "context" "log/slog" "time" @@ -57,17 +58,18 @@ type ttlInlineMigrationPlanState struct { } // openTTLInlineMigrationPassLocked returns the open pass of the plan named -// name, starting one if needed. fromStart reports that the page being recorded -// starts at the beginning of the prefix; such a page always opens a new pass -// unless the open pass has not completed a page yet. Callers hold cursorMu. -func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, fromStart bool) *ttlInlineMigrationPass { +// name, starting one if needed. A page that starts at the beginning of the +// prefix always opens a new pass unless the open pass has not completed a page +// yet. A pass the page opens starts when the page's scan began. Callers hold +// cursorMu. +func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, page ttlInlineMigrationPage) *ttlInlineMigrationPass { state := c.ttlInlinePlans[name] if state == nil { state = &ttlInlineMigrationPlanState{} c.ttlInlinePlans[name] = state } - if state.pass == nil || (fromStart && state.pass.pages > 0) { - state.pass = &ttlInlineMigrationPass{full: fromStart, startedAt: time.Now()} + if state.pass == nil || (page.fromStart && state.pass.pages > 0) { + state.pass = &ttlInlineMigrationPass{full: page.fromStart, startedAt: page.scannedAt} } return state.pass } @@ -75,19 +77,20 @@ func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, fromStart // noteTTLInlineMigrationPageFailed adds the keys a failed page already // rewrote to its plan's pass. The page itself does not count: its cursor stays, // and the next tick reads it again. -func (c *DeltaCompactor) noteTTLInlineMigrationPageFailed(name string, fromStart bool, migrated int) { +func (c *DeltaCompactor) noteTTLInlineMigrationPageFailed(name string, page ttlInlineMigrationPage, migrated int) { if migrated == 0 { return } c.cursorMu.Lock() defer c.cursorMu.Unlock() - c.openTTLInlineMigrationPassLocked(name, fromStart).keysMigrated += migrated + c.openTTLInlineMigrationPassLocked(name, page).keysMigrated += migrated } // finishTTLInlineMigrationPage adds a migrated page to its plan's pass and // moves the plan's cursor past it. When the page reached the end of the prefix // it closes the pass and returns it with done set, and whether the plan's -// previous pass was clean. +// previous pass was clean. A page is dropped when the plan's cursor moved after +// its scan began: another tick that read the same cursor already finished it. func (c *DeltaCompactor) finishTTLInlineMigrationPage( name string, page ttlInlineMigrationPage, @@ -95,7 +98,10 @@ func (c *DeltaCompactor) finishTTLInlineMigrationPage( ) (pass ttlInlineMigrationPass, prevClean bool, done bool) { c.cursorMu.Lock() defer c.cursorMu.Unlock() - open := c.openTTLInlineMigrationPassLocked(name, page.fromStart) + if !bytes.Equal(c.cursors[name], page.startCursor) { + return ttlInlineMigrationPass{}, false, false + } + open := c.openTTLInlineMigrationPassLocked(name, page) open.add(tally) open.pages++ if !c.advanceCursorLocked(name, page.cursorKey, page.truncated) { diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 728fdd4eb..5f12b1891 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -221,7 +221,7 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( if err != nil { // The next tick reads this page again, so only the keys its earlier // batches already rewrote count toward the pass. - c.noteTTLInlineMigrationPageFailed(plan.cursorName, page.fromStart, tally.keysMigrated) + c.noteTTLInlineMigrationPageFailed(plan.cursorName, page, tally.keysMigrated) return err } if pass, prevClean, done := c.finishTTLInlineMigrationPage(plan.cursorName, page, tally); done { @@ -340,6 +340,11 @@ type ttlInlineMigrationPage struct { // fromStart reports that the page starts at the beginning of the prefix, // so it is the first page of a pass. fromStart bool + // startCursor is the plan's cursor when the scan began. A page whose plan + // cursor has moved since was also finished by another tick. + startCursor []byte + // scannedAt is when the scan began; a pass opened by this page starts then. + scannedAt time.Time } func (c *DeltaCompactor) scanTTLInlineMigrationRaw( @@ -348,15 +353,16 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( plan ttlInlineMigrationScanPlan, readTS uint64, ) (ttlInlineMigrationPage, error) { + scannedAt := time.Now() c.cursorMu.Lock() start := h.prefix - cur := c.cursors[plan.cursorName] + cur := bytes.Clone(c.cursors[plan.cursorName]) if len(cur) > 0 { start = append(bytes.Clone(cur), cursorNextByte) } c.cursorMu.Unlock() - page := ttlInlineMigrationPage{fromStart: len(cur) == 0} + page := ttlInlineMigrationPage{fromStart: len(cur) == 0, startCursor: cur, scannedAt: scannedAt} end := store.PrefixScanEnd(h.prefix) if plan.scanner != nil { kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) From 9eb1c1f7911f3cf01fbedf66bf4ea6e21f69b70c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 13:41:35 +0000 Subject: [PATCH 5/9] monitoring, adapter, docs: gate Phase 3 on the latest pass outcome, not the last clean pass Codex review on #1283: a last-clean-pass gauge keeps its value after a later not_clean pass, and max across nodes can pick a former leader's stale clean sample. Either let the Phase 3 gate pass while the current leader was still finding legacy anchors. Passes now report one of three outcomes: clean, partial (found nothing but did not read the whole prefix) and not_clean. The gauge becomes last_pass_timestamp_seconds{type,group,outcome}. The doc's gate requires the latest clean pass for every type and group to be later than T0 (the last upgrade, route move between groups, or restart) and later than every not_clean pass on any node. The doc also states that a leadership loss and regain between two ticks is not noticed. It says why that cannot make a clean pass wrong while no live build writes the legacy format, and lists that as an explicit gate condition. The tests were updated first to the new outcome API and failed to build before the change. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor_test.go | 24 +++++-- adapter/redis_ttl_inline_migration_pass.go | 30 +++++++-- ...2026_04_17_implemented_ttl_inline_value.md | 65 +++++++++++++------ monitoring/redis_ttl_inline.go | 49 ++++++++------ monitoring/redis_ttl_inline_test.go | 31 ++++++--- 5 files changed, 140 insertions(+), 59 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 440b4a47d..977ab7e22 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -15,6 +15,7 @@ import ( "github.com/bootjp/elastickv/distribution" "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/monitoring" "github.com/bootjp/elastickv/store" "github.com/cockroachdb/errors" "github.com/stretchr/testify/require" @@ -2332,7 +2333,7 @@ func (c *ttlInlinePassTestCoordinator) Dispatch(ctx context.Context, req *kv.Ope type ttlInlinePassObservation struct { typeName string groupID uint64 - clean bool + outcome string } type recordingTTLInlinePassObserver struct { @@ -2340,10 +2341,10 @@ type recordingTTLInlinePassObserver struct { passes []ttlInlinePassObservation } -func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) { +func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) { o.mu.Lock() defer o.mu.Unlock() - o.passes = append(o.passes, ttlInlinePassObservation{typeName: typeName, groupID: groupID, clean: clean}) + o.passes = append(o.passes, ttlInlinePassObservation{typeName: typeName, groupID: groupID, outcome: outcome}) } func (o *recordingTTLInlinePassObserver) forType(typeName string) []ttlInlinePassObservation { @@ -2405,6 +2406,20 @@ func (b *lockedLogBuffer) ttlInlinePasses(t *testing.T, typeName string) []ttlIn return out } +// wantTTLInlinePassOutcome is the monitoring outcome a pass should report: a +// pass that is not clean but found nothing to rewrite, skip or fail did not +// read its whole prefix, so it is partial. +func wantTTLInlinePassOutcome(clean, foundNothing bool) string { + switch { + case clean: + return monitoring.RedisTTLInlinePassClean + case foundNothing: + return monitoring.RedisTTLInlinePassPartial + default: + return monitoring.RedisTTLInlinePassNotClean + } +} + func seedTTLInlinePassStrings(t *testing.T, st store.MVCCStore, n uint64, legacy bool) { t.Helper() ctx := context.Background() @@ -2563,7 +2578,8 @@ func TestDeltaCompactor_TTLInlineReportsCompletedPasses(t *testing.T) { wantObserved := make([]ttlInlinePassObservation, 0, len(tc.want)) for _, p := range tc.want { - wantObserved = append(wantObserved, ttlInlinePassObservation{typeName: typeName, groupID: 1, clean: p.clean}) + outcome := wantTTLInlinePassOutcome(p.clean, p.keysMigrated+p.keysSkipped+p.keysFailed == 0) + wantObserved = append(wantObserved, ttlInlinePassObservation{typeName: typeName, groupID: 1, outcome: outcome}) } require.Equal(t, wantObserved, observer.forType(typeName)) }) diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go index 4c0f9b7c2..807cc8a36 100644 --- a/adapter/redis_ttl_inline_migration_pass.go +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -5,6 +5,8 @@ import ( "context" "log/slog" "time" + + "github.com/bootjp/elastickv/monitoring" ) // ttlInlineMigrationTally counts what the TTL-inline migrator did with the @@ -33,9 +35,9 @@ func (t *ttlInlineMigrationTally) add(o ttlInlineMigrationTally) { // from the start of the prefix through the page that reaches its end. type ttlInlineMigrationPass struct { ttlInlineMigrationTally - // full reports that the pass started at the beginning of the prefix in - // this leadership term. A pass that resumes a cursor left by an earlier - // term is not full: this term never read the anchors before the cursor. + // full reports that the pass started at the beginning of the prefix and + // its group stayed in every tick's plans since. A pass that resumes a + // cursor left before its group dropped out of the plans is not full. full bool pages int startedAt time.Time @@ -45,7 +47,25 @@ type ttlInlineMigrationPass struct { // already in the inline format: it read the whole prefix, rewrote nothing, and // left nothing unchecked. func (p ttlInlineMigrationPass) clean() bool { - return p.full && p.keysMigrated == 0 && p.keysSkipped == 0 && p.keysFailed == 0 + return p.full && p.foundNothing() +} + +// foundNothing reports that the pass rewrote, skipped and failed no anchor. +func (p ttlInlineMigrationPass) foundNothing() bool { + return p.keysMigrated == 0 && p.keysSkipped == 0 && p.keysFailed == 0 +} + +// outcome is the pass's monitoring outcome: clean, partial (found nothing but +// did not read the whole prefix), or not_clean. +func (p ttlInlineMigrationPass) outcome() string { + switch { + case p.clean(): + return monitoring.RedisTTLInlinePassClean + case p.foundNothing(): + return monitoring.RedisTTLInlinePassPartial + default: + return monitoring.RedisTTLInlinePassNotClean + } } // ttlInlineMigrationPlanState is a scan plan's pass bookkeeping. @@ -165,6 +185,6 @@ func (c *DeltaCompactor) reportTTLInlineMigrationPass( "duration", time.Since(pass.startedAt), ) if c.ttlInlineObserver != nil { - c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, clean) + c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, pass.outcome()) } } diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index 80651213d..2c875254b 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -333,7 +333,7 @@ completed pass, `ttl inline migrator: pass complete`, with these keys: |---|---| | `type` | Anchor type. | | `group_id` | Data group. `0` is the whole-prefix fallback scan, used only when the coordinator does not report its local leader groups. | -| `full_pass` | The pass started at the top of the prefix in this leadership term. A plan that loses its group mid-pass drops its counters. If the group comes back, the resumed pass is not full. | +| `full_pass` | The pass started at the top of the prefix, and every tick since planned its group on this node. A plan whose group is missing from a tick's plans drops its counters; if the group comes back, the resumed pass is not full. Leadership lost and regained between two ticks is not noticed (see "What a clean pass does not prove"). | | `pages` | Pages the pass read. | | `keys_scanned` | Anchors the pass read. | | `keys_migrated` | Anchors rewritten by committed batches. A page retried after a failed batch still counts the keys its earlier batches rewrote. | @@ -345,36 +345,63 @@ completed pass, `ttl inline migrator: pass complete`, with these keys: A pass that is not clean, and the first clean pass after one that was not, are logged at info. Repeated clean passes are logged at debug. -The same outcome is exported per node with labels `type` and `group`: -- `elastickv_redis_ttl_inline_migration_passes_total{outcome="clean"|"not_clean"}`; -- `elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds`. +Each pass also has a monitoring outcome: +- `clean`: as in the log. +- `partial`: rewrote, skipped and failed nothing, but was not full. +- `not_clean`: rewrote, skipped or failed at least one anchor. + +Two metrics per node carry it, labelled `type`, `group` and `outcome`: +- `elastickv_redis_ttl_inline_migration_passes_total`; +- `elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds`, the time + this node last completed a pass with that outcome. **Condition to check before disabling the fallback.** -1. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is +Let T0 be the latest of: +- the time the last node was upgraded to a build that reports passes (it also + migrates every locally led group, bootjp/elastickv#1278); +- the last route catalog change that moved a range between groups, since a + pass covers only the rows its group served at its read timestamps; +- the last restart of any node, since the metrics are process-local and a + restart drops that node's history. + +Then all of the following must hold: +1. No live build writes the legacy format. Every node runs a build that writes + inline TTLs (Phase 1 or later). +2. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is enabled on every node. A group whose leader runs no compactor gets no pass. -2. Every anchor type in every data group has completed a clean pass. - Leadership moves, so check across nodes: - `max by (type, group) (elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds)` - must have a series for all eight types and every group. The gauge exists - only in builds that also migrate every locally led group - (bootjp/elastickv#1278), so an older build cannot produce it. It is - process-local and gone after a restart: a missing series means no clean - pass is known, not that one failed. -3. No route catalog change has moved a range between groups since those - passes. A pass covers only the rows its group served at its read - timestamps. After such a change, repeat step 2. +3. For all eight types and every data group, the latest clean pass across all + nodes is later than T0: + `max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="clean"})` + has a series for every type and group, each later than T0. +4. For each type and group, that clean pass is later than every `not_clean` + pass on any node. Leadership moves, so a clean sample from a former leader + does not outrank a later `not_clean` pass on the current one. This query + must return nothing: + + ``` + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="not_clean"}) + >= on (type, group) + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="clean"}) + ``` + +`partial` passes do not affect the gate. **What a clean pass does not prove.** -- Live writers only produce the inline format. A clean pass therefore shows - that the anchors its group's scan serves were all inline when read. +- Live writers only produce the inline format (condition 1). A clean pass + therefore shows that the anchors its group's scan serves were all inline + when read. +- Leadership lost and regained between two ticks leaves the pass open and + full. The anchors behind the cursor were read inline earlier in the pass, and + without a legacy writer (condition 1) the interim leader could not have made + them legacy again. - The scan skips some legacy-route hash, set, zset and stream rows without counting them: those whose user key belongs to a group another node leads. They are not migrated, and they do not keep a pass from being clean. The gate for them is milestone M4 of the cross-leader design (bootjp/elastickv#1280), which Phase 3 must also pass. - The scan also defers, without counting, rows whose route is under staged - visibility. Step 3 excludes them: staging only happens while a route moves. + visibility. T0 excludes them: staging only happens while a route moves. - Legacy bare string keys are outside the anchor prefixes, as noted above. ### Phase 4 — Remove TTL buffer diff --git a/monitoring/redis_ttl_inline.go b/monitoring/redis_ttl_inline.go index 0e8e2c42d..1108c60a8 100644 --- a/monitoring/redis_ttl_inline.go +++ b/monitoring/redis_ttl_inline.go @@ -6,25 +6,36 @@ import ( "github.com/prometheus/client_golang/prometheus" ) +// Outcomes of a Redis TTL-inline migration pass. const ( - redisTTLInlinePassClean = "clean" - redisTTLInlinePassNotClean = "not_clean" + // RedisTTLInlinePassClean is a pass that read its whole prefix while its + // group stayed in this node's plans and found every anchor already in the + // inline format. + RedisTTLInlinePassClean = "clean" + // RedisTTLInlinePassPartial is a pass that rewrote, skipped and failed + // nothing but did not read its whole prefix: it resumed a cursor left + // before its group dropped out of this node's plans. + RedisTTLInlinePassPartial = "partial" + // RedisTTLInlinePassNotClean is a pass that rewrote, skipped or failed at + // least one anchor. + RedisTTLInlinePassNotClean = "not_clean" ) // RedisTTLInlineMigrationObserver records the passes the Redis TTL-inline // migrator completes. A pass covers one anchor type (typeName, such as // "ttl-inline-string") in one data group; group 0 is the whole-prefix scan the // migrator falls back to when the coordinator does not report its groups. +// outcome is one of the RedisTTLInlinePass constants. type RedisTTLInlineMigrationObserver interface { - ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) + ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) } // RedisTTLInlineMigrationMetrics exports the TTL-inline migrator's passes. -// Labels are the anchor type and data group only: eight types times the data -// groups a node leads. +// Labels are the anchor type, data group and outcome only: eight types times +// the data groups a node leads times three outcomes. type RedisTTLInlineMigrationMetrics struct { - passes *prometheus.CounterVec - lastCleanPass *prometheus.GaugeVec + passes *prometheus.CounterVec + lastPass *prometheus.GaugeVec } func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisTTLInlineMigrationMetrics { @@ -32,35 +43,31 @@ func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisT passes: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "elastickv_redis_ttl_inline_migration_passes_total", - Help: "Completed Redis TTL-inline migration passes by anchor type, data group, and outcome. A clean pass read the whole prefix and found every anchor already in the inline format.", + Help: "Completed Redis TTL-inline migration passes by anchor type, data group, and outcome (clean, partial, not_clean).", }, []string{"type", "group", "outcome"}, ), - lastCleanPass: prometheus.NewGaugeVec( + lastPass: prometheus.NewGaugeVec( prometheus.GaugeOpts{ - Name: "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds", - Help: "Unix time at which this node last completed a clean Redis TTL-inline migration pass, by anchor type and data group.", + Name: "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds", + Help: "Unix time at which this node last completed a Redis TTL-inline migration pass, by anchor type, data group, and outcome (clean, partial, not_clean).", }, - []string{"type", "group"}, + []string{"type", "group", "outcome"}, ), } if registerer != nil { - registerer.MustRegister(m.passes, m.lastCleanPass) + registerer.MustRegister(m.passes, m.lastPass) } return m } -// ObserveRedisTTLInlineMigrationPass counts a completed pass and, when it was -// clean, stamps the type and group's last clean pass time. -func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, clean bool) { +// ObserveRedisTTLInlineMigrationPass counts a completed pass and stamps the +// type, group and outcome's last pass time. +func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) { if m == nil { return } group := strconv.FormatUint(groupID, 10) - outcome := redisTTLInlinePassNotClean - if clean { - outcome = redisTTLInlinePassClean - m.lastCleanPass.WithLabelValues(typeName, group).SetToCurrentTime() - } m.passes.WithLabelValues(typeName, group, outcome).Inc() + m.lastPass.WithLabelValues(typeName, group, outcome).SetToCurrentTime() } diff --git a/monitoring/redis_ttl_inline_test.go b/monitoring/redis_ttl_inline_test.go index 5075f7aaa..f5f41558a 100644 --- a/monitoring/redis_ttl_inline_test.go +++ b/monitoring/redis_ttl_inline_test.go @@ -9,34 +9,45 @@ import ( "github.com/stretchr/testify/require" ) +// Each outcome keeps its own last pass time, so a not_clean pass after a clean +// one stays visible next to it: a gate that compares the two can tell that the +// clean pass is no longer the latest word on the plan. func TestRedisTTLInlineMigrationMetricsExportPasses(t *testing.T) { t.Parallel() reg := prometheus.NewRegistry() metrics := newRedisTTLInlineMigrationMetrics(reg) before := float64(time.Now().Unix()) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, false) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, true) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, true) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, false) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassNotClean) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, RedisTTLInlinePassPartial) require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "not_clean"))) require.Equal(t, 2.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "clean"))) - require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-hash", "3", "not_clean"))) - require.GreaterOrEqual(t, testutil.ToFloat64(metrics.lastCleanPass.WithLabelValues("ttl-inline-string", "2")), before) - require.Equal(t, 1, testutil.CollectAndCount(metrics.lastCleanPass), - "only a clean pass stamps the last clean pass time") + require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-hash", "3", "partial"))) + for _, labels := range [][]string{ + {"ttl-inline-string", "2", "not_clean"}, + {"ttl-inline-string", "2", "clean"}, + {"ttl-inline-hash", "3", "partial"}, + } { + require.GreaterOrEqual(t, testutil.ToFloat64(metrics.lastPass.WithLabelValues(labels...)), before, labels) + } + require.Equal(t, 3, testutil.CollectAndCount(metrics.lastPass)) families, err := reg.Gather() require.NoError(t, err) require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_passes_total")) - require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds")) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds")) + require.False(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds")) } func TestRedisTTLInlineMigrationObserverIsNilSafe(t *testing.T) { t.Parallel() var metrics *RedisTTLInlineMigrationMetrics - require.NotPanics(t, func() { metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, true) }) + require.NotPanics(t, func() { + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, RedisTTLInlinePassClean) + }) var r *Registry require.Nil(t, r.RedisTTLInlineMigrationObserver()) } From aaee3d9998c139f603fd6b9ac8e689ec7d89be2c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 22:27:49 +0000 Subject: [PATCH 6/9] monitoring, adapter, docs: record when each TTL-inline pass began Codex review on #1283: a pass that ends after T0 may have read its first pages before T0. A route change at T0 can then put a legacy anchor behind its cursor, and the remaining pages still finish clean. The observer now receives the pass's start, which is when its first page's scan began. The gauge becomes last_pass_start_timestamp_seconds{type,group,outcome}. The doc's gate now requires the latest clean pass to have started after T0 and after every not_clean pass. The updated tests (the gauge value and the start the observer receives) failed to build before the change. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor_test.go | 35 ++++++++++++++++--- adapter/redis_ttl_inline_migration_pass.go | 2 +- ...2026_04_17_implemented_ttl_inline_value.md | 20 ++++++----- monitoring/redis_ttl_inline.go | 28 ++++++++------- monitoring/redis_ttl_inline_test.go | 35 ++++++++++--------- 5 files changed, 78 insertions(+), 42 deletions(-) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 977ab7e22..858f11dd5 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -2336,21 +2336,37 @@ type ttlInlinePassObservation struct { outcome string } +type recordedTTLInlinePass struct { + ttlInlinePassObservation + startedAt time.Time +} + type recordingTTLInlinePassObserver struct { mu sync.Mutex - passes []ttlInlinePassObservation + passes []recordedTTLInlinePass } -func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) { +func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) { o.mu.Lock() defer o.mu.Unlock() - o.passes = append(o.passes, ttlInlinePassObservation{typeName: typeName, groupID: groupID, outcome: outcome}) + o.passes = append(o.passes, recordedTTLInlinePass{ + ttlInlinePassObservation: ttlInlinePassObservation{typeName: typeName, groupID: groupID, outcome: outcome}, + startedAt: startedAt, + }) } func (o *recordingTTLInlinePassObserver) forType(typeName string) []ttlInlinePassObservation { + var out []ttlInlinePassObservation + for _, p := range o.recordedForType(typeName) { + out = append(out, p.ttlInlinePassObservation) + } + return out +} + +func (o *recordingTTLInlinePassObserver) recordedForType(typeName string) []recordedTTLInlinePass { o.mu.Lock() defer o.mu.Unlock() - var out []ttlInlinePassObservation + var out []recordedTTLInlinePass for _, p := range o.passes { if p.typeName == typeName { out = append(out, p) @@ -2663,12 +2679,23 @@ func TestDeltaCompactor_TTLInlinePassDurationIncludesFirstPage(t *testing.T) { } seedTTLInlinePassStrings(t, st, 3, false) c, logs := newHookedTTLInlinePassCompactor(t, st) + observer := &recordingTTLInlinePassObserver{} + c.ttlInlineObserver = observer + before := time.Now() require.NoError(t, c.SyncOnce(context.Background())) + after := time.Now() durations := logs.ttlInlinePassDurations(t, "ttl-inline-string") require.Len(t, durations, 1) require.GreaterOrEqual(t, durations[0], scanDelay) + + // The observer gets the pass's start, which precedes its first page's + // scan: a gate can then require a pass that started after an event. + passes := observer.recordedForType("ttl-inline-string") + require.Len(t, passes, 1) + require.False(t, passes[0].startedAt.Before(before)) + require.LessOrEqual(t, passes[0].startedAt, after.Add(-scanDelay)) } func (b *lockedLogBuffer) ttlInlinePassDurations(t *testing.T, typeName string) []time.Duration { diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go index 807cc8a36..b37e381d3 100644 --- a/adapter/redis_ttl_inline_migration_pass.go +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -185,6 +185,6 @@ func (c *DeltaCompactor) reportTTLInlineMigrationPass( "duration", time.Since(pass.startedAt), ) if c.ttlInlineObserver != nil { - c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, pass.outcome()) + c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, pass.outcome(), pass.startedAt) } } diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index 2c875254b..b08edeb7e 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -352,8 +352,10 @@ Each pass also has a monitoring outcome: Two metrics per node carry it, labelled `type`, `group` and `outcome`: - `elastickv_redis_ttl_inline_migration_passes_total`; -- `elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds`, the time - this node last completed a pass with that outcome. +- `elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds`, + the time the latest pass this node completed with that outcome began. It + records the start, not the end, because a pass spanning several ticks read + its first pages at its start. **Condition to check before disabling the fallback.** @@ -371,18 +373,20 @@ Then all of the following must hold: 2. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is enabled on every node. A group whose leader runs no compactor gets no pass. 3. For all eight types and every data group, the latest clean pass across all - nodes is later than T0: - `max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="clean"})` - has a series for every type and group, each later than T0. -4. For each type and group, that clean pass is later than every `not_clean` + nodes started after T0: + `max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"})` + has a series for every type and group, each later than T0. A pass that only + ended after T0 does not count: it read its first pages before T0, so a + route change at T0 could have put a legacy anchor behind its cursor. +4. For each type and group, that clean pass started after every `not_clean` pass on any node. Leadership moves, so a clean sample from a former leader does not outrank a later `not_clean` pass on the current one. This query must return nothing: ``` - max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="not_clean"}) + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="not_clean"}) >= on (type, group) - max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds{outcome="clean"}) + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"}) ``` `partial` passes do not affect the gate. diff --git a/monitoring/redis_ttl_inline.go b/monitoring/redis_ttl_inline.go index 1108c60a8..4c32522ff 100644 --- a/monitoring/redis_ttl_inline.go +++ b/monitoring/redis_ttl_inline.go @@ -2,6 +2,7 @@ package monitoring import ( "strconv" + "time" "github.com/prometheus/client_golang/prometheus" ) @@ -25,17 +26,18 @@ const ( // migrator completes. A pass covers one anchor type (typeName, such as // "ttl-inline-string") in one data group; group 0 is the whole-prefix scan the // migrator falls back to when the coordinator does not report its groups. -// outcome is one of the RedisTTLInlinePass constants. +// outcome is one of the RedisTTLInlinePass constants; startedAt is when the +// pass's first page was scanned. type RedisTTLInlineMigrationObserver interface { - ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) + ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) } // RedisTTLInlineMigrationMetrics exports the TTL-inline migrator's passes. // Labels are the anchor type, data group and outcome only: eight types times // the data groups a node leads times three outcomes. type RedisTTLInlineMigrationMetrics struct { - passes *prometheus.CounterVec - lastPass *prometheus.GaugeVec + passes *prometheus.CounterVec + lastPassStart *prometheus.GaugeVec } func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisTTLInlineMigrationMetrics { @@ -47,27 +49,29 @@ func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisT }, []string{"type", "group", "outcome"}, ), - lastPass: prometheus.NewGaugeVec( + lastPassStart: prometheus.NewGaugeVec( prometheus.GaugeOpts{ - Name: "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds", - Help: "Unix time at which this node last completed a Redis TTL-inline migration pass, by anchor type, data group, and outcome (clean, partial, not_clean).", + Name: "elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds", + Help: "Unix time at which the latest Redis TTL-inline migration pass this node completed began, by anchor type, data group, and outcome (clean, partial, not_clean).", }, []string{"type", "group", "outcome"}, ), } if registerer != nil { - registerer.MustRegister(m.passes, m.lastPass) + registerer.MustRegister(m.passes, m.lastPassStart) } return m } -// ObserveRedisTTLInlineMigrationPass counts a completed pass and stamps the -// type, group and outcome's last pass time. -func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string) { +// ObserveRedisTTLInlineMigrationPass counts a completed pass and records when +// it began. The start, not the end, is what a gate compares against: a pass +// that ended after an event but began before it read part of its prefix +// before the event. +func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) { if m == nil { return } group := strconv.FormatUint(groupID, 10) m.passes.WithLabelValues(typeName, group, outcome).Inc() - m.lastPass.WithLabelValues(typeName, group, outcome).SetToCurrentTime() + m.lastPassStart.WithLabelValues(typeName, group, outcome).Set(float64(startedAt.UnixNano()) / float64(time.Second)) } diff --git a/monitoring/redis_ttl_inline_test.go b/monitoring/redis_ttl_inline_test.go index f5f41558a..ed1ba45d4 100644 --- a/monitoring/redis_ttl_inline_test.go +++ b/monitoring/redis_ttl_inline_test.go @@ -9,44 +9,45 @@ import ( "github.com/stretchr/testify/require" ) -// Each outcome keeps its own last pass time, so a not_clean pass after a clean -// one stays visible next to it: a gate that compares the two can tell that the -// clean pass is no longer the latest word on the plan. +// Each outcome keeps the start time of its latest pass, so a not_clean pass +// after a clean one stays visible next to it, and a gate can require a clean +// pass that started, not just ended, after an event. func TestRedisTTLInlineMigrationMetricsExportPasses(t *testing.T) { t.Parallel() reg := prometheus.NewRegistry() metrics := newRedisTTLInlineMigrationMetrics(reg) - before := float64(time.Now().Unix()) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassNotClean) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean) - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, RedisTTLInlinePassPartial) + start := time.Unix(1_700_000_000, 0) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassNotClean, start) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean, start.Add(time.Minute)) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean, start.Add(2*time.Minute)) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, RedisTTLInlinePassPartial, start) require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "not_clean"))) require.Equal(t, 2.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "clean"))) require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-hash", "3", "partial"))) - for _, labels := range [][]string{ - {"ttl-inline-string", "2", "not_clean"}, - {"ttl-inline-string", "2", "clean"}, - {"ttl-inline-hash", "3", "partial"}, + for labels, want := range map[[3]string]time.Time{ + {"ttl-inline-string", "2", "not_clean"}: start, + {"ttl-inline-string", "2", "clean"}: start.Add(2 * time.Minute), + {"ttl-inline-hash", "3", "partial"}: start, } { - require.GreaterOrEqual(t, testutil.ToFloat64(metrics.lastPass.WithLabelValues(labels...)), before, labels) + got := testutil.ToFloat64(metrics.lastPassStart.WithLabelValues(labels[0], labels[1], labels[2])) + require.InDelta(t, float64(want.Unix()), got, 0, labels) } - require.Equal(t, 3, testutil.CollectAndCount(metrics.lastPass)) + require.Equal(t, 3, testutil.CollectAndCount(metrics.lastPassStart)) families, err := reg.Gather() require.NoError(t, err) require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_passes_total")) - require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds")) - require.False(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_clean_pass_timestamp_seconds")) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds")) + require.False(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds")) } func TestRedisTTLInlineMigrationObserverIsNilSafe(t *testing.T) { t.Parallel() var metrics *RedisTTLInlineMigrationMetrics require.NotPanics(t, func() { - metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, RedisTTLInlinePassClean) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, RedisTTLInlinePassClean, time.Now()) }) var r *Registry require.Nil(t, r.RedisTTLInlineMigrationObserver()) From 91df14c46fd08cb9bd5086a585b38bfd56b470f6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 22:37:52 +0000 Subject: [PATCH 7/9] adapter: start a TTL-inline pass no later than its read timestamp Codex review on #1283: SyncOnce acquires readTS at the start of the tick, but a pass's start was stamped only when its first page's scan began, after the delta handlers had run. If a route moved in between, a pass that read a snapshot from before the move reported a start after it, so it could satisfy the Phase 3 gate's started-after-T0 condition. SyncOnce now notes the time before it acquires the read timestamp and hands it to every page of the tick. A pass opened by such a page starts then. Regression test: the reported start precedes the coordinator's first Clock() call, which is where the read timestamp is taken. Before the fix it came about 10ms after. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor.go | 5 +- adapter/redis_delta_compactor_test.go | 52 +++++++++++++++++-- adapter/redis_ttl_inline_migration_pass.go | 6 +-- adapter/redis_ttl_inline_migrator.go | 19 ++++--- ...2026_04_17_implemented_ttl_inline_value.md | 7 +-- monitoring/redis_ttl_inline.go | 4 +- 6 files changed, 74 insertions(+), 19 deletions(-) diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index 06120e42c..bfa49b1a5 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -334,6 +334,9 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error { } tickCtx, cancel := context.WithTimeout(ctx, c.timeout) defer cancel() + // snapshotAt precedes the read timestamp, so a TTL-inline pass that reads + // at readTS never reports a start later than its snapshot. + snapshotAt := time.Now() readTimestamp, err := kv.BeginReadTimestampThrough(tickCtx, c.coord, snapshotTS(c.coord.Clock(), c.st), "redis delta compactor: begin read timestamp") if err != nil { @@ -344,7 +347,7 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error { combined := c.compactBackgroundHandlers(tickCtx, readTS) if tickCtx.Err() == nil { - if err := c.migrateTTLInlineOnce(tickCtx, readTS); err != nil && !errors.Is(err, context.Canceled) { + if err := c.migrateTTLInlineOnce(tickCtx, readTS, snapshotAt); err != nil && !errors.Is(err, context.Canceled) { combined = errors.CombineErrors(combined, err) } } diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 858f11dd5..0bec9e1e3 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -669,7 +669,7 @@ func TestDeltaCompactor_TTLInlineRotatesPlansAcrossTicksThatRunOutOfBudget(t *te for range plans { ctx, cancel := context.WithCancel(context.Background()) st.endTick = cancel - require.ErrorIs(t, c.migrateTTLInlineOnce(ctx, 10), context.Canceled) + require.ErrorIs(t, c.migrateTTLInlineOnce(ctx, 10, time.Now()), context.Canceled) cancel() } @@ -2666,8 +2666,8 @@ func TestDeltaCompactor_TTLInlinePassIgnoresAPageAnotherTickFinished(t *testing. require.True(t, passes[0].Clean) } -// A pass's duration includes its first page: the pass starts when that page's -// scan starts, not when the page has been migrated. +// A pass's duration includes its first page: the pass starts before that +// page's scan, not when the page has been migrated. func TestDeltaCompactor_TTLInlinePassDurationIncludesFirstPage(t *testing.T) { t.Parallel() @@ -2719,3 +2719,49 @@ func (b *lockedLogBuffer) ttlInlinePassDurations(t *testing.T, typeName string) } return out } + +// snapshotTimingCoordinator records when the compactor first asks for its +// clock, which is when SyncOnce acquires the tick's read timestamp, and then +// delays, as a slow read-timestamp acquisition would. +type snapshotTimingCoordinator struct { + *ttlInlinePassTestCoordinator + + once sync.Once + snapshotAt time.Time +} + +func (c *snapshotTimingCoordinator) Clock() *kv.HLC { + c.once.Do(func() { + c.snapshotAt = time.Now() + time.Sleep(10 * time.Millisecond) + }) + return c.ttlInlinePassTestCoordinator.Clock() +} + +// A pass reads its pages at the tick's read timestamp, so it starts no later +// than that timestamp is acquired. Stamping it when the scan began let a route +// change between the two give a pass that read a snapshot from before the +// change a start after it. +func TestDeltaCompactor_TTLInlinePassStartsNoLaterThanItsReadTimestamp(t *testing.T) { + t.Parallel() + + st := &recordingTTLInlineGroupScanStore{MVCCStore: store.NewMVCCStore()} + seedTTLInlinePassStrings(t, st, 3, false) + coord := &snapshotTimingCoordinator{ttlInlinePassTestCoordinator: &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + groupIDs: []uint64{1}, + leadsKeys: true, + }} + observer := &recordingTTLInlinePassObserver{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorTTLInlineMigrationObserver(observer), + ) + + require.NoError(t, c.SyncOnce(context.Background())) + + passes := observer.recordedForType("ttl-inline-string") + require.Len(t, passes, 1) + require.False(t, coord.snapshotAt.IsZero(), "SyncOnce must read the coordinator's clock") + require.LessOrEqual(t, passes[0].startedAt, coord.snapshotAt) +} diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go index b37e381d3..7b2144473 100644 --- a/adapter/redis_ttl_inline_migration_pass.go +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -80,8 +80,8 @@ type ttlInlineMigrationPlanState struct { // openTTLInlineMigrationPassLocked returns the open pass of the plan named // name, starting one if needed. A page that starts at the beginning of the // prefix always opens a new pass unless the open pass has not completed a page -// yet. A pass the page opens starts when the page's scan began. Callers hold -// cursorMu. +// yet. A pass the page opens starts at the page's snapshot time, no later than +// its read timestamp was acquired. Callers hold cursorMu. func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, page ttlInlineMigrationPage) *ttlInlineMigrationPass { state := c.ttlInlinePlans[name] if state == nil { @@ -89,7 +89,7 @@ func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, page ttlI c.ttlInlinePlans[name] = state } if state.pass == nil || (page.fromStart && state.pass.pages > 0) { - state.pass = &ttlInlineMigrationPass{full: page.fromStart, startedAt: page.scannedAt} + state.pass = &ttlInlineMigrationPass{full: page.fromStart, startedAt: page.snapshotAt} } return state.pass } diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 5f12b1891..56a302b56 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -49,7 +49,10 @@ type ttlInlineMigrationTickPlan struct { // 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 { +// +// snapshotAt is a wall-clock time no later than the acquisition of readTS. A +// pass that opens this tick reports it as its start. +func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64, snapshotAt time.Time) error { plans := c.ttlInlineMigrationTickPlans() c.pruneTTLInlineMigrationPlanStates(plans) start := c.rotationStart(&c.ttlInlinePlanCursor, len(plans)) @@ -59,7 +62,7 @@ func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64 return errors.WithStack(errors.CombineErrors(combined, ctx.Err())) } idx := (start + offset) % len(plans) - err := c.migrateTTLInlineScanPlan(ctx, plans[idx].handler, plans[idx].plan, readTS) + err := c.migrateTTLInlineScanPlan(ctx, plans[idx].handler, plans[idx].plan, readTS, snapshotAt) c.advanceRotation(&c.ttlInlinePlanCursor, idx, len(plans)) if err != nil { combined = errors.CombineErrors(combined, err) @@ -212,8 +215,9 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, + snapshotAt time.Time, ) error { - page, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) + page, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS, snapshotAt) if err != nil { return err } @@ -343,8 +347,9 @@ type ttlInlineMigrationPage struct { // startCursor is the plan's cursor when the scan began. A page whose plan // cursor has moved since was also finished by another tick. startCursor []byte - // scannedAt is when the scan began; a pass opened by this page starts then. - scannedAt time.Time + // snapshotAt is no later than the acquisition of the read timestamp the + // page was scanned at; a pass opened by this page starts then. + snapshotAt time.Time } func (c *DeltaCompactor) scanTTLInlineMigrationRaw( @@ -352,8 +357,8 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, + snapshotAt time.Time, ) (ttlInlineMigrationPage, error) { - scannedAt := time.Now() c.cursorMu.Lock() start := h.prefix cur := bytes.Clone(c.cursors[plan.cursorName]) @@ -362,7 +367,7 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( } c.cursorMu.Unlock() - page := ttlInlineMigrationPage{fromStart: len(cur) == 0, startCursor: cur, scannedAt: scannedAt} + page := ttlInlineMigrationPage{fromStart: len(cur) == 0, startCursor: cur, snapshotAt: snapshotAt} end := store.PrefixScanEnd(h.prefix) if plan.scanner != nil { kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index b08edeb7e..84a555419 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -353,9 +353,10 @@ Each pass also has a monitoring outcome: Two metrics per node carry it, labelled `type`, `group` and `outcome`: - `elastickv_redis_ttl_inline_migration_passes_total`; - `elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds`, - the time the latest pass this node completed with that outcome began. It - records the start, not the end, because a pass spanning several ticks read - its first pages at its start. + the time the latest pass this node completed with that outcome began. A pass + begins when the tick that reads its first page starts, before that tick + acquires the read timestamp the page is read at. The gauge records the start, + not the end, because a pass spanning several ticks read its first pages then. **Condition to check before disabling the fallback.** diff --git a/monitoring/redis_ttl_inline.go b/monitoring/redis_ttl_inline.go index 4c32522ff..9162ef812 100644 --- a/monitoring/redis_ttl_inline.go +++ b/monitoring/redis_ttl_inline.go @@ -26,8 +26,8 @@ const ( // migrator completes. A pass covers one anchor type (typeName, such as // "ttl-inline-string") in one data group; group 0 is the whole-prefix scan the // migrator falls back to when the coordinator does not report its groups. -// outcome is one of the RedisTTLInlinePass constants; startedAt is when the -// pass's first page was scanned. +// outcome is one of the RedisTTLInlinePass constants; startedAt is no later +// than the read timestamp of the pass's first page was acquired. type RedisTTLInlineMigrationObserver interface { ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) } From 33ab9415120584bff3ae68102bb5a6493041bca1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 22:48:41 +0000 Subject: [PATCH 8/9] adapter: let one tick at a time run the TTL-inline migration Codex review on #1283 found three more ways for overlapping ticks to corrupt pass reports: - a failed stale page still added its migrated keys; - reports from overlapping ticks could reach the gauge out of order; - on a one-page plan the cursor resets to empty, so a stale page passed the cursor check and was reported as a second pass (ABA). All three come from ticks overlapping, which only happens when SyncOnce is called concurrently (tests); Run calls it serially. The migration now holds ttlInlineMu for the whole tick, and a tick that finds it held skips the migration. That makes the per-page cursor check from 4e2a406 redundant, so it is removed. Regression test: TestDeltaCompactor_TTLInlineTicksDoNotOverlap runs a tick inside the last page's scan, for a one-page plan and a middle page. Before the change the overlapping tick scanned again, and the one-page plan reported two passes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- adapter/redis_delta_compactor.go | 5 ++ adapter/redis_delta_compactor_test.go | 72 +++++++++++++--------- adapter/redis_ttl_inline_migration_pass.go | 7 +-- adapter/redis_ttl_inline_migrator.go | 12 ++-- 4 files changed, 57 insertions(+), 39 deletions(-) diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index bfa49b1a5..4b5ea2f6a 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -88,6 +88,11 @@ type DeltaCompactor struct { ttlInlinePlanCursor int ttlInlineGroupsUnknownOnce sync.Once + + // ttlInlineMu lets one tick at a time run the TTL-inline migration, so a + // plan's pages are scanned, tallied and reported in order. A tick that + // finds it held skips the migration. + ttlInlineMu sync.Mutex } // DeltaCompactorOption configures a DeltaCompactor. diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 0bec9e1e3..b6fedb4e6 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -2632,38 +2632,54 @@ func newHookedTTLInlinePassCompactor(t *testing.T, st *hookedTTLInlineScanStore) return c, logs } -// Two ticks that run at once read the same cursor and migrate the same page. -// Only the first to finish may add that page to the pass; the other's page is -// stale. Counting both inflated the pass's tallies, and on the last page opened -// a second, spurious pass. -func TestDeltaCompactor_TTLInlinePassIgnoresAPageAnotherTickFinished(t *testing.T) { +// TTL-inline migration ticks never overlap. A tick that starts while another +// is migrating skips migration, so a page is scanned, tallied and reported by +// one tick only. Overlapping ticks double-counted a middle page; on a one-page +// plan, where the cursor resets to empty, the second tick even reported the +// same page as a second pass. +func TestDeltaCompactor_TTLInlineTicksDoNotOverlap(t *testing.T) { t.Parallel() - ctx := context.Background() - const keys = 2*ttlInlineMigrationTickScanLimit + 88 - st := &hookedTTLInlineScanStore{MVCCStore: store.NewMVCCStore(), prefix: []byte(redisStrPrefix)} - seedTTLInlinePassStrings(t, st, keys, false) - c, logs := newHookedTTLInlinePassCompactor(t, st) - - // The scan of the second string page runs a whole tick in between, as a - // concurrent SyncOnce would: both ticks read the cursor before page two. - scans, nested := 0, false - st.onScan = func() { - scans++ - if scans == 2 && !nested { - nested = true - require.NoError(t, c.SyncOnce(ctx)) - } - } - for range 3 { - require.NoError(t, c.SyncOnce(ctx)) + tests := []struct { + name string + keys uint64 + ticks int + pages int + }{ + {name: "one-page plan", keys: 3, ticks: 1, pages: 1}, + {name: "middle page", keys: 2*ttlInlineMigrationTickScanLimit + 88, ticks: 3, pages: 3}, } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() - passes := logs.ttlInlinePasses(t, "ttl-inline-string") - require.Len(t, passes, 1) - require.Equal(t, 3, passes[0].Pages) - require.Equal(t, keys, passes[0].KeysScanned) - require.True(t, passes[0].Clean) + ctx := context.Background() + st := &hookedTTLInlineScanStore{MVCCStore: store.NewMVCCStore(), prefix: []byte(redisStrPrefix)} + seedTTLInlinePassStrings(t, st, tc.keys, false) + c, logs := newHookedTTLInlinePassCompactor(t, st) + + // The scan of the last page runs a whole tick in between, as a + // concurrent SyncOnce would. + scans, nested := 0, false + st.onScan = func() { + scans++ + if scans == tc.pages && !nested { + nested = true + require.NoError(t, c.SyncOnce(ctx)) + } + } + for range tc.ticks { + require.NoError(t, c.SyncOnce(ctx)) + } + + require.Equal(t, tc.pages, scans, "the overlapping tick must not scan") + passes := logs.ttlInlinePasses(t, "ttl-inline-string") + require.Len(t, passes, 1) + require.Equal(t, tc.pages, passes[0].Pages) + require.EqualValues(t, tc.keys, passes[0].KeysScanned) + require.True(t, passes[0].Clean) + }) + } } // A pass's duration includes its first page: the pass starts before that diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go index 7b2144473..05d84d81b 100644 --- a/adapter/redis_ttl_inline_migration_pass.go +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -1,7 +1,6 @@ package adapter import ( - "bytes" "context" "log/slog" "time" @@ -109,8 +108,7 @@ func (c *DeltaCompactor) noteTTLInlineMigrationPageFailed(name string, page ttlI // finishTTLInlineMigrationPage adds a migrated page to its plan's pass and // moves the plan's cursor past it. When the page reached the end of the prefix // it closes the pass and returns it with done set, and whether the plan's -// previous pass was clean. A page is dropped when the plan's cursor moved after -// its scan began: another tick that read the same cursor already finished it. +// previous pass was clean. func (c *DeltaCompactor) finishTTLInlineMigrationPage( name string, page ttlInlineMigrationPage, @@ -118,9 +116,6 @@ func (c *DeltaCompactor) finishTTLInlineMigrationPage( ) (pass ttlInlineMigrationPass, prevClean bool, done bool) { c.cursorMu.Lock() defer c.cursorMu.Unlock() - if !bytes.Equal(c.cursors[name], page.startCursor) { - return ttlInlineMigrationPass{}, false, false - } open := c.openTTLInlineMigrationPassLocked(name, page) open.add(tally) open.pages++ diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 56a302b56..6d0e4466e 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -53,6 +53,11 @@ type ttlInlineMigrationTickPlan struct { // snapshotAt is a wall-clock time no later than the acquisition of readTS. A // pass that opens this tick reports it as its start. func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64, snapshotAt time.Time) error { + if !c.ttlInlineMu.TryLock() { + c.logger.DebugContext(ctx, "ttl inline migrator: another tick is migrating; skipping") + return nil + } + defer c.ttlInlineMu.Unlock() plans := c.ttlInlineMigrationTickPlans() c.pruneTTLInlineMigrationPlanStates(plans) start := c.rotationStart(&c.ttlInlinePlanCursor, len(plans)) @@ -344,9 +349,6 @@ type ttlInlineMigrationPage struct { // fromStart reports that the page starts at the beginning of the prefix, // so it is the first page of a pass. fromStart bool - // startCursor is the plan's cursor when the scan began. A page whose plan - // cursor has moved since was also finished by another tick. - startCursor []byte // snapshotAt is no later than the acquisition of the read timestamp the // page was scanned at; a pass opened by this page starts then. snapshotAt time.Time @@ -361,13 +363,13 @@ func (c *DeltaCompactor) scanTTLInlineMigrationRaw( ) (ttlInlineMigrationPage, error) { c.cursorMu.Lock() start := h.prefix - cur := bytes.Clone(c.cursors[plan.cursorName]) + cur := c.cursors[plan.cursorName] if len(cur) > 0 { start = append(bytes.Clone(cur), cursorNextByte) } c.cursorMu.Unlock() - page := ttlInlineMigrationPage{fromStart: len(cur) == 0, startCursor: cur, snapshotAt: snapshotAt} + page := ttlInlineMigrationPage{fromStart: len(cur) == 0, snapshotAt: snapshotAt} end := store.PrefixScanEnd(h.prefix) if plan.scanner != nil { kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) From eedce55632d3b05bef6a47cbfdb80bb53156f06c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 22:48:42 +0000 Subject: [PATCH 9/9] docs: keep a clock-skew margin in the TTL-inline Phase 3 gate Codex review on #1283: pass starts come from each node's wall clock, so a node whose clock runs ahead can record a pass that began before T0 as starting after it. The gate's comparisons, start after T0 and clean after not_clean, now require a margin of S, the deployment's guaranteed maximum clock offset. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01YZANJ6GDMbc1ZaTeoBVXii --- ...2026_04_17_implemented_ttl_inline_value.md | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index 84a555419..5b36ceecd 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -373,19 +373,25 @@ Then all of the following must hold: inline TTLs (Phase 1 or later). 2. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is enabled on every node. A group whose leader runs no compactor gets no pass. +Pass starts come from each node's wall clock, and T0 from whichever clock +recorded the event. Let S be the largest wall-clock offset between any two +nodes (or between a node and the clock T0 was read from) that the deployment +guarantees, for example its NTP bound. Every comparison below keeps a margin +of S, so skew cannot order a pass on the wrong side of an event. + 3. For all eight types and every data group, the latest clean pass across all - nodes started after T0: + nodes started more than S after T0: `max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"})` - has a series for every type and group, each later than T0. A pass that only - ended after T0 does not count: it read its first pages before T0, so a + has a series for every type and group, each later than T0 + S. A pass that + only ended after T0 does not count: it read its first pages before T0, so a route change at T0 could have put a legacy anchor behind its cursor. -4. For each type and group, that clean pass started after every `not_clean` - pass on any node. Leadership moves, so a clean sample from a former leader - does not outrank a later `not_clean` pass on the current one. This query - must return nothing: +4. For each type and group, that clean pass started more than S after every + `not_clean` pass on any node. Leadership moves, so a clean sample from a + former leader does not outrank a later `not_clean` pass on the current one. + With S in seconds, this query must return nothing: ``` - max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="not_clean"}) + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="not_clean"}) + S >= on (type, group) max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"}) ```