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.