From b393ab3bf80d894242246370efd6d9b2ec91efca Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Tue, 20 Oct 2026 02:15:40 +0100 Subject: [PATCH] fix(purchase): fail a plan step's root row atomically with its reason and suppression release failExistingPlanStep committed the pinned failed CAS on its own, then stamped Error/NoSubmission and released the row's suppressions in a second transaction. An error or crash between the two left the root failed with no reason, no no-submission proof and its suppressions held, and adoptableRootRow never adopts a failed root, so nothing repaired it. Add TransitionExecutionStatusPinnedTx, which runs the same pinned CAS (and its rejection probe) on a caller transaction, and do the CAS, the stamp and the suppression release in one transaction. An error before commit now leaves the row pending for the next tick, and the CAS's row lock keeps concurrent approve/cancel/expire writers from landing between the status change and the stamp. The failure is logged only after commit. Real-Postgres integration tests inject an error at the release, hold the fail mid-transaction to check what other connections see and that an approval waits and loses, and race the fail against approve, cancel, the expiry sweep and a second tick. Closes #1052 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 4 + internal/config/interfaces.go | 4 + internal/config/store_postgres.go | 49 +++- internal/config/store_postgres_db_test.go | 52 ++++ internal/mocks/stores.go | 20 ++ internal/purchase/notifications.go | 60 +++- internal/purchase/notifications_test.go | 4 +- internal/purchase/plan_step_attach_test.go | 6 +- .../plan_step_fail_atomic_integration_test.go | 264 ++++++++++++++++++ internal/server/test_helpers_test.go | 4 + .../36_plan_step_fail_cas_not_atomic.md | 9 +- 11 files changed, 444 insertions(+), 32 deletions(-) create mode 100644 internal/purchase/plan_step_fail_atomic_integration_test.go rename known_issues/{ => resolved}/36_plan_step_fail_cas_not_atomic.md (75%) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index f5422c072..44ecc53eb 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -315,6 +315,10 @@ func (m *mockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, e return nil, nil } +func (m *mockConfigStore) TransitionExecutionStatusPinnedTx(ctx context.Context, tx pgx.Tx, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStore) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*config.PurchaseExecution, error) { return nil, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 6483836dc..e837dad9d 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -234,6 +234,10 @@ type StoreInterface interface { // row whose recommendations were attached after the caller's gate check. A // lost pin returns ErrExecutionNotInExpectedStatus like any lost CAS. TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) + // TransitionExecutionStatusPinnedTx is TransitionExecutionStatusPinned + // inside a caller-owned transaction, so the status change commits or + // rolls back with the caller's other writes (issue #1052). + TransitionExecutionStatusPinnedTx(ctx context.Context, tx pgx.Tx, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) // TransitionExecutionStatusTokenPinned is TransitionExecutionStatus that // also requires the row's stored approval token hash to match rawToken and // to be unexpired, in the same statement as the status change, so a token diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index c90c71ab5..bd8c75ef8 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1242,7 +1242,7 @@ func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executio // and the legacy cancelled_by (issue #119), matching what the other cancel paths // record and what History reads. func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error) { - return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil, nil) + return s.transitionExecutionStatus(ctx, s.db, executionID, fromStatuses, toStatus, actor, nil, nil) } // TransitionExecutionStatusPinned is TransitionExecutionStatus that also @@ -1251,7 +1251,15 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution // claim makes the pin fail with ErrExecutionNotInExpectedStatus, so a row is // never run on a snapshot different from the one that was gated. func (s *PostgresStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) { - return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations, nil) + return s.transitionExecutionStatus(ctx, s.db, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations, nil) +} + +// TransitionExecutionStatusPinnedTx is TransitionExecutionStatusPinned inside +// the caller's transaction. The UPDATE's row lock is held until that +// transaction ends, so the status change commits or rolls back together with +// the caller's other writes and no other writer can touch the row in between. +func (s *PostgresStore) TransitionExecutionStatusPinnedTx(ctx context.Context, tx pgx.Tx, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) { + return s.transitionExecutionStatus(ctx, tx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations, nil) } // TransitionExecutionStatusTokenPinned is TransitionExecutionStatus that also @@ -1265,7 +1273,7 @@ func (s *PostgresStore) TransitionExecutionStatusTokenPinned(ctx context.Context return nil, fmt.Errorf("%w: execution %s: no approval token supplied", ErrApprovalTokenChanged, executionID) } tokenHash := HashApprovalToken(rawToken) - return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil, &tokenHash) + return s.transitionExecutionStatus(ctx, s.db, executionID, fromStatuses, toStatus, actor, nil, &tokenHash) } // RescheduleAwaitingExecution is a conditional single-column update: the @@ -1302,7 +1310,7 @@ func (s *PostgresStore) RescheduleAwaitingExecution(ctx context.Context, executi return updated, nil } -func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool, tokenHash *string) (*PurchaseExecution, error) { +func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, q executionQuerier, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool, tokenHash *string) (*PurchaseExecution, error) { var cancelActor *string if toStatus == StatusCanceled { cancelActor = actor @@ -1327,25 +1335,26 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, execution idempotency_key, scheduled_execution_at, no_submission, created_by_system ` - records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty, tokenHash) + records, err := queryExecutionsOn(ctx, q, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty, tokenHash) if err != nil { return nil, err } if len(records) == 0 { - return nil, s.transitionRejection(ctx, executionID, fromStatuses, toStatus, tokenHash != nil) + return nil, transitionRejection(ctx, q, executionID, fromStatuses, toStatus, tokenHash != nil) } return &records[0], nil } // transitionRejection explains a zero-row transitionExecutionStatus by probing -// the row: gone (ErrNotFound), a hard DB error, a token pin lost on a row still -// in an allowed status (ErrApprovalTokenChanged), or a status CAS lost +// the row through q (the caller's transaction for the Tx variant): gone +// (ErrNotFound), a hard DB error, a token pin lost on a row still in an +// allowed status (ErrApprovalTokenChanged), or a status CAS lost // (ErrExecutionNotInExpectedStatus). Extracted to keep the caller under the // gocyclo threshold. -func (s *PostgresStore) transitionRejection(ctx context.Context, executionID string, fromStatuses []string, toStatus string, tokenPinned bool) error { - existing, existErr := s.GetExecutionByID(ctx, executionID) +func transitionRejection(ctx context.Context, q executionQuerier, executionID string, fromStatuses []string, toStatus string, tokenPinned bool) error { + existing, existErr := getExecutionByIDOn(ctx, q, executionID) if errors.Is(existErr, ErrNotFound) { // Wrap ErrNotFound so callers (e.g. the purchase reaper) can // use errors.Is to distinguish "row vanished mid-flight" — a @@ -1875,6 +1884,11 @@ func (s *PostgresStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, crea // from a real DB failure (any other non-nil error). A nil error guarantees // a non-nil execution (fail-loud contract; issues #976, #1339). func (s *PostgresStore) GetExecutionByID(ctx context.Context, executionID string) (*PurchaseExecution, error) { + return getExecutionByIDOn(ctx, s.db, executionID) +} + +// getExecutionByIDOn is GetExecutionByID on q, the pool or a transaction. +func getExecutionByIDOn(ctx context.Context, q executionQuerier, executionID string) (*PurchaseExecution, error) { query := ` SELECT plan_id, execution_id, status, step_number, scheduled_date, notification_sent, approval_token_hash, recommendations, @@ -1888,7 +1902,7 @@ func (s *PostgresStore) GetExecutionByID(ctx context.Context, executionID string WHERE execution_id = $1 ` - executions, err := s.queryExecutions(ctx, query, executionID) + executions, err := queryExecutionsOn(ctx, q, query, executionID) if err != nil { return nil, err } @@ -2016,9 +2030,20 @@ func (s *PostgresStore) ListPendingExecutionIDsForAccount(ctx context.Context, a return ids, nil } +// executionQuerier is the query method shared by the pool (dbConn) and a +// pgx.Tx, so the execution read and transition helpers can run on either. +type executionQuerier interface { + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) +} + // queryExecutions is a helper to query and scan purchase executions. func (s *PostgresStore) queryExecutions(ctx context.Context, query string, args ...any) ([]PurchaseExecution, error) { - rows, err := s.db.Query(ctx, query, args...) + return queryExecutionsOn(ctx, s.db, query, args...) +} + +// queryExecutionsOn is queryExecutions on q, the pool or a transaction. +func queryExecutionsOn(ctx context.Context, q executionQuerier, query string, args ...any) ([]PurchaseExecution, error) { + rows, err := q.Query(ctx, query, args...) if err != nil { return nil, fmt.Errorf("failed to query executions: %w", err) } diff --git a/internal/config/store_postgres_db_test.go b/internal/config/store_postgres_db_test.go index 04a458be0..fb7ad2672 100644 --- a/internal/config/store_postgres_db_test.go +++ b/internal/config/store_postgres_db_test.go @@ -7,6 +7,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "path/filepath" "runtime" @@ -1776,6 +1777,57 @@ func TestPostgresStoreDB_TransitionExecutionStatusPinned(t *testing.T) { }) } +// TestPostgresStoreDB_TransitionExecutionStatusPinnedTx (issue #1052): the tx +// variant commits or rolls back with the caller's transaction, and a lost pin, +// a lost status CAS or a missing row is reported without aborting it. +func TestPostgresStoreDB_TransitionExecutionStatusPinnedTx(t *testing.T) { + conn := setupTestContainerDB(t) + if conn == nil { + return + } + store := NewPostgresStore(conn) + ctx := context.Background() + plan := &PurchasePlan{Name: "pinned claim tx", Enabled: true, Services: map[string]ServiceConfig{}, RampSchedule: PresetRampSchedules["immediate"]} + require.NoError(t, store.CreatePurchasePlan(ctx, plan)) + exec := &PurchaseExecution{PlanID: plan.ID, Status: "notified", StepNumber: 1, ScheduledDate: time.Now()} + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + from := []string{"pending", "notified"} + status := func() string { + t.Helper() + reread, err := store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + return reread.Status + } + + rollback := errors.New("caller's later write failed") + err := store.WithTx(ctx, func(tx pgx.Tx) error { + got, txErr := store.TransitionExecutionStatusPinnedTx(ctx, tx, exec.ExecutionID, from, "failed", nil, true) + require.NoError(t, txErr) + assert.Equal(t, "failed", got.Status) + return rollback + }) + require.ErrorIs(t, err, rollback) + assert.Equal(t, "notified", status(), "the transition rolls back with the caller's transaction") + + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + _, pinErr := store.TransitionExecutionStatusPinnedTx(ctx, tx, exec.ExecutionID, from, "failed", nil, false) + require.ErrorIs(t, pinErr, ErrExecutionNotInExpectedStatus, "a lost pin is a lost CAS") + _, goneErr := store.TransitionExecutionStatusPinnedTx(ctx, tx, uuid.NewString(), from, "failed", nil, true) + require.ErrorIs(t, goneErr, ErrNotFound) + got, txErr := store.TransitionExecutionStatusPinnedTx(ctx, tx, exec.ExecutionID, from, "failed", nil, true) + require.NoError(t, txErr, "the rejections did not abort the transaction") + assert.Equal(t, "failed", got.Status) + return nil + })) + assert.Equal(t, "failed", status()) + + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + _, casErr := store.TransitionExecutionStatusPinnedTx(ctx, tx, exec.ExecutionID, from, "failed", nil, true) + require.ErrorIs(t, casErr, ErrExecutionNotInExpectedStatus) + return nil + })) +} + // TestPostgresStoreDB_TransitionExecutionStatusTokenPinned (issue #590): the // delayed email-link approval re-checks the approval token inside the status // CAS, so a token rotated (re-notification) or expired after the caller read diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index a4c9546f7..0860a82d6 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -484,6 +484,26 @@ func (m *MockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, e return v, args.Error(1) } +// TransitionExecutionStatusPinnedTx mocks the tx variant of the pinned claim. +// With no expectation registered it delegates to TransitionExecutionStatusPinned +// (and from there to TransitionExecutionStatus), so tests that mock the +// non-tx transition keep working. +func (m *MockConfigStore) TransitionExecutionStatusPinnedTx(ctx context.Context, tx pgx.Tx, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + m.record("TransitionExecutionStatusPinnedTx", ctx, tx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations) + if !isExpected(&m.Mock, "TransitionExecutionStatusPinnedTx") { + return m.TransitionExecutionStatusPinned(ctx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations) + } + args := m.Called(ctx, tx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.PurchaseExecution) + if !ok { + panic(fmt.Sprintf("mock: expected *config.PurchaseExecution, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // TransitionExecutionStatusTokenPinned mocks the token-pinned transition // (issue #590). It does not delegate to TransitionExecutionStatus: a caller // that must pin the approval token is expected to call this method, so a test diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index 5b34502fc..45b1b264c 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -298,34 +298,68 @@ func adoptableRootRow(rows []config.PurchaseExecution) (*config.PurchaseExecutio // is no longer awaiting approval and the tick skips it, instead of an upsert // from the stale read overwriting the approval. A won CAS returns the fresh // row, which is the one the error text is written onto. +// +// The CAS, the Error/NoSubmission stamp and the suppression release commit in +// one transaction (issue #1052). A failed root is never adopted again, so a +// CAS committed on its own could strand the row failed with no reason and its +// suppressions still held; an error before commit now leaves the row pending +// for the next tick. func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.PurchaseExecution, reason string) error { - failed, err := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, len(exec.Recommendations) == 0) - if errors.Is(err, config.ErrNotFound) || errors.Is(err, config.ErrExecutionNotInExpectedStatus) { - return fmt.Errorf("%w: %s changed state before it could be failed", errExecutionNotNotifiable, exec.ExecutionID) + var failed *config.PurchaseExecution + err := m.config.WithTx(ctx, func(tx pgx.Tx) error { + var txErr error + failed, txErr = m.failPlanStepTx(ctx, tx, exec, reason) + return txErr + }) + if errors.Is(err, errExecutionNotNotifiable) { + return err } if err != nil { return fmt.Errorf("failed to fail plan step %d of plan %s: %w", exec.StepNumber, exec.PlanID, err) } - stampFailedPlanStep(failed, reason) + logFailedPlanStep(failed) + return bareStepFailure(failed.PlanID, failed.StepNumber, reason) +} + +// failPlanStepTx is failExistingPlanStep's transaction: the pinned CAS, then +// the stamp and the suppression release on the row it returned. Only a lost +// CAS is the quiet changed-state skip; any later error is a failed write. +func (m *Manager) failPlanStepTx(ctx context.Context, tx pgx.Tx, exec *config.PurchaseExecution, reason string) (*config.PurchaseExecution, error) { + failed, err := m.config.TransitionExecutionStatusPinnedTx(ctx, tx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, len(exec.Recommendations) == 0) + if errors.Is(err, config.ErrNotFound) || errors.Is(err, config.ErrExecutionNotInExpectedStatus) { + return nil, fmt.Errorf("%w: %s changed state before it could be failed", errExecutionNotNotifiable, exec.ExecutionID) + } + if err != nil { + return nil, err + } + markFailedPlanStep(failed, reason) + if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil { + return nil, fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", failed.StepNumber, failed.PlanID, err) + } // The row is failed with NoSubmission, so its suppressions are released as // for any proven not-submitted failure (saveTerminalExecution); a bare row has none. - if err := m.config.WithTx(ctx, func(tx pgx.Tx) error { - if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil { - return err - } - return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, failed.ExecutionID) - }); err != nil { - return fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", failed.StepNumber, failed.PlanID, err) + if err := m.config.DeleteSuppressionsByExecutionTx(ctx, tx, failed.ExecutionID); err != nil { + return nil, err } - return bareStepFailure(failed.PlanID, failed.StepNumber, reason) + return failed, nil } // stampFailedPlanStep marks exec failed with reason and logs it once. func stampFailedPlanStep(exec *config.PurchaseExecution, reason string) { + markFailedPlanStep(exec, reason) + logFailedPlanStep(exec) +} + +// markFailedPlanStep sets the failed status, NoSubmission and reason on exec. +func markFailedPlanStep(exec *config.PurchaseExecution, reason string) { exec.Status = "failed" exec.NoSubmission = true // a pending/notified step never reaches the executor exec.Error = reason - logging.Errorf("purchase[%s]: plan %s step %d was marked failed: %s", exec.ExecutionID, exec.PlanID, exec.StepNumber, reason) +} + +// logFailedPlanStep logs a plan step that was marked failed. +func logFailedPlanStep(exec *config.PurchaseExecution) { + logging.Errorf("purchase[%s]: plan %s step %d was marked failed: %s", exec.ExecutionID, exec.PlanID, exec.StepNumber, exec.Error) } // stampFailedBarePlanStep marks exec failed with the no-recommendations reason. diff --git a/internal/purchase/notifications_test.go b/internal/purchase/notifications_test.go index 8475ac5df..c8a39fc26 100644 --- a/internal/purchase/notifications_test.go +++ b/internal/purchase/notifications_test.go @@ -280,7 +280,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowIsFailedNotRotated(t *testi expectStepRows(mockStore, plan, *bare) failed := *bare failed.Status = "failed" - mockStore.On("TransitionExecutionStatusPinned", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil) + mockStore.On("TransitionExecutionStatusPinnedTx", ctx, mock.Anything, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil) mockStore.On("SavePurchaseExecution", ctx, &failed).Return(nil) manager := &Manager{config: mockStore, resolver: &fakeResolver{err: planstep.ErrNoMatch}} @@ -309,7 +309,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowApprovedMeanwhileIsNotOverw ExecutionID: "bare-exec-id", PlanID: "plan-123", Status: "pending", StepNumber: 1, ScheduledDate: nextExec, } expectStepRows(mockStore, plan, *stale) - mockStore.On("TransitionExecutionStatusPinned", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true). + mockStore.On("TransitionExecutionStatusPinnedTx", ctx, mock.Anything, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true). Return(nil, config.ErrExecutionNotInExpectedStatus) // No SavePurchaseExecution expectation: an upsert here would clobber the approval. diff --git a/internal/purchase/plan_step_attach_test.go b/internal/purchase/plan_step_attach_test.go index 876041468..60513aa56 100644 --- a/internal/purchase/plan_step_attach_test.go +++ b/internal/purchase/plan_step_attach_test.go @@ -291,7 +291,7 @@ func TestAttach_ManualPlanRequiresExplicitRootAndHonorsOptionalCap(t *testing.T) store.On("CreateSuppressionTx", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() failed := root failed.Status = "failed" - store.On("TransitionExecutionStatusPinned", mock.Anything, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil).Maybe() + store.On("TransitionExecutionStatusPinnedTx", mock.Anything, mock.Anything, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil).Maybe() store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil).Maybe() exec, _, _, err := m.getOrCreateExecution(context.Background(), plan) if tc.wantAttach { @@ -342,9 +342,9 @@ func TestManualDueStepCleanupWithoutEmailAfterLongDowntime(t *testing.T) { failed.Status = "failed" if !tc.future { if tc.lostClaim { - store.On("TransitionExecutionStatusPinned", ctx, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(nil, config.ErrExecutionNotInExpectedStatus).Once() + store.On("TransitionExecutionStatusPinnedTx", ctx, mock.Anything, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(nil, config.ErrExecutionNotInExpectedStatus).Once() } else { - store.On("TransitionExecutionStatusPinned", ctx, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil).Once() + store.On("TransitionExecutionStatusPinnedTx", ctx, mock.Anything, root.ExecutionID, []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil).Once() store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(row *config.PurchaseExecution) bool { return row.ExecutionID == root.ExecutionID && row.NoSubmission && row.Status == "failed" && len(row.Recommendations) == 0 && row.ScheduledDate.Equal(root.ScheduledDate) })).Return(nil).Once() diff --git a/internal/purchase/plan_step_fail_atomic_integration_test.go b/internal/purchase/plan_step_fail_atomic_integration_test.go new file mode 100644 index 000000000..e0c3864b3 --- /dev/null +++ b/internal/purchase/plan_step_fail_atomic_integration_test.go @@ -0,0 +1,264 @@ +//go:build integration + +package purchase + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +const failAtomicReason = "plan step can never be bought" + +// releaseHookStore runs hook at the start of the suppression release that +// failExistingPlanStep does after its failed CAS. A returned error stands in +// for a crash or DB error at that point; a blocking hook holds the writer +// there so a test can look at what other connections see. +type releaseHookStore struct { + *config.PostgresStore + hook func() error +} + +func (s *releaseHookStore) DeleteSuppressionsByExecutionTx(ctx context.Context, tx pgx.Tx, executionID string) error { + if err := s.hook(); err != nil { + return err + } + return s.PostgresStore.DeleteSuppressionsByExecutionTx(ctx, tx, executionID) +} + +// failRowState is what a plan-step root looks like from outside the writer. +type failRowState struct { + Status string + Error string + NoSubmission bool + Suppressions int +} + +func (f *cumFixture) failRowState(t *testing.T, executionID string) failRowState { + t.Helper() + var st failRowState + require.NoError(t, f.pool.QueryRow(f.ctx, ` + SELECT status, COALESCE(error, ''), COALESCE(no_submission, false), + (SELECT COUNT(*) FROM purchase_suppressions WHERE execution_id = $1) + FROM purchase_executions WHERE execution_id = $1`, executionID). + Scan(&st.Status, &st.Error, &st.NoSubmission, &st.Suppressions)) + return st +} + +// seedAttachedRoot saves a pending root row carrying recommendations and one +// suppression, as a refused attached step looks before it is failed. +func (f *cumFixture) seedAttachedRoot(t *testing.T) *config.PurchaseExecution { + t.Helper() + rec := cumRec(f.account, "m5.t0", 100) + rec.Selected = true + root := &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: "pending", StepNumber: 1, + ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}} + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, root)) + require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: root.ExecutionID, AccountID: "sup-acct", + Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)})) + return root +} + +var stampedFailure = failRowState{Status: "failed", Error: failAtomicReason, NoSubmission: true, Suppressions: 0} + +// Issue #1052: an error between the failed CAS and the stamp/release used to +// leave the row failed with no reason, no NoSubmission and its suppressions in +// place, and no later tick adopts a failed root. Now nothing commits, so the +// row is still pending and the next attempt fails it properly. +func TestFailExistingPlanStep_ReleaseErrorCommitsNothing(t *testing.T) { + f := newCumFixture(t, 100) + root := f.seedAttachedRoot(t) + injected := errors.New("connection lost before the suppression release") + f.m.config = &releaseHookStore{PostgresStore: f.store, hook: func() error { return injected }} + + err := f.m.failExistingPlanStep(f.ctx, root, failAtomicReason) + require.ErrorIs(t, err, injected) + require.NotErrorIs(t, err, errExecutionNotNotifiable, "a failed write is not a quiet skip") + require.Equal(t, failRowState{Status: "pending", Suppressions: 1}, f.failRowState(t, root.ExecutionID), + "the CAS rolls back with the stamp and the release") + + f.m.config = f.store + err = f.m.failExistingPlanStep(f.ctx, root, failAtomicReason) + require.ErrorIs(t, err, errExecutionNotNotifiable) + require.Contains(t, err.Error(), failAtomicReason) + require.Equal(t, stampedFailure, f.failRowState(t, root.ExecutionID), "the retry fails the still-pending row") +} + +// While the fail is between its CAS and its commit, other connections still +// see the pending row, and an approval waits on the row lock and then loses to +// the committed failure instead of landing on a half-written one. +func TestFailExistingPlanStep_ConcurrentApproveWaitsForStampedFailure(t *testing.T) { + f := newCumFixture(t, 100) + root := f.seedAttachedRoot(t) + reached, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(unblock) // a failed assertion must not leave the writer parked + f.m.config = &releaseHookStore{PostgresStore: f.store, hook: func() error { + close(reached) + <-release + return nil + }} + failErr := make(chan error, 1) + go func() { failErr <- f.m.failExistingPlanStep(f.ctx, root, failAtomicReason) }() + <-reached + + require.Equal(t, failRowState{Status: "pending", Suppressions: 1}, f.failRowState(t, root.ExecutionID), + "no other connection sees a failed row before its reason and release commit") + approveErr := make(chan error, 1) + go func() { + _, err := f.store.TransitionExecutionStatus(f.ctx, root.ExecutionID, []string{"pending", "notified"}, "approved", nil) + approveErr <- err + }() + f.requireApproveBlocked(t, approveErr) + unblock() + + err := <-failErr + require.ErrorIs(t, err, errExecutionNotNotifiable) + require.Contains(t, err.Error(), failAtomicReason) + require.ErrorIs(t, <-approveErr, config.ErrExecutionNotInExpectedStatus) + require.Equal(t, stampedFailure, f.failRowState(t, root.ExecutionID)) +} + +// requireApproveBlocked waits until the approval's UPDATE is waiting on a row +// lock. An approval that returns first did not wait for the fail to commit. +func (f *cumFixture) requireApproveBlocked(t *testing.T, approveErr chan error) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-approveErr: + t.Fatalf("approval returned (%v) while the fail was still uncommitted", err) + default: + } + var waiting int + require.NoError(t, f.pool.QueryRow(f.ctx, ` + SELECT COUNT(*) FROM pg_stat_activity + WHERE datname = current_database() AND wait_event_type = 'Lock' + AND query LIKE '%UPDATE purchase_executions%'`).Scan(&waiting)) + if waiting > 0 { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("approval never waited on the row lock") +} + +// failRacer is a writer that competes with failExistingPlanStep for a pending +// root. It reports whether it won and the state its win leaves behind. +type failRacer struct { + name string + run func(f *cumFixture, root *config.PurchaseExecution) (won bool, err error) + want failRowState +} + +var failRacers = []failRacer{ + {name: "approve", want: failRowState{Status: "approved", Suppressions: 1}, + run: func(f *cumFixture, root *config.PurchaseExecution) (bool, error) { + _, err := f.store.TransitionExecutionStatus(f.ctx, root.ExecutionID, []string{"pending", "notified"}, "approved", nil) + if errors.Is(err, config.ErrExecutionNotInExpectedStatus) { + return false, nil + } + return err == nil, err + }}, + {name: "cancel", want: failRowState{Status: config.StatusCanceled, Suppressions: 0}, + run: func(f *cumFixture, root *config.PurchaseExecution) (bool, error) { + var won bool + err := f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + var cancelErr error + if won, _, cancelErr = f.store.CancelExecutionAtomic(f.ctx, tx, root.ExecutionID, nil); cancelErr != nil || !won { + return cancelErr + } + return f.store.DeleteSuppressionsByExecutionTx(f.ctx, tx, root.ExecutionID) + }) + return won, err + }}, + {name: "expiry sweep", want: failRowState{Status: config.StatusExpired, Suppressions: 0}, + run: func(f *cumFixture, root *config.PurchaseExecution) (bool, error) { + // The same transaction as the history handler's expireStaleExecutionsSweep. + var won bool + err := f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + var expErr error + if won, expErr = f.store.ExpireExecutionAtomic(f.ctx, tx, root.ExecutionID); expErr != nil || !won { + return expErr + } + return f.store.DeleteSuppressionsByExecutionTx(f.ctx, tx, root.ExecutionID) + }) + return won, err + }}, + {name: "second tick", want: stampedFailure, + run: func(f *cumFixture, root *config.PurchaseExecution) (bool, error) { + won, err := failStepOutcome(f.m.failExistingPlanStep(f.ctx, root, failAtomicReason)) + return won, err + }}, +} + +// failStepOutcome classifies a failExistingPlanStep result: a won fail is the +// bare step failure, a lost CAS is the changed-state skip. +func failStepOutcome(err error) (won bool, unexpected error) { + switch { + case err != nil && errors.Is(err, errExecutionNotNotifiable) && strings.Contains(err.Error(), "changed state before it could be failed"): + return false, nil + case err != nil && errors.Is(err, errExecutionNotNotifiable) && strings.Contains(err.Error(), failAtomicReason): + return true, nil + default: + return false, fmt.Errorf("unexpected fail outcome: %w", err) + } +} + +// Real-Postgres race of the fail against each writer that can take a pending +// root (approval, cancel, the stale-execution expiry sweep, another tick): exactly one wins every round, the row ends in that winner's complete +// state, and no round deadlocks or leaves a failed row unstamped. +func TestFailExistingPlanStep_RacesEndInOneCompleteState(t *testing.T) { + f := newCumFixture(t, 100) + const rounds = 25 + for _, racer := range failRacers { + t.Run(racer.name, func(t *testing.T) { + failWins := 0 + defer func() { t.Logf("fail won %d of %d rounds against %s", failWins, rounds, racer.name) }() + for i := 0; i < rounds; i++ { + root := f.seedAttachedRoot(t) + var wg sync.WaitGroup + start := make(chan struct{}) + var failWon, racerWon bool + var failErr, racerErr error + wg.Add(2) + go func() { + defer wg.Done() + <-start + failWon, failErr = failStepOutcome(f.m.failExistingPlanStep(f.ctx, root, failAtomicReason)) + }() + // Staggering the racer by 0-2ms lets both sides win some rounds: + // the fail opens its transaction first, so without it the + // single-statement approval always lands first. + stagger := time.Duration(i%3) * time.Millisecond + go func() { + defer wg.Done() + <-start + time.Sleep(stagger) + racerWon, racerErr = racer.run(f, root) + }() + close(start) + wg.Wait() + require.NoError(t, failErr, "round %d", i) + require.NoError(t, racerErr, "round %d", i) + require.True(t, failWon != racerWon, "round %d: exactly one writer wins (fail=%v %s=%v)", i, failWon, racer.name, racerWon) + want := racer.want + if failWon { + failWins++ + want = stampedFailure + } + require.Equal(t, want, f.failRowState(t, root.ExecutionID), "round %d (fail won: %v)", i, failWon) + } + }) + } +} diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index bf5d1ee90..397aed9a4 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -166,6 +166,10 @@ func (m *mockConfigStoreForHealth) TransitionExecutionStatusPinned(ctx context.C return nil, nil } +func (m *mockConfigStoreForHealth) TransitionExecutionStatusPinnedTx(ctx context.Context, tx pgx.Tx, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStoreForHealth) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*config.PurchaseExecution, error) { return nil, nil } diff --git a/known_issues/36_plan_step_fail_cas_not_atomic.md b/known_issues/resolved/36_plan_step_fail_cas_not_atomic.md similarity index 75% rename from known_issues/36_plan_step_fail_cas_not_atomic.md rename to known_issues/resolved/36_plan_step_fail_cas_not_atomic.md index 9a654d04c..fb4227231 100644 --- a/known_issues/36_plan_step_fail_cas_not_atomic.md +++ b/known_issues/resolved/36_plan_step_fail_cas_not_atomic.md @@ -1,6 +1,6 @@ # Known Issues: plan step failure is not atomic with its stamp and suppression release -> **Audit status (2026-10-17):** `1 needs triage · 0 resolved` +> **Audit status (2026-10-20):** `0 needs triage · 1 resolved` ## LOW: crash between the failed CAS and the stamp tx strands the row @@ -27,4 +27,9 @@ CAS, the stamp and `DeleteSuppressionsByExecutionTx` commit together. **Why deferred:** out of scope for the retention-ledger runtime change (#905); the window is a crash between two statements and the failure mode is safe. -**Status:** Open. Tracked in #1052. +**Status:** Resolved in #1052. `TransitionExecutionStatusPinnedTx` runs the +pinned CAS in a caller transaction, and `failExistingPlanStep` now commits the +CAS, the `Error`/`NoSubmission` stamp and the suppression release together. An +error before commit leaves the row pending for the next tick. Real-Postgres +integration tests inject an error at the release and race the fail against an +approval, a cancel and a second tick.