migration: add target readiness guard

adapter/internal.go

@@ -254,6 +254,31 @@ func (i *Internal) PromoteStagedVersions(ctx context.Context, req *pb.PromoteSta
 	}, nil
 }
 
+func (i *Internal) ApplyTargetStagedReadiness(ctx context.Context, req *pb.TargetStagedReadinessRequest) (*pb.TargetStagedReadinessResponse, error) {
+	if req == nil {
+		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "target staged readiness request is nil"))
+	}
+	if req.GetJobId() == 0 {
+		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "target staged readiness job_id is required"))
+	}
+	if req.GetMigrationJobId() == 0 {
+		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "target staged readiness migration_job_id is required"))
+	}
+	if i.migrationProposer == nil {
+		return nil, errors.WithStack(status.Error(codes.FailedPrecondition, "target staged readiness proposer is not configured"))
+	}
+	if err := i.verifyInternalLeader(ctx); err != nil {
+		return nil, err
+	}
+	if err := i.verifyMigrationPromoteEnabled(ctx); err != nil {
+		return nil, err
+	}
+	if err := i.proposeTargetStagedReadiness(ctx, req); err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return &pb.TargetStagedReadinessResponse{}, nil
+}
+
 func (i *Internal) verifyMigrationPromoteEnabled(ctx context.Context) error {
 	if i.migrationPromoteGate == nil {
 		return nil
@@ -349,6 +374,24 @@ func (i *Internal) proposeMigrationPromote(ctx context.Context, req *pb.PromoteS
 	}
 }
 
+func (i *Internal) proposeTargetStagedReadiness(ctx context.Context, req *pb.TargetStagedReadinessRequest) error {
+	cmd, err := kv.MarshalTargetStagedReadinessCommand(req)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	resp, err := i.proposeMigrationCommand(ctx, cmd, "target staged readiness")
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if resp == nil {
+		return nil
+	}
+	if err, ok := resp.(error); ok {
+		return errors.WithStack(err)
+	}
+	return errors.WithStack(errors.Newf("unexpected target readiness apply response type %T", resp))
+}
+
 func (i *Internal) proposeMigrationCommand(ctx context.Context, cmd []byte, label string) (any, error) {
 	result, err := i.migrationProposer.Propose(ctx, cmd)
 	if err != nil {

adapter/internal_migration_test.go

@@ -498,3 +498,75 @@ func TestInternalPromoteStagedVersionsAppliesStoreBatch(t *testing.T) {
 	require.ErrorIs(t, err, store.ErrKeyNotFound)
 	require.GreaterOrEqual(t, clock.Current(), uint64(30))
 }
+
+func TestInternalApplyTargetStagedReadinessRejectsWhenOpcodeGateClosed(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	proposer := &applyingMigrationProposer{
+		fsm: kv.NewKvFSMWithHLC(st, nil),
+	}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalStore(st),
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationPromoteGate(func(context.Context) error {
+			return status.Error(codes.FailedPrecondition, "migration opcode disabled for test")
+		}),
+	)
+
+	resp, err := internal.ApplyTargetStagedReadiness(ctx, &pb.TargetStagedReadinessRequest{
+		JobId:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobId:         7,
+		MinWriteTsExclusive:    100,
+		Armed:                  true,
+	})
+	require.Nil(t, resp)
+	require.Error(t, err)
+	require.Equal(t, codes.FailedPrecondition, status.Code(err))
+	require.Equal(t, uint64(0), proposer.calls)
+}
+
+func TestInternalApplyTargetStagedReadinessProposesThroughRaft(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	proposer := &applyingMigrationProposer{
+		fsm: kv.NewKvFSMWithHLC(st, nil),
+	}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalStore(st),
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationPromoteGate(func(context.Context) error { return nil }),
+	)
+
+	_, err := internal.ApplyTargetStagedReadiness(ctx, &pb.TargetStagedReadinessRequest{
+		JobId:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobId:         7,
+		MinWriteTsExclusive:    100,
+		Armed:                  true,
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint64(1), proposer.calls)
+
+	reader, ok := st.(store.MigrationTargetReadinessReader)
+	require.True(t, ok)
+	states, err := reader.MigrationTargetReadinessStates(ctx)
+	require.NoError(t, err)
+	require.Equal(t, []store.TargetStagedReadinessState{{
+		JobID:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobID:         7,
+		MinWriteTSExclusive:    100,
+		Armed:                  true,
+	}}, states)
+}

adapter/redis_compat_commands_stream_test.go

@@ -279,7 +279,7 @@ func TestRedis_StreamXReadLatencyIsConstant(t *testing.T) {
 
 	rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
 	defer func() { _ = rdb.Close() }()
-	ctx := context.Background()
+	ctx := t.Context()
 
 	const (
 		total  = 10_000
@@ -291,13 +291,21 @@ func TestRedis_StreamXReadLatencyIsConstant(t *testing.T) {
 	}
 
 	measure := func() time.Duration {
-		start := time.Now()
-		streams, err := rdb.XRead(ctx, &redis.XReadArgs{
-			Streams: []string{"stream-lat", lastID},
-			Count:   10,
-			Block:   10 * time.Millisecond,
-		}).Result()
-		elapsed := time.Since(start)
+		var (
+			streams []redis.XStream
+			elapsed time.Duration
+		)
+		err := retryNotLeader(ctx, func() error {
+			start := time.Now()
+			var xerr error
+			streams, xerr = rdb.XRead(ctx, &redis.XReadArgs{
+				Streams: []string{"stream-lat", lastID},
+				Count:   10,
+				Block:   10 * time.Millisecond,
+			}).Result()
+			elapsed = time.Since(start)
+			return xerr
+		})
 		require.True(t, errors.Is(err, redis.Nil) || err == nil)
 		require.Empty(t, streams)
 		return elapsed

adapter/redis_delta_compactor_test.go

@@ -499,27 +499,20 @@ func TestDeltaCompactor_UrgentCompactionPagination(t *testing.T) {
 		require.NoError(t, st.PutAt(ctx, dKey, delta, i+1, 0))
 	}
 
-	// Queue and process the urgent compaction.
-	c.TriggerUrgentCompaction("hash", userKey)
-
-	runCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	go func() { _ = c.Run(runCtx) }()
+	// Exercise the urgent pagination loop directly. Running through c.Run would
+	// also start an initial background SyncOnce; the local test coordinator applies
+	// elems one-by-one instead of atomically, so a test read can observe the meta
+	// update before all delete elems have been applied under the race detector.
+	c.compactUrgentKey(ctx, urgentCompactionRequest{typeName: "hash", userKey: userKey})
 
-	// Wait until the base meta holds the accumulated total.
-	// The pagination loop should take two passes: first 1025, then 9.
-	require.Eventually(t, func() bool {
-		readTS := st.LastCommitTS()
-		raw, err := st.GetAt(ctx, store.HashMetaKey(userKey), readTS)
-		if err != nil {
-			return false
-		}
-		got, err := store.UnmarshalHashMeta(raw)
-		return err == nil && got.Len == int64(totalDeltasU64)
-	}, 5*time.Second, 20*time.Millisecond, "all %d delta keys should be compacted into base meta", totalDeltasU64)
+	readTS := st.LastCommitTS()
+	raw, err := st.GetAt(ctx, store.HashMetaKey(userKey), readTS)
+	require.NoError(t, err)
+	got, err := store.UnmarshalHashMeta(raw)
+	require.NoError(t, err)
+	require.Equal(t, int64(totalDeltasU64), got.Len, "all %d delta keys should be compacted into base meta", totalDeltasU64)
 
 	// No delta keys should remain after pagination compaction.
-	readTS := st.LastCommitTS()
 	prefix := store.HashMetaDeltaScanPrefix(userKey)
 	end := store.PrefixScanEnd(prefix)
 	remaining, err := st.ScanAt(ctx, prefix, end, int(totalDeltasU64)+1, readTS)

distribution/split_job_catalog.go

@@ -36,6 +36,7 @@ var (
 	ErrCatalogSplitJobConflict            = errors.New("catalog split job conflict")
 	ErrCatalogSplitJobTerminalRequired    = errors.New("catalog split job terminal state is required")
 	ErrSplitJobOverlap                    = errors.New("split job overlaps requested route")
+	ErrTooManyInFlightSplitJobs           = errors.New("too many in-flight split jobs")
 )
 
 // SplitJobPhase is the durable phase of a split migration job.
@@ -844,6 +845,11 @@ func CloneSplitJob(job SplitJob) SplitJob {
 	}
 }
 
+// SplitJobToProto converts a catalog SplitJob into its wire representation.
+func SplitJobToProto(job SplitJob) *pb.SplitJob {
+	return splitJobToProto(job)
+}
+
 func splitJobToProto(job SplitJob) *pb.SplitJob {
 	job = CloneSplitJob(job)
 	return &pb.SplitJob{

distribution/split_job_catalog_test.go

@@ -242,6 +242,98 @@ func TestCatalogStoreSaveSplitJobRejectsStaleExpectedJob(t *testing.T) {
 	assertSplitJobEqual(t, advanced, got)
 }
 
+func TestCatalogStoreRetrySplitJobUsesDurableRetryPhase(t *testing.T) {
+	cs := NewCatalogStore(store.NewMVCCStore())
+	ctx := context.Background()
+	job := sampleSplitJob(18)
+	job.Phase = SplitJobPhaseFailed
+	job.RetryPhase = SplitJobPhaseDeltaCopy
+	job.LastError = "transient"
+
+	if err := cs.CreateSplitJob(ctx, job); err != nil {
+		t.Fatalf("create split job: %v", err)
+	}
+	retried, err := cs.RetrySplitJob(ctx, job.JobID, 1200)
+	if err != nil {
+		t.Fatalf("retry split job: %v", err)
+	}
+	if retried.Phase != SplitJobPhaseDeltaCopy || retried.RetryPhase != SplitJobPhaseNone {
+		t.Fatalf("unexpected retry transition: %+v", retried)
+	}
+	if retried.AbandonFromPhase != SplitJobPhaseNone || retried.LastError != "" || retried.UpdatedAtMs != 1200 {
+		t.Fatalf("unexpected retry witness cleanup: %+v", retried)
+	}
+	got, found, err := cs.SplitJob(ctx, job.JobID)
+	if err != nil {
+		t.Fatalf("load retried split job: %v", err)
+	}
+	if !found {
+		t.Fatal("expected retried split job")
+	}
+	assertSplitJobEqual(t, retried, got)
+}
+
+func TestCatalogStoreRetrySplitJobRejectsMissingRetryWitness(t *testing.T) {
+	cs := NewCatalogStore(store.NewMVCCStore())
+	ctx := context.Background()
+	job := sampleSplitJob(19)
+	job.Phase = SplitJobPhaseFailed
+	job.RetryPhase = SplitJobPhaseNone
+
+	if err := cs.CreateSplitJob(ctx, job); err != nil {
+		t.Fatalf("create split job: %v", err)
+	}
+	if _, err := cs.RetrySplitJob(ctx, job.JobID, 1200); !errors.Is(err, ErrCatalogSplitJobCannotRetry) {
+		t.Fatalf("expected ErrCatalogSplitJobCannotRetry, got %v", err)
+	}
+}
+
+func TestCatalogStoreBeginSplitJobAbandonRecordsPreCutoverPhase(t *testing.T) {
+	cs := NewCatalogStore(store.NewMVCCStore())
+	ctx := context.Background()
+	job := sampleSplitJob(20)
+	job.Phase = SplitJobPhaseFailed
+	job.RetryPhase = SplitJobPhaseFence
+	job.AbandonFromPhase = SplitJobPhaseNone
+
+	if err := cs.CreateSplitJob(ctx, job); err != nil {
+		t.Fatalf("create split job: %v", err)
+	}
+	abandoning, err := cs.BeginSplitJobAbandon(ctx, job.JobID, 1300)
+	if err != nil {
+		t.Fatalf("begin split job abandon: %v", err)
+	}
+	if abandoning.Phase != SplitJobPhaseAbandoning ||
+		abandoning.RetryPhase != SplitJobPhaseNone ||
+		abandoning.AbandonFromPhase != SplitJobPhaseFence ||
+		abandoning.UpdatedAtMs != 1300 {
+		t.Fatalf("unexpected abandon transition: %+v", abandoning)
+	}
+	got, found, err := cs.SplitJob(ctx, job.JobID)
+	if err != nil {
+		t.Fatalf("load abandoning split job: %v", err)
+	}
+	if !found {
+		t.Fatal("expected abandoning split job")
+	}
+	assertSplitJobEqual(t, abandoning, got)
+}
+
+func TestCatalogStoreBeginSplitJobAbandonRejectsPostCutoverPhases(t *testing.T) {
+	cs := NewCatalogStore(store.NewMVCCStore())
+	ctx := context.Background()
+	job := sampleSplitJob(21)
+	job.Phase = SplitJobPhaseCleanup
+	job.RetryPhase = SplitJobPhaseNone
+
+	if err := cs.CreateSplitJob(ctx, job); err != nil {
+		t.Fatalf("create split job: %v", err)
+	}
+	if _, err := cs.BeginSplitJobAbandon(ctx, job.JobID, 1300); !errors.Is(err, ErrCatalogSplitJobCannotAbandon) {
+		t.Fatalf("expected ErrCatalogSplitJobCannotAbandon, got %v", err)
+	}
+}
+
 func TestCatalogStoreListSplitJobsIncludesLiveAndHistory(t *testing.T) {
 	cs := NewCatalogStore(store.NewMVCCStore())
 	ctx := context.Background()