fix(tern): settle grouped applies when the engine loses in-flight work

pkg/tern/local_apply_grouped.go

@@ -560,14 +560,18 @@ func (c *LocalClient) deriveAggregateApplyState(ctx context.Context, apply *stor
 // pollForCompletionAtomic polls the engine for progress in atomic mode (all tasks share state).
 func (c *LocalClient) pollForCompletionAtomic(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, creds *engine.Credentials, resumeState *engine.ResumeState, options map[string]string, releaseAtCutoverBarrier bool) {
 	eng := c.getEngine()
-	ticker := time.NewTicker(defaultTaskPollInterval)
+	ticker := time.NewTicker(c.taskPollInterval())
 	defer ticker.Stop()
 
 	// Seed revertSkipped from the durable signal so a driver that picks this apply
 	// up after a restart treats skip-revert as already accepted: it won't re-attempt
 	// SkipRevert, and it keeps surfacing skipping_revert while finalization is in
 	// flight rather than falling back to revert_window.
-	ps := &atomicPollState{lastProgressLog: time.Now(), revertSkipped: apply.RevertSkippedAt != nil}
+	ps := &atomicPollState{
+		lastProgressLog: time.Now(),
+		revertSkipped:   apply.RevertSkippedAt != nil,
+		lostWork:        lostEngineWorkTracker{budget: c.lostEngineWorkPendingBudget(eng)},
+	}
 
 	for {
 		select {
@@ -652,6 +656,45 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E
 		}
 		return false
 	}
+	now := time.Now()
+	newState := taskStateFromProgressResult(result)
+
+	settled := settledTaskSet{}
+	if engineReportsLostApplyWork(newState, tasks) {
+		pendingFor, exhausted := ps.lostWork.observePending(now)
+		if exhausted {
+			var settleErr error
+			if settled, settleErr = c.settleLostEngineWorkForTasks(ctx, apply, tasks, result.State); settleErr != nil {
+				// Neither the engine nor the target has answered what happened
+				// to the work, so count the failed verification against the
+				// same bounded error budget as a failed poll — this must never
+				// become an unbounded loop. Return before the reset below so
+				// the healthy poll that carried the pending report cannot
+				// clear the count.
+				ps.consecutiveErrors++
+				logger.Warn("engine reports no active schema change for an in-flight apply and target verification failed; the drive re-verifies at the next poll",
+					append(apply.MutableLogAttrs(), "engine_state", result.State, "consecutive_errors", ps.consecutiveErrors, "error", settleErr)...)
+				if ps.consecutiveErrors >= maxConsecutiveProgressPollErrors {
+					c.markApplyRetryableWithTasks(ctx, apply, tasks, fmt.Sprintf("engine reports no active schema change for an in-flight apply and the target could not be verified; %d consecutive errors across progress polls and target verification; see server logs", ps.consecutiveErrors))
+					return true
+				}
+				return false
+			}
+			// The settled tasks are terminal or resting now, and the progress
+			// sync below leaves them out, so the tick falls through and the
+			// apply-state derivation quiesces the apply from the settled task
+			// states.
+		} else {
+			// Inside the budget the engine is still trusted: it may be
+			// serving a stale snapshot after a restart, or reporting
+			// pending for real work it has not begun executing yet.
+			logger.Debug("engine reports no active schema change for an in-flight apply; still inside the trust budget",
+				append(apply.MutableLogAttrs(), "engine_state", result.State, "pending_for", pendingFor.Round(time.Second))...)
+		}
+	} else {
+		ps.lostWork.reset()
+	}
+
 	ps.consecutiveErrors = 0
 	c.logEngineResumeOnce(ctx, logger, apply, result.ResumedFromCheckpoint, &ps.resumeEventLogged)
 
@@ -683,9 +726,6 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E
 		}
 	}
 
-	now := time.Now()
-	newState := taskStateFromProgressResult(result)
-
 	// Log state transitions and track when waiting states are entered (for timeouts)
 	if newState != ps.lastTaskState {
 		msg := fmt.Sprintf("State changed to %s", newState)
@@ -717,7 +757,7 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E
 	c.logAtomicProgress(ctx, apply, result, ps, now)
 
 	// Update all tasks with engine progress
-	c.syncAtomicTaskProgress(ctx, logger, tasks, result, newState, now)
+	c.syncAtomicTaskProgress(ctx, logger, tasks, result, newState, now, settled)
 	if handled, err := c.processPendingCancelOrStopControlRequest(ctx, apply); err != nil {
 		logger.Warn("pending stop request processing failed after progress sync; current apply owner will exit for operator retry",
 			"error", err)
@@ -1079,6 +1119,94 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E
 	return false
 }
 
+// engineReportsLostApplyWork reports whether a grouped progress poll came back
+// with no active schema change (a state that maps to pending) while durable
+// storage says at least one of the apply's tasks is in flight. A grouped apply
+// is a single engine operation, so the engine-vs-storage divergence is
+// detected at the apply level and one trust budget covers all of its tasks;
+// the per-task stored states still decide which tasks a settlement touches.
+// Like the sequential detection, this shape is ambiguous on its own — the
+// caller treats it as the start of a timed trust budget, not as evidence (see
+// engineReportsLostWork).
+func engineReportsLostApplyWork(engineTaskState string, tasks []*storage.Task) bool {
+	for _, task := range tasks {
+		if engineReportsLostWork(task.State, engineTaskState) {
+			return true
+		}
+	}
+	return false
+}
+
+// settleLostEngineWorkForTasks resolves a grouped apply's in-flight tasks after
+// the engine has stopped reporting on them, by verifying the target schema
+// directly — the grouped counterpart of settleLostEngineWork. The engine says
+// no schema change is active while durable storage says the apply's work is in
+// flight, and that divergence outlasted the tolerated staleness window, so
+// engine progress can never terminalize these tasks and the target itself is
+// the only remaining authority.
+//
+// Revert-phase tasks rest retryable without a target read — a schema read
+// cannot settle a revert — so they settle even when the re-plan below fails.
+// Every other in-flight task settles from one shared re-plan of the reviewed
+// schema set: completed when its (namespace, shard, table) no longer needs a
+// change, retryable when it still does. Tasks already at rest or terminal are
+// left untouched, so a settlement retried after a verification error never
+// re-settles what an earlier pass already decided. A verification error is
+// returned for the caller's consecutive-error budget to count.
+//
+// Returns the tasks it settled, so the caller can keep the same tick's engine
+// report from claiming a state for work the target already answered for.
+func (c *LocalClient) settleLostEngineWorkForTasks(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, engineState engine.State) (settledTaskSet, error) {
+	settled := settledTaskSet{}
+	var unverified []*storage.Task
+	for _, task := range tasks {
+		if !state.IsInFlightTaskState(task.State) {
+			c.logger.Debug("leaving task out of lost-work settlement; its stored state has no active engine work",
+				append(task.LogAttrs(), "apply_id", apply.ApplyIdentifier, "engine_state", engineState)...)
+			continue
+		}
+		if taskInRevertPhase(task) {
+			c.settleLostRevertPhaseTask(ctx, apply, task, engineState)
+			settled.add(task)
+			continue
+		}
+		unverified = append(unverified, task)
+	}
+	if len(unverified) == 0 {
+		return settled, nil
+	}
+	plan, err := c.storage.Plans().GetByID(ctx, apply.PlanID)
+	if err != nil {
+		return settled, fmt.Errorf("load plan for apply %s to verify target schema: %w", apply.ApplyIdentifier, err)
+	}
+	if plan == nil {
+		return settled, fmt.Errorf("plan not found for apply %s while verifying target schema", apply.ApplyIdentifier)
+	}
+	replanDDL, err := c.replanTargetSchema(ctx, apply, plan)
+	if err != nil {
+		return settled, fmt.Errorf("verify target schema for apply %s: %w", apply.ApplyIdentifier, err)
+	}
+	for _, task := range unverified {
+		_, needsChange := replanDDL[shardTableKey{namespace: task.Namespace, shard: task.Shard, table: task.TableName}]
+		c.settleLostVerifiedTask(ctx, apply, task, needsChange, engineState)
+		settled.add(task)
+	}
+	return settled, nil
+}
+
+// settledTaskSet holds the tasks a drive resolved from a more authoritative
+// source than the engine during a single tick — here, the live target schema.
+// Their state is decided and persisted, so the same tick's engine report is no
+// longer entitled to claim one for them.
+type settledTaskSet map[*storage.Task]struct{}
+
+func (s settledTaskSet) add(task *storage.Task) { s[task] = struct{}{} }
+
+func (s settledTaskSet) contains(task *storage.Task) bool {
+	_, ok := s[task]
+	return ok
+}
+
 // autoTriggerCutover fires the engine cutover for a drive that is not
 // deferring cutover, pacing the operator-visible signal around the backend's
 // staging window: the trigger event is recorded once per drive so retries do
@@ -1241,12 +1369,13 @@ func (c *LocalClient) logAtomicProgress(ctx context.Context, apply *storage.Appl
 // always refresh, and the state the poll claims for the task, which is a
 // correctness decision.
 //
-// Every task ends its tick with a persisted write, even when no field moved:
-// the operator reads tasks.updated_at as the drive's liveness signal
-// (ApplyDriveStallAfter) and cancels a drive whose rows stop advancing, so the
-// write must stay unconditional — including through parked states such as
-// deferred cutovers and revert windows, where nothing changes tick to tick.
-func (c *LocalClient) syncAtomicTaskProgress(ctx context.Context, logger *slog.Logger, tasks []*storage.Task, result *engine.ProgressResult, newState string, now time.Time) {
+// Every task the poll speaks for ends its tick with a persisted write, even
+// when no field moved: the operator reads tasks.updated_at as the drive's
+// liveness signal (ApplyDriveStallAfter) and cancels a drive whose rows stop
+// advancing, so the write must stay unconditional — including through parked
+// states such as deferred cutovers and revert windows, where nothing changes
+// tick to tick.
+func (c *LocalClient) syncAtomicTaskProgress(ctx context.Context, logger *slog.Logger, tasks []*storage.Task, result *engine.ProgressResult, newState string, now time.Time, settled settledTaskSet) {
 	tableProgress := indexEngineTableProgress(result.Tables)
 	retryableFailure := state.IsState(newState, state.Task.FailedRetryable)
 	instantFromMetadata := false
@@ -1260,6 +1389,15 @@ func (c *LocalClient) syncAtomicTaskProgress(ctx context.Context, logger *slog.L
 		if retryableFailure && state.IsTerminalTaskState(task.State) {
 			continue
 		}
+		if settled.contains(task) {
+			// The target schema answered for this task earlier in the tick and
+			// settlement persisted the result. The poll that sent the drive to
+			// the target reports no active schema change, so it carries neither
+			// progress to display nor a state this task may take.
+			logger.Debug("leaving settled task out of the engine progress projection",
+				append(task.LogAttrs(), "engine_state", result.State)...)
+			continue
+		}
 		tp, _ := engineProgressForTask(tableProgress, task)
 		c.refreshTaskDisplayFromEngine(ctx, logger, task, tp, result, instantFromMetadata, retryableFailure, now)
 		c.advanceTaskFromEngineProgress(ctx, task, newState, tp, result, retryableFailure, now)