fix(storage): terminal apply states are immutable

Applies().Update was last-writer-wins: a caller holding a stale in-memory
snapshot could write an active state over a terminal row, resurrecting a
completed, failed, or stopped apply. Update now refuses terminal-to-active
transitions in the WHERE clause and surfaces the refusal as a distinct
ErrApplyTerminalStateImmutable, resolved from the ambiguous zero-rows
result by a read-committed re-read of the row.

Terminal-to-terminal writes (including same-state refreshes) stay allowed;
a settled apply re-enters the active lifecycle only through the dedicated
guarded transition of claiming a stopped apply. The stop-before-start
normalization in the API layer now reloads the row and proceeds from the
newer verdict instead of overwriting it, and guard errors surface the
apply identifier rather than the internal row ID.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: aparajon

pkg/api/control_handlers.go

@@ -1388,14 +1388,39 @@ func (s *Service) completeResolvedStopBeforeStart(ctx context.Context, client te
 		return nil
 	}
 
+	// The remote progress check above is an arbitrary time window: a driver can
+	// advance the stored row while the RPC is in flight. Write stopped only if
+	// the row still holds the state this decision was made from, so a stale
+	// snapshot never overwrites a newer state.
 	now := time.Now()
 	oldState := apply.State
+	swapped, err := s.storage.Applies().UpdateDerivedState(ctx, apply.ID, oldState, state.Apply.Stopped, apply.ErrorMessage, nil, &now)
+	if err != nil {
+		return fmt.Errorf("sync remote stopped apply %s before start: %w", apply.ApplyIdentifier, err)
+	}
+	if !swapped {
+		fresh, err := s.storage.Applies().Get(ctx, apply.ID)
+		if err != nil {
+			return fmt.Errorf("reload apply %s after concurrent state change before start: %w", apply.ApplyIdentifier, err)
+		}
+		if fresh == nil {
+			return fmt.Errorf("reload apply %s after concurrent state change before start: %w", apply.ApplyIdentifier, storage.ErrApplyNotFound)
+		}
+		*apply = *fresh
+		s.logger.Info("stored apply advanced while checking remote state; leaving pending stop request for the current owner and starting from the reloaded state",
+			"apply_id", apply.ApplyIdentifier,
+			"external_apply_id", apply.ExternalID,
+			"database", apply.Database,
+			"environment", apply.Environment,
+			"requested_by", stopCaller,
+			"start_requested_by", caller,
+			"expected_state", oldState,
+			"state", apply.State)
+		return nil
+	}
 	apply.State = state.Apply.Stopped
 	apply.CompletedAt = &now
 	apply.UpdatedAt = now
-	if err := s.storage.Applies().Update(ctx, apply); err != nil {
-		return fmt.Errorf("sync remote stopped apply %s before start: %w", apply.ApplyIdentifier, err)
-	}
 	if err := controlStore.CompletePending(ctx, apply.ID, storage.ControlOperationStop); err != nil {
 		return fmt.Errorf("complete pending remote stop control request for apply %s before start: %w", apply.ApplyIdentifier, err)
 	}

pkg/api/control_handlers_test.go

@@ -7,10 +7,14 @@ import (
 	"log/slog"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
+	ternv1 "github.com/block/schemabot/pkg/proto/ternv1"
 	"github.com/block/schemabot/pkg/state"
 	"github.com/block/schemabot/pkg/storage"
 )
@@ -78,3 +82,80 @@ func TestWriteControlError_LogsCarryFullApplyAttrs(t *testing.T) {
 	assert.Equal(t, "remote-apply-7", line["external_id"])
 	assert.Contains(t, line, "error")
 }
+
+// TestCompleteResolvedStopBeforeStart verifies the stop-request normalization
+// that runs before a start of a remote apply: when the data plane reports the
+// apply stopped, the stored row is written to stopped only if it still holds
+// the state the handler read before the remote check. If a driver advanced the
+// row while the check was in flight, the stale write is skipped, the pending
+// stop request stays with the current owner, and the handler proceeds from the
+// reloaded state instead of overwriting a newer verdict.
+func TestCompleteResolvedStopBeforeStart(t *testing.T) {
+	newSnapshot := func() *storage.Apply {
+		return &storage.Apply{
+			ID:              7,
+			ApplyIdentifier: "apply_stop_sync",
+			Database:        "testdb",
+			DatabaseType:    storage.DatabaseTypeMySQL,
+			Environment:     "staging",
+			ExternalID:      "remote-apply-7",
+			State:           state.Apply.Running,
+		}
+	}
+	newService := func(t *testing.T, applies *staticApplyStore) (*Service, *memoryControlRequestStore) {
+		t.Helper()
+		controls := &memoryControlRequestStore{}
+		_, _, err := controls.RequestPending(t.Context(), &storage.ApplyControlRequest{
+			ApplyID:     7,
+			Operation:   storage.ControlOperationStop,
+			RequestedBy: "alice",
+			Status:      storage.ControlRequestPending,
+		})
+		require.NoError(t, err)
+		logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError}))
+		return New(&mockStorageWithApplyStores{
+			applies:   applies,
+			controls:  controls,
+			applyLogs: &noopApplyLogStore{},
+		}, testServerConfig(), nil, logger), controls
+	}
+	remoteStopped := &mockTernClient{
+		isRemote:     true,
+		progressResp: &ternv1.ProgressResponse{State: ternv1.State_STATE_STOPPED},
+	}
+
+	t.Run("stored row still matches the handler's read", func(t *testing.T) {
+		snapshot := newSnapshot()
+		stored := *snapshot
+		applies := &staticApplyStore{apply: &stored}
+		svc, controls := newService(t, applies)
+
+		require.NoError(t, svc.completeResolvedStopBeforeStart(t.Context(), remoteStopped, snapshot, "bob"))
+
+		assert.Equal(t, state.Apply.Stopped, stored.State)
+		assert.Equal(t, state.Apply.Stopped, snapshot.State)
+		assert.NotNil(t, snapshot.CompletedAt)
+		pending, err := controls.GetPending(t.Context(), 7, storage.ControlOperationStop)
+		require.NoError(t, err)
+		assert.Nil(t, pending, "resolved stop request must be completed")
+	})
+
+	t.Run("stored row advanced while the remote check was in flight", func(t *testing.T) {
+		snapshot := newSnapshot()
+		completedAt := time.Now().Add(-time.Minute)
+		stored := *snapshot
+		stored.State = state.Apply.Completed
+		stored.CompletedAt = &completedAt
+		applies := &staticApplyStore{apply: &stored}
+		svc, controls := newService(t, applies)
+
+		require.NoError(t, svc.completeResolvedStopBeforeStart(t.Context(), remoteStopped, snapshot, "bob"))
+
+		assert.Equal(t, state.Apply.Completed, stored.State, "a stale stop write must not overwrite the newer verdict")
+		assert.Equal(t, state.Apply.Completed, snapshot.State, "handler must proceed from the reloaded state")
+		pending, err := controls.GetPending(t.Context(), 7, storage.ControlOperationStop)
+		require.NoError(t, err)
+		require.NotNil(t, pending, "pending stop request must stay with the current owner")
+		assert.Equal(t, "alice", pending.RequestedBy)
+	})
+}

pkg/api/handlers_test.go

@@ -255,6 +255,32 @@ func (s *staticApplyStore) Update(_ context.Context, apply *storage.Apply) error
 	return s.err
 }
 
+// UpdateDerivedState compare-and-swaps against the stored row like the real
+// store: the write lands only when the row still holds expectedState, so tests
+// can advance the row independently of the snapshot a handler is working from.
+func (s *staticApplyStore) UpdateDerivedState(_ context.Context, applyID int64, expectedState, newState, errorMessage string, startedAt, completedAt *time.Time) (bool, error) {
+	if s.err != nil {
+		return false, s.err
+	}
+	apply := s.apply
+	for _, candidate := range s.applies {
+		if candidate.ID == applyID {
+			apply = candidate
+			break
+		}
+	}
+	if apply == nil || !state.IsState(apply.State, expectedState) {
+		return false, nil
+	}
+	apply.State = newState
+	apply.ErrorMessage = errorMessage
+	if apply.StartedAt == nil {
+		apply.StartedAt = startedAt
+	}
+	apply.CompletedAt = completedAt
+	return true, nil
+}
+
 type recentApplyStore struct {
 	storage.ApplyStore
 	filters []storage.RecentAppliesFilter

pkg/storage/errors.go

@@ -37,6 +37,13 @@ var (
 	// matches the apply lease token stored by the latest operator claimant.
 	ErrApplyLeaseLost = errors.New("apply lease lost")
 
+	// ErrApplyTerminalStateImmutable is returned when a general update would
+	// move an apply from a terminal state back to an active state. A settled
+	// apply re-enters the active lifecycle only through the dedicated guarded
+	// transition of claiming a stopped apply, never through an update written
+	// from a caller's in-memory snapshot.
+	ErrApplyTerminalStateImmutable = errors.New("terminal apply state is immutable")
+
 	// ErrPlanNotFound is returned when a plan does not exist.
 	ErrPlanNotFound = errors.New("plan not found")
 

pkg/storage/internal/sqlstore/applies.go

@@ -462,6 +462,33 @@ func ensureApplyLeaseStillOwned(ctx context.Context, db queryRower, lease storag
 	return nil
 }
 
+// ensureApplyStillActiveOnZeroRows resolves a zero-rows result from an update
+// that carried the terminal-state guard (new state is active). Zero rows is
+// ambiguous: the write can be an idempotent no-op (every column already held
+// its target value), the row can be gone, or the guard refused because the row
+// is already terminal. It re-reads the row to distinguish the three so callers
+// can log the right cause: a missing row returns ErrApplyNotFound, a terminal
+// row returns ErrApplyTerminalStateImmutable, and a still-active row is a
+// benign no-op.
+func ensureApplyStillActiveOnZeroRows(ctx context.Context, db queryRower, apply *storage.Apply) error {
+	// The re-read must observe the latest committed row, not the transaction's
+	// repeatable-read snapshot: a concurrent terminal write committed after the
+	// snapshot is exactly what the guard refused. FOR UPDATE reads current data
+	// on the row lock the guarded UPDATE already examined.
+	var currentState string
+	err := db.QueryRowContext(ctx, `SELECT state FROM applies WHERE id = ? FOR UPDATE`, apply.ID).Scan(&currentState)
+	if errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("apply %s no longer exists for update to state %s: %w", apply.ApplyIdentifier, apply.State, storage.ErrApplyNotFound)
+	}
+	if err != nil {
+		return fmt.Errorf("re-read apply %s after guarded update to state %s: %w", apply.ApplyIdentifier, apply.State, err)
+	}
+	if state.IsTerminalApplyState(currentState) {
+		return fmt.Errorf("apply %s is %s; update to active state %s refused: %w", apply.ApplyIdentifier, currentState, apply.State, storage.ErrApplyTerminalStateImmutable)
+	}
+	return nil
+}
+
 // confirmLeaseOnZeroRows fails closed when a lease-scoped write changed no rows.
 // Zero rows is ambiguous: either a legitimate idempotent no-op (the lease is
 // still valid) or the lease token no longer matches because ownership was lost.
@@ -951,24 +978,44 @@ func (s *applyStore) Update(ctx context.Context, apply *storage.Apply) error {
 		leasePredicate = " AND lease_token = ?"
 		args = append(args, lease.Token)
 	}
+	// Terminal states are immutable through a general update: a caller writing
+	// from a stale snapshot must not resurrect a settled apply back into the
+	// active lifecycle. When the new state is active, match only rows that are
+	// still active; terminal rows re-enter the lifecycle solely through the
+	// dedicated guarded transition of claiming a stopped apply.
+	// Terminal-to-terminal writes (including same-state refreshes) stay allowed.
+	terminalGuard := isActiveApplyState(apply.State)
+	terminalGuardPredicate := ""
+	if terminalGuard {
+		predicate, guardArgs := nonTerminalApplyStatePredicate("state")
+		terminalGuardPredicate = " AND " + predicate
+		args = append(args, guardArgs...)
+	}
 
 	result, err := writeTx.tx.ExecContext(ctx, fmt.Sprintf(`
 		UPDATE applies
 		SET state = ?, error_message = ?, attempt = ?,
 		    external_id = ?%s, started_at = ?, completed_at = ?, updated_at = NOW()
-		WHERE id = ?%s
-	`, optionsUpdate, leasePredicate), args...)
+		WHERE id = ?%s%s
+	`, optionsUpdate, leasePredicate, terminalGuardPredicate), args...)
 	if err != nil {
 		return fmt.Errorf("update apply %d: %w", apply.ID, err)
 	}
-	if hasLease {
+	if hasLease || terminalGuard {
 		rows, err := result.RowsAffected()
 		if err != nil {
 			return fmt.Errorf("read apply update rows affected for apply %d: %w", apply.ID, err)
 		}
 		if rows == 0 {
-			if err := ensureApplyLeaseStillOwned(ctx, writeTx.tx, lease); err != nil {
-				return err
+			if hasLease {
+				if err := ensureApplyLeaseStillOwned(ctx, writeTx.tx, lease); err != nil {
+					return err
+				}
+			}
+			if terminalGuard {
+				if err := ensureApplyStillActiveOnZeroRows(ctx, writeTx.tx, apply); err != nil {
+					return err
+				}
 			}
 		}
 	}

pkg/storage/internal/sqlstore/applies_test.go

@@ -1881,7 +1881,9 @@ func TestApplyStore_UpdateBlocksActiveApplyForSameTarget(t *testing.T) {
 
 	active.State = state.Apply.Completed
 	require.NoError(t, store.Applies().Update(ctx, active))
-	require.NoError(t, store.Applies().Update(ctx, completed))
+	// With the target free, the terminal row is still not resurrectable through
+	// a general update: only the dedicated transitions reopen a settled apply.
+	require.ErrorIs(t, store.Applies().Update(ctx, completed), storage.ErrApplyTerminalStateImmutable)
 }
 
 func TestApplyStore_UpdateNonExistent(t *testing.T) {
@@ -1894,9 +1896,88 @@ func TestApplyStore_UpdateNonExistent(t *testing.T) {
 		State: state.Apply.Running,
 	}
 
-	// Update on a non-existent row is a no-op (0 rows affected), not an error.
-	// MySQL UPDATE with WHERE id=? succeeds even when no row matches.
-	require.NoError(t, store.Applies().Update(ctx, apply))
+	// An update to an active state re-reads the row when nothing matched, so a
+	// missing row surfaces as ErrApplyNotFound instead of a silent no-op.
+	require.ErrorIs(t, store.Applies().Update(ctx, apply), storage.ErrApplyNotFound)
+}
+
+// TestApplyStore_UpdateRefusesTerminalToActive verifies that a settled apply is
+// immutable through a general update: a caller writing an active state from a
+// stale in-memory snapshot must not resurrect a completed, failed, or stopped
+// apply. A settled apply re-enters the active lifecycle only through the
+// dedicated guarded transition of claiming a stopped apply.
+func TestApplyStore_UpdateRefusesTerminalToActive(t *testing.T) {
+	clearTables(t)
+	ctx := t.Context()
+	store := New(testDB)
+
+	lock := createTestLock(t, store, "testdb", "mysql", "staging")
+
+	cases := []struct {
+		name          string
+		applyID       string
+		planID        int64
+		terminalState string
+		activeState   string
+	}{
+		{"completed to running", "apply_guard_completed", 401, state.Apply.Completed, state.Apply.Running},
+		{"failed to failed_retryable", "apply_guard_failed", 402, state.Apply.Failed, state.Apply.FailedRetryable},
+		{"stopped to resuming", "apply_guard_stopped", 403, state.Apply.Stopped, state.Apply.Resuming},
+		{"cancelled to pending", "apply_guard_cancelled", 404, state.Apply.Cancelled, state.Apply.Pending},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			// A driver's stale snapshot: taken while the apply was active, used
+			// for a write after another writer settled the row.
+			apply := createTestApply(t, store, lock, tc.applyID, tc.planID)
+			snapshot := *apply
+
+			now := time.Now()
+			apply.State = tc.terminalState
+			apply.CompletedAt = &now
+			require.NoError(t, store.Applies().Update(ctx, apply))
+
+			snapshot.State = tc.activeState
+			require.ErrorIs(t, store.Applies().Update(ctx, &snapshot), storage.ErrApplyTerminalStateImmutable)
+
+			stored, err := store.Applies().Get(ctx, apply.ID)
+			require.NoError(t, err)
+			assert.Equal(t, tc.terminalState, stored.State)
+			assert.NotNil(t, stored.CompletedAt)
+		})
+	}
+}
+
+// TestApplyStore_UpdateTerminalToTerminalAllowed verifies that terminal rows
+// stay writable for terminal outcomes: a failed apply can refresh its error
+// message, and a stopped apply can be corrected to another terminal verdict,
+// without reopening the apply.
+func TestApplyStore_UpdateTerminalToTerminalAllowed(t *testing.T) {
+	clearTables(t)
+	ctx := t.Context()
+	store := New(testDB)
+
+	lock := createTestLock(t, store, "testdb", "mysql", "staging")
+
+	failed := createTestApplyWithStateAndEnv(t, store, lock, "apply_terminal_refresh", 405, state.Apply.Failed, "staging")
+	failed.ErrorMessage = "engine failure: table copy interrupted"
+	require.NoError(t, store.Applies().Update(ctx, failed))
+
+	stored, err := store.Applies().Get(ctx, failed.ID)
+	require.NoError(t, err)
+	assert.Equal(t, state.Apply.Failed, stored.State)
+	assert.Equal(t, "engine failure: table copy interrupted", stored.ErrorMessage)
+
+	stopped := createTestApplyWithStateAndEnv(t, store, lock, "apply_terminal_correct", 406, state.Apply.Stopped, "staging")
+	now := time.Now()
+	stopped.State = state.Apply.Cancelled
+	stopped.CompletedAt = &now
+	require.NoError(t, store.Applies().Update(ctx, stopped))
+
+	stored, err = store.Applies().Get(ctx, stopped.ID)
+	require.NoError(t, err)
+	assert.Equal(t, state.Apply.Cancelled, stored.State)
+	assert.NotNil(t, stored.CompletedAt)
 }
 
 // TestApplyStore_UpdateDerivedState verifies the rollout-projection compare-and-
@@ -3746,7 +3827,7 @@ func TestApplyStore_UpdateOptions(t *testing.T) {
 		PullRequest:     123,
 		Environment:     "staging",
 		Engine:          "spirit",
-		State:           state.Apply.Stopped,
+		State:           state.Apply.Pending,
 	}
 	apply.SetOptions(storage.ApplyOptions{Target: "testdb"})
 
@@ -3755,7 +3836,7 @@ func TestApplyStore_UpdateOptions(t *testing.T) {
 
 	retrieved, err := store.Applies().Get(ctx, id)
 	require.NoError(t, err)
-	retrieved.State = state.Apply.Pending
+	retrieved.State = state.Apply.Running
 
 	require.NoError(t, store.Applies().Update(ctx, retrieved))