fix(storage): terminal apply states are immutable

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))