ddl: preserve cloud storage mode across owner failover

Project: pingcap/tidb | Language: Go | Concepts: consistency-models, error-handling, testing-strategy | Difficulty: mid
Link: https://github.com/pingcap/tidb/pull/70495

The problem

When an ADD INDEX DDL job in TiDB is configured to use cloud storage for ingest, and the DDL owner (the node responsible for schema changes) fails over before the distributed backfill task is submitted, the new owner loses the cloud storage URI that was only cached in memory. As a result, the retrying owner might submit a local-sort task instead of a cloud-storage task, even though the job requires cloud storage. This can leave the job stuck and cause inconsistencies, especially for multi-schema change jobs that create proxy jobs.

Why it's non-trivial

A naive fix might only persist the cloud storage URI or check for its presence, but:

The URI is not persisted in the job itself, only in the in-memory context of the owner.
On failover, the new owner must recover the correct URI from cluster configuration, not just rely on memory.
Multi-schema changes involve proxy jobs that must propagate the cloud-storage mode back to the parent job, or else later steps can silently revert to local mode.
Error handling must ensure that if the URI is missing, the job fails fast rather than hanging in a retry loop.

The approach

1. Recover the cloud storage URI on failover

A new method ensures that when a new DDL owner resumes a job, it reloads the cloud storage URI from the cluster configuration if needed:

+// resolveCloudStorageURI reloads the URI when a new DDL owner resumes the job.
+// UseCloudStorage is persisted in the job, but the URI loaded by loadCloudStorageURI
+// is cached only in the previous owner's in-memory ReorgContext.
+func (w *worker) resolveCloudStorageURI(job *model.Job, mergeTempIndex bool) (string, error) {
+	jc := w.jobContext(job.ID, job.ReorgMeta)
+	if mergeTempIndex || !job.ReorgMeta.UseCloudStorage || jc.cloudStorageURI != "" {
+		return jc.cloudStorageURI, nil
+	}
+
+	jc.cloudStorageURI = handle.GetCloudStorageURI(w.workCtx, w.store)
+	if jc.cloudStorageURI == "" {
+		// Recovering requires the user to restore the cloud storage URI. Retrying could leave the DDL job
+		// in write reorganization for a long time before the configuration problem is noticed, so fail it.
+		return "", dbterror.ErrIngestFailed.GenWithStackByArgs(
+			fmt.Sprintf("cloud storage URI is empty for add-index job %d with cloud storage enabled", job.ID))
+	}
+	return jc.cloudStorageURI, nil
+}

What this does:

If the job requires cloud storage and the cached URI is missing, it fetches the URI from configuration.
If the URI is still missing, it returns a non-retryable error, failing the job fast.

2. Use the resolved URI when creating backfill tasks

The URI is now always resolved before submitting a distributed backfill task:

+		cloudStorageURI, err := w.resolveCloudStorageURI(job, reorgInfo.mergingTmpIdx)
+		if err != nil {
+			return err
+		}
 		taskMeta := &BackfillTaskMeta{
 			Job:             *job.Clone(),
 			EleIDs:          extractElemIDs(reorgInfo),
 			EleTypeKey:      reorgInfo.currElement.TypeKey,
-			CloudStorageURI: w.jobContext(job.ID, job.ReorgMeta).cloudStorageURI,
+			CloudStorageURI: cloudStorageURI,
 			MergeTempIndex:  reorgInfo.mergingTmpIdx,
 			EstimateRowSize: rowSize,
 			Version:         BackfillTaskMetaVersion1,

What this does:

Ensures the correct URI is always used, even after failover.

3. Propagate cloud-storage mode in multi-schema changes

A helper copies the cloud-storage mode from proxy jobs back to the parent job, so subsequent proxies keep the correct mode:

+// updateParentJobFromProxy copies state discovered while executing a temporary
+// proxy job that belongs to the durable parent rather than the SubJob. ToProxyJob
+// gives each proxy a copy of the parent's ReorgMeta, and FromProxyJob does not
+// copy UseCloudStorage back. Without this step, later proxies can revert to local
+// sort, including after a DDL owner failover.
+func updateParentJobFromProxy(parentJob, proxyJob *model.Job) {
+	if parentJob.ReorgMeta != nil && proxyJob.ReorgMeta != nil && proxyJob.ReorgMeta.UseCloudStorage {
+		parentJob.ReorgMeta.UseCloudStorage = true
+	}
+}

And this is called in all relevant multi-schema change paths:

+				updateParentJobFromProxy(job, &proxyJob)

What this does:

Ensures that once any proxy job selects cloud storage, the parent and future proxies remember this choice.

4. Regression and unit tests

The PR adds thorough tests for:

Recovering the URI from configuration after owner cache loss.
Failing fast if the URI is missing.
Ensuring cached URIs take precedence.
Local sort and merge-temp-index behaviors.
Multi-schema change jobs propagating cloud-storage mode.

Example (abbreviated):

+func TestResolveCloudStorageURI(t *testing.T) {
+	// ...setup...
+	t.Run("configured URI recovers empty owner cache", func(t *testing.T) {
+		vardef.CloudStorageURI.Store("s3://bucket")
+		w, jc := newTestWorker("")
+		uri, err := w.resolveCloudStorageURI(newJob(true), false)
+		require.NoError(t, err)
+		require.Equal(t, "s3://bucket/dxf/", uri)
+		require.Equal(t, uri, jc.cloudStorageURI)
+	})
+	// ...other cases...
+}

From the review thread

No notable review discussion on this PR.

Takeaways

Always persist or recover critical configuration needed for job resumption after failover.
Relying on in-memory state is not sufficient for distributed systems where ownership can change.
Propagate operational modes from sub-tasks or proxies back to parent jobs to ensure consistency across retries and failovers.
This prevents silent fallback to incorrect defaults in multi-step or multi-part jobs.
Fail fast and clearly on missing critical configuration, rather than retrying indefinitely.
This helps operators detect and fix misconfigurations quickly, avoiding jobs stuck in limbo.