From 1faae48dc59fac17914b94915bcc49b10070ebc5 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 19 Oct 2026 13:11:45 +0100 Subject: [PATCH] fix(purchase): guard the remaining post-CAS full-row saves on status Six saves that follow a status CAS (or a fresh read) still wrote the whole row with SavePurchaseExecution, so a status another writer set in between was put back. The reachable cases: a revoke landing between the delayed approve's CAS and its scheduled_execution_at stamp was undone and the scheduler then bought the revoked purchase; a session revoke landing after the terminal save was erased by the revocation-token rotation or the ramp-refusal note. Convert scheduleApprovedExecution, safeFail, failExistingPlanStep, rotateApprovalToken, recordRampAdvanceRefusal and clearApprovalToken to SaveExecutionIfStatus(Tx), each expecting the status its own CAS or read established. The executors' post-claim audit stamp now fails closed when its running guard is lost (the claim is provably gone): approve/run-now/direct execute return a 409-mapped error and the scheduler counts a race loss, with no provider call. Other store errors stay best-effort as before. Closes #835 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_purchases.go | 11 +- ...s_schedule_revoke_race_integration_test.go | 84 +++++++ internal/api/handler_purchases_test.go | 34 ++- internal/purchase/approvals.go | 22 +- internal/purchase/approvals_test.go | 12 +- internal/purchase/claim.go | 31 +++ internal/purchase/claim_guard_test.go | 10 +- internal/purchase/manager.go | 8 +- internal/purchase/notifications.go | 5 +- .../post_cas_guarded_save_integration_test.go | 207 ++++++++++++++++++ .../purchase/post_cas_guarded_save_test.go | 174 +++++++++++++++ internal/purchase/scheduled_fire.go | 12 +- 12 files changed, 580 insertions(+), 30 deletions(-) create mode 100644 internal/api/handler_purchases_schedule_revoke_race_integration_test.go create mode 100644 internal/purchase/post_cas_guarded_save_integration_test.go create mode 100644 internal/purchase/post_cas_guarded_save_test.go diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index af2c99a92..9ee27af9f 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -1255,15 +1255,18 @@ func (h *Handler) scheduleApprovedExecution(ctx context.Context, execution *conf } // Stamp the scheduled time and actor onto the post-CAS row, then persist. - // SavePurchaseExecution is a full-row upsert, so we set all extra fields on - // the returned (freshly-transitioned) execution to avoid overwriting fields - // a concurrent writer may have set between the CAS and here. + // The write is a full-row upsert, so the extra fields go on the returned + // (freshly-transitioned) execution, and it is guarded on "scheduled": a + // revoke that lands between the CAS and here flips the row to canceled, + // and an unguarded upsert would put "scheduled" back, so the scheduler + // would buy a purchase the user revoked (issue #835). A lost guard returns + // ErrExecutionNotInExpectedStatus, which approveWithDelay maps to a 409. scheduledAt := time.Now().Add(delay) updated.ScheduledExecutionAt = &scheduledAt if actor != "" { updated.ApprovedBy = &actor } - if err := h.config.SavePurchaseExecution(ctx, updated); err != nil { + if err := h.config.SaveExecutionIfStatus(ctx, updated, []string{"scheduled"}); err != nil { return nil, fmt.Errorf("failed to stamp scheduled_execution_at on execution %s: %w", execution.ExecutionID, err) } return updated, nil diff --git a/internal/api/handler_purchases_schedule_revoke_race_integration_test.go b/internal/api/handler_purchases_schedule_revoke_race_integration_test.go new file mode 100644 index 000000000..55718c608 --- /dev/null +++ b/internal/api/handler_purchases_schedule_revoke_race_integration_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package api + +// Issue #835 on real PostgreSQL: approveWithDelay's CAS to "scheduled" wins, +// then a revoke cancels the row before the scheduled_execution_at stamp is +// saved. Real store and handler; the revoke is the exact store work +// revokeScheduledExecution does (CancelScheduledExecutionAtomic plus the +// suppression release, in one transaction). Nothing here reaches a cloud API. + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// revokeBeforeStampStore runs revoke once, ahead of the first execution save +// of either kind, so the revoke lands between the CAS and the stamp whether +// the stamp is guarded or not. +type revokeBeforeStampStore struct { + *config.PostgresStore + once sync.Once + revoke func() +} + +func (s *revokeBeforeStampStore) SavePurchaseExecution(ctx context.Context, e *config.PurchaseExecution) error { + s.once.Do(s.revoke) + return s.PostgresStore.SavePurchaseExecution(ctx, e) +} + +func (s *revokeBeforeStampStore) SaveExecutionIfStatus(ctx context.Context, e *config.PurchaseExecution, expected []string) error { + s.once.Do(s.revoke) + return s.PostgresStore.SaveExecutionIfStatus(ctx, e, expected) +} + +// Before the fix the stamp's full-row upsert wrote "scheduled" back over the +// canceled row, with a scheduled_execution_at the scheduler then picks up: a +// purchase the user revoked for free would be bought anyway. +func TestApproveWithDelay_Postgres_RevokeBeforeStampStaysCanceled(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + t.Cleanup(cancel) + f := newCreateConcurrencyFixture(ctx, t) + + row := &config.PurchaseExecution{ + ExecutionID: uuid.NewString(), PlanID: f.planID, Status: "notified", StepNumber: 3, ScheduledDate: time.Now(), + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, row)) + revokedBy := "revoker@example.com" + var canceled bool + f.handler.config = &revokeBeforeStampStore{PostgresStore: f.store, revoke: func() { + require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error { + var err error + if canceled, _, err = f.store.CancelScheduledExecutionAtomic(ctx, tx, row.ExecutionID, &revokedBy); err != nil { + return err + } + return f.store.DeleteSuppressionsByExecutionTx(ctx, tx, row.ExecutionID) + })) + }} + + // A delay already in the past: had the stamp landed, the scheduler's due + // query would return the row on its next tick. + _, err := f.handler.approveWithDelay(ctx, row, -time.Minute, "approver@example.com", nil, "") + + require.True(t, canceled, "the revoke must have won its CAS on the scheduled row") + got, getErr := f.store.GetExecutionByID(ctx, row.ExecutionID) + require.NoError(t, getErr) + assert.Equal(t, config.StatusCanceled, got.Status, "the revoke must survive the approve's stamp") + assert.Nil(t, got.ScheduledExecutionAt) + due, dueErr := f.store.GetScheduledExecutionsDue(ctx) + require.NoError(t, dueErr) + for i := range due { + assert.NotEqual(t, row.ExecutionID, due[i].ExecutionID, "a revoked row must never be due for the scheduler") + } + ce, ok := IsClientError(err) + require.True(t, ok, "expected a client error, got %v", err) + assert.Equal(t, 409, ce.code) +} diff --git a/internal/api/handler_purchases_test.go b/internal/api/handler_purchases_test.go index 6bd68c935..7bb579cd9 100644 --- a/internal/api/handler_purchases_test.go +++ b/internal/api/handler_purchases_test.go @@ -5677,11 +5677,13 @@ func TestHandler_scheduleApprovedExecution_HappyPath(t *testing.T) { mockConfig := new(MockConfigStore) mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "scheduled", mock.Anything). Return(transitioned, nil) - mockConfig.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + // The stamp is guarded on "scheduled" (issue #835); the mock falls back to + // SavePurchaseExecution without a guarded expectation, so this pins it. + mockConfig.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { return e.ExecutionID == execID && e.ScheduledExecutionAt != nil && *e.ApprovedBy == "actor@example.com" - })).Return(nil) + }), []string{"scheduled"}).Return(nil) handler := &Handler{config: mockConfig} @@ -5717,6 +5719,34 @@ func TestApproveWithDelay_CASLostMaps409(t *testing.T) { mockConfig.AssertExpectations(t) } +// TestApproveWithDelay_StampLostToRevokeMaps409 covers issue #835: the CAS to +// scheduled wins, then a revoke flips the row to canceled before the +// scheduled_execution_at stamp. The stamp is guarded on "scheduled", so it +// writes nothing and the approver gets the same 409 as a lost CAS, instead of +// an unguarded upsert putting "scheduled" back for the scheduler to buy. +func TestApproveWithDelay_StampLostToRevokeMaps409(t *testing.T) { + ctx := context.Background() + execID := "exec-stamp-409" + exec := &config.PurchaseExecution{ExecutionID: execID, Status: "pending"} + transitioned := &config.PurchaseExecution{ExecutionID: execID, Status: "scheduled"} + revoked := fmt.Errorf("%w: execution %s is %q, expected one of [scheduled]", + config.ErrExecutionNotInExpectedStatus, execID, config.StatusCanceled) + + mockConfig := new(MockConfigStore) + mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "scheduled", mock.Anything). + Return(transitioned, nil) + mockConfig.On("SaveExecutionIfStatus", ctx, transitioned, []string{"scheduled"}).Return(revoked).Once() + + handler := &Handler{config: mockConfig} + + _, err := handler.approveWithDelay(ctx, exec, 48*time.Hour, "actor@example.com", nil, "") + ce, ok := IsClientError(err) + require.True(t, ok, "a stamp lost to a revoke must be a ClientError, got %v", err) + assert.Equal(t, 409, ce.code) + mockConfig.AssertExpectations(t) + mockConfig.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + // --------------------------------------------------------------------------- // revokePurchase handler tests (issue #291) // --------------------------------------------------------------------------- diff --git a/internal/purchase/approvals.go b/internal/purchase/approvals.go index b43544d12..f11ac04c0 100644 --- a/internal/purchase/approvals.go +++ b/internal/purchase/approvals.go @@ -389,7 +389,8 @@ func (m *Manager) RunPlannedPurchaseNow(ctx context.Context, executionID, actor // rotation lives here, in the one funnel all approve paths share. // // A non-empty skipReason stamps the executed_* audit fields once the CAS wins. -// The returned row is nil when the CAS was never won. +// The returned row is nil when the CAS was never won, or when the audit stamp +// found the claim already lost (nothing is bought then; issue #835). // // check runs against the row the CAS returned, which is the row // executeAndFinalize buys, before any audit stamp or provider call (issue @@ -437,11 +438,10 @@ func (m *Manager) transitionApproveAndExecute(ctx context.Context, executionID, updated.ApprovedBy = &a } if actor != "" || skipReason != "" { - if saveErr := m.config.SaveExecutionIfStatus(ctx, updated, executorOwnedStatuses); saveErr != nil { - // Attribution is best-effort once the atomic flip has landed -- - // dropping these audit fields must not stop the purchase from firing. - // Log loudly so the audit gap is visible. - logging.Errorf("AUDIT GAP: failed to stamp approval audit fields on %s: %v", executionID, saveErr) + // Guarded on "running". A store error stays best-effort, but a lost + // guard means the claim is gone, so nothing is bought (issue #835). + if stampErr := m.stampClaimedAudit(ctx, updated, "approval audit fields"); stampErr != nil { + return nil, "", stampErr } } @@ -568,7 +568,10 @@ func (m *Manager) rotateApprovalToken(ctx context.Context, exec *config.Purchase expiry := time.Now().Add(ttl) exec.ApprovalToken = config.HashApprovalToken(tok) exec.ApprovalTokenExpiresAt = &expiry - if saveErr := m.config.SavePurchaseExecution(ctx, exec); saveErr != nil { + // Guarded on the terminal status the executor just wrote, so this + // full-row write cannot put back a status another writer set since + // (issue #835). + if saveErr := m.config.SaveExecutionIfStatus(ctx, exec, []string{exec.Status}); saveErr != nil { return "", saveErr } return tok, nil @@ -592,7 +595,10 @@ func (m *Manager) clearApprovalToken(ctx context.Context, executionID string) er } exec.ApprovalToken = "" exec.ApprovalTokenExpiresAt = nil - return m.config.SavePurchaseExecution(ctx, exec) + // Guarded on "canceled": the row was read after the cancel landed, and + // this full-row write must not undo a status change made since (issue + // #835). + return m.config.SaveExecutionIfStatus(ctx, exec, []string{config.StatusCanceled}) } // loadCancelableExecution fetches an execution, validates the approval diff --git a/internal/purchase/approvals_test.go b/internal/purchase/approvals_test.go index 78a2d8fe0..11266865e 100644 --- a/internal/purchase/approvals_test.go +++ b/internal/purchase/approvals_test.go @@ -1312,9 +1312,12 @@ func TestClearApprovalToken_PersistsEmpty(t *testing.T) { ApprovalTokenExpiresAt: &future, } store.On("GetExecutionByID", ctx, "exec-tok-clear").Return(execution, nil) - store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + // Guarded on "canceled" (issue #835). The mock falls back to + // SavePurchaseExecution only when no guarded expectation is set, so this + // expectation is what pins the guard. + store.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { return e.ExecutionID == "exec-tok-clear" && e.ApprovalToken == "" && e.ApprovalTokenExpiresAt == nil - })).Return(nil) + }), []string{"canceled"}).Return(nil) err := manager.clearApprovalToken(ctx, "exec-tok-clear") require.NoError(t, err) @@ -1334,13 +1337,14 @@ func TestMintRevocationToken_PersistsFreshToken(t *testing.T) { Status: "completed", ApprovalToken: "old-approval-token", } - store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + // Guarded on the terminal status the row already carries (issue #835). + store.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { // Fresh token must be non-empty, different from old, and expiry set. return e.ExecutionID == "exec-mint" && e.ApprovalToken != "" && e.ApprovalToken != "old-approval-token" && e.ApprovalTokenExpiresAt != nil - })).Return(nil) + }), []string{"completed"}).Return(nil) rawToken, err := manager.rotateApprovalToken(ctx, execution, config.RevocationWindow) require.NoError(t, err) diff --git a/internal/purchase/claim.go b/internal/purchase/claim.go index 70d6a2216..25e60e9dd 100644 --- a/internal/purchase/claim.go +++ b/internal/purchase/claim.go @@ -59,3 +59,34 @@ func (m *Manager) claimAndCheck(ctx context.Context, executionID string, from [] } return updated, nil } + +// errClaimLostBeforePurchase marks a run that stopped before any provider call +// because its post-claim audit stamp found the row no longer "running": another +// writer moved it on (or it vanished) after the claim, so this run no longer +// owns it and must not buy (issue #835). It wraps the store's lost-guard error, +// so errors.Is(err, config.ErrExecutionNotInExpectedStatus) (or ErrNotFound) +// still holds for callers that map those to a 409 or a race loss. +var errClaimLostBeforePurchase = errors.New("claim lost before the purchase; nothing was bought") + +// stampClaimedAudit persists the audit fields (approved_by, executed_at, ...) +// the caller set on its freshly claimed row, guarded on "running" so a stale +// copy never rewrites a status another writer set. +// +// Losing the guard is fail-closed: it returns errClaimLostBeforePurchase and +// the caller must stop before the provider call, because the guard has just +// proved the claim is gone (issue #835). Any other save error stays +// best-effort, as before: the claim is still this run's, the terminal save is +// guarded on "running" too, and missing attribution must not block a purchase. +// what names the fields for the AUDIT GAP log line. +func (m *Manager) stampClaimedAudit(ctx context.Context, exec *config.PurchaseExecution, what string) error { + err := m.config.SaveExecutionIfStatus(ctx, exec, executorOwnedStatuses) + if err == nil { + return nil + } + if errors.Is(err, config.ErrExecutionNotInExpectedStatus) || errors.Is(err, config.ErrNotFound) { + logging.Warnf("purchase[%s]: %s stamp found the claim gone; not purchasing: %v", exec.ExecutionID, what, err) + return fmt.Errorf("%w: %w", errClaimLostBeforePurchase, err) + } + logging.Errorf("AUDIT GAP: failed to stamp %s on %s: %v", what, exec.ExecutionID, err) + return nil +} diff --git a/internal/purchase/claim_guard_test.go b/internal/purchase/claim_guard_test.go index 6baa35803..27354b261 100644 --- a/internal/purchase/claim_guard_test.go +++ b/internal/purchase/claim_guard_test.go @@ -107,12 +107,14 @@ func TestApprove_ClaimsRunningAndStampsGuarded(t *testing.T) { stubExecuteChain(t, store, sender, "plan-g") store.On("SaveExecutionIfStatus", ctx, mock.Anything, []string{"running"}).Return(nil).Once() store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"running"}).Return(nil).Once() + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"completed"}).Return(nil).Once() _, _, err := manager.ApproveAndExecute(ctx, "exec-g", "approver@example.invalid", nil, nil) require.NoError(t, err) store.AssertCalled(t, "TransitionExecutionStatus", ctx, "exec-g", approveFromStatuses, "running", (*string)(nil)) - // Two guarded writes (audit stamp, terminal save); the only unguarded one is - // the post-execution revocation-token rotation. - store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 2) - store.AssertNumberOfCalls(t, "SavePurchaseExecution", 1) + // Three guarded writes: the audit stamp and the terminal save on running, + // and the post-execution revocation-token rotation on the terminal status + // it follows (issue #835). No unguarded full-row write remains. + store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 3) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index dbd9579e4..593058658 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -380,7 +380,9 @@ func (m *Manager) recordRampAdvanceRefusal(ctx context.Context, exec *config.Pur } logging.Errorf("Failed to update plan progress: %v", cause) exec.Error = appendErrNote(exec.Error, fmt.Sprintf("ramp not advanced: %v", cause)) - if saveErr := m.config.SavePurchaseExecution(ctx, exec); saveErr != nil { + // Guarded on the terminal status this run just wrote, so the full-row + // write cannot put back a status another writer set since (issue #835). + if saveErr := m.config.SaveExecutionIfStatus(ctx, exec, []string{exec.Status}); saveErr != nil { logging.Errorf("AUDIT LOSS: failed to persist ramp-advance refusal for execution %s: %v", exec.ExecutionID, saveErr) } @@ -666,7 +668,9 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) next = "Retry is refused for this purchase because a repeat could buy a second commitment. Verify on the cloud provider whether the commitment was created; if it was not, start a new purchase." } updated.Error = "execution was approved but its purchase run was interrupted before completing and never finalized; failed by the recovery sweep so it is not silently stuck (issue #632). " + next - if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil { + if saveErr := m.config.SaveExecutionIfStatus(ctx, updated, []string{config.StatusFailed}); saveErr != nil { + // Guarded on "failed" (issue #835): the stamp lands only while the row + // is still in the status this CAS set, never over a later writer's. // The atomic flip to "failed" already landed via TransitionExecutionStatus; // only the explanatory error string failed to persist. Log loudly but // still count the recovery - the row is no longer stranded in "approved". diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index 5b34502fc..fb8b6a382 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -309,8 +309,11 @@ func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.Purchas stampFailedPlanStep(failed, reason) // The row is failed with NoSubmission, so its suppressions are released as // for any proven not-submitted failure (saveTerminalExecution); a bare row has none. + // The save is guarded on "failed", the status this CAS set, so the + // full-row write never puts back a status another writer set since + // (issue #835); a lost guard rolls the suppression release back too. if err := m.config.WithTx(ctx, func(tx pgx.Tx) error { - if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil { + if err := m.config.SaveExecutionIfStatusTx(ctx, tx, failed, []string{config.StatusFailed}); err != nil { return err } return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, failed.ExecutionID) diff --git a/internal/purchase/post_cas_guarded_save_integration_test.go b/internal/purchase/post_cas_guarded_save_integration_test.go new file mode 100644 index 000000000..ead6efb93 --- /dev/null +++ b/internal/purchase/post_cas_guarded_save_integration_test.go @@ -0,0 +1,207 @@ +//go:build integration + +package purchase + +// Issue #835 on real PostgreSQL: each post-CAS save replayed against a +// concurrent status change landing between the CAS (or the caller's read) and +// the save. Real store, real Manager, counting fake provider (labeled fixture; +// nothing here reaches a cloud API). The concurrent writers are the production +// store calls that make them (the late-recovery write, the session revoke's +// CAS, the reaper's CAS) except where noted as simulated. + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// interleaveStore runs before exactly once, ahead of the first execution save +// of any kind (guarded or not, own transaction or caller's), so the concurrent +// write lands after the caller's CAS or read and before its save. Hooking every +// save variant keeps the replay identical before and after the fix. +type interleaveStore struct { + *config.PostgresStore + once sync.Once + before func() +} + +func (s *interleaveStore) fire() { s.once.Do(s.before) } + +func (s *interleaveStore) SavePurchaseExecution(ctx context.Context, e *config.PurchaseExecution) error { + s.fire() + return s.PostgresStore.SavePurchaseExecution(ctx, e) +} + +func (s *interleaveStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, e *config.PurchaseExecution) error { + s.fire() + return s.PostgresStore.SavePurchaseExecutionTx(ctx, tx, e) +} + +func (s *interleaveStore) SaveExecutionIfStatus(ctx context.Context, e *config.PurchaseExecution, expected []string) error { + s.fire() + return s.PostgresStore.SaveExecutionIfStatus(ctx, e, expected) +} + +func (s *interleaveStore) SaveExecutionIfStatusTx(ctx context.Context, tx pgx.Tx, e *config.PurchaseExecution, expected []string) error { + s.fire() + return s.PostgresStore.SaveExecutionIfStatusTx(ctx, tx, e, expected) +} + +// interleave points the manager at a store that runs before ahead of its next +// execution save, and restores the plain store when the subtest ends. +func (f *rampStepFixture) interleave(t *testing.T, before func()) { + t.Helper() + f.manager.config = &interleaveStore{PostgresStore: f.store, before: before} + t.Cleanup(func() { f.manager.config = f.store }) +} + +// mustTransition is the concurrent writer: one production status CAS. +func mustTransition(ctx context.Context, t *testing.T, f *rampStepFixture, id string, from []string, to string) { + t.Helper() + _, err := f.store.TransitionExecutionStatus(ctx, id, from, to, nil) + require.NoError(t, err) +} + +// Writers that exist today land between the CAS and the save, and the save +// used to put the old status back. +func TestPostCASSave_ConcurrentStatusChangeSurvives(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + + // safeFail flips a stranded approved row to failed; the interrupted + // original run then finishes late and its purchased outcome is recovered + // over the failed row (RecoverReapedExecution, the D1 path). The stamp must + // not turn the purchased row back into "failed", which invites a Retry + // that buys a second commitment. + t.Run("safeFail stamp after a late recovery", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, "approved") + f.interleave(t, func() { + late := rowStatusCopy(row) + late.Status = config.StatusCompleted + now := time.Now() + late.CompletedAt = &now + late.Recommendations[0].Purchased = true + require.NoError(t, f.store.RecoverReapedExecution(ctx, late)) + }) + + counted, err := f.manager.safeFail(ctx, rowStatusCopy(row)) + + require.NoError(t, err) + assert.True(t, counted, "the flip to failed landed, so the sweep still counts it") + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, config.StatusCompleted, got.Status, "the recovered purchase must not be reverted to failed") + assert.NotContains(t, got.Error, "failed by the recovery sweep") + require.Len(t, got.Recommendations, 1) + assert.True(t, got.Recommendations[0].Purchased) + }) + + // A session revoke (revokeViaSession's CAS) lands right after the terminal + // save; the revocation-token rotation and the ramp-refusal note then write + // the caller's in-memory "completed" row. Neither may erase the request. + for _, tc := range []struct { + name string + write func(t *testing.T, exec *config.PurchaseExecution) + }{ + {"revocation token rotation", func(t *testing.T, exec *config.PurchaseExecution) { + _, err := f.manager.rotateApprovalToken(ctx, exec, config.RevocationWindow) + assert.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + }}, + {"ramp-advance refusal note", func(_ *testing.T, exec *config.PurchaseExecution) { + f.manager.recordRampAdvanceRefusal(ctx, exec, errors.New("plan row changed")) + }}, + } { + t.Run(tc.name+" after a revoke", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, config.StatusCompleted) + mustTransition(ctx, t, f, row.ExecutionID, []string{config.StatusCompleted, config.StatusPartiallyCompleted}, config.StatusRevocationRequested) + + tc.write(t, rowStatusCopy(row)) + + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, config.StatusRevocationRequested, got.Status, "the revocation request must survive") + assert.NotContains(t, got.Error, "ramp not advanced") + }) + } +} + +// The executors' post-claim audit stamp finds the claim gone (here the +// reaper's own CAS, running -> failed, lands right after the claim; the reaper +// waits 10 minutes today, so this ordering is not reachable in production yet). +// Nothing may be bought. Before the fix the stamp failure was logged and the +// purchase went ahead. +func TestPostCASSave_ClaimLostBeforeStampBuysNothing(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + reap := func(t *testing.T, row *config.PurchaseExecution) { + f.interleave(t, func() { mustTransition(ctx, t, f, row.ExecutionID, []string{"running"}, config.StatusFailed) }) + } + + t.Run("approve", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, "notified") + before := f.purchaseCount() + reap(t, row) + + got, _, err := f.manager.ApproveAndExecute(ctx, row.ExecutionID, "approver@example.invalid", nil, nil) + + assert.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + assert.Nil(t, got) + assert.Equal(t, 0, f.purchaseCount()-before, "a run that lost its claim must not reach the provider") + assert.Equal(t, config.StatusFailed, rowStatus(ctx, t, f, row.ExecutionID).Status) + }) + + t.Run("scheduled fire", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, "scheduled") + before := f.purchaseCount() + reap(t, row) + + fired, raceLost := f.manager.fireOneDue(ctx, rowStatusCopy(row)) + + assert.False(t, fired) + assert.True(t, raceLost) + assert.Equal(t, 0, f.purchaseCount()-before, "a run that lost its claim must not reach the provider") + assert.Equal(t, config.StatusFailed, rowStatus(ctx, t, f, row.ExecutionID).Status) + }) +} + +// No production writer moves a failed plan-step row or a canceled row today, +// so the concurrent write here is simulated (a status CAS to expired). The +// guard is defense in depth: the full-row save must not resurrect the status +// its own CAS set once another writer has moved the row on. +func TestPostCASSave_SimulatedWriterSurvives(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + + t.Run("failExistingPlanStep", func(t *testing.T) { + row := &config.PurchaseExecution{ + PlanID: f.plan.ID, ExecutionID: uuid.New().String(), Status: "notified", StepNumber: 3, ScheduledDate: time.Now(), + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, row)) + f.interleave(t, func() { + mustTransition(ctx, t, f, row.ExecutionID, []string{config.StatusFailed}, config.StatusExpired) + }) + + err := f.manager.failExistingPlanStep(ctx, rowStatusCopy(row), "no recommendations") + + assert.Equal(t, config.StatusExpired, rowStatus(ctx, t, f, row.ExecutionID).Status) + assert.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + }) + + t.Run("clearApprovalToken", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, config.StatusCanceled) + f.interleave(t, func() { + mustTransition(ctx, t, f, row.ExecutionID, []string{config.StatusCanceled}, config.StatusExpired) + }) + + err := f.manager.clearApprovalToken(ctx, row.ExecutionID) + + assert.Equal(t, config.StatusExpired, rowStatus(ctx, t, f, row.ExecutionID).Status) + assert.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + }) +} diff --git a/internal/purchase/post_cas_guarded_save_test.go b/internal/purchase/post_cas_guarded_save_test.go new file mode 100644 index 000000000..71047ee85 --- /dev/null +++ b/internal/purchase/post_cas_guarded_save_test.go @@ -0,0 +1,174 @@ +package purchase + +// Issue #835: the post-CAS saves outside the executors are status-guarded, and +// a lost claim found by the post-claim audit stamp stops the run before any +// provider call. Mock store and mock provider (labeled fixtures; nothing here +// reaches a cloud API). The Postgres replays of the races are in +// post_cas_guarded_save_integration_test.go. +// +// These assert on the guarded store methods themselves: the mock falls back to +// SavePurchaseExecution(Tx) when no guarded expectation is set, so a +// regression back to the unguarded save would otherwise pass unnoticed. + +import ( + "context" + "errors" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestSafeFail_StampIsGuardedOnFailed(t *testing.T) { + for _, tc := range []struct { + name string + saveErr error + }{ + {"guard holds", nil}, + // The flip to failed already landed; a lost stamp is an audit gap, not + // a sweep error, and the row still counts as recovered. + {"guard lost", lostGuard()}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + stale := &config.PurchaseExecution{ExecutionID: "exec-sf", Status: "approved"} + failed := &config.PurchaseExecution{ExecutionID: "exec-sf", Status: config.StatusFailed} + store.On("TransitionExecutionStatus", ctx, "exec-sf", []string{"approved"}, config.StatusFailed, (*string)(nil)).Return(failed, nil) + store.On("SaveExecutionIfStatus", ctx, failed, []string{config.StatusFailed}).Return(tc.saveErr).Once() + + counted, err := (&Manager{config: store}).safeFail(ctx, stale) + + require.NoError(t, err) + assert.True(t, counted) + assert.Contains(t, failed.Error, "failed by the recovery sweep") + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) + }) + } +} + +func TestFailExistingPlanStep_SaveIsGuardedOnFailed(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + row := &config.PurchaseExecution{ExecutionID: "exec-fps", PlanID: "plan-fps", StepNumber: 2, Status: "notified"} + failed := *row + failed.Status = config.StatusFailed + store.On("TransitionExecutionStatusPinned", ctx, "exec-fps", []string{"pending", "notified"}, config.StatusFailed, (*string)(nil), true).Return(&failed, nil) + store.On("SaveExecutionIfStatusTx", ctx, mock.Anything, &failed, []string{config.StatusFailed}).Return(nil).Once() + store.On("DeleteSuppressionsByExecutionTx", ctx, mock.Anything, "exec-fps").Return(nil).Once() + + err := (&Manager{config: store}).failExistingPlanStep(ctx, row, "no recommendations") + + require.ErrorIs(t, err, errExecutionNotNotifiable, "a recorded failure is still the tick's quiet skip") + assert.True(t, failed.NoSubmission) + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecutionTx", mock.Anything, mock.Anything, mock.Anything) +} + +// A lost guard writes nothing: the suppression release shares the transaction +// and is never reached. +func TestFailExistingPlanStep_LostGuardReleasesNothing(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + row := &config.PurchaseExecution{ExecutionID: "exec-fps", PlanID: "plan-fps", StepNumber: 2, Status: "pending"} + failed := *row + failed.Status = config.StatusFailed + store.On("TransitionExecutionStatusPinned", ctx, "exec-fps", []string{"pending", "notified"}, config.StatusFailed, (*string)(nil), true).Return(&failed, nil) + store.On("SaveExecutionIfStatusTx", ctx, mock.Anything, &failed, []string{config.StatusFailed}).Return(lostGuard()).Once() + + err := (&Manager{config: store}).failExistingPlanStep(ctx, row, "no recommendations") + + require.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + store.AssertExpectations(t) + store.AssertNotCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, mock.Anything) +} + +func TestRecordRampAdvanceRefusal_SaveIsGuardedOnTerminalStatus(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + exec := &config.PurchaseExecution{ExecutionID: "exec-rr", PlanID: "plan-rr", StepNumber: 1, Status: config.StatusCompleted} + store.On("SaveExecutionIfStatus", ctx, exec, []string{config.StatusCompleted}).Return(lostGuard()).Once() + + (&Manager{config: store}).recordRampAdvanceRefusal(ctx, exec, errors.New("plan row changed")) + + assert.Contains(t, exec.Error, "ramp not advanced") + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +// stampLostManager is an approve manager whose post-claim audit stamp loses +// its guard: another writer moved the claimed row on before the stamp. +func stampLostManager(t *testing.T, factory *MockProviderFactory, stampErr error) (*Manager, *MockConfigStore) { + t.Helper() + store := new(MockConfigStore) + failOnProviderCall(factory) + claimed := &config.PurchaseExecution{ExecutionID: "exec-st", PlanID: "plan-st", Status: "running", StepNumber: 1, Recommendations: approvalTestRecs()} + store.On("GetExecutionByID", mock.Anything, "exec-st").Return(&config.PurchaseExecution{ExecutionID: "exec-st", PlanID: "plan-st", Status: "notified"}, nil).Maybe() + store.On("TransitionExecutionStatus", mock.Anything, "exec-st", mock.Anything, "running", (*string)(nil)).Return(claimed, nil).Once() + store.On("SaveExecutionIfStatus", mock.Anything, claimed, executorOwnedStatuses).Return(stampErr).Once() + return &Manager{config: store, providerFactory: factory}, store +} + +// failOnProviderCall makes any provider construction fail AND be recorded, so +// AssertNotCalled can see an attempt. +func failOnProviderCall(factory *MockProviderFactory) { + factory.On("CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything). + Return(nil, errors.New("the provider must not be reached")).Maybe() +} + +func TestApprove_StampFindsClaimLostBuysNothing(t *testing.T) { + ctx := context.Background() + factory := new(MockProviderFactory) + m, store := stampLostManager(t, factory, lostGuard()) + + row, _, err := m.ApproveAndExecute(ctx, "exec-st", "approver@example.invalid", nil, nil) + + require.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus, "callers map a lost claim to a 409") + require.ErrorIs(t, err, errClaimLostBeforePurchase) + assert.Nil(t, row, "no row is returned for a run that never reached the provider") + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + store.AssertExpectations(t) + store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 1) +} + +func TestFireOneDue_StampFindsClaimLostBuysNothing(t *testing.T) { + ctx := context.Background() + factory := new(MockProviderFactory) + m, store := stampLostManager(t, factory, lostGuard()) + + fired, raceLost := m.fireOneDue(ctx, &config.PurchaseExecution{ExecutionID: "exec-st", Status: "scheduled"}) + + assert.False(t, fired) + assert.True(t, raceLost, "a claim lost before the purchase is a race loss, not an error") + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + store.AssertExpectations(t) + store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 1) +} + +// A store error that is not a lost guard leaves the claim this run's, so the +// stamp stays best-effort and the purchase goes ahead. +func TestStampClaimedAudit_StoreErrorStaysBestEffort(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + exec := &config.PurchaseExecution{ExecutionID: "exec-be", Status: "running"} + store.On("SaveExecutionIfStatus", ctx, exec, executorOwnedStatuses).Return(errors.New("connection reset")).Once() + + require.NoError(t, (&Manager{config: store}).stampClaimedAudit(ctx, exec, "approved_by")) + store.AssertExpectations(t) +} + +func TestStampClaimedAudit_MissingRowIsClaimLost(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + exec := &config.PurchaseExecution{ExecutionID: "exec-gone", Status: "running"} + store.On("SaveExecutionIfStatus", ctx, exec, executorOwnedStatuses).Return(config.ErrNotFound).Once() + + err := (&Manager{config: store}).stampClaimedAudit(ctx, exec, "approved_by") + + require.ErrorIs(t, err, errClaimLostBeforePurchase) + require.ErrorIs(t, err, config.ErrNotFound) + store.AssertExpectations(t) +} diff --git a/internal/purchase/scheduled_fire.go b/internal/purchase/scheduled_fire.go index 4340593f5..df04b3f6e 100644 --- a/internal/purchase/scheduled_fire.go +++ b/internal/purchase/scheduled_fire.go @@ -72,7 +72,9 @@ func (m *Manager) FireScheduledDelayedPurchases(ctx context.Context) (*FireResul // fireOneDue attempts to fire a single due scheduled execution. Returns // (true, false) on success, (false, true) when the CAS was lost to a -// concurrent revoke, or (false, false) on a real error. +// concurrent revoke or the claim was lost before the provider call (the +// guarded audit stamp found the row no longer running), or (false, false) on +// a real error. func (m *Manager) fireOneDue(ctx context.Context, exec *config.PurchaseExecution) (fired, raceLost bool) { // CAS: scheduled -> running (the executor claim). If this fails with ErrExecutionNotInExpectedStatus // the revoke handler already transitioned the row to "canceled" — that is @@ -91,10 +93,10 @@ func (m *Manager) fireOneDue(ctx context.Context, exec *config.PurchaseExecution // Stamp ApprovedBy for the audit trail before executing. actor := "scheduler" updated.ApprovedBy = &actor - if saveErr := m.config.SaveExecutionIfStatus(ctx, updated, executorOwnedStatuses); saveErr != nil { - // Best-effort audit stamp: attribution failure must not block the - // purchase from firing. Log loudly so the audit gap is visible. - logging.Errorf("AUDIT GAP: fireOneDue[%s]: failed to stamp approved_by: %v", exec.ExecutionID, saveErr) + // Guarded on "running". A store error stays best-effort, but a lost guard + // means the claim is gone, so nothing is bought (issue #835). + if stampErr := m.stampClaimedAudit(ctx, updated, "approved_by"); stampErr != nil { + return false, true } if execErr := m.executeAndFinalize(ctx, updated); execErr != nil {