fix: harden copy-segment restore pipeline against partial failures and transient errors (#51527)

issue: #51191

Fixes three robustness defects in the copy-segment restore pipeline
(`internal/datacoord/copy_segment_*.go`), each with a unit test covering
the failure mode.

### 1. `checkPendingJob` cannot resume a partially created job (high)

Task creation is a multi-step sequence (per-group `AllocID` + `AddTask`,
each persisted individually, then the `Pending → Executing` transition).
If any step fails mid-way (etcd hiccup, DataCoord restart), the job
stayed `Pending` with only a subset of tasks persisted, and the old
early-return `if len(tasks) > 0 { return }` fired on every subsequent
round — the restore hung until the job timeout with no way to progress.

Now the transition is resume-safe: it computes the source segments
already covered by persisted tasks, creates tasks only for the uncovered
mappings, and always (re-)applies the idempotent `Executing` transition
(so a round that persisted all tasks but failed the job update also
recovers).

### 2. Transient `QueryCopySegment` RPC error failed the whole restore
(medium)

A single network blip or DataNode restart during a status poll
permanently failed the task and its parent restore job. This was
inconsistent with the import pipeline, where a `QueryImport` RPC error
resets the task for retry instead of failing the job.

Now a query RPC error is logged and the task is kept `InProgress` for
re-query on the next tick. The job-level timeout still bounds a
permanently unreachable node; worker-reported task failures keep the
fail-fast behavior.

### 3. `copySegmentTask.Clone` mutated shared state before persisting
(medium)

`Clone` copied the `atomic.Pointer` value, not the protobuf, so the
"clone" used by `UpdateTask` shared the same `*datapb.CopySegmentTask`
as the cached entry. Update actions mutate the proto in place, so the
in-memory task was mutated *before* `SaveCopySegmentTask` — a failed
catalog save left the coordinator cache reflecting the new state while
etcd kept the old one. Now `Clone` deep-copies with `proto.Clone`,
matching `copySegmentJob.Clone`.

### Testing

- `TestCheckPendingJob_ResumesPartialTaskCreation`,
`TestCheckPendingJob_AllTasksExistTransitionsToExecuting`
- `TestQueryTaskOnWorker_RPCErrorKeepsTaskInProgress`
- `TestClone_DeepCopiesProto`,
`TestUpdateTask_SaveFailureLeavesCacheUnchanged`
- Full `./internal/datacoord/` package: PASS

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

internal/datacoord/copy_segment_checker.go

@@ -286,19 +286,35 @@ func (c *copySegmentChecker) LogTaskStats() {
 //   - Full segment metadata (binlogs, indexes) is fetched by DataNode when executing
 //   - Keeps task metadata small and efficient to persist
 //
-// Idempotency:
-//   - Safe to call multiple times - only creates tasks on first call
-//   - Subsequent calls return early if tasks already exist
+// Idempotency and crash recovery:
+//   - Task creation is a multi-step sequence (per-group AllocID + AddTask, then
+//     the Executing transition), each step persisted individually. A failure
+//     mid-way (etcd hiccup, DataCoord restart) leaves the job Pending with only
+//     a subset of tasks persisted.
+//   - To be resume-safe, each round creates tasks only for source segments not
+//     yet covered by persisted tasks, then (re-)applies the idempotent
+//     Pending → Executing transition.
 func (c *copySegmentChecker) checkPendingJob(job CopySegmentJob) {
 	log := mlog.With(mlog.FieldJobID(job.GetJobId()))
 
-	// Step 1: Check if tasks already created (idempotent operation)
-	tasks := c.copyMeta.GetTasksByJobID(c.ctx, job.GetJobId())
-	if len(tasks) > 0 {
+	// Step 0: Re-read the cached job. The `job` argument is a snapshot taken
+	// before this function ran, but tasks of a Pending job can already be
+	// dispatched (the inspector does not filter by job state) and a concurrent
+	// failure (markTaskAndJobFailed) may have moved the job to a terminal
+	// state and released its snapshot pin. Creating more tasks for such a job
+	// would be wasted work at best.
+	current := c.copyMeta.GetJob(c.ctx, job.GetJobId())
+	if current == nil {
+		log.Info(c.ctx, "job no longer exists, skip pending check")
+		return
+	}
+	if current.GetState() != datapb.CopySegmentJobState_CopySegmentJobPending {
+		log.Info(c.ctx, "job is no longer pending, skip pending check",
+			mlog.String("currentState", current.GetState().String()))
 		return
 	}
 
-	// Step 2: Validate job has segment mappings
+	// Step 1: Validate job has segment mappings
 	idMappings := job.GetIdMappings()
 	if len(idMappings) == 0 {
 		log.Warn(c.ctx, "no id mappings to copy, mark job as completed")
@@ -310,9 +326,23 @@ func (c *copySegmentChecker) checkPendingJob(job CopySegmentJob) {
 		return
 	}
 
-	// Step 3: Split mappings into groups (max segments per task)
+	// Step 2: Compute source segments already covered by persisted tasks,
+	// so a partially created job resumes instead of duplicating tasks.
+	tasks := c.copyMeta.GetTasksByJobID(c.ctx, job.GetJobId())
+	coveredSourceIDs := make(map[int64]struct{})
+	for _, task := range tasks {
+		for _, mapping := range task.GetIdMappings() {
+			coveredSourceIDs[mapping.GetSourceSegmentId()] = struct{}{}
+		}
+	}
+	pendingMappings := lo.Filter(idMappings, func(mapping *datapb.CopySegmentIDMapping, _ int) bool {
+		_, covered := coveredSourceIDs[mapping.GetSourceSegmentId()]
+		return !covered
+	})
+
+	// Step 3: Split uncovered mappings into groups (max segments per task)
 	maxSegmentsPerTask := Params.DataCoordCfg.MaxSegmentsPerCopyTask.GetAsInt()
-	groups := lo.Chunk(idMappings, maxSegmentsPerTask)
+	groups := lo.Chunk(pendingMappings, maxSegmentsPerTask)
 
 	// Step 4: Create task for each group
 	for i, group := range groups {
@@ -357,16 +387,28 @@ func (c *copySegmentChecker) checkPendingJob(job CopySegmentJob) {
 			mlog.Int("segmentCount", len(group)))
 	}
 
-	// Step 5: Update job state to Executing
-	err := c.copyMeta.UpdateJob(c.ctx, job.GetJobId(),
+	// Step 5: Update job state to Executing. This also runs when all segments
+	// were already covered (groups is empty), retrying a transition that a
+	// previous round failed to persist.
+	// The transition is state-guarded (Pending -> Executing only): a task
+	// dispatched during Step 4 can fail concurrently, and markTaskAndJobFailed
+	// then moves the job to Failed and releases its snapshot pin. An
+	// unconditional update here would resurrect that Failed job as Executing.
+	updated, err := c.copyMeta.UpdateJobInState(c.ctx, job.GetJobId(),
+		datapb.CopySegmentJobState_CopySegmentJobPending,
 		UpdateCopyJobState(datapb.CopySegmentJobState_CopySegmentJobExecuting),
-		UpdateCopyJobProgress(0, int64(len(idMappings))))
+		UpdateCopyJobTotalSegments(int64(len(idMappings))))
 	if err != nil {
 		log.Warn(c.ctx, "failed to update job state to Executing", mlog.Err(err))
 		return
 	}
+	if !updated {
+		log.Info(c.ctx, "job left Pending state concurrently, skip transition to Executing")
+		return
+	}
 	log.Info(c.ctx, "copy segment job started",
-		mlog.Int("taskCount", len(groups)),
+		mlog.Int("newTaskCount", len(groups)),
+		mlog.Int("resumedTaskCount", len(tasks)),
 		mlog.Int("totalSegments", len(idMappings)))
 }
 

internal/datacoord/copy_segment_checker_test.go

@@ -180,6 +180,155 @@ func (s *CopySegmentCheckerSuite) TestCheckPendingJob_CreateTasks() {
 	s.Equal([]int{1, 2}, mappingCounts, "should have one task with 1 mapping and one with 2 mappings")
 }
 
+func (s *CopySegmentCheckerSuite) TestCheckPendingJob_ResumesPartialTaskCreation() {
+	// Simulate a previous round that persisted only the first task before
+	// failing (etcd hiccup / DataCoord restart): checkPendingJob must create
+	// tasks for the remaining uncovered segments and move the job to Executing
+	// instead of returning early and leaving the job Pending forever.
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
+	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(nil).Times(2)
+	s.alloc.EXPECT().AllocID(mock.Anything).Return(int64(1002), nil).Times(1)
+
+	idMappings := []*datapb.CopySegmentIDMapping{
+		{SourceSegmentId: 1, TargetSegmentId: 101, PartitionId: 10},
+		{SourceSegmentId: 2, TargetSegmentId: 102, PartitionId: 10},
+		{SourceSegmentId: 3, TargetSegmentId: 103, PartitionId: 10},
+	}
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        s.jobID,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobPending,
+			IdMappings:   idMappings,
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), job))
+
+	// Pre-existing task from the failed round covers the first two mappings.
+	partialTask := &copySegmentTask{
+		tr:    timerecord.NewTimeRecorder("test task"),
+		times: taskcommon.NewTimes(),
+	}
+	partialTask.task.Store(&datapb.CopySegmentTask{
+		TaskId:       1001,
+		JobId:        s.jobID,
+		CollectionId: s.collectionID,
+		NodeId:       NullNodeID,
+		TaskSlot:     1,
+		State:        datapb.CopySegmentTaskState_CopySegmentTaskPending,
+		IdMappings:   idMappings[:2],
+	})
+	s.NoError(s.copyMeta.AddTask(context.TODO(), partialTask))
+
+	Params.DataCoordCfg.MaxSegmentsPerCopyTask.SwapTempValue("2")
+	defer Params.DataCoordCfg.MaxSegmentsPerCopyTask.SwapTempValue("10")
+
+	s.checker.checkPendingJob(job)
+
+	// Job must reach Executing.
+	updatedJob := s.copyMeta.GetJob(context.TODO(), s.jobID)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting, updatedJob.GetState())
+
+	// The uncovered mapping (source segment 3) must now have a task.
+	tasks := s.copyMeta.GetTasksByJobID(context.TODO(), s.jobID)
+	s.Len(tasks, 2)
+	coveredSources := make(map[int64]int)
+	for _, task := range tasks {
+		for _, mapping := range task.GetIdMappings() {
+			coveredSources[mapping.GetSourceSegmentId()]++
+		}
+	}
+	s.Equal(map[int64]int{1: 1, 2: 1, 3: 1}, coveredSources,
+		"each source segment must be covered exactly once")
+}
+
+func (s *CopySegmentCheckerSuite) TestCheckPendingJob_AllTasksExistTransitionsToExecuting() {
+	// Simulate a previous round that created all tasks but failed to persist
+	// the Pending→Executing transition: checkPendingJob must not create any
+	// new task (no AllocID expectation) and must retry the transition.
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
+	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(nil).Times(1)
+
+	idMappings := []*datapb.CopySegmentIDMapping{
+		{SourceSegmentId: 1, TargetSegmentId: 101, PartitionId: 10},
+		{SourceSegmentId: 2, TargetSegmentId: 102, PartitionId: 10},
+	}
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        s.jobID,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobPending,
+			IdMappings:   idMappings,
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), job))
+
+	fullTask := &copySegmentTask{
+		tr:    timerecord.NewTimeRecorder("test task"),
+		times: taskcommon.NewTimes(),
+	}
+	fullTask.task.Store(&datapb.CopySegmentTask{
+		TaskId:       1001,
+		JobId:        s.jobID,
+		CollectionId: s.collectionID,
+		NodeId:       NullNodeID,
+		TaskSlot:     1,
+		State:        datapb.CopySegmentTaskState_CopySegmentTaskPending,
+		IdMappings:   idMappings,
+	})
+	s.NoError(s.copyMeta.AddTask(context.TODO(), fullTask))
+
+	s.checker.checkPendingJob(job)
+
+	updatedJob := s.copyMeta.GetJob(context.TODO(), s.jobID)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting, updatedJob.GetState())
+	s.Len(s.copyMeta.GetTasksByJobID(context.TODO(), s.jobID), 1)
+}
+
+func (s *CopySegmentCheckerSuite) TestCheckPendingJob_StaleSnapshotDoesNotResurrectFailedJob() {
+	// Regression test: checkPendingJob receives a job snapshot taken before it
+	// runs, while tasks of a Pending job can already be dispatched and fail
+	// concurrently — markTaskAndJobFailed then moves the job to Failed and
+	// releases its snapshot pin. The checker, still holding the stale Pending
+	// snapshot, must neither create more tasks nor resurrect the job as
+	// Executing.
+	// AddJob + the concurrent Failed transition each persist once; the stale
+	// checkPendingJob call must not persist anything (and must not AllocID —
+	// the allocator mock has no expectation and would fail the test).
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
+
+	staleJob := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        s.jobID,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobPending,
+			IdMappings: []*datapb.CopySegmentIDMapping{
+				{SourceSegmentId: 1, TargetSegmentId: 101, PartitionId: 10},
+			},
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), staleJob))
+
+	// Concurrent failure path moves the cached job to Failed. The cached entry
+	// is replaced with a Failed clone; `staleJob` keeps its Pending state and
+	// plays the stale snapshot below.
+	s.NoError(s.copyMeta.UpdateJobStateAndReleaseRef(context.TODO(), s.jobID,
+		UpdateCopyJobState(datapb.CopySegmentJobState_CopySegmentJobFailed),
+		UpdateCopyJobReason("task failed concurrently")))
+
+	s.checker.checkPendingJob(staleJob)
+
+	updatedJob := s.copyMeta.GetJob(context.TODO(), s.jobID)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobFailed, updatedJob.GetState())
+	s.Equal("task failed concurrently", updatedJob.GetReason())
+	s.Empty(s.copyMeta.GetTasksByJobID(context.TODO(), s.jobID))
+}
+
 func (s *CopySegmentCheckerSuite) TestCheckCopyingJob_UpdateProgress() {
 	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
 	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(nil).Times(2)
@@ -424,6 +573,55 @@ func (s *CopySegmentCheckerSuite) TestTryTimeoutJob() {
 	s.Equal("timeout", updatedJob.GetReason())
 }
 
+func (s *CopySegmentCheckerSuite) TestTryTimeoutJob_FiresWithProductionTimeoutTs() {
+	// End-to-end unit check across the write site and the read site: the job
+	// creation path composes TimeoutTs via CopyJobTimeoutTs, and tryTimeoutJob
+	// must actually fire once that deadline has elapsed. Guards against the
+	// regression where the write site stored UnixNano while the reader decoded
+	// a hybrid TSO, so no job ever timed out.
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        s.jobID,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobExecuting,
+			// Same composition as the creation path in snapshot_manager, with
+			// an already-elapsed deadline.
+			TimeoutTs: CopyJobTimeoutTs(-time.Minute),
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), job))
+
+	s.checker.tryTimeoutJob(job)
+
+	updatedJob := s.copyMeta.GetJob(context.TODO(), s.jobID)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobFailed, updatedJob.GetState())
+	s.Equal("timeout", updatedJob.GetReason())
+}
+
+func (s *CopySegmentCheckerSuite) TestTryTimeoutJob_NotElapsedKeepsExecuting() {
+	// A freshly composed production deadline must NOT trigger the timeout.
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(1)
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        s.jobID,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobExecuting,
+			TimeoutTs:    CopyJobTimeoutTs(time.Hour),
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), job))
+
+	s.checker.tryTimeoutJob(job)
+
+	updatedJob := s.copyMeta.GetJob(context.TODO(), s.jobID)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting, updatedJob.GetState())
+}
+
 func (s *CopySegmentCheckerSuite) TestCheckGC_RemoveCompletedJob() {
 	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil)
 	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(nil)

internal/datacoord/copy_segment_job.go

@@ -59,6 +59,15 @@ func WithoutCopyJobStates(states ...datapb.CopySegmentJobState) CopySegmentJobFi
 	}
 }
 
+// CopyJobTimeoutTs returns the job timeout deadline as a hybrid TSO timestamp
+// (physical milliseconds << 18), matching how tryTimeoutJob reads it back via
+// tsoutil.PhysicalTime and how CleanupTs is composed. Storing UnixNano here
+// instead would be interpreted as a TSO landing centuries in the future, so
+// the job timeout would never fire.
+func CopyJobTimeoutTs(timeout time.Duration) uint64 {
+	return tsoutil.AddPhysicalDurationOnTs(tsoutil.ComposeTSByTime(time.Now()), timeout)
+}
+
 type UpdateCopySegmentJobAction func(job CopySegmentJob)
 
 func UpdateCopyJobState(state datapb.CopySegmentJobState) UpdateCopySegmentJobAction {
@@ -92,6 +101,16 @@ func UpdateCopyJobProgress(copied, total int64) UpdateCopySegmentJobAction {
 	}
 }
 
+// UpdateCopyJobTotalSegments sets only the segment total, leaving CopiedSegments
+// untouched. Used when (re-)entering Executing: on a resume after a restart some
+// tasks may already be completed, and zeroing the copied count here would make
+// the reported progress transiently drop until the next checkCopyingJob tick.
+func UpdateCopyJobTotalSegments(total int64) UpdateCopySegmentJobAction {
+	return func(job CopySegmentJob) {
+		job.(*copySegmentJob).TotalSegments = total
+	}
+}
+
 func UpdateCopyJobCompleteTs(completeTs uint64) UpdateCopySegmentJobAction {
 	return func(job CopySegmentJob) {
 		job.(*copySegmentJob).CompleteTs = completeTs

internal/datacoord/copy_segment_job_test.go

@@ -503,3 +503,16 @@ func (s *CopySegmentJobSuite) TestJobWithEmptyOptions() {
 	options := job.GetOptions()
 	s.Nil(options)
 }
+
+func (s *CopySegmentJobSuite) TestCopyJobTimeoutTs_MatchesPhysicalTimeReader() {
+	// Regression test for the TimeoutTs unit mismatch: the deadline is read
+	// back by tryTimeoutJob via tsoutil.PhysicalTime, which expects a hybrid
+	// TSO timestamp (physical ms << 18). Storing UnixNano here made the
+	// decoded deadline land centuries in the future, so the job timeout never
+	// fired. The composed value must round-trip to now+timeout.
+	timeout := 24 * time.Hour
+	deadline := tsoutil.PhysicalTime(CopyJobTimeoutTs(timeout))
+
+	expected := time.Now().Add(timeout)
+	s.WithinDuration(expected, deadline, 10*time.Second)
+}

internal/datacoord/copy_segment_meta.go

@@ -81,6 +81,7 @@ type CopySegmentMeta interface {
 	// Job operations
 	AddJob(ctx context.Context, job CopySegmentJob) error
 	UpdateJob(ctx context.Context, jobID int64, actions ...UpdateCopySegmentJobAction) error
+	UpdateJobInState(ctx context.Context, jobID int64, expectedState datapb.CopySegmentJobState, actions ...UpdateCopySegmentJobAction) (bool, error)
 	UpdateJobStateAndReleaseRef(ctx context.Context, jobID int64, actions ...UpdateCopySegmentJobAction) error
 	GetJob(ctx context.Context, jobID int64) CopySegmentJob
 	GetJobBy(ctx context.Context, filters ...CopySegmentJobFilter) []CopySegmentJob
@@ -393,6 +394,32 @@ func (m *copySegmentMeta) UpdateJob(ctx context.Context, jobID int64, actions ..
 	return err
 }
 
+// UpdateJobInState applies the actions only if the cached job is currently in
+// expectedState, with the check and the update under the same write lock.
+//
+// Callers that hold a job snapshot taken before a slow operation (e.g. the
+// checker creating tasks) must use this instead of UpdateJob for state
+// transitions: a concurrent failure path (markTaskAndJobFailed) may have moved
+// the job to a terminal state in the meantime, and an unconditional update
+// would resurrect it — e.g. Failed -> Executing after the job's snapshot pin
+// was already released.
+//
+// Returns (false, nil) when the job is missing or not in expectedState (the
+// update is skipped), (true, nil) on success.
+func (m *copySegmentMeta) UpdateJobInState(ctx context.Context, jobID int64, expectedState datapb.CopySegmentJobState, actions ...UpdateCopySegmentJobAction) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	job, ok := m.jobs[jobID]
+	if !ok || job.GetState() != expectedState {
+		return false, nil
+	}
+	_, _, err := m.updateJob(ctx, jobID, actions...)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 // GetJob retrieves a job by ID from in-memory cache.
 //
 // Thread safety: Protected by read lock (allows concurrent reads)

internal/datacoord/copy_segment_meta_test.go

@@ -294,6 +294,47 @@ func (s *CopySegmentMetaSuite) TestUpdateJob_Success() {
 	s.Equal("executing", updatedJob.GetReason())
 }
 
+func (s *CopySegmentMetaSuite) TestUpdateJobInState_AppliesOnlyWhenStateMatches() {
+	s.catalog.EXPECT().SaveCopySegmentJob(mock.Anything, mock.Anything).Return(nil).Times(2)
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        100,
+			CollectionId: s.collectionID,
+			State:        datapb.CopySegmentJobState_CopySegmentJobFailed,
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+	s.NoError(s.copyMeta.AddJob(context.TODO(), job))
+
+	// Expected state mismatch (job is Failed, expected Pending): the update
+	// must be skipped without touching the catalog, so a stale caller cannot
+	// resurrect a terminal job.
+	updated, err := s.copyMeta.UpdateJobInState(context.TODO(), 100,
+		datapb.CopySegmentJobState_CopySegmentJobPending,
+		UpdateCopyJobState(datapb.CopySegmentJobState_CopySegmentJobExecuting))
+	s.NoError(err)
+	s.False(updated)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobFailed,
+		s.copyMeta.GetJob(context.TODO(), 100).GetState())
+
+	// Missing job: skipped, no error.
+	updated, err = s.copyMeta.UpdateJobInState(context.TODO(), 999,
+		datapb.CopySegmentJobState_CopySegmentJobPending,
+		UpdateCopyJobState(datapb.CopySegmentJobState_CopySegmentJobExecuting))
+	s.NoError(err)
+	s.False(updated)
+
+	// Matching expected state: the update applies.
+	updated, err = s.copyMeta.UpdateJobInState(context.TODO(), 100,
+		datapb.CopySegmentJobState_CopySegmentJobFailed,
+		UpdateCopyJobState(datapb.CopySegmentJobState_CopySegmentJobExecuting))
+	s.NoError(err)
+	s.True(updated)
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting,
+		s.copyMeta.GetJob(context.TODO(), 100).GetState())
+}
+
 func (s *CopySegmentMetaSuite) TestUpdateJob_NotFound() {
 	// Try to update non-existent job (should not error, just no-op)
 	err := s.copyMeta.UpdateJob(context.TODO(), 999,
@@ -553,6 +594,36 @@ func (s *CopySegmentMetaSuite) TestUpdateTask_Success() {
 	s.Equal("executing", updatedTask.GetReason())
 }
 
+func (s *CopySegmentMetaSuite) TestUpdateTask_SaveFailureLeavesCacheUnchanged() {
+	// AddTask persists fine; the subsequent UpdateTask save fails.
+	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(nil).Once()
+	s.catalog.EXPECT().SaveCopySegmentTask(mock.Anything, mock.Anything).Return(errors.New("etcd unavailable")).Once()
+
+	task := &copySegmentTask{
+		copyMeta: s.copyMeta,
+		tr:       timerecord.NewTimeRecorder("task"),
+		times:    taskcommon.NewTimes(),
+	}
+	task.task.Store(&datapb.CopySegmentTask{
+		TaskId:       1001,
+		JobId:        100,
+		CollectionId: s.collectionID,
+		State:        datapb.CopySegmentTaskState_CopySegmentTaskPending,
+	})
+	err := s.copyMeta.AddTask(context.TODO(), task)
+	s.NoError(err)
+
+	err = s.copyMeta.UpdateTask(context.TODO(), 1001,
+		UpdateCopyTaskState(datapb.CopySegmentTaskState_CopySegmentTaskFailed),
+		UpdateCopyTaskReason("should not stick"))
+	s.Error(err)
+
+	// The in-memory cache must still reflect the persisted (old) state.
+	cachedTask := s.copyMeta.GetTask(context.TODO(), 1001)
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskPending, cachedTask.GetState())
+	s.Empty(cachedTask.GetReason())
+}
+
 func (s *CopySegmentMetaSuite) TestUpdateTask_NotFound() {
 	// Try to update non-existent task (should not error, just no-op)
 	err := s.copyMeta.UpdateTask(context.TODO(), 9999,

internal/datacoord/copy_segment_task.go

@@ -22,6 +22,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cockroachdb/errors"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
@@ -225,6 +228,11 @@ func (t *copySegmentTask) GetTR() *timerecord.TimeRecorder {
 // Why needed:
 // - UpdateTask clones before applying actions to avoid race conditions
 // - Original task remains accessible to other goroutines during update
+//
+// The protobuf payload must be deep-copied (proto.Clone): update actions
+// mutate the proto in place, so sharing the pointer would leak mutations
+// into the cached task before the catalog save succeeds — a failed save
+// would leave memory and etcd out of sync.
 func (t *copySegmentTask) Clone() CopySegmentTask {
 	cloned := &copySegmentTask{
 		copyMeta:     t.copyMeta,
@@ -234,7 +242,7 @@ func (t *copySegmentTask) Clone() CopySegmentTask {
 		tr:           t.tr,
 		times:        t.times,
 	}
-	cloned.task.Store(t.task.Load())
+	cloned.task.Store(proto.Clone(t.task.Load()).(*datapb.CopySegmentTask))
 	return cloned
 }
 
@@ -370,6 +378,26 @@ func (t *copySegmentTask) markTaskAndJobFailed(reason string) {
 		WrapCopySegmentTaskLog(t, mlog.String("reason", reason))...)
 }
 
+// isCopyTaskLostOnWorker reports whether a QueryCopySegment error means the
+// worker-side task is confirmed lost, as opposed to a transient transport error.
+//
+// Confirmed-loss signals (audited against every construction site on the
+// QueryCopySegment path):
+//   - merr.ErrNodeNotFound: the node manager no longer knows the assigned
+//     DataNode (its session was removed after a restart/replacement), so its
+//     in-memory task manager — and the task with it — is gone.
+//   - merr.ErrImportSysFailed: on this RPC the code is produced only by the
+//     DataNode's task-not-found branch (importv2.WrapTaskNotFoundError when the
+//     queried task is absent from its task manager), i.e. the DataNode is alive
+//     but restarted and lost the task.
+//
+// Everything else (gRPC transport errors, node briefly not serving/not ready,
+// response decode failures) may coexist with a still-running worker task and
+// must be retried by polling, not by re-dispatching.
+func isCopyTaskLostOnWorker(err error) bool {
+	return errors.Is(err, merr.ErrNodeNotFound) || errors.Is(err, merr.ErrImportSysFailed)
+}
+
 // QueryTaskOnWorker polls the DataNode for task execution status.
 //
 // Process flow:
@@ -381,9 +409,15 @@ func (t *copySegmentTask) markTaskAndJobFailed(reason string) {
 //  3. Update task state accordingly
 //
 // Failure handling:
-// - RPC errors and worker failure responses trigger immediate failure
-// - Task failure immediately marks parent job as failed (fail-fast)
-// - Enables quick feedback to user without waiting for timeout
+//   - A query RPC error is either a transient transport failure or a confirmed loss of the
+//     worker-side task; the two must be handled differently (see isCopyTaskLostOnWorker):
+//     confirmed loss resets the task to Pending for re-dispatch, transient errors keep the
+//     task InProgress so the next check round simply queries again
+//   - Re-dispatch is the only way a node restart gets retried: the scheduler only
+//     re-dispatches Pending(Init) tasks, never ones left InProgress
+//   - Worker failure responses trigger immediate failure
+//   - Task failure immediately marks parent job as failed (fail-fast)
+//   - Enables quick feedback to user without waiting for timeout
 //
 // Success handling:
 // - Calls SyncCopySegmentTask to update segment metadata
@@ -401,9 +435,37 @@ func (t *copySegmentTask) QueryTaskOnWorker(cluster session.Cluster) {
 		TaskID: t.GetTaskId(),
 	}
 	resp, err := cluster.QueryCopySegment(nodeID, req)
-	// Handle RPC error separately to avoid nil resp dereference
+	// Handle RPC error separately to avoid nil resp dereference.
 	if err != nil {
-		t.markTaskAndJobFailed(fmt.Sprintf("query copy segment RPC failed: %v", err))
+		if !isCopyTaskLostOnWorker(err) {
+			// Transient transport failure (network blip, RPC timeout, node briefly
+			// not ready). The worker-side task may well still be running, so keep
+			// the task InProgress and let the next check round query again.
+			// Resetting here would re-dispatch a task that is possibly still
+			// executing on a live node, starting a concurrent duplicate copy.
+			mlog.Warn(context.TODO(), "transient error querying copy segment task on datanode, will retry",
+				WrapCopySegmentTaskLog(t, mlog.FieldNodeID(nodeID), mlog.Err(err))...)
+			return
+		}
+		// Confirmed loss: the worker-side task no longer exists (DataNode
+		// restarted/replaced, or its in-memory task manager lost the task).
+		// Leaving the task InProgress would make the scheduler poll a dead
+		// node until the job-level timeout, since only Pending tasks are
+		// re-dispatched. Reset to Pending with NullNodeID so the scheduler
+		// re-dispatches it to a live node.
+		// Re-dispatch is idempotent: target binlog paths are deterministic
+		// transforms of the source paths (same content on overwrite), and each
+		// dispatch allocates fresh buildIDs, so index files from a partial
+		// earlier attempt are never referenced by meta and are removed by GC.
+		if resetErr := t.copyMeta.UpdateTask(context.TODO(), t.GetTaskId(),
+			UpdateCopyTaskState(datapb.CopySegmentTaskState_CopySegmentTaskPending),
+			UpdateCopyTaskNodeID(NullNodeID)); resetErr != nil {
+			mlog.Warn(context.TODO(), "failed to reset copy segment task to pending after worker loss",
+				WrapCopySegmentTaskLog(t, mlog.FieldNodeID(nodeID), mlog.Err(resetErr))...)
+			return
+		}
+		mlog.Info(context.TODO(), "reset copy segment task to pending due to worker loss, will re-dispatch",
+			WrapCopySegmentTaskLog(t, mlog.FieldNodeID(nodeID), mlog.Err(err))...)
 		return
 	}
 

internal/datacoord/copy_segment_task_test.go

@@ -18,6 +18,7 @@ package datacoord
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -31,6 +32,7 @@ import (
 	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
 	"github.com/milvus-io/milvus/internal/datacoord/session"
+	task2 "github.com/milvus-io/milvus/internal/datacoord/task"
 	"github.com/milvus-io/milvus/internal/metastore"
 	kvdatacoord "github.com/milvus-io/milvus/internal/metastore/kv/datacoord"
 	catalogmocks "github.com/milvus-io/milvus/internal/metastore/mocks"
@@ -40,6 +42,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v3/taskcommon"
 	"github.com/milvus-io/milvus/pkg/v3/util/lock"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
@@ -495,21 +498,139 @@ func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_NotCompletedKeepsTaskInProg
 	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskInProgress, task.GetState())
 }
 
-func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_MarksFailedOnRPCError() {
+func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_TransientRPCErrorKeepsInProgress() {
+	// A transient transport error (network blip, RPC timeout, node briefly not
+	// ready) must NOT fail the task or its parent job, and must NOT re-dispatch
+	// it either: the worker-side task may still be running, and re-dispatching
+	// would start a concurrent duplicate copy. The task stays InProgress on the
+	// same node and is simply queried again on the next check round.
+	transientErrs := []error{
+		errors.New("rpc failed"),
+		merr.WrapErrServiceNotReady("datanode", 10, "Initializing"),
+	}
+	for _, transientErr := range transientErrs {
+		cluster := session.NewMockCluster(s.T())
+		cluster.EXPECT().QueryCopySegment(mock.Anything, mock.Anything).Return(nil, transientErr)
+
+		task := createTestCopyTask(100, 2001).(*copySegmentTask)
+		task.task.Load().NodeId = 10
+		copyMeta, _ := newCopySegmentTaskTestMeta(s.T(), task)
+		// A parent job in Executing state must remain untouched by the retry.
+		s.NoError(copyMeta.AddJob(context.Background(), newTestCopyJob(100, datapb.CopySegmentJobState_CopySegmentJobExecuting)))
+
+		task.QueryTaskOnWorker(cluster)
+
+		updatedTask := copyMeta.GetTask(context.Background(), 1001)
+		s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskInProgress, updatedTask.GetState())
+		s.EqualValues(10, updatedTask.GetNodeId())
+		s.Empty(updatedTask.GetReason())
+		// The parent job must NOT be failed - the task retries by polling.
+		s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting,
+			copyMeta.GetJob(context.Background(), 100).GetState())
+	}
+}
+
+func (s *CopySegmentTaskSuite) TestScheduler_OneOffTransientErrorDoesNotRedispatch() {
+	// Regression test at the global-scheduler level: a one-off transport error
+	// while querying an InProgress copy-segment task must NOT trigger a second
+	// CreateCopySegment dispatch. The task stays in runningTasks (InProgress on
+	// the same node) and is simply queried again on the next check tick.
+	cluster := session.NewMockCluster(s.T())
+	var queries atomic.Int32
+	cluster.EXPECT().QueryCopySegment(mock.Anything, mock.Anything).RunAndReturn(
+		func(nodeID int64, req *datapb.QueryCopySegmentRequest) (*datapb.QueryCopySegmentResponse, error) {
+			s.EqualValues(10, nodeID)
+			if queries.Add(1) == 1 {
+				return nil, errors.New("one-off transport error")
+			}
+			return &datapb.QueryCopySegmentResponse{
+				TaskID: 1001,
+				State:  datapb.CopySegmentTaskState_CopySegmentTaskInProgress,
+			}, nil
+		})
+	var creates atomic.Int32
+	cluster.EXPECT().CreateCopySegment(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
+		func(int64, *datapb.CopySegmentRequest, int64) error {
+			creates.Add(1)
+			return nil
+		}).Maybe()
+	// Only reached if the task is wrongly reset to Pending and re-scheduled.
+	cluster.EXPECT().QuerySlot().Return(map[int64]*session.WorkerSlots{
+		10: {NodeID: 10, AvailableSlots: 16},
+	}).Maybe()
+
+	copyTask := createTestCopyTask(100, 2001).(*copySegmentTask)
+	copyTask.task.Load().NodeId = 10
+	newCopySegmentTaskTestMeta(s.T(), copyTask)
+
+	scheduler := task2.NewGlobalTaskScheduler(context.Background(), cluster)
+	scheduler.Enqueue(copyTask)
+	scheduler.Start()
+	defer scheduler.Stop()
+
+	// The task must survive the transient error and keep being polled on the
+	// same node across multiple check ticks.
+	s.Eventually(func() bool { return queries.Load() >= 3 }, 10*time.Second, 20*time.Millisecond)
+	s.EqualValues(0, creates.Load())
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskInProgress, copyTask.GetState())
+	s.EqualValues(10, copyTask.GetNodeId())
+}
+
+func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_NodeGoneResetsToPending() {
+	// Regression test for DataNode restart/replacement: the node manager
+	// returns NodeNotFound, surfaced as a query error. The worker-side task is
+	// confirmed lost, so the task must be reset to Pending with NullNodeID so
+	// the scheduler re-dispatches it to a live node rather than polling the
+	// dead node until the job-level timeout.
 	cluster := session.NewMockCluster(s.T())
 	cluster.EXPECT().QueryCopySegment(mock.Anything, mock.Anything).Return(
 		nil,
-		errors.New("rpc failed"),
+		merr.WrapErrNodeNotFound(10),
 	)
 
 	task := createTestCopyTask(100, 2001).(*copySegmentTask)
+	task.task.Load().NodeId = 10
 	copyMeta, _ := newCopySegmentTaskTestMeta(s.T(), task)
+	s.NoError(copyMeta.AddJob(context.Background(), newTestCopyJob(100, datapb.CopySegmentJobState_CopySegmentJobExecuting)))
 
 	task.QueryTaskOnWorker(cluster)
 
 	updatedTask := copyMeta.GetTask(context.Background(), 1001)
-	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskFailed, updatedTask.GetState())
-	s.Contains(updatedTask.GetReason(), "rpc failed")
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskPending, updatedTask.GetState())
+	s.EqualValues(NullNodeID, updatedTask.GetNodeId())
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting,
+		copyMeta.GetJob(context.Background(), 100).GetState())
+}
+
+func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_TaskNotFoundOnWorkerResetsToPending() {
+	// Regression test for a DataNode that restarted but kept (or re-acquired)
+	// its session: the node is reachable, but its in-memory task manager no
+	// longer has the task, so QueryCopySegment surfaces the DataNode's
+	// task-not-found status (importv2.WrapTaskNotFoundError -> ErrImportSysFailed)
+	// as an error. The task is confirmed lost and must be reset to Pending with
+	// NullNodeID for re-dispatch instead of polling until the job-level timeout.
+	cluster := session.NewMockCluster(s.T())
+	// Round-trip the error through Status -> Error exactly like
+	// cluster.queryTask's merr.CheckRPCCall does with the DataNode's response
+	// status, so the test also covers the code-preservation assumption.
+	taskNotFoundErr := merr.Error(merr.Status(
+		merr.WrapErrImportSysFailedMsg("cannot find import task with id %d", 1001)))
+	cluster.EXPECT().QueryCopySegment(mock.Anything, mock.Anything).Return(nil, taskNotFoundErr)
+
+	task := createTestCopyTask(100, 2001).(*copySegmentTask)
+	task.task.Load().NodeId = 10
+	copyMeta, _ := newCopySegmentTaskTestMeta(s.T(), task)
+	s.NoError(copyMeta.AddJob(context.Background(), newTestCopyJob(100, datapb.CopySegmentJobState_CopySegmentJobExecuting)))
+
+	task.QueryTaskOnWorker(cluster)
+
+	updatedTask := copyMeta.GetTask(context.Background(), 1001)
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskPending, updatedTask.GetState())
+	s.EqualValues(NullNodeID, updatedTask.GetNodeId())
+	s.Empty(updatedTask.GetReason())
+	// The parent job must NOT be failed - the task is re-dispatched.
+	s.Equal(datapb.CopySegmentJobState_CopySegmentJobExecuting,
+		copyMeta.GetJob(context.Background(), 100).GetState())
 }
 
 func (s *CopySegmentTaskSuite) TestQueryTaskOnWorker_MarksFailedOnWorkerFailure() {
@@ -687,6 +808,22 @@ func createTestIndexMeta(t *testing.T, collectionID int64, indexes map[int64]*mo
 	return im
 }
 
+func (s *CopySegmentTaskSuite) TestClone_DeepCopiesProto() {
+	task := createTestCopyTask(100, 2001).(*copySegmentTask)
+
+	cloned := task.Clone()
+	// Mutating the clone (as UpdateTask actions do) must not leak into the
+	// original task; otherwise a failed catalog save during UpdateTask leaves
+	// the in-memory cache already mutated while etcd keeps the old state.
+	UpdateCopyTaskState(datapb.CopySegmentTaskState_CopySegmentTaskFailed)(cloned)
+	UpdateCopyTaskReason("mutated on clone")(cloned)
+
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskInProgress, task.GetState())
+	s.Empty(task.GetReason())
+	s.Equal(datapb.CopySegmentTaskState_CopySegmentTaskFailed, cloned.GetState())
+	s.Equal("mutated on clone", cloned.GetReason())
+}
+
 // createTestCopyTask creates a minimal CopySegmentTask for testing syncVectorScalarIndexes.
 func createTestCopyTask(collectionID int64, segmentID int64) CopySegmentTask {
 	task := &copySegmentTask{
@@ -705,6 +842,17 @@ func createTestCopyTask(collectionID int64, segmentID int64) CopySegmentTask {
 	return task
 }
 
+func newTestCopyJob(jobID int64, state datapb.CopySegmentJobState) CopySegmentJob {
+	return &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        jobID,
+			CollectionId: 100,
+			State:        state,
+		},
+		tr: timerecord.NewTimeRecorder("test job"),
+	}
+}
+
 func newCopySegmentTaskTestMeta(t *testing.T, task *copySegmentTask) (CopySegmentMeta, *meta) {
 	ctx := context.Background()
 	catalog := kvdatacoord.NewCatalog(NewMetaMemoryKV(), "", "")
@@ -1328,6 +1476,92 @@ func TestAssembleCopySegmentRequest_AllocatesTextAndJsonBuildIDs(t *testing.T) {
 	}
 }
 
+func TestAssembleCopySegmentRequest_RedispatchAllocatesFreshBuildIDs(t *testing.T) {
+	// Regression test for re-dispatch idempotency: when a task is reset to
+	// Pending after a worker loss and dispatched again, the new attempt must
+	// allocate a fresh set of buildIDs. Index files written by the partial
+	// earlier attempt (under the old buildIDs) are then never referenced by
+	// meta — they stay orphaned and are removed by GC — while the new attempt
+	// cannot collide with them.
+	snapshotData := &SnapshotData{
+		SnapshotInfo: &datapb.SnapshotInfo{
+			Id:           1,
+			CollectionId: 100,
+			Name:         "test_snapshot",
+		},
+		Segments: []*datapb.SegmentDescription{
+			{
+				SegmentId:   1,
+				PartitionId: 10,
+				IndexFiles: []*indexpb.IndexFilePathInfo{
+					{BuildID: 3001, FieldID: 100, IndexID: 1001},
+				},
+			},
+		},
+	}
+
+	sm := &snapshotMeta{}
+	mock1 := mockey.Mock((*snapshotMeta).ReadSnapshotData).Return(snapshotData, nil).Build()
+	defer mock1.UnPatch()
+
+	nextID := int64(9001)
+	alloc := &embeddedAllocator{}
+	mock2 := mockey.Mock((*embeddedAllocator).AllocID).To(func(ctx context.Context) (typeutil.UniqueID, error) {
+		id := nextID
+		nextID++
+		return id, nil
+	}).Build()
+	defer mock2.UnPatch()
+
+	mock3 := mockey.Mock(createStorageConfig).Return(nil).Build()
+	defer mock3.UnPatch()
+
+	task := &copySegmentTask{
+		snapshotMeta: sm,
+		alloc:        alloc,
+		tr:           timerecord.NewTimeRecorder("test"),
+		times:        taskcommon.NewTimes(),
+	}
+	task.task.Store(&datapb.CopySegmentTask{
+		TaskId:       1001,
+		JobId:        100,
+		CollectionId: 100,
+		IdMappings: []*datapb.CopySegmentIDMapping{
+			{SourceSegmentId: 1, TargetSegmentId: 2001, PartitionId: 10},
+		},
+	})
+
+	job := &copySegmentJob{
+		CopySegmentJob: &datapb.CopySegmentJob{
+			JobId:        100,
+			CollectionId: 100,
+			SnapshotName: "test_snapshot",
+		},
+		tr: timerecord.NewTimeRecorder("test_job"),
+	}
+
+	// First dispatch attempt
+	firstReq, err := AssembleCopySegmentRequest(task, job)
+	assert.NoError(t, err)
+	// Second dispatch attempt after a simulated worker loss + reset
+	secondReq, err := AssembleCopySegmentRequest(task, job)
+	assert.NoError(t, err)
+
+	firstIDs := firstReq.Targets[0].GetNewBuildIds()
+	secondIDs := secondReq.Targets[0].GetNewBuildIds()
+	assert.Len(t, firstIDs, 1)
+	assert.Len(t, secondIDs, 1)
+	// The re-dispatched attempt must not reuse any buildID from the first
+	// attempt, otherwise its index files could collide with partial files
+	// written by the lost worker.
+	for srcID, newFirst := range firstIDs {
+		newSecond, ok := secondIDs[srcID]
+		assert.True(t, ok)
+		assert.NotEqual(t, newFirst, newSecond,
+			"re-dispatch must allocate a fresh buildID for source buildID %d", srcID)
+	}
+}
+
 // embeddedAllocator: named type for mockey interface-method patching; avoids a
 // go1.26 `go vet` printf-pass panic on method expressions of anonymous structs.
 type embeddedAllocator struct{ allocator.Allocator }

internal/datacoord/snapshot_manager.go

@@ -1249,7 +1249,7 @@ func (sm *snapshotManager) createRestoreJob(
 			CollectionId: targetCollection,
 			State:        datapb.CopySegmentJobState_CopySegmentJobPending,
 			IdMappings:   idMappings,
-			TimeoutTs:    uint64(time.Now().Add(jobTimeout).UnixNano()),
+			TimeoutTs:    CopyJobTimeoutTs(jobTimeout),
 			StartTs:      uint64(time.Now().UnixNano()),
 			Options: []*commonpb.KeyValuePair{
 				{Key: "copy_index", Value: "true"},