ddl/ingest: unregister backend after job cancelled

ddl/index.go

@@ -924,7 +924,6 @@ func runIngestReorgJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job,
 			logutil.BgLogger().Warn("[ddl] run reorg job failed, convert job to rollback",
 				zap.String("job", job.String()), zap.Error(err))
 			ver, err = convertAddIdxJob2RollbackJob(d, t, job, tbl.Meta(), indexInfo, err)
-			ingest.LitBackCtxMgr.Unregister(job.ID)
 		}
 		return false, ver, errors.Trace(err)
 	}
@@ -943,7 +942,6 @@ func runIngestReorgJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job,
 			logutil.BgLogger().Warn("[ddl] lightning import error", zap.Error(err))
 			if !errorIsRetryable(err, job) {
 				ver, err = convertAddIdxJob2RollbackJob(d, t, job, tbl.Meta(), indexInfo, err)
-				ingest.LitBackCtxMgr.Unregister(job.ID)
 			}
 		}
 		return false, ver, errors.Trace(err)
@@ -1113,9 +1111,6 @@ func onDropIndex(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		// Finish this job.
 		if job.IsRollingback() {
 			job.FinishTableJob(model.JobStateRollbackDone, model.StateNone, ver, tblInfo)
-			if job.ReorgMeta.ReorgTp == model.ReorgTypeLitMerge {
-				ingest.LitBackCtxMgr.Unregister(job.ID)
-			}
 			job.Args[0] = indexInfo.ID
 		} else {
 			// the partition ids were append by convertAddIdxJob2RollbackJob, it is weird, but for the compatibility,

ddl/ingest/BUILD.bazel

@@ -64,10 +64,11 @@ go_test(
     embed = [":ingest"],
     flaky = True,
     race = "on",
-    shard_count = 12,
+    shard_count = 13,
     deps = [
         "//config",
         "//ddl",
+        "//ddl/internal/callback",
         "//ddl/internal/session",
         "//ddl/testutil",
         "//domain",

ddl/ingest/integration_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/ddl/ingest"
+	"github.com/pingcap/tidb/ddl/internal/callback"
 	"github.com/pingcap/tidb/ddl/testutil"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/errno"
@@ -248,3 +249,29 @@ func TestAddIndexIngestClientError(t *testing.T) {
 	tk.MustExec(`insert into t1(f1) values (cast("null" as json));`)
 	tk.MustGetErrCode("create index i1 on t1((cast(f1 as unsigned array)));", errno.ErrInvalidJSONValueForFuncIndex)
 }
+
+func TestAddIndexCancelOnNoneState(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tkCancel := testkit.NewTestKit(t, store)
+	defer injectMockBackendMgr(t, store)()
+
+	tk.MustExec("use test")
+	tk.MustExec(`create table t (c1 int, c2 int, c3 int)`)
+	tk.MustExec("insert into t values(1, 1, 1);")
+
+	hook := &callback.TestDDLCallback{Do: dom}
+	first := true
+	hook.OnJobRunBeforeExported = func(job *model.Job) {
+		if job.SchemaState == model.StateNone && first {
+			_, err := tkCancel.Exec(fmt.Sprintf("admin cancel ddl jobs %d", job.ID))
+			assert.NoError(t, err)
+			first = false
+		}
+	}
+	dom.DDL().SetHook(hook.Clone())
+	tk.MustGetErrCode("alter table t add index idx1(c1)", errno.ErrCancelledDDLJob)
+	available, err := ingest.LitBackCtxMgr.CheckAvailable()
+	require.NoError(t, err)
+	require.True(t, available)
+}

ddl/ingest/mock.go

@@ -42,8 +42,8 @@ func NewMockBackendCtxMgr(sessCtxProvider func() sessionctx.Context) *MockBacken
 }
 
 // CheckAvailable implements BackendCtxMgr.Available interface.
-func (*MockBackendCtxMgr) CheckAvailable() (bool, error) {
-	return true, nil
+func (m *MockBackendCtxMgr) CheckAvailable() (bool, error) {
+	return len(m.runningJobs) == 0, nil
 }
 
 // Register implements BackendCtxMgr.Register interface.

ddl/rollingback.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/tidb/ddl/ingest"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/parser/ast"
 	"github.com/pingcap/tidb/parser/model"
@@ -76,12 +77,14 @@ func convertAddIdxJob2RollbackJob(d *ddlCtx, t *meta.Meta, job *model.Job, tblIn
 	}
 	job.State = model.JobStateRollingback
 	err = completeErr(err, indexInfo)
+	ingest.LitBackCtxMgr.Unregister(job.ID)
 	return ver, errors.Trace(err)
 }
 
 // convertNotReorgAddIdxJob2RollbackJob converts the add index job that are not started workers to rollingbackJob,
 // to rollback add index operations. job.SnapshotVer == 0 indicates the workers are not started.
 func convertNotReorgAddIdxJob2RollbackJob(d *ddlCtx, t *meta.Meta, job *model.Job, occuredErr error) (ver int64, err error) {
+	defer ingest.LitBackCtxMgr.Unregister(job.ID)
 	schemaID := job.SchemaID
 	tblInfo, err := GetTableInfoAndCancelFaultJob(t, job, schemaID)
 	if err != nil {