diff --git a/pkg/tern/local_apply_grouped.go b/pkg/tern/local_apply_grouped.go index 87baad1c8..ac2d2ea1f 100644 --- a/pkg/tern/local_apply_grouped.go +++ b/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) diff --git a/pkg/tern/local_apply_grouped_progress_test.go b/pkg/tern/local_apply_grouped_progress_test.go new file mode 100644 index 000000000..092d0f445 --- /dev/null +++ b/pkg/tern/local_apply_grouped_progress_test.go @@ -0,0 +1,313 @@ +package tern + +import ( + "fmt" + "log/slog" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/engine" + "github.com/block/schemabot/pkg/schema" + "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" +) + +// A pending grouped progress report only signals lost work when at least one +// stored task is genuinely in flight. Resting states — stopped, +// failed_retryable — have no active engine work by design, terminal states are +// never re-verified, and an apply with no tasks has nothing to settle. +func TestEngineReportsLostApplyWork(t *testing.T) { + cases := []struct { + name string + engineTaskState string + taskStates []string + want bool + }{ + {"running task with pending engine report is lost work", state.Task.Pending, []string{state.Task.Running, state.Task.Completed}, true}, + {"cutting-over task with pending engine report is lost work", state.Task.Pending, []string{state.Task.CuttingOver}, true}, + {"all tasks at rest is not divergence", state.Task.Pending, []string{state.Task.FailedRetryable, state.Task.Stopped}, false}, + {"all tasks terminal is never re-verified", state.Task.Pending, []string{state.Task.Completed, state.Task.Cancelled}, false}, + {"running engine report is not lost work", state.Task.Running, []string{state.Task.Running}, false}, + {"no tasks means nothing to settle", state.Task.Pending, nil, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + tasks := make([]*storage.Task, len(tc.taskStates)) + for i, s := range tc.taskStates { + tasks[i] = &storage.Task{State: s} + } + assert.Equal(t, tc.want, engineReportsLostApplyWork(tc.engineTaskState, tasks)) + }) + } +} + +// lostWorkAtomicPollFixture builds a running grouped (Vitess) apply with two +// running tasks and a LocalClient polling the given engine with a fast poll +// cadence, so lost-work scenarios can drive many polls without waiting out the +// real cadence. trustBudget sets how long the drive keeps trusting an engine +// reporting no active schema change: a tiny budget reaches target +// verification, and a large one proves a short pending run is tolerated. +func lostWorkAtomicPollFixture(eng engine.Engine, trustBudget time.Duration) (*LocalClient, *storage.Apply, []*storage.Task, *stateRecordingTaskStore) { + return lostWorkAtomicPollFixtureInState(eng, trustBudget, state.Task.Running) +} + +// lostWorkAtomicPollFixtureInState is lostWorkAtomicPollFixture with the +// tasks' stored state chosen by the caller, for the states whose settlement +// rules differ. +func lostWorkAtomicPollFixtureInState(eng engine.Engine, trustBudget time.Duration, taskState string) (*LocalClient, *storage.Apply, []*storage.Task, *stateRecordingTaskStore) { + apply := &storage.Apply{ + ID: 1, PlanID: 7, ApplyIdentifier: "apply-1", Database: "appdb", + DatabaseType: storage.DatabaseTypeVitess, Environment: "staging", + State: state.Apply.Running, + } + tasks := []*storage.Task{ + { + ID: 1, ApplyID: 1, PlanID: 7, TaskIdentifier: "task-orders", + Database: "appdb", DatabaseType: storage.DatabaseTypeVitess, + Namespace: "appdb", TableName: "orders", + Environment: "staging", State: taskState, + }, + { + ID: 2, ApplyID: 1, PlanID: 7, TaskIdentifier: "task-payments", + Database: "appdb", DatabaseType: storage.DatabaseTypeVitess, + Namespace: "appdb", TableName: "payments", + Environment: "staging", State: taskState, + }, + } + taskStore := &stateRecordingTaskStore{ + exactProgressTaskStore: &exactProgressTaskStore{tasks: tasks}, + } + client := &LocalClient{ + config: LocalConfig{Database: "appdb", Type: storage.DatabaseTypeVitess}, + planetscaleEngine: eng, + storage: &exactProgressStorage{ + applies: &snapshotApplyStore{stored: *apply}, + tasks: taskStore, + controlRequests: &testControlRequestStore{}, + logs: &mockApplyLogStore{}, + plans: &scriptedPlanStore{plan: &storage.Plan{ID: 7, SchemaFiles: schema.SchemaFiles{}}}, + }, + logger: slog.Default(), + taskPollIntervalOverride: time.Millisecond, + lostEngineWorkPendingBudgetOverride: trustBudget, + } + return client, apply, tasks, taskStore +} + +// An engine can complete a grouped apply's work and then lose all record of it +// — after that, every progress poll reports no active schema change while +// durable storage still says the tasks are running. Once the tolerated +// staleness window is exhausted the drive must verify the target directly: a +// target that already has the desired schema means the change landed and only +// its outcome was lost, so every task completes and the apply terminalizes +// instead of polling forever and holding the database's active-apply slot. +func TestPollForCompletionAtomic_LostEngineWorkTargetConverged(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StateRunning}, + {State: engine.StatePending}, + }}, + // The re-plan reports no remaining change for any table: the target + // already has the desired schema. + planResult: &engine.PlanResult{NoChanges: true}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixture(eng, lostWorkTrustBudgetReached) + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.Completed, apply.State, "a converged target settles the apply through the normal completed flow") + require.NotNil(t, apply.CompletedAt) + for _, task := range tasks { + assert.Equal(t, state.Task.Completed, task.State, "table %s", task.TableName) + assert.Equal(t, 100, task.ProgressPercent, "table %s", task.TableName) + require.NotNil(t, task.CompletedAt, "table %s", task.TableName) + } + assert.Equal(t, 1, eng.planCalls, "one shared re-plan verifies the target for every task of the grouped apply") + assert.GreaterOrEqual(t, eng.calls, 3, "a single pending report never settles the apply; the engine is polled again first") +} + +// An engine that never had (or irrecoverably lost) grouped work reports no +// active schema change forever while the target still needs some of it. The +// drive settles each task on its own verification verdict: a table whose +// change already landed completes, a table the target still needs rests +// retryable — never permanently failed, because nothing about the target is +// broken — and the apply pauses for a fresh claim to re-drive the rest. +func TestPollForCompletionAtomic_LostEngineWorkTargetNotConverged(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + }}, + // The re-plan still contains orders but not payments: the target has + // payments' change and still needs orders'. + planResult: &engine.PlanResult{Changes: []engine.SchemaChange{{ + Namespace: "appdb", + TableChanges: []engine.TableChange{{ + Table: "orders", + DDL: "ALTER TABLE `orders` ADD COLUMN `note` VARCHAR(255)", + }}, + }}}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixture(eng, lostWorkTrustBudgetReached) + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.FailedRetryable, apply.State, "a lost change the target still needs pauses the apply for retry, never fails it permanently") + assert.Nil(t, apply.CompletedAt, "a retryable apply carries no completion timestamp") + orders, payments := tasks[0], tasks[1] + assert.Equal(t, state.Task.FailedRetryable, orders.State, "the table the target still needs is retryable") + assert.Contains(t, orders.ErrorMessage, "orders") + assert.Contains(t, orders.ErrorMessage, "still needs the change") + assert.Nil(t, orders.CompletedAt, "a retryable task carries no completion timestamp") + assert.Equal(t, state.Task.Completed, payments.State, "the table whose change already landed completes") + require.NotNil(t, payments.CompletedAt) +} + +// When the engine reports no active schema change and the target plan cannot +// be read either, neither side can answer what happened to the work. The drive +// must not spin between the two forever: each failed verification counts +// against the same bounded error budget as a failed poll, and exhausting it +// pauses the apply retryable for a fresh claim to re-drive — never permanently +// failed, because nothing proved the target is broken. +func TestPollForCompletionAtomic_LostEngineWorkVerificationErrorsAreBounded(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + }}, + planResult: &engine.PlanResult{NoChanges: true}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixture(eng, lostWorkTrustBudgetReached) + client.storage.(*exactProgressStorage).plans = &scriptedPlanStore{err: fmt.Errorf("storage read failed")} + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.FailedRetryable, apply.State, "an unverifiable target pauses the apply retryable, never permanently failed") + assert.Contains(t, apply.ErrorMessage, "could not be verified") + assert.Contains(t, apply.ErrorMessage, "consecutive errors") + for _, task := range tasks { + assert.Equal(t, state.Task.FailedRetryable, task.State, "table %s", task.TableName) + assert.Nil(t, task.CompletedAt, "a retryable task carries no completion timestamp (table %s)", task.TableName) + } + assert.Equal(t, 0, eng.planCalls, "a failed plan read settles nothing; the engine re-plan is never reached") +} + +// A freshly restarted engine can serve a stale snapshot that omits in-flight +// grouped work for a few polls before it catches up. A short run of pending +// reports inside the tolerated window must self-heal: the drive keeps polling, +// never distrusts the engine, and the apply completes through the normal flow. +func TestPollForCompletionAtomic_StaleEngineSnapshotSelfHeals(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + {State: engine.StatePending}, + {State: engine.StatePending}, + {State: engine.StateRunning}, + {State: engine.StateCompleted}, + }}, + planResult: &engine.PlanResult{NoChanges: true}, + } + client, apply, tasks, taskStore := lostWorkAtomicPollFixture(eng, lostWorkTrustBudgetAmple) + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.Completed, apply.State) + for _, task := range tasks { + assert.Equal(t, state.Task.Completed, task.State, "table %s", task.TableName) + } + assert.Equal(t, 0, eng.planCalls, "a self-healing stale snapshot never triggers target verification") + assert.NotContains(t, taskStore.states, state.Task.FailedRetryable) + assert.NotContains(t, taskStore.states, state.Task.Failed) +} + +// Settlement touches only the tasks whose stored state says work is in +// flight. A task that already reached its terminal state — here a table that +// cut over and completed before the engine lost the rest of the apply — is a +// durable final answer: the target read must not re-settle it or rewrite its +// completion record. +func TestPollForCompletionAtomic_LostEngineWorkLeavesSettledTasksUntouched(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + }}, + // The re-plan still contains orders: only the in-flight task is settled + // from it; the already-completed payments is not consulted at all. + planResult: &engine.PlanResult{Changes: []engine.SchemaChange{{ + Namespace: "appdb", + TableChanges: []engine.TableChange{{ + Table: "orders", + DDL: "ALTER TABLE `orders` ADD COLUMN `note` VARCHAR(255)", + }}, + }}}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixture(eng, lostWorkTrustBudgetReached) + orders, payments := tasks[0], tasks[1] + completedEarlier := time.Now().Add(-time.Hour) + payments.State = state.Task.Completed + payments.CompletedAt = &completedEarlier + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Task.FailedRetryable, orders.State, "the in-flight task still settles from the target read") + assert.Equal(t, state.Task.Completed, payments.State, "a terminal task is never re-settled") + require.NotNil(t, payments.CompletedAt) + assert.Equal(t, completedEarlier, *payments.CompletedAt, "a terminal task's completion record is never rewritten by settlement") + assert.Equal(t, state.Apply.FailedRetryable, apply.State) +} + +// Once a grouped schema change has cut over, the live schema matches the +// reviewed target whether or not the revert that was undoing it ever ran — so +// tasks in their revert phase can never be settled by reading the target. An +// engine that loses a revert must leave them retryable for a fresh claim to +// re-drive; completing them would report the apply as a successful schema +// change while the change it was reverting is still in place. +func TestPollForCompletionAtomic_LostEngineWorkNeverCompletesRevertPhaseTasks(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + }}, + // A converged target is exactly what a post-cutover re-plan reports, and + // it must not be read as the revert having finished. + planResult: &engine.PlanResult{NoChanges: true}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixtureInState(eng, lostWorkTrustBudgetReached, state.Task.Reverting) + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.FailedRetryable, apply.State, "a lost revert pauses the apply for retry, never completes it") + for _, task := range tasks { + assert.Equal(t, state.Task.FailedRetryable, task.State, "a lost revert is retryable, never a completed schema change (table %s)", task.TableName) + assert.Nil(t, task.CompletedAt, "a retryable task carries no completion timestamp (table %s)", task.TableName) + assert.Contains(t, task.ErrorMessage, "revert phase") + } + assert.Equal(t, 0, eng.planCalls, "the target schema is never consulted for revert-phase tasks") +} + +// A revert-phase task needs no target read to settle, so a plan the drive +// cannot read must not strand it alongside the tasks that do need verifying. +// The revert rests retryable on its own reason while the forward task falls to +// the bounded error budget — an operator re-driving the apply has to be able to +// tell a lost revert from a forward change whose target could not be read. +func TestPollForCompletionAtomic_LostEngineWorkSettlesRevertPhaseTasksWhenVerificationFails(t *testing.T) { + eng := &lostWorkEngine{ + phaseSequenceEngine: phaseSequenceEngine{results: []*engine.ProgressResult{ + {State: engine.StatePending}, + }}, + planResult: &engine.PlanResult{NoChanges: true}, + } + client, apply, tasks, _ := lostWorkAtomicPollFixtureInState(eng, lostWorkTrustBudgetReached, state.Task.Reverting) + reverting, forward := tasks[0], tasks[1] + forward.State = state.Task.Running + client.storage.(*exactProgressStorage).plans = &scriptedPlanStore{err: fmt.Errorf("storage read failed")} + + client.pollForCompletionAtomic(t.Context(), apply, tasks, nil, nil, map[string]string{}, false) + + assert.Equal(t, state.Apply.FailedRetryable, apply.State, "an unverifiable target pauses the apply retryable, never permanently failed") + assert.Equal(t, state.Task.FailedRetryable, reverting.State, "a lost revert rests retryable without ever reading the target") + assert.Contains(t, reverting.ErrorMessage, "revert phase", "the revert-phase task keeps the reason that names its phase") + assert.Nil(t, reverting.CompletedAt, "a retryable task carries no completion timestamp") + assert.Equal(t, state.Task.FailedRetryable, forward.State, "the task that needed verifying falls to the bounded error budget") + assert.Contains(t, forward.ErrorMessage, "could not be verified") + assert.Equal(t, 0, eng.planCalls, "a failed plan read settles nothing by re-plan; the engine is never consulted") +} diff --git a/pkg/tern/local_apply_grouped_sync_test.go b/pkg/tern/local_apply_grouped_sync_test.go index 507943594..6bb9d0673 100644 --- a/pkg/tern/local_apply_grouped_sync_test.go +++ b/pkg/tern/local_apply_grouped_sync_test.go @@ -56,7 +56,7 @@ func TestSyncAtomicTaskProgress_RefinesPhaseAndDisplayPerTask(t *testing.T) { } client := groupedSyncClient(taskStore) - client.syncAtomicTaskProgress(t.Context(), slog.Default(), []*storage.Task{catchingUp, copying}, result, state.Task.Running, time.Now()) + client.syncAtomicTaskProgress(t.Context(), slog.Default(), []*storage.Task{catchingUp, copying}, result, state.Task.Running, time.Now(), settledTaskSet{}) assert.Equal(t, state.Task.CatchingUp, catchingUp.State, "a table applying its changeset renders as catching up") assert.EqualValues(t, 100, catchingUp.ProgressPercent) @@ -90,10 +90,42 @@ func TestSyncAtomicTaskProgress_UnreportedTableKeepsApplyStateAndLastProgress(t } client := groupedSyncClient(taskStore) - client.syncAtomicTaskProgress(t.Context(), slog.Default(), []*storage.Task{unreported}, result, state.Task.Running, time.Now()) + client.syncAtomicTaskProgress(t.Context(), slog.Default(), []*storage.Task{unreported}, result, state.Task.Running, time.Now(), settledTaskSet{}) assert.Equal(t, state.Task.Running, unreported.State, "a sibling table's phase never refines a task the engine did not report on") assert.EqualValues(t, 500, unreported.RowsCopied, "an unreported table keeps its last known progress") assert.EqualValues(t, 50, unreported.ProgressPercent) assert.Len(t, taskStore.states, 1) } + +// A task the drive already settled from the live target schema is out of the +// engine's hands: the poll that sent the drive to the target reports no active +// schema change, so it carries neither progress to display nor a state the +// task may take. The projection must leave such a task alone outright rather +// than write it again and lean on the no-backward guard to reject the claim. +func TestSyncAtomicTaskProgress_SettledTaskTakesNothingFromThePoll(t *testing.T) { + settledTask := &storage.Task{ + ID: 1, ApplyID: 1, TaskIdentifier: "task-1", + Database: "appdb", DatabaseType: storage.DatabaseTypeMySQL, + TableName: "mutes", State: state.Task.Completed, + RowsCopied: 900, RowsTotal: 900, ProgressPercent: 100, + } + taskStore := &stateRecordingTaskStore{ + exactProgressTaskStore: &exactProgressTaskStore{tasks: []*storage.Task{settledTask}}, + } + result := &engine.ProgressResult{ + State: engine.StateRunning, + Tables: []engine.TableProgress{{Table: "mutes", State: spiritstatus.CopyRows.String(), RowsCopied: 3, RowsTotal: 900, Progress: 1, Throttled: true, ThrottleReason: "replica lag"}}, + } + client := groupedSyncClient(taskStore) + settled := settledTaskSet{} + settled.add(settledTask) + + client.syncAtomicTaskProgress(t.Context(), slog.Default(), []*storage.Task{settledTask}, result, state.Task.Running, time.Now(), settled) + + assert.Equal(t, state.Task.Completed, settledTask.State, "the target's verdict stands; the poll claims nothing") + assert.EqualValues(t, 900, settledTask.RowsCopied, "a settled task keeps the progress its settlement left") + assert.EqualValues(t, 100, settledTask.ProgressPercent) + assert.False(t, settledTask.Throttled, "an engine that lost the work reports nothing a settled task should show") + assert.Empty(t, taskStore.states, "settlement already persisted the task; the projection writes it no further") +} diff --git a/pkg/tern/local_apply_sequential.go b/pkg/tern/local_apply_sequential.go index 285341e10..da468c482 100644 --- a/pkg/tern/local_apply_sequential.go +++ b/pkg/tern/local_apply_sequential.go @@ -320,6 +320,13 @@ type atomicPollState struct { // engine is unreachable (e.g., branch deleted mid-apply). consecutiveErrors int + // lostWork measures how long the engine has been reporting no active + // schema change while stored tasks say the work is in flight, so the drive + // can stop trusting the engine and settle the tasks from the target + // schema. One tracker covers the whole apply: grouped work is a single + // engine operation, so the engine loses or keeps all of it together. + lostWork lostEngineWorkTracker + // warnedPerShardUnavailable is set after the drive warns that a sharded // engine could not report per-shard/row-copy progress, so the warning is // emitted once per apply rather than on every poll. @@ -483,7 +490,8 @@ const ( defaultTaskStallWarnInterval = 5 * time.Minute ) -// taskPollInterval returns the sequential drive's progress poll cadence. +// taskPollInterval returns a drive's progress poll cadence, shared by the +// sequential and grouped polls. func (c *LocalClient) taskPollInterval() time.Duration { if c.taskPollIntervalOverride > 0 { return c.taskPollIntervalOverride @@ -797,24 +805,13 @@ func (t *lostEngineWorkTracker) reset() { // active while durable storage says this task's change is in flight, and that // divergence outlasted the tolerated staleness window — so engine progress can // never terminalize the task and the target itself is the only remaining -// authority. A target that already has the desired schema means the work -// finished and only its outcome was lost: the task completes. A target that -// still needs the change means the work is genuinely gone: the task is marked -// retryable so a fresh claim re-drives it — never permanently failed, because -// nothing about the target is known to be broken. A verification error is -// returned for the caller's consecutive-error budget to count. +// authority. Revert-phase tasks rest retryable without a target read; every +// other task settles from the verification verdict (see +// settleLostRevertPhaseTask and settleLostVerifiedTask). A verification error +// is returned for the caller's consecutive-error budget to count. func (c *LocalClient) settleLostEngineWork(ctx context.Context, apply *storage.Apply, task *storage.Task, engineState engine.State) (taskAction, error) { - // A revert-phase task can never be settled by reading the target. The - // forward change has already cut over, so the live schema matches the - // reviewed target by definition and a match says nothing about whether the - // revert this task was driving ever finished. Completing on it would report - // the apply as a successful schema change while the revert it was undoing is - // gone. Retryable is the only answer a schema read supports here. if taskInRevertPhase(task) { - c.logger.Warn("engine reports no active schema change for a revert-phase task; marking it retryable because the target schema cannot settle a revert", - append(task.LogAttrs(), "apply_id", apply.ApplyIdentifier, "engine_state", engineState)...) - c.markTaskRetryable(ctx, task, - fmt.Sprintf("engine reports no active schema change while table %s was in its revert phase; a fresh claim will re-drive it", task.TableName)) + c.settleLostRevertPhaseTask(ctx, apply, task, engineState) return taskFailed, nil } plan, err := c.storage.Plans().GetByID(ctx, apply.PlanID) @@ -828,6 +825,35 @@ func (c *LocalClient) settleLostEngineWork(ctx context.Context, apply *storage.A if err != nil { return taskContinue, fmt.Errorf("verify target schema for task %s table %s: %w", task.TaskIdentifier, task.TableName, err) } + c.settleLostVerifiedTask(ctx, apply, task, needsChange, engineState) + if needsChange { + return taskFailed, nil + } + return taskContinue, nil +} + +// settleLostRevertPhaseTask rests a revert-phase task retryable after the +// engine stopped reporting on it. A revert-phase task can never be settled by +// reading the target: the forward change has already cut over, so the live +// schema matches the reviewed target by definition and a match says nothing +// about whether the revert this task was driving ever finished. Completing on +// it would report the apply as a successful schema change while the revert it +// was undoing is gone. Retryable is the only answer a schema read supports. +func (c *LocalClient) settleLostRevertPhaseTask(ctx context.Context, apply *storage.Apply, task *storage.Task, engineState engine.State) { + c.logger.Warn("engine reports no active schema change for a revert-phase task; marking it retryable because the target schema cannot settle a revert", + append(task.LogAttrs(), "apply_id", apply.ApplyIdentifier, "engine_state", engineState)...) + c.markTaskRetryable(ctx, task, + fmt.Sprintf("engine reports no active schema change while table %s was in its revert phase; a fresh claim will re-drive it", task.TableName)) +} + +// settleLostVerifiedTask settles a task from its target-verification verdict +// after the engine stopped reporting on it. A target that already has the +// desired schema means the work finished and only its outcome was lost: the +// task completes. A target that still needs the change means the work is +// genuinely gone: the task rests retryable so a fresh claim re-drives it — +// never permanently failed, because nothing about the target is known to be +// broken. +func (c *LocalClient) settleLostVerifiedTask(ctx context.Context, apply *storage.Apply, task *storage.Task, needsChange bool, engineState engine.State) { if !needsChange { now := time.Now() task.ProgressPercent = 100 @@ -836,13 +862,12 @@ func (c *LocalClient) settleLostEngineWork(ctx context.Context, apply *storage.A append(task.LogAttrs(), "apply_id", apply.ApplyIdentifier, "engine_state", engineState)...) c.transitionTaskState(ctx, task, task.ApplyID, state.Task.Completed, fmt.Sprintf("Task %s completed: engine no longer reports the schema change and the target has the desired schema", task.TaskIdentifier)) - return taskContinue, nil + return } c.logger.Warn("engine reports no active schema change but the target still needs it; marking the task retryable for a fresh claim to re-drive", append(task.LogAttrs(), "apply_id", apply.ApplyIdentifier, "engine_state", engineState)...) c.markTaskRetryable(ctx, task, fmt.Sprintf("engine reports no active schema change for table %s but the target still needs the change; a fresh claim will re-drive it", task.TableName)) - return taskFailed, nil } // taskWaitsForOperatorAction reports whether a task's state is one the drive is diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go index 1c1aa16c9..90364a519 100644 --- a/pkg/tern/local_control_resume.go +++ b/pkg/tern/local_control_resume.go @@ -443,6 +443,17 @@ func replanShardTableDDL(result *engine.PlanResult) map[shardTableKey]string { return out } +// replanTargetSchema re-plans the reviewed schema set against the live target +// and indexes the remaining changes by (namespace, shard, table), so callers +// can look up whether each task's table still needs its change. +func (c *LocalClient) replanTargetSchema(ctx context.Context, apply *storage.Apply, plan *storage.Plan) (map[shardTableKey]string, error) { + result, err := c.planWithEngine(ctx, &ternv1.PlanRequest{}, apply.Database, plan.SchemaFiles) + if err != nil { + return nil, fmt.Errorf("re-plan check failed: %w", err) + } + return replanShardTableDDL(result), nil +} + // tableStillNeedsChange re-plans the full schema set and then looks up whether // this task's table still needs a change on its (namespace, shard). Returns // false if it already has the desired schema (e.g., Spirit's cutover completed @@ -450,11 +461,11 @@ func replanShardTableDDL(result *engine.PlanResult) map[shardTableKey]string { // the DDL the re-plan would now apply so the caller can confirm it still matches // the reviewed DDL before applying it. func (c *LocalClient) tableStillNeedsChange(ctx context.Context, apply *storage.Apply, plan *storage.Plan, task *storage.Task) (string, bool, error) { - result, err := c.planWithEngine(ctx, &ternv1.PlanRequest{}, apply.Database, plan.SchemaFiles) + replanDDL, err := c.replanTargetSchema(ctx, apply, plan) if err != nil { - return "", false, fmt.Errorf("re-plan check failed: %w", err) + return "", false, err } - ddl, stillNeeded := replanShardTableDDL(result)[shardTableKey{namespace: task.Namespace, shard: task.Shard, table: task.TableName}] + ddl, stillNeeded := replanDDL[shardTableKey{namespace: task.Namespace, shard: task.Shard, table: task.TableName}] return ddl, stillNeeded, nil }