fix(tern): transient storage errors no longer terminalize in-flight applies

pkg/tern/local_apply_failure.go

@@ -11,7 +11,9 @@ import (
 
 // failApplyWithTasks marks all tasks and the apply as failed with the given error.
 // If the apply is already in a terminal state (e.g., cancelled by Stop()), the
-// apply state is not overwritten.
+// stored state is not overwritten; the settled state is adopted into the
+// in-memory apply so callers that notify observers afterwards report the
+// concurrent verdict instead of the stale pre-failure state.
 func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, errMsg string) {
 	now := time.Now()
 	for _, task := range tasks {
@@ -31,6 +33,7 @@ func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.App
 	if err == nil && fresh != nil && state.IsTerminalApplyState(fresh.State) {
 		c.logger.Debug("apply already in terminal state, not overwriting",
 			"apply_id", apply.ApplyIdentifier, "state", fresh.State)
+		*apply = *fresh
 		return
 	}
 
@@ -65,6 +68,7 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 	if err == nil && fresh != nil && state.IsTerminalApplyState(fresh.State) {
 		c.logger.Debug("apply already in terminal state, not marking retryable",
 			"apply_id", apply.ApplyIdentifier, "state", fresh.State)
+		*apply = *fresh
 		return
 	}
 

pkg/tern/local_apply_grouped.go

@@ -157,9 +157,15 @@ func (c *LocalClient) executeGroupedApply(ctx context.Context, apply *storage.Ap
 	if result.ResumeState != nil {
 		resumeState = result.ResumeState
 		if c.config.Type == storage.DatabaseTypeVitess {
+			// The engine has already accepted the apply, so a deploy request is
+			// live on the provider. A failure to persist its resume state is
+			// storage uncertainty, not a failed schema change: pause for operator
+			// retry so the resume path reattaches to the in-flight work instead
+			// of abandoning it as terminal.
 			if saveErr := c.saveEngineResumeState(ctx, apply, tasks, resumeState); saveErr != nil {
-				c.logger.Error("failed to save opaque engine resume state", append(apply.LogAttrs(), "error", saveErr)...)
-				c.failApplyWithTasks(ctx, apply, tasks, fmt.Sprintf("failed to save engine resume state: %v", saveErr))
+				c.logger.Warn("failed to save engine resume state after accepted apply; pausing apply for operator retry",
+					append(apply.LogAttrs(), "error", saveErr)...)
+				c.markApplyRetryableWithTasks(ctx, apply, tasks, fmt.Sprintf("failed to save engine resume state: %v", saveErr))
 				return
 			}
 		}

pkg/tern/local_client_test.go

@@ -466,10 +466,12 @@ type exactProgressStorage struct {
 	logs            storage.ApplyLogStore
 	controlRequests storage.ControlRequestStore
 	applyOperations storage.ApplyOperationStore
+	plans           storage.PlanStore
 }
 
 func (s *exactProgressStorage) Applies() storage.ApplyStore { return s.applies }
 func (s *exactProgressStorage) Tasks() storage.TaskStore    { return s.tasks }
+func (s *exactProgressStorage) Plans() storage.PlanStore    { return s.plans }
 func (s *exactProgressStorage) ApplyLogs() storage.ApplyLogStore {
 	if s.logs != nil {
 		return s.logs
@@ -936,6 +938,70 @@ func TestLaunchAtomicResumeMissingMetadataStaysRecoverable(t *testing.T) {
 	assert.NotEqual(t, state.Apply.Failed, applyStore.apply.State)
 }
 
+// After the engine accepts a grouped Vitess apply, a deploy request is live on
+// the provider. A failure to persist the engine resume state is storage
+// uncertainty, not a failed schema change: the apply pauses for operator retry
+// so the resume path reattaches to the in-flight deploy request instead of
+// permanently failing an apply whose remote work is still running.
+func TestExecuteGroupedApplySaveFailureAfterAcceptPausesForRetry(t *testing.T) {
+	operationID := int64(7)
+	apply := &storage.Apply{
+		ID:              42,
+		ApplyIdentifier: "apply-grouped-save-failure",
+		Database:        "testdb",
+		DatabaseType:    storage.DatabaseTypeVitess,
+		Engine:          storage.EnginePlanetScale,
+		State:           state.Apply.Pending,
+	}
+	tasks := []*storage.Task{{
+		ID:               7,
+		ApplyID:          apply.ID,
+		ApplyOperationID: &operationID,
+		TaskIdentifier:   "task-grouped-save-failure",
+		Database:         "testdb",
+		DatabaseType:     storage.DatabaseTypeVitess,
+		Namespace:        "commerce",
+		TableName:        "users",
+		DDL:              "ALTER TABLE `users` ADD COLUMN `email` varchar(255)",
+		DDLAction:        "alter",
+		State:            state.Task.Pending,
+	}}
+	plan := &storage.Plan{
+		PlanIdentifier: "plan-grouped-save-failure",
+		Namespaces:     map[string]*storage.NamespacePlanData{"commerce": {}},
+	}
+	applyStore := &exactProgressApplyStore{apply: apply}
+	client := &LocalClient{
+		config: LocalConfig{Database: "testdb", Type: storage.DatabaseTypeVitess},
+		storage: &exactProgressStorage{
+			applies:         applyStore,
+			tasks:           &exactProgressTaskStore{tasks: tasks},
+			controlRequests: &testControlRequestStore{},
+			applyOperations: &exactProgressApplyOperationStore{saveErr: errors.New("storage unavailable")},
+		},
+		planetscaleEngine: &fakeControlEngine{applyResult: &engine.ApplyResult{
+			Accepted: true,
+			ResumeState: &engine.ResumeState{
+				MigrationContext: "ctx-grouped",
+				Metadata:         `{"branch_name":"branch-1","deploy_request_id":5}`,
+			},
+		}},
+		heartbeatInterval: time.Hour,
+		logger:            slog.Default(),
+	}
+
+	client.executeGroupedApply(t.Context(), apply, tasks, plan, nil, false)
+
+	assert.True(t, state.IsState(applyStore.apply.State, state.Apply.FailedRetryable),
+		"apply must pause for operator retry, got %s", applyStore.apply.State)
+	assert.Nil(t, applyStore.apply.CompletedAt)
+	assert.Contains(t, applyStore.apply.ErrorMessage, "failed to save engine resume state")
+	for _, task := range tasks {
+		assert.True(t, state.IsState(task.State, state.Task.FailedRetryable),
+			"task must pause for operator retry, got %s", task.State)
+	}
+}
+
 // Rebuilt resume changes must preserve each task's namespace and table so
 // engines key per-table progress on the same identity the stored tasks carry.
 func TestGroupedResumeChangesGroupsTasksByNamespace(t *testing.T) {

pkg/tern/local_control_resume.go

@@ -1320,17 +1320,23 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		return err
 	}
 
-	// Get the plan to retrieve original DDLs
+	// Get the plan to retrieve original DDLs. A storage read failure says
+	// nothing about whether the plan row exists, so it must not become terminal
+	// apply state — the engine-side work (a checkpointed copy or a live deploy
+	// request) is untouched. The recovery attempt exits with an error so the
+	// claim is released and a later attempt retries against intact storage.
 	plan, err := c.storage.Plans().GetByID(ctx, apply.PlanID)
-	if err != nil || plan == nil {
-		c.logger.Warn("plan not found for apply, marking as failed",
-			"apply_id", apply.ApplyIdentifier,
-			"plan_id", apply.PlanID)
-		apply.State = state.Apply.Failed
-		apply.ErrorMessage = "plan not found during recovery"
-		if err := c.storage.Applies().Update(ctx, apply); err != nil {
-			c.logger.Error("failed to update apply state", append(apply.LogAttrs(), "error", err)...)
-		}
+	if err != nil {
+		c.logger.Warn("failed to load plan during recovery; current apply owner will exit for operator retry",
+			append(apply.LogAttrs(), "error", err)...)
+		return fmt.Errorf("load plan for recovery of apply %s (database %s): %w", apply.ApplyIdentifier, apply.Database, err)
+	}
+	// A confirmed-missing plan row is unrecoverable: the reviewed DDL cannot be
+	// rebuilt, so the apply fails and its observer is notified.
+	if plan == nil {
+		c.logger.Warn("plan row does not exist for apply; recovery cannot rebuild the reviewed DDL, marking apply failed",
+			apply.LogAttrs()...)
+		c.failApplyWithTasks(ctx, apply, tasks, "plan not found during recovery")
 		c.notifyTerminalObserver(apply, tasks)
 		return nil
 	}

pkg/tern/local_control_resume_test.go

@@ -1,12 +1,16 @@
 package tern
 
 import (
+	"context"
+	"errors"
+	"log/slog"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/block/schemabot/pkg/engine"
+	"github.com/block/schemabot/pkg/state"
 	"github.com/block/schemabot/pkg/storage"
 	"github.com/block/spirit/pkg/statement"
 )
@@ -262,3 +266,137 @@ func TestReplanAndFilterTasks_MatchKeepsTaskActive(t *testing.T) {
 	assert.Equal(t, "ALTER TABLE `users` ADD COLUMN `email` varchar(255)", rp.ActiveTasks[0].DDL)
 	assert.Zero(t, rp.CompletedCount)
 }
+
+// scriptedPlanStore scripts plan reads for recovery tests: a nil plan with a
+// nil error models a confirmed-missing plan row, while a non-nil error models
+// a storage read failure.
+type scriptedPlanStore struct {
+	storage.PlanStore
+	plan *storage.Plan
+	err  error
+}
+
+func (s *scriptedPlanStore) GetByID(context.Context, int64) (*storage.Plan, error) {
+	return s.plan, s.err
+}
+
+// terminalRecordingObserver records terminal notifications so recovery tests
+// can assert whether an apply's registered waiter (e.g. the PR check/comment)
+// was told the apply reached a terminal state.
+type terminalRecordingObserver struct {
+	terminal []*storage.Apply
+}
+
+func (o *terminalRecordingObserver) OnProgress(*storage.Apply, []*storage.Task) {}
+func (o *terminalRecordingObserver) OnTerminal(apply *storage.Apply, _ []*storage.Task) {
+	o.terminal = append(o.terminal, apply)
+}
+
+// recoveryPlanLoadFixture builds an in-flight Vitess apply whose recovery is
+// about to load its plan, with the plan store scripted by the caller.
+func recoveryPlanLoadFixture(plans storage.PlanStore) (*LocalClient, *storage.Apply, []*storage.Task, *exactProgressApplyStore) {
+	operationID := int64(3)
+	apply := &storage.Apply{
+		ID:              21,
+		ApplyIdentifier: "apply-recover-plan",
+		PlanID:          5,
+		Database:        "testdb",
+		DatabaseType:    storage.DatabaseTypeVitess,
+		Engine:          storage.EnginePlanetScale,
+		State:           state.Apply.Running,
+	}
+	tasks := []*storage.Task{{
+		ID:               2,
+		ApplyID:          apply.ID,
+		ApplyOperationID: &operationID,
+		TaskIdentifier:   "task-recover-plan",
+		Database:         "testdb",
+		DatabaseType:     storage.DatabaseTypeVitess,
+		Namespace:        "commerce",
+		TableName:        "users",
+		DDL:              "ALTER TABLE `users` ADD COLUMN `email` varchar(255)",
+		DDLAction:        "alter",
+		State:            state.Task.Running,
+	}}
+	applyStore := &exactProgressApplyStore{apply: apply}
+	client := &LocalClient{
+		config: LocalConfig{Database: "testdb", Type: storage.DatabaseTypeVitess},
+		storage: &exactProgressStorage{
+			applies:         applyStore,
+			tasks:           &exactProgressTaskStore{tasks: tasks},
+			controlRequests: &testControlRequestStore{},
+			plans:           plans,
+		},
+		logger: slog.Default(),
+	}
+	return client, apply, tasks, applyStore
+}
+
+// Recovery must not convert a transient storage failure on the plan load into
+// terminal apply state: the engine-side work (a checkpointed copy or a live
+// deploy request) is untouched. The recovery attempt exits with an error so
+// the claim is released and a later attempt retries against intact storage,
+// and no terminal notification reaches the apply's observer.
+func TestResumeApplyPlanLoadStorageErrorStaysRecoverable(t *testing.T) {
+	storageErr := errors.New("storage unavailable")
+	client, apply, tasks, applyStore := recoveryPlanLoadFixture(&scriptedPlanStore{err: storageErr})
+	observer := &terminalRecordingObserver{}
+	client.SetObserver(apply.ID, observer)
+
+	err := client.resumeApplyWithTasks(t.Context(), apply, tasks, nil, false, false)
+
+	require.ErrorIs(t, err, storageErr)
+	assert.ErrorContains(t, err, "apply-recover-plan")
+	assert.True(t, state.IsState(applyStore.apply.State, state.Apply.Running),
+		"in-flight apply must stay recoverable, not terminal: got %s", applyStore.apply.State)
+	assert.False(t, state.IsTerminalApplyState(applyStore.apply.State))
+	assert.Empty(t, applyStore.apply.ErrorMessage)
+	assert.Empty(t, observer.terminal, "a transient plan-load failure must not notify the terminal observer")
+}
+
+// A confirmed-missing plan row (a nil plan with no read error) is
+// unrecoverable — the reviewed DDL cannot be rebuilt — so recovery fails the
+// apply with an operator-facing reason and notifies its terminal observer.
+func TestResumeApplyMissingPlanFailsApply(t *testing.T) {
+	client, apply, tasks, applyStore := recoveryPlanLoadFixture(&scriptedPlanStore{})
+	observer := &terminalRecordingObserver{}
+	client.SetObserver(apply.ID, observer)
+
+	err := client.resumeApplyWithTasks(t.Context(), apply, tasks, nil, false, false)
+
+	require.NoError(t, err)
+	assert.True(t, state.IsState(applyStore.apply.State, state.Apply.Failed),
+		"apply with no plan row must fail, got %s", applyStore.apply.State)
+	assert.Equal(t, "plan not found during recovery", applyStore.apply.ErrorMessage)
+	assert.NotNil(t, applyStore.apply.CompletedAt, "a failed apply must record its completion time")
+	assert.True(t, state.IsState(tasks[0].State, state.Task.Failed),
+		"in-flight task must fail with its apply, got %s", tasks[0].State)
+	assert.Equal(t, "plan not found during recovery", tasks[0].ErrorMessage)
+	require.Len(t, observer.terminal, 1)
+	assert.True(t, state.IsState(observer.terminal[0].State, state.Apply.Failed))
+}
+
+// When another actor settles the apply between the recovery claim and the
+// plan-missing terminalization (e.g. a raced Stop()), the stored terminal
+// state wins: it is not overwritten, and the observer is notified with the
+// settled verdict rather than the stale in-flight state this recovery
+// attempt was holding.
+func TestResumeApplyMissingPlanAdoptsConcurrentTerminalState(t *testing.T) {
+	client, apply, tasks, applyStore := recoveryPlanLoadFixture(&scriptedPlanStore{})
+	settled := *apply
+	settled.State = state.Apply.Stopped
+	settled.ErrorMessage = "stopped by operator"
+	applyStore.apply = &settled
+	observer := &terminalRecordingObserver{}
+	client.SetObserver(apply.ID, observer)
+
+	err := client.resumeApplyWithTasks(t.Context(), apply, tasks, nil, false, false)
+
+	require.NoError(t, err)
+	assert.True(t, state.IsState(applyStore.apply.State, state.Apply.Stopped),
+		"concurrently-settled state must not be overwritten, got %s", applyStore.apply.State)
+	assert.Equal(t, "stopped by operator", applyStore.apply.ErrorMessage)
+	require.Len(t, observer.terminal, 1)
+	assert.True(t, state.IsState(observer.terminal[0].State, state.Apply.Stopped),
+		"observer must see the settled verdict, got %s", observer.terminal[0].State)
+}