diff --git a/pkg/api/control_handlers.go b/pkg/api/control_handlers.go index d3de18199..21f42417b 100644 --- a/pkg/api/control_handlers.go +++ b/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) } diff --git a/pkg/api/control_handlers_test.go b/pkg/api/control_handlers_test.go index 79047344d..ef78113ac 100644 --- a/pkg/api/control_handlers_test.go +++ b/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) + }) +} diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index 6202d2d80..67d98fa2b 100644 --- a/pkg/api/handlers_test.go +++ b/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 diff --git a/pkg/storage/errors.go b/pkg/storage/errors.go index 98fc2a04d..a087c3026 100644 --- a/pkg/storage/errors.go +++ b/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") diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go index 52918fff0..508bf96f0 100644 --- a/pkg/storage/internal/sqlstore/applies.go +++ b/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(¤tState) + 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 + } } } } diff --git a/pkg/storage/internal/sqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go index d05e4f761..3dcabeea5 100644 --- a/pkg/storage/internal/sqlstore/applies_test.go +++ b/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)) diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index d95864e78..338e580b2 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -364,6 +364,13 @@ type ApplyStore interface { // Returns ErrActiveApplyExists when moving an apply into an active state // would overlap another active apply for the same database, database type, // and environment. + // Returns ErrApplyTerminalStateImmutable when the update would move an + // apply from a terminal state 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 Update. + // Terminal-to-terminal writes, including same-state refreshes, are allowed. + // Returns ErrApplyNotFound when an update to an active state matches no + // rows because the apply row no longer exists. Update(ctx context.Context, apply *Apply) error // UpdateDerivedState compare-and-swaps the rollout-projected applies.state.