TTL-inline: implement cross-leader legacy-row migration

adapter/redis_background_rotation.go

@@ -0,0 +1,46 @@
+package adapter
+
+import (
+	"context"
+	"time"
+
+	"github.com/cockroachdb/errors"
+)
+
+// Delta and migration work share both the timeout and its rotation. Separate
+// rotations still starve migration when any delta handler exhausts every tick.
+func (c *DeltaCompactor) compactBackgroundRotation(ctx context.Context, readTS uint64, snapshotAt time.Time) error {
+	if !c.ttlInlineMu.TryLock() {
+		return nil
+	}
+	defer c.ttlInlineMu.Unlock()
+	plans := c.ttlInlineMigrationTickPlans()
+	c.pruneTTLInlineMigrationPlanStates(plans)
+	var handlers []collectionDeltaHandler
+	if c.coord.IsLeader() {
+		handlers = c.allHandlers()
+	}
+	total := len(handlers) + len(plans)
+	start := c.rotationStart(&c.backgroundCursor, total)
+	var combined error
+	for offset := range total {
+		if ctx.Err() != nil {
+			break
+		}
+		index := (start + offset) % total
+		err := c.runBackgroundSlot(ctx, handlers, plans, index, readTS, snapshotAt)
+		c.advanceRotation(&c.backgroundCursor, index, total)
+		if err != nil && !errors.Is(err, context.Canceled) {
+			combined = errors.CombineErrors(combined, err)
+		}
+	}
+	return errors.WithStack(errors.CombineErrors(combined, ctx.Err()))
+}
+
+func (c *DeltaCompactor) runBackgroundSlot(ctx context.Context, handlers []collectionDeltaHandler, plans []ttlInlineMigrationTickPlan, index int, readTS uint64, snapshotAt time.Time) error {
+	if index < len(handlers) {
+		return c.compactHandlerSafely(ctx, handlers[index], readTS)
+	}
+	plan := plans[index-len(handlers)]
+	return c.migrateTTLInlineScanPlan(ctx, plan.handler, plan.plan, readTS, snapshotAt)
+}

adapter/redis_delta_compactor.go

@@ -81,10 +81,9 @@ type DeltaCompactor struct {
 
 	stateMu      sync.Mutex
 	backoffUntil time.Time
-	// handlerCursor and ttlInlinePlanCursor are where the next tick's
-	// background handlers and TTL-inline scan plans start, so a tick that runs
-	// out of budget hands the rest of the rotation to the next one.
-	handlerCursor       int
+	// backgroundCursor rotates delta handlers and migration plans together.
+	// ttlInlinePlanCursor supports standalone migration passes.
+	backgroundCursor    int
 	ttlInlinePlanCursor int
 
 	ttlInlineGroupsUnknownOnce sync.Once
@@ -350,12 +349,7 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error {
 	tickCtx = readTimestamp.WithDispatchVoucher(tickCtx)
 	readTS := readTimestamp.Timestamp()
 
-	combined := c.compactBackgroundHandlers(tickCtx, readTS)
-	if tickCtx.Err() == nil {
-		if err := c.migrateTTLInlineOnce(tickCtx, readTS, snapshotAt); err != nil && !errors.Is(err, context.Canceled) {
-			combined = errors.CombineErrors(combined, err)
-		}
-	}
+	combined := c.compactBackgroundRotation(tickCtx, readTS, snapshotAt)
 	if deltaCompactorBudgetExhausted(combined, tickCtx, ctx) {
 		c.backoffBackgroundCompaction(c.timeoutBackoff)
 		c.logger.WarnContext(ctx, "delta compactor: backing off after timeout",
@@ -365,28 +359,6 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error {
 	return errors.WithStack(combined)
 }
 
-func (c *DeltaCompactor) compactBackgroundHandlers(ctx context.Context, readTS uint64) error {
-	if !c.coord.IsLeader() {
-		return nil
-	}
-	handlers := c.allHandlers()
-	start := c.rotationStart(&c.handlerCursor, len(handlers))
-	var combined error
-	for offset := range handlers {
-		if ctx.Err() != nil {
-			return errors.WithStack(errors.CombineErrors(combined, ctx.Err()))
-		}
-		idx := (start + offset) % len(handlers)
-		h := handlers[idx]
-		err := c.compactHandlerSafely(ctx, h, readTS)
-		c.advanceRotation(&c.handlerCursor, idx, len(handlers))
-		if err != nil && !errors.Is(err, context.Canceled) {
-			combined = errors.CombineErrors(combined, err)
-		}
-	}
-	return errors.WithStack(combined)
-}
-
 // rotationStart returns where a rotation over total items starts this tick.
 // cursor is one of the compactor's rotation cursors, guarded by stateMu.
 func (c *DeltaCompactor) rotationStart(cursor *int, total int) int {

adapter/redis_delta_compactor_test.go

@@ -1636,7 +1636,7 @@ func TestDeltaCompactor_TTLInlineStopsAfterTimeout(t *testing.T) {
 	require.True(t, bytes.HasPrefix(st.starts[0], []byte(redisStrPrefix)))
 }
 
-func TestDeltaCompactor_RotatesHandlerAfterTimeout(t *testing.T) {
+func TestDeltaCompactor_RotatesAllBackgroundWorkAfterTimeout(t *testing.T) {
 	t.Parallel()
 
 	base := store.NewMVCCStore()
@@ -1655,8 +1655,11 @@ func TestDeltaCompactor_RotatesHandlerAfterTimeout(t *testing.T) {
 		store.HashMetaDeltaPrefix,
 		store.SetMetaDeltaPrefix,
 		store.ZSetMetaDeltaPrefix,
-		store.ListMetaDeltaPrefix,
 	}
+	for _, handler := range c.ttlInlineMigrationHandlers() {
+		wantPrefixes = append(wantPrefixes, string(handler.prefix))
+	}
+	wantPrefixes = append(wantPrefixes, store.ListMetaDeltaPrefix)
 	for _, prefix := range wantPrefixes {
 		err := c.SyncOnce(context.Background())
 		require.ErrorIs(t, err, context.DeadlineExceeded)

adapter/redis_ttl_inline_migrator.go

@@ -46,8 +46,8 @@ type ttlInlineMigrationTickPlan struct {
 
 // migrateTTLInlineOnce runs every handler's scan plans under the tick's
 // budget. The plans share that budget, so each tick starts at the plan after
-// the last one the previous tick ran, as compactBackgroundHandlers rotates its
-// handlers; starting from the first plan every tick starved the later handlers
+// the last one the previous tick ran. SyncOnce uses the combined background
+// rotation instead; starting from the first plan every tick starved later handlers
 // and groups whenever the earlier plans used up the budget.
 //
 // snapshotAt is a wall-clock time no later than the acquisition of readTS. A

distribution/migrator.go

@@ -357,7 +357,7 @@ func SplitJobBracketProgressForPlan(brackets []MigrationBracket, phase SplitJobE
 // interval after applying bracket-local exclusions. Route ownership still
 // requires the caller's RouteKeyFilter for every bracket.
 func (b MigrationBracket) ContainsRawKey(rawKey []byte) bool {
-	if hasAnyExactKey(rawKey, migrationInternalExactKeys) {
+	if store.IsReplicatedControlKey(rawKey) || hasAnyExactKey(rawKey, migrationInternalExactKeys) {
 		return false
 	}
 	if bytes.Compare(rawKey, b.Start) < 0 {

docs/design/2026_09_26_proposed_positioning_and_roadmap.md

@@ -245,7 +245,7 @@ one activated but still draining included, runs
 without these guarantees. The claims also exclude Phase-D clusters
 (`--tsoPhaseDEnabled`) until both TSO admission proposals
 (`2026_08_29_proposed_tso_batch_slot_claims.md`,
-`2026_09_02_proposed_prephase_d_resolution_evidence.md`) have landed:
+`2026_09_02_partial_prephase_d_resolution_evidence.md`) have landed:
 under Phase D the batch-window validator accepts any timestamp inside a
 committed window whether or not it was issued, and the pre-Phase-D
 resolution carve-out accepts a caller-asserted legacy timestamp, so a
@@ -330,7 +330,7 @@ failure on every surface (A6, G11). The A0 reproduction test is the first thing
 its result decides what README may say.
 
 The two open TSO proposals (`2026_08_29_proposed_tso_batch_slot_claims.md`,
-`2026_09_02_proposed_prephase_d_resolution_evidence.md`) protect the
+`2026_09_02_partial_prephase_d_resolution_evidence.md`) protect the
 timestamp-uniqueness assumption the OCC check relies on. They belong to the
 same correctness track and are listed as active in §7. Their order relative
 to the audit's fixes is open (§9), but they gate the README claims for
@@ -381,7 +381,7 @@ more. Can proceed in parallel with 6.2 to 6.4.
 
 | Track | Docs |
 |---|---|
-| Active, correctness | this doc; `2026_09_26_proposed_serializable_isolation_audit.md`; `2026_08_29_proposed_tso_batch_slot_claims.md`; `2026_09_02_proposed_prephase_d_resolution_evidence.md`; the security and benchmark docs to be written |
+| Active, correctness | this doc; `2026_09_26_proposed_serializable_isolation_audit.md`; `2026_08_29_proposed_tso_batch_slot_claims.md`; `2026_09_02_partial_prephase_d_resolution_evidence.md`; the security and benchmark docs to be written |
 | Active, features | `2026_07_19_partial_physical_snapshot_object_offload.md` (M3); `2026_04_25_partial_s3_raft_blob_offload.md` (M3, M4); `2026_06_11_partial_hotspot_split_milestone2_migration.md` with parent `2026_02_18_partial_hotspot_shard_split.md` |
 | Future track (horizontal scale) | `2026_06_23_proposed_scaling_roadmap.md` (its 06/12 predecessor is already marked superseded). Referenced from README's future goals; re-evaluated when the active list is empty. |
 | Deferred | Remaining Stage 9 of `2026_04_29_partial_data_at_rest_encryption.md` (9C+: rotation budget, rewrap / retire / rewrite, `last_proposed_index_per_raft_dek`, encrypted Jepsen). Stage 5E (the capability fan-out helper) is shipped and wired (`buildEncryptionCapabilityFanout` in `main.go`) although that doc's header still says deferred; the header is corrected in the A5 documentation pass. |

docs/design/2026_09_26_proposed_serializable_isolation_audit.md

@@ -5,7 +5,7 @@ Author: bootjp
 Date: 2026-09-26
 Related: `2026_09_26_proposed_positioning_and_roadmap.md` §5 and §6.3 (the
 claim this audit backs), `2026_08_29_proposed_tso_batch_slot_claims.md` and
-`2026_09_02_proposed_prephase_d_resolution_evidence.md` (timestamp
+`2026_09_02_partial_prephase_d_resolution_evidence.md` (timestamp
 uniqueness, which the conflict check assumes).
 
 ## 1. Claim and scope