From 10cd30c66e50ce06e625f54f692adcaf8c578265 Mon Sep 17 00:00:00 2001 From: "yihao.dai" Date: Mon, 20 Jul 2026 10:48:41 +0800 Subject: [PATCH] fix: allow aborting a failed CDC 2PC import to release the peer (#51083) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary In CDC replication mode, a 2PC bulk import (`auto_commit=false`) runs **independently** on the primary (source) and standby clusters against the same files. Only the source can broadcast `Commit`/`Rollback`; the standby replays them via CDC. When the import fails on **one** cluster but the other holds a replicated `Uncommitted` job, that job had no recovery path: - **Mode 1 (source fails, standby `Uncommitted`):** the failure never broadcast a `RollbackImport`, and `AbortImport` on a terminal-`Failed` source was rejected — so the standby stayed `Uncommitted` indefinitely, holding invisible imported segments. (Observed in production: standby stuck 2+ days.) - **Mode 2 (standby fails, source `Uncommitted`):** if the platform commits the source before confirming the standby is prepared, the source commits while the standby's commit ack no-ops on its `Failed` job → silent divergence. issue: #51070 ## What this changes (datacoord) - **Abort a failed source (`AbortImport`):** a `Failed` source (its own import failed) is now abortable — it broadcasts `RollbackImport` to release the peer's `Uncommitted` job. `Committing`/`Completed` stay rejected. Lets an operator recover **immediately**, instead of waiting for GC. - **GC self-heal (`importChecker.checkGC`):** before GC removes a `Failed` replicate-source job, it broadcasts `RollbackImport` so the peer is released **automatically at retention expiry**. Transient broadcast failure keeps the job and retries on the next tick; a standby's `ErrNotPrimary` proceeds; **job removal itself is the idempotency guard** (once gone, never re-broadcast). This closes the recovery-window gap without a persisted flag or proto change. - **Divergence signal (`CommitImport` ack):** a commit ack landing on a `Failed` replica now logs at WARN (Mode 2). - `isReplicatingClusterNow`: the GC decision now **propagates** replication-check errors instead of tolerating them — an indeterminate status (transient balancer error, `OnShutdownError`) retains the job and retries next tick, since a false "not replicating" would irreversibly strand the replicating peer; only permanent broadcast errors (`ErrNotPrimary` from a standby, `ErrCollectionNotFound` after a replicated drop) proceed to GC. The balancer access is bounded by a 10s timeout so an unregistered balancer cannot park the checker loop. - Minor: the coordinator callbacks the checker needs are bundled into an `importCheckerHooks` struct instead of a growing positional-arg list. ## Recovery guarantee The peer's `Uncommitted` job is released either **immediately** via `AbortImport`, or **automatically** when the failed source reaches GC retention — no permanent stranding, and no dependence on a manual abort within the 3h retention window. ## Not in scope (follow-ups) - **Idempotent re-broadcast dedup:** repeated `AbortImport` on a real-failure source re-broadcasts each time (harmless — flusher/standby-ack no-op). A persisted `rollback_broadcasted` flag would dedup this. - **Eager self-heal:** broadcasting the rollback at *failure* time (in `checkFailedJob`) rather than at GC time, to shorten peer-stranding to seconds. - **Mode 2 prevention:** a standby→coordinator prepared-vote so the source refuses to commit unless all replicas are prepared. ## Test plan - `checkGC` self-heal: failed replicate-source broadcasts-then-removes with transient-retry; `ErrNotPrimary` (standby) proceeds; non-replicating skips the broadcast. - `AbortImport`: failed-source broadcasts rollback; `Committing` and `Completed` rejected; previously-user-aborted is idempotent. - CI `ci-v2/ut-go` compiles and runs these tests. (Local datacoord UT was not run for this revision due to cgo core drift in the worktree.) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: bigsheeper Co-authored-by: Claude Opus 4.8 (1M context) --- internal/datacoord/ddl_callbacks_import.go | 28 +- internal/datacoord/import_checker.go | 157 +++++++++-- internal/datacoord/import_checker_test.go | 295 ++++++++++++++++++++- internal/datacoord/server.go | 6 +- internal/datacoord/services.go | 37 ++- internal/datacoord/services_test.go | 66 +++++ 6 files changed, 543 insertions(+), 46 deletions(-) diff --git a/internal/datacoord/ddl_callbacks_import.go b/internal/datacoord/ddl_callbacks_import.go index 2dd65aef4d..1033f7a579 100644 --- a/internal/datacoord/ddl_callbacks_import.go +++ b/internal/datacoord/ddl_callbacks_import.go @@ -145,6 +145,27 @@ func isReplicatingCluster(cfg *commonpb.ReplicateConfiguration) bool { return cfg != nil && (len(cfg.GetCrossClusterTopology()) > 0 || len(cfg.GetClusters()) > 1) } +// isReplicatingClusterNow reports whether this cluster is currently part of a CDC +// replication topology. A non-nil error means the status could not be determined (e.g. a +// transient balancer error, or OnShutdownError while streamingcoord is stopping before +// datacoord); the caller must treat that as indeterminate rather than "not replicating", +// because at GC time a false "not replicating" would irreversibly drop a replicating job +// without releasing the peer. A nil assignment is an unambiguous "not replicating". +func (s *Server) isReplicatingClusterNow(ctx context.Context) (bool, error) { + balancer, err := balance.GetWithContext(ctx) + if err != nil { + return false, err + } + assignment, err := balancer.GetLatestChannelAssignment() + if err != nil { + return false, err + } + if assignment == nil { + return false, nil + } + return isReplicatingCluster(assignment.ReplicateConfiguration), nil +} + // broadcastImport broadcasts the import message to all vchannels. // This method is called from the new ImportV2 flow where proxy calls DataCoord directly. func (s *Server) broadcastImport(ctx context.Context, @@ -235,8 +256,11 @@ func (c *DDLCallbacks) commitImportV2AckCallback(ctx context.Context, result mes mlog.FieldJobID(jobID), mlog.String("state", job.GetState().String())) return nil case internalpb.ImportJobState_Failed: - mlog.Info(ctx, "CommitImport: job already failed, no-op", - mlog.FieldJobID(jobID)) + // Divergence signal: the source committed but this replica already failed, so + // this replica will NOT make the data visible. Left as a no-op here; surfaced + // at WARN for alerting. + mlog.Warn(ctx, "CommitImport ack landed on a Failed import job; this replica will NOT commit while the source commits — potential primary/standby divergence", + mlog.FieldJobID(jobID), mlog.String("reason", job.GetReason())) return nil default: // CommitImport may be replicated before the local import task reaches diff --git a/internal/datacoord/import_checker.go b/internal/datacoord/import_checker.go index e74985d113..89060f3638 100644 --- a/internal/datacoord/import_checker.go +++ b/internal/datacoord/import_checker.go @@ -22,16 +22,19 @@ import ( "sync" "time" + "github.com/cockroachdb/errors" "github.com/samber/lo" "github.com/milvus-io/milvus/internal/datacoord/allocator" "github.com/milvus-io/milvus/internal/datacoord/broker" + "github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster" "github.com/milvus-io/milvus/internal/util/importutilv2" "github.com/milvus-io/milvus/pkg/v3/metrics" "github.com/milvus-io/milvus/pkg/v3/mlog" "github.com/milvus-io/milvus/pkg/v3/proto/datapb" "github.com/milvus-io/milvus/pkg/v3/proto/internalpb" "github.com/milvus-io/milvus/pkg/v3/util/funcutil" + "github.com/milvus-io/milvus/pkg/v3/util/merr" "github.com/milvus-io/milvus/pkg/v3/util/tsoutil" ) @@ -40,6 +43,23 @@ type ImportChecker interface { Close() } +// importCheckerHooks bundles the coordinator callbacks the import checker invokes, +// injected as one named unit (instead of a growing positional-arg list) so the checker +// does not depend on *Server. A nil callback disables the corresponding behavior; tests +// inject only the hooks they exercise. +type importCheckerHooks struct { + // commitImport broadcasts a CommitImport WAL message. Required in production; a nil + // value is a programming error only when reached on the auto_commit=true path. + commitImport func(ctx context.Context, job ImportJob) error + // rollbackImport broadcasts a RollbackImport WAL message. nil disables GC self-heal. + rollbackImport func(ctx context.Context, job ImportJob) error + // isReplicatingCluster reports whether this cluster is currently replicating. A + // non-nil error means the status is indeterminate (e.g. a transient balancer error + // during shutdown) and the caller must not make an irreversible GC decision. nil hook + // is treated as "not replicating" (GC self-heal disabled). + isReplicatingCluster func(ctx context.Context) (bool, error) +} + type importChecker struct { ctx context.Context meta *meta @@ -49,9 +69,7 @@ type importChecker struct { ci CompactionInspector handler Handler - // commitImportFn broadcasts a CommitImport WAL message. - // Injected at construction time so the checker does not depend on *Server. - commitImportFn func(ctx context.Context, job ImportJob) error + hooks importCheckerHooks closeOnce sync.Once closeChan chan struct{} @@ -64,35 +82,44 @@ func NewImportChecker(ctx context.Context, importMeta ImportMeta, ci CompactionInspector, handler Handler, - commitImportFn func(ctx context.Context, job ImportJob) error, // required; nil OK for tests + hooks importCheckerHooks, ) ImportChecker { return &importChecker{ - ctx: ctx, - meta: meta, - broker: broker, - alloc: alloc, - importMeta: importMeta, - ci: ci, - handler: handler, - commitImportFn: commitImportFn, - closeChan: make(chan struct{}), + ctx: ctx, + meta: meta, + broker: broker, + alloc: alloc, + importMeta: importMeta, + ci: ci, + handler: handler, + hooks: hooks, + closeChan: make(chan struct{}), } } +// Start runs the checker loops until Close. The state-machine loop and the +// timeout/GC loop deliberately run on separate goroutines: checkGC's rollback +// broadcast can park on the ctx-insensitive resource-key lock (see checkGC), and +// isolating it guarantees the state machine keeps making progress no matter how +// long GC blocks. All state shared by the two loops lives behind importMeta's +// mutex (which already serves concurrent RPC and ack-callback goroutines), and +// UpdateJob refuses transitions out of Completed/Failed, so the loops cannot +// resurrect or regress each other's terminal states. func (c *importChecker) Start() { mlog.Info(c.ctx, "start import checker") - var ( - ticker1 = time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalHigh.GetAsDuration(time.Second)) // 2s - ticker2 = time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalLow.GetAsDuration(time.Second)) // 2min - ) - defer ticker1.Stop() - defer ticker2.Stop() + go c.runGCLoop() + c.runStateMachineLoop() +} + +func (c *importChecker) runStateMachineLoop() { + ticker := time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalHigh.GetAsDuration(time.Second)) // 2s + defer ticker.Stop() for { select { case <-c.closeChan: - mlog.Info(c.ctx, "import checker exited") + mlog.Info(c.ctx, "import checker state-machine loop exited") return - case <-ticker1.C: + case <-ticker.C: jobs := c.importMeta.GetJobBy(c.ctx) for _, job := range jobs { if !funcutil.SliceSetEqual[string](job.GetVchannels(), job.GetReadyVchannels()) { @@ -122,7 +149,19 @@ func (c *importChecker) Start() { c.checkFailedJob(job) } } - case <-ticker2.C: + } + } +} + +func (c *importChecker) runGCLoop() { + ticker := time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalLow.GetAsDuration(time.Second)) // 2min + defer ticker.Stop() + for { + select { + case <-c.closeChan: + mlog.Info(c.ctx, "import checker gc loop exited") + return + case <-ticker.C: jobs := c.importMeta.GetJobBy(c.ctx) for _, job := range jobs { c.tryTimeoutJob(job) @@ -469,11 +508,11 @@ func (c *importChecker) checkUncommittedJob(job ImportJob) { // collection-level resource-key lock serializes overlapping broadcasts, the // ack callback only transitions when the job is still Uncommitted, and // HandleCommitVchannel is idempotent on committed_vchannels. - if c.commitImportFn == nil { - log.Error(c.ctx, "commitImportFn is nil but auto_commit=true; this is a programming error") + if c.hooks.commitImport == nil { + log.Error(c.ctx, "commit hook is nil but auto_commit=true; this is a programming error") return } - if err := c.commitImportFn(c.ctx, job); err != nil { + if err := c.hooks.commitImport(c.ctx, job); err != nil { log.Warn(c.ctx, "auto-commit broadcast failed, will retry on next tick", mlog.Err(err)) } } @@ -601,6 +640,57 @@ func (c *importChecker) checkGC(job ImportJob) { if !shouldRemoveJob { return } + // In a CDC replicating cluster, a failed 2PC source import must release the + // peer cluster's replicated Uncommitted job before we drop it — otherwise the + // peer is stranded with invisible imported segments and no recovery path, since + // source GC never touches the peer. Removal of the job is itself the idempotency + // guard: once gone we never re-broadcast. Auto-commit jobs have no 2PC peer to + // release, so they skip the gate entirely. + if c.hooks.rollbackImport != nil && c.hooks.isReplicatingCluster != nil && + job.GetState() == internalpb.ImportJobState_Failed && !job.GetAutoCommit() { + // The check reaches the streaming balancer future, which blocks until the + // balancer is registered — under the server-lifetime c.ctx that would park + // the GC loop during the window before streamingcoord registers + // it (e.g. a restart recovering a job already past retention). Bound it like + // checkCollection does; a timeout is just another indeterminate status. + replicateCheckCtx, cancel := context.WithTimeout(c.ctx, 10*time.Second) + replicating, err := c.hooks.isReplicatingCluster(replicateCheckCtx) + cancel() + switch { + case err != nil: + // Indeterminate replication status (e.g. a transient balancer error during + // shutdown, when streamingcoord stops before datacoord). Removing the job now + // could strand a replicating peer's Uncommitted job with no recovery path, + // which is irreversible — a false "not replicating" costs nothing but a retry, + // so keep the job and re-evaluate on the next GC tick. + log.Warn(c.ctx, "cannot determine replication status before GC of failed import job, will retry", mlog.Err(err)) + return + case replicating: + // Broadcast the RollbackImport to release the peer. A transient error keeps + // the job to retry next tick; a permanent error (standby ErrNotPrimary, or the + // collection was dropped — itself a replicated DDL, so the peer fails its own + // job independently) falls through to GC, since retrying it forever would leak + // the job's metadata. + // + // Bound the broadcast like the replication check above: it blocks in + // BlockUntilDone until every vchannel append succeeds, and under the + // server-lifetime c.ctx an unavailable streamingnode would park this + // loop until shutdown. A timeout is just another transient status — + // keep the job and retry on the next GC tick. The resource-key lock on + // the broadcast path is still ctx-insensitive (making it fail-fast is a + // follow-up), which is one reason this GC loop runs on its own + // goroutine (see Start): even an unbounded park here can only delay + // GC, never the import state machine. + rollbackCtx, rollbackCancel := context.WithTimeout(c.ctx, 10*time.Second) + err := c.hooks.rollbackImport(rollbackCtx, job) + rollbackCancel() + if err != nil && !isPermanentRollbackErr(err) { + log.Warn(c.ctx, "failed to broadcast rollback before GC of failed replicate import job, will retry", mlog.Err(err)) + return + } + log.Info(c.ctx, "proceeding with GC of failed replicate import job after rollback attempt") + } + } err := c.importMeta.RemoveJob(c.ctx, job.GetJobID()) if err != nil { log.Warn(c.ctx, "remove import job failed", mlog.Err(err)) @@ -609,3 +699,20 @@ func (c *importChecker) checkGC(job ImportJob) { log.Info(c.ctx, "import job removed") } } + +// isPermanentRollbackErr reports whether a RollbackImport broadcast error is permanent, +// i.e. retrying it can never succeed, so the failed job should still be GC'd rather than +// retried forever (which would leak its metadata). Everything else is treated as transient +// and retried on the next GC tick — misclassifying a transient error as permanent would +// drop a replicating job without releasing the peer, which is irreversible. +func isPermanentRollbackErr(err error) bool { + // ErrNotPrimary: this cluster is a replication standby, not the primary that owns the + // broadcast; its own failed job is independent and safe to drop. + // ErrCollectionNotFound: the collection was dropped. DropCollection is itself a + // replicated DDL, so the peer marks its own import job Failed independently — there is + // no peer left to release, and the broadcast can never succeed. + // errRollbackImportNoVchannels: the job carries no vchannels (fixed at creation), so + // the broadcast has no peer to address and can never succeed. + return errors.Is(err, broadcaster.ErrNotPrimary) || errors.Is(err, merr.ErrCollectionNotFound) || + errors.Is(err, errRollbackImportNoVchannels) +} diff --git a/internal/datacoord/import_checker_test.go b/internal/datacoord/import_checker_test.go index 798a1fecfd..15a37dcdb6 100644 --- a/internal/datacoord/import_checker_test.go +++ b/internal/datacoord/import_checker_test.go @@ -33,10 +33,12 @@ import ( "github.com/milvus-io/milvus/internal/datacoord/allocator" broker2 "github.com/milvus-io/milvus/internal/datacoord/broker" "github.com/milvus-io/milvus/internal/metastore/mocks" + "github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster" "github.com/milvus-io/milvus/pkg/v3/mlog" "github.com/milvus-io/milvus/pkg/v3/proto/datapb" "github.com/milvus-io/milvus/pkg/v3/proto/internalpb" "github.com/milvus-io/milvus/pkg/v3/proto/rootcoordpb" + "github.com/milvus-io/milvus/pkg/v3/util/merr" "github.com/milvus-io/milvus/pkg/v3/util/paramtable" "github.com/milvus-io/milvus/pkg/v3/util/timerecord" "github.com/milvus-io/milvus/pkg/v3/util/tsoutil" @@ -88,7 +90,7 @@ func (s *ImportCheckerSuite) SetupTest() { }, nil }).Maybe() - checker := NewImportChecker(context.TODO(), meta, broker, s.alloc, importMeta, ci, handler, nil).(*importChecker) + checker := NewImportChecker(context.TODO(), meta, broker, s.alloc, importMeta, ci, handler, importCheckerHooks{}).(*importChecker) s.checker = checker job := &importJob{ @@ -508,6 +510,277 @@ func (s *ImportCheckerSuite) TestCheckGC() { s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) } +// setupGCReadyFailedJob puts the suite's job into a Failed, past-cleanup-ts state with +// a single import task that has no live segments and is unassigned, so checkGC is one +// step away from removing the job (the only remaining gate is the replicate rollback). +func (s *ImportCheckerSuite) setupGCReadyFailedJob() { + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().SaveImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).Return(nil) + + taskProto := &datapb.ImportTaskV2{ + JobID: s.jobID, + TaskID: 1, + State: datapb.ImportTaskStateV2_Failed, + NodeID: NullNodeID, + } + task := &importTask{tr: timerecord.NewTimeRecorder("import task")} + task.task.Store(taskProto) + s.NoError(s.importMeta.AddTask(context.TODO(), task)) + + s.NoError(s.importMeta.UpdateJob(context.TODO(), s.jobID, UpdateJobState(internalpb.ImportJobState_Failed))) + GCRetention := Params.DataCoordCfg.ImportTaskRetention.GetAsDuration(time.Second) + job := s.importMeta.GetJob(context.TODO(), s.jobID) + job.(*importJob).CleanupTs = tsoutil.AddPhysicalDurationOnTs(job.GetCleanupTs(), GCRetention*-2) + s.NoError(s.importMeta.AddJob(context.TODO(), job)) +} + +// A failed source in a replicating cluster must broadcast RollbackImport before its job +// is GC'd. A transient broadcast error keeps the job alive to retry; once it succeeds the +// job is removed. +func (s *ImportCheckerSuite) TestCheckGCReplicateSourceBroadcastsRollback() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + rollbackCalls := 0 + rollbackErr := errors.New("broadcast failed") + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + rollbackCalls++ + return rollbackErr + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + // First tick: broadcast fails → job retained (tasks already removed), rollback attempted. + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(1, rollbackCalls) + s.Equal(1, len(s.importMeta.GetJobBy(context.TODO()))) + + // Next tick: broadcast succeeds → job removed. + rollbackErr = nil + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(2, rollbackCalls) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// A standby (not the replication primary) gets ErrNotPrimary from the broadcast; that is +// treated as success so its own failed job is still GC'd. +func (s *ImportCheckerSuite) TestCheckGCReplicateNotPrimaryProceeds() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { return broadcaster.ErrNotPrimary } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// A non-replicating cluster must not broadcast any rollback; the failed job is GC'd as before. +func (s *ImportCheckerSuite) TestCheckGCNonReplicatingSkipsRollback() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + rollbackCalls := 0 + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + rollbackCalls++ + return nil + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return false, nil } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(0, rollbackCalls) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// The replication check reaches the streaming balancer future, which blocks until the +// balancer is registered; checkGC must pass a deadline-bounded ctx so an unregistered +// balancer (datacoord ready before streamingcoord) cannot park the whole checker loop. +func (s *ImportCheckerSuite) TestCheckGCReplicateCheckCtxIsBounded() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { return nil } + hasDeadline := false + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { + _, hasDeadline = ctx.Deadline() + return false, nil + } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.True(hasDeadline) +} + +// When the replication status cannot be determined (e.g. a transient balancer error during +// shutdown), GC must NOT drop the job: a false "not replicating" would strand a replicating +// peer with no recovery path. The job is retained and no rollback is broadcast. +func (s *ImportCheckerSuite) TestCheckGCReplicateIndeterminateRetainsJob() { + s.setupGCReadyFailedJob() + // The task-cleanup loop runs before the replication gate, so the task is removed + // even though the job itself is retained; DropImportJob must NOT be called. + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + + rollbackCalls := 0 + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + rollbackCalls++ + return nil + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { + return false, errors.New("balancer not ready") + } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(0, rollbackCalls) + s.Equal(1, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// A permanent rollback error (e.g. the collection was dropped → ErrCollectionNotFound) must +// NOT be retried forever, which would leak the job's metadata. The job is GC'd instead. +func (s *ImportCheckerSuite) TestCheckGCReplicatePermanentRollbackErrRemovesJob() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + rollbackCalls := 0 + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + rollbackCalls++ + return merr.WrapErrCollectionNotFound(job.GetCollectionID()) + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(1, rollbackCalls) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// The rollback broadcast can block until every vchannel append succeeds (or forever on +// an unavailable streamingnode) under the server-lifetime c.ctx; checkGC must pass a +// deadline-bounded ctx so a stuck broadcast cannot park the whole checker loop. +func (s *ImportCheckerSuite) TestCheckGCReplicateRollbackCtxIsBounded() { + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + hasDeadline := false + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + _, hasDeadline = ctx.Deadline() + return nil + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.True(hasDeadline) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// A job without vchannels can never deliver its rollback (Vchannels are fixed at +// creation), so the error must be classified permanent and the job GC'd — while a +// generic ImportSysFailed error must stay transient. +func (s *ImportCheckerSuite) TestCheckGCReplicateNoVchannelsRollbackErrRemovesJob() { + server := &Server{} + err := server.broadcastRollbackImportMessage(context.TODO(), &importJob{ImportJob: &datapb.ImportJob{JobID: 1}}) + s.Error(err) + s.True(isPermanentRollbackErr(err)) + s.False(isPermanentRollbackErr(merr.WrapErrImportSysFailedMsg("some transient failure"))) + + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil) + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil) + + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + return server.broadcastRollbackImportMessage(ctx, &importJob{ImportJob: &datapb.ImportJob{JobID: job.GetJobID()}}) + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID)) + s.Equal(0, len(s.importMeta.GetJobBy(context.TODO()))) +} + +// The GC loop and the state-machine loop must run on separate goroutines: a +// rollback broadcast parked on the ctx-insensitive resource-key lock (or any +// other stall inside checkGC) must delay only GC, never the import state +// machine. This test parks checkGC's rollback forever and asserts the state +// machine still processes another job meanwhile. +func (s *ImportCheckerSuite) TestStateMachineProgressesWhileGCRollbackParked() { + params := paramtable.Get() + params.Save(params.DataCoordCfg.ImportCheckIntervalHigh.Key, "0.05") + params.Save(params.DataCoordCfg.ImportCheckIntervalLow.Key, "0.05") + defer params.Reset(params.DataCoordCfg.ImportCheckIntervalHigh.Key) + defer params.Reset(params.DataCoordCfg.ImportCheckIntervalLow.Key) + + s.setupGCReadyFailedJob() + catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog) + catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil).Maybe() + // Teardown releases the parked rollback, letting the in-flight checkGC run to + // completion (RemoveJob, then checkCollection) while the loops shut down. + catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil).Maybe() + s.checker.broker.(*broker2.MockBroker).EXPECT().HasCollection(mock.Anything, mock.Anything).Return(true, nil).Maybe() + + rollbackEntered := make(chan struct{}, 1) + rollbackRelease := make(chan struct{}) + s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { + select { + case rollbackEntered <- struct{}{}: + default: + } + // Park forever, ignoring ctx — simulates the ctx-insensitive lock. + <-rollbackRelease + return nil + } + s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil } + + go s.checker.Start() + defer s.checker.Close() + defer close(rollbackRelease) + + // Wait until the GC loop is parked inside the rollback broadcast. + select { + case <-rollbackEntered: + case <-time.After(10 * time.Second): + s.FailNow("GC loop never reached the rollback broadcast") + } + + // Feed the state machine a Failed job with a live task; only the + // state-machine loop (checkFailedJob → tryFailingTasks) can fail the task. + jobB := &importJob{ + ImportJob: &datapb.ImportJob{ + JobID: s.jobID + 100, + CollectionID: 1, + Vchannels: []string{"ch1"}, + ReadyVchannels: []string{"ch1"}, + State: internalpb.ImportJobState_Failed, + CleanupTs: tsoutil.ComposeTSByTime(time.Now().Add(24 * time.Hour)), // never GC-ready + }, + tr: timerecord.NewTimeRecorder("import job"), + } + s.NoError(s.importMeta.AddJob(context.TODO(), jobB)) + taskProto := &datapb.ImportTaskV2{ + JobID: jobB.GetJobID(), + TaskID: 999, + State: datapb.ImportTaskStateV2_Pending, + NodeID: NullNodeID, + } + task := &importTask{tr: timerecord.NewTimeRecorder("import task")} + task.task.Store(taskProto) + s.NoError(s.importMeta.AddTask(context.TODO(), task)) + + s.Eventually(func() bool { + tasks := s.importMeta.GetTaskBy(context.TODO(), WithJob(jobB.GetJobID())) + return len(tasks) == 1 && tasks[0].GetState() == datapb.ImportTaskStateV2_Failed + }, 10*time.Second, 20*time.Millisecond) +} + func (s *ImportCheckerSuite) TestCheckCollection() { mockErr := errors.New("mock err") @@ -603,7 +876,7 @@ func TestImportCheckerCompaction(t *testing.T) { cim := NewMockCompactionInspector(t) handler := NewNMockHandler(t) - checker := NewImportChecker(context.TODO(), meta, broker, alloc, importMeta, cim, handler, nil).(*importChecker) + checker := NewImportChecker(context.TODO(), meta, broker, alloc, importMeta, cim, handler, importCheckerHooks{}).(*importChecker) job := &importJob{ ImportJob: &datapb.ImportJob{ @@ -814,13 +1087,13 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitTrue() { }) commitCalled := false - s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error { + s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error { commitCalled = true return nil } s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID)) - s.True(commitCalled, "commitImportFn should be called when auto_commit=true") + s.True(commitCalled, "commit hook should be called when auto_commit=true") } func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitFalse() { @@ -831,25 +1104,25 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitFalse() { }) commitCalled := false - s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error { + s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error { commitCalled = true return nil } s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID)) - s.False(commitCalled, "commitImportFn must NOT be called when auto_commit=false") + s.False(commitCalled, "commit hook must NOT be called when auto_commit=false") // Job state must remain Uncommitted. s.Equal(internalpb.ImportJobState_Uncommitted, s.importMeta.GetJob(context.TODO(), s.jobID).GetState()) } func (s *ImportCheckerSuite) TestCheckUncommittedJob_NilFn_AutoCommitTrue() { - // commitImportFn=nil with auto_commit=true is a programming error; the checker + // commit hook=nil with auto_commit=true is a programming error; the checker // must log an error and return without crashing (no panic in the ticker goroutine). s.manuallyUpdateJob(s.jobID, func(job ImportJob) { job.(*importJob).State = internalpb.ImportJobState_Uncommitted job.(*importJob).AutoCommit = true }) - s.checker.commitImportFn = nil + s.checker.hooks.commitImport = nil s.NotPanics(func() { s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID)) @@ -859,7 +1132,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_NilFn_AutoCommitTrue() { // TestCheckUncommittedJob_RepeatedTicks_Safe verifies that ticker re-entry into // checkUncommittedJob before the ack callback transitions the job state is safe. -// commitImportFn is invoked once per tick; correctness against the resulting +// the commit hook is invoked once per tick; correctness against the resulting // duplicate broadcasts is guaranteed by the broadcaster's resource-key lock, // the ack callback's state guard, and HandleCommitVchannel's idempotency. func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() { @@ -869,7 +1142,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() { }) callCount := 0 - s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error { + s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error { callCount++ return nil } @@ -877,7 +1150,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() { for i := 0; i < 3; i++ { s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID)) } - s.Equal(3, callCount, "each tick must call commitImportFn; broadcaster handles dedup") + s.Equal(3, callCount, "each tick must call the commit hook; broadcaster handles dedup") s.Equal(internalpb.ImportJobState_Uncommitted, s.importMeta.GetJob(context.TODO(), s.jobID).GetState(), "state must remain Uncommitted until the ack callback fires") } diff --git a/internal/datacoord/server.go b/internal/datacoord/server.go index 2bab76d1ab..74a7c76b1f 100644 --- a/internal/datacoord/server.go +++ b/internal/datacoord/server.go @@ -343,7 +343,11 @@ func (s *Server) initDataCoord() error { s.importInspector = NewImportInspector(s.ctx, s.meta, s.importMeta, s.globalScheduler) - s.importChecker = NewImportChecker(s.ctx, s.meta, s.broker, s.allocator, s.importMeta, s.compactionInspector, s.handler, s.broadcastCommitImportMessage) + s.importChecker = NewImportChecker(s.ctx, s.meta, s.broker, s.allocator, s.importMeta, s.compactionInspector, s.handler, importCheckerHooks{ + commitImport: s.broadcastCommitImportMessage, + rollbackImport: s.broadcastRollbackImportMessage, + isReplicatingCluster: s.isReplicatingClusterNow, + }) // init file resource observer if s.fileResourceObserver != nil { diff --git a/internal/datacoord/services.go b/internal/datacoord/services.go index d7d10db03b..53aa4341b8 100644 --- a/internal/datacoord/services.go +++ b/internal/datacoord/services.go @@ -2864,12 +2864,20 @@ func (s *Server) broadcastCommitImportMessage(ctx context.Context, job ImportJob return err } +// errRollbackImportNoVchannels marks a rollback that can never be delivered: the job +// carries no vchannels, so there is no peer to address. A job's Vchannels are fixed at +// creation, so retrying can never succeed — isPermanentRollbackErr classifies this as +// permanent so GC proceeds instead of retaining the job forever. A plain sentinel +// attached via errors.Mark, NOT a merr error: merr's errors.Is matches by error code, +// which would make every ImportSysFailed error (mostly transient) match it. +var errRollbackImportNoVchannels = errors.New("import job has no vchannels") + // broadcastRollbackImportMessage broadcasts a RollbackImport WAL message for the given import job. // Targets the job's data vchannels, matching the CommitImport routing. func (s *Server) broadcastRollbackImportMessage(ctx context.Context, job ImportJob) error { vchannels := job.GetVchannels() if len(vchannels) == 0 { - return merr.WrapErrImportSysFailedMsg("job %d has no vchannels", job.GetJobID()) + return errors.Mark(merr.WrapErrImportSysFailedMsg("job %d has no vchannels", job.GetJobID()), errRollbackImportNoVchannels) } broadcaster, err := s.startBroadcastWithCollectionID(ctx, job.GetCollectionID()) @@ -2954,23 +2962,38 @@ func (s *Server) CommitImport(ctx context.Context, req *datapb.CommitImportReque ) } -// AbortImport aborts a 2PC import job that has not yet been committed. -// It broadcasts a RollbackImport WAL message to cancel the job. -// Returns an error if the job is already committed or committing. +// AbortImport rolls back a 2PC import job that has not been committed by +// broadcasting a RollbackImport WAL message. A job that has already Failed (for +// example because its own import failed) is still abortable, so the control plane +// can proactively release the peer cluster's replicated Uncommitted job instead of +// waiting for the failed source's GC self-heal (see importChecker.checkGC). +// Committing/Completed jobs are terminal and rejected. +// +// Behavior change for non-CDC clusters: aborting an already-Failed 2PC job used to +// return an error; it now succeeds and (re)broadcasts a RollbackImport, which the +// flusher no-ops. Repeated aborts on a real-failure source therefore re-broadcast +// each time — harmless, but not deduplicated (there is no persisted "rolled back" +// flag in this change; that idempotency is a follow-up). func (s *Server) AbortImport(ctx context.Context, req *datapb.AbortImportRequest) (*commonpb.Status, error) { return s.validateAndExecuteImportAction(ctx, req.GetJobId(), func(job ImportJob) *commonpb.Status { state := job.GetState() + // Idempotent only for a job that was previously Uncommitted and then + // user-aborted (its reason is rewritten to importJobReasonAbortedByUser). A + // source that failed on its own keeps its real failure reason, so this does + // NOT fire for it and each abort re-broadcasts (see the note above). if state == internalpb.ImportJobState_Failed && job.GetReason() == importJobReasonAbortedByUser { return merr.Success() } - if state == internalpb.ImportJobState_Failed || - state == internalpb.ImportJobState_Committing || + // Committed states are truly terminal and cannot be rolled back. + if state == internalpb.ImportJobState_Committing || state == internalpb.ImportJobState_Completed { return merr.Status(merr.WrapErrImportFailed( fmt.Sprintf("job %d is in terminal/committed state %s, abort not allowed", req.GetJobId(), state))) } - return nil // proceed + // Uncommitted, or a Failed source (its own import failed) → broadcast the + // rollback so the peer cluster's replicated Uncommitted job is released. + return nil }, func(ctx context.Context, job ImportJob) error { mlog.Info(context.TODO(), "aborting import job via WAL broadcast") diff --git a/internal/datacoord/services_test.go b/internal/datacoord/services_test.go index a5716d8eb2..64b8474640 100644 --- a/internal/datacoord/services_test.go +++ b/internal/datacoord/services_test.go @@ -5811,3 +5811,69 @@ type ( embeddedBroadcastAPI struct{ broadcaster.BroadcastAPI } embeddedBroker struct{ broker.Broker } ) + +func TestAbortImport_FailedSourceBroadcastsRollback(t *testing.T) { + ctx := context.Background() + // A source whose own import failed (real reason, not user-aborted) must still be + // abortable, so the control plane can release the peer cluster's Uncommitted job. + job := &importJob{ImportJob: &datapb.ImportJob{ + JobID: 2101, CollectionID: 300, + State: internalpb.ImportJobState_Failed, Reason: "disk quota exceeded", + }} + im := NewMockImportMeta(t) + im.EXPECT().GetJob(mock.Anything, int64(2101)).Return(job).Times(2) + + server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()} + server.stateCode.Store(commonpb.StateCode_Healthy) + // The broadcast is the assertion of record: a regression that short-circuits + // AbortImport on a Failed job (returning success without running the action) + // would skip the rollback and strand the peer, so success alone is not enough. + rollbackCalls := 0 + var rollbackJobID int64 + bm := mockey.Mock((*Server).broadcastRollbackImportMessage).To( + func(s *Server, ctx context.Context, job ImportJob) error { + rollbackCalls++ + rollbackJobID = job.GetJobID() + return nil + }).Build() + defer bm.UnPatch() + + resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2101}) + assert.NoError(t, err) + assert.True(t, merr.Ok(resp)) + assert.Equal(t, 1, rollbackCalls) + assert.Equal(t, int64(2101), rollbackJobID) +} + +func TestAbortImport_CompletedRejected(t *testing.T) { + ctx := context.Background() + job := &importJob{ImportJob: &datapb.ImportJob{ + JobID: 2103, State: internalpb.ImportJobState_Completed, + }} + im := NewMockImportMeta(t) + im.EXPECT().GetJob(mock.Anything, int64(2103)).Return(job).Once() + + server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()} + server.stateCode.Store(commonpb.StateCode_Healthy) + + resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2103}) + assert.NoError(t, err) + assert.False(t, merr.Ok(resp)) +} + +func TestAbortImport_CommittingRejected(t *testing.T) { + ctx := context.Background() + // Committing is mid-commit and cannot be rolled back. + job := &importJob{ImportJob: &datapb.ImportJob{ + JobID: 2104, State: internalpb.ImportJobState_Committing, + }} + im := NewMockImportMeta(t) + im.EXPECT().GetJob(mock.Anything, int64(2104)).Return(job).Once() + + server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()} + server.stateCode.Store(commonpb.StateCode_Healthy) + + resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2104}) + assert.NoError(t, err) + assert.False(t, merr.Ok(resp)) +}