distribution: add split job management RPCs

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()

distribution/split_job_lifecycle.go

@@ -0,0 +1,147 @@
+package distribution
+
+import (
+	"bytes"
+	"context"
+
+	"github.com/cockroachdb/errors"
+)
+
+var (
+	ErrCatalogSplitJobNotFound      = errors.New("catalog split job is not found")
+	ErrCatalogSplitJobCannotRetry   = errors.New("catalog split job cannot be retried")
+	ErrCatalogSplitJobCannotAbandon = errors.New("catalog split job cannot be abandoned")
+)
+
+// RetrySplitJobState returns the durable state transition for RetrySplitJob.
+// The retry target must come from the recorded RetryPhase witness; callers must
+// not infer a phase from cursors, timestamps, or diagnostics after restart.
+func RetrySplitJobState(job SplitJob, nowMs int64) (SplitJob, error) {
+	out := CloneSplitJob(job)
+	if out.Phase != SplitJobPhaseFailed || !splitJobRetryPhaseAllowed(out.RetryPhase) {
+		return SplitJob{}, errors.WithStack(ErrCatalogSplitJobCannotRetry)
+	}
+	out.Phase = out.RetryPhase
+	out.RetryPhase = SplitJobPhaseNone
+	out.AbandonFromPhase = SplitJobPhaseNone
+	out.LastError = ""
+	out.UpdatedAtMs = nowMs
+	return out, nil
+}
+
+// BeginSplitJobAbandon records the durable ABANDONING witness before any
+// pre-CUTOVER cleanup side effect is removed.
+func BeginSplitJobAbandon(job SplitJob, nowMs int64) (SplitJob, error) {
+	out := CloneSplitJob(job)
+	from, ok := splitJobAbandonFromPhase(out)
+	if !ok {
+		return SplitJob{}, errors.WithStack(ErrCatalogSplitJobCannotAbandon)
+	}
+	if out.Phase == SplitJobPhaseAbandoning {
+		return out, nil
+	}
+	out.Phase = SplitJobPhaseAbandoning
+	out.RetryPhase = SplitJobPhaseNone
+	out.AbandonFromPhase = from
+	out.UpdatedAtMs = nowMs
+	return out, nil
+}
+
+// RetrySplitJob CASes a FAILED live job back to its recorded retry phase.
+func (s *CatalogStore) RetrySplitJob(ctx context.Context, jobID uint64, nowMs int64) (SplitJob, error) {
+	expected, _, err := s.LiveSplitJobForUpdate(ctx, jobID)
+	if err != nil {
+		return SplitJob{}, err
+	}
+	next, err := RetrySplitJobState(expected, nowMs)
+	if err != nil {
+		return SplitJob{}, err
+	}
+	return next, s.SaveSplitJob(ctx, expected, next)
+}
+
+// BeginSplitJobAbandon CASes a live pre-CUTOVER job into ABANDONING.
+func (s *CatalogStore) BeginSplitJobAbandon(ctx context.Context, jobID uint64, nowMs int64) (SplitJob, error) {
+	expected, _, err := s.LiveSplitJobForUpdate(ctx, jobID)
+	if err != nil {
+		return SplitJob{}, err
+	}
+	next, err := BeginSplitJobAbandon(expected, nowMs)
+	if err != nil {
+		return SplitJob{}, err
+	}
+	if SplitJobsEquivalent(expected, next) {
+		return next, nil
+	}
+	return next, s.SaveSplitJob(ctx, expected, next)
+}
+
+// LiveSplitJobForUpdate reads the latest live split job and the snapshot
+// timestamp used for the read. RPC callers use readTS as a CAS StartTS when
+// proposing the update through Raft.
+func (s *CatalogStore) LiveSplitJobForUpdate(ctx context.Context, jobID uint64) (SplitJob, uint64, error) {
+	if err := ensureCatalogStore(s); err != nil {
+		return SplitJob{}, 0, err
+	}
+	if jobID == 0 {
+		return SplitJob{}, 0, errors.WithStack(ErrCatalogSplitJobIDRequired)
+	}
+	ctx = contextOrBackground(ctx)
+	readTS := s.store.LastCommitTS()
+	job, found, err := s.liveSplitJobAt(ctx, jobID, readTS)
+	if err != nil {
+		return SplitJob{}, 0, err
+	}
+	if !found {
+		return SplitJob{}, 0, errors.WithStack(ErrCatalogSplitJobNotFound)
+	}
+	return job, readTS, nil
+}
+
+func splitJobRetryPhaseAllowed(phase SplitJobPhase) bool {
+	switch phase {
+	case SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy, SplitJobPhaseCutover, SplitJobPhaseCleanup:
+		return true
+	case SplitJobPhaseNone, SplitJobPhasePlanned, SplitJobPhaseDone, SplitJobPhaseFailed, SplitJobPhaseAbandoning, SplitJobPhaseAbandoned:
+		return false
+	}
+	return false
+}
+
+func splitJobAbandonFromPhase(job SplitJob) (SplitJobPhase, bool) {
+	switch job.Phase {
+	case SplitJobPhasePlanned, SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy:
+		return job.Phase, true
+	case SplitJobPhaseFailed:
+		if splitJobAbandonRetryPhaseAllowed(job.RetryPhase) {
+			return job.RetryPhase, true
+		}
+	case SplitJobPhaseAbandoning:
+		if splitJobAbandonRetryPhaseAllowed(job.AbandonFromPhase) {
+			return job.AbandonFromPhase, true
+		}
+	case SplitJobPhaseNone, SplitJobPhaseCutover, SplitJobPhaseCleanup, SplitJobPhaseDone, SplitJobPhaseAbandoned:
+	}
+	return SplitJobPhaseNone, false
+}
+
+func splitJobAbandonRetryPhaseAllowed(phase SplitJobPhase) bool {
+	switch phase {
+	case SplitJobPhasePlanned, SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy:
+		return true
+	case SplitJobPhaseNone, SplitJobPhaseCutover, SplitJobPhaseCleanup, SplitJobPhaseDone, SplitJobPhaseFailed, SplitJobPhaseAbandoning, SplitJobPhaseAbandoned:
+		return false
+	}
+	return false
+}
+
+// SplitJobsEquivalent reports whether two split jobs encode to the same
+// durable catalog value.
+func SplitJobsEquivalent(left, right SplitJob) bool {
+	leftRaw, leftErr := EncodeSplitJob(left)
+	rightRaw, rightErr := EncodeSplitJob(right)
+	if leftErr != nil || rightErr != nil {
+		return false
+	}
+	return bytes.Equal(leftRaw, rightRaw)
+}