diff --git a/internal/api/handler_purchases_test.go b/internal/api/handler_purchases_test.go index b169cbbc..396f4162 100644 --- a/internal/api/handler_purchases_test.go +++ b/internal/api/handler_purchases_test.go @@ -6644,7 +6644,7 @@ func TestHandler_approvePurchaseViaSession_FourEyesOn_DifferentApproverSucceeds( approved := &config.PurchaseExecution{ExecutionID: execID, Status: "approved"} mockConfig.On("GetExecutionByID", ctx, execID).Return(exec, nil) mockConfig.On("GetGlobalConfig", ctx).Return(fourEyesCfgOn(), nil) - mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "approved", &approverID).Return(approved, nil) + mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "running", &approverID).Return(approved, nil) mockConfig.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) mockAuth := new(MockAuthService) diff --git a/internal/api/handler_purchases_token_errors_test.go b/internal/api/handler_purchases_token_errors_test.go index 7d9bd9ef..7993072f 100644 --- a/internal/api/handler_purchases_token_errors_test.go +++ b/internal/api/handler_purchases_token_errors_test.go @@ -103,7 +103,7 @@ func TestHandleRequest_ApproveViaToken_NonPendingExecutionIs409(t *testing.T) { exec := &config.PurchaseExecution{Status: "completed", ApprovalToken: config.HashApprovalToken(tokenErrRawToken), ApprovalTokenExpiresAt: &future} h := tokenErrHandler(t, exec, nil) store := h.config.(*MockConfigStore) - store.On("TransitionExecutionStatus", mock.Anything, tokenErrExecID, mock.Anything, "approved", mock.Anything). + store.On("TransitionExecutionStatus", mock.Anything, tokenErrExecID, mock.Anything, "running", mock.Anything). Return(nil, config.ErrExecutionNotInExpectedStatus) resp, err := h.HandleRequest(context.Background(), tokenErrRequest("approve", tokenErrRawToken)) diff --git a/internal/config/store_postgres_execution_guarded_integration_test.go b/internal/config/store_postgres_execution_guarded_integration_test.go index 12992523..7a9084f9 100644 --- a/internal/config/store_postgres_execution_guarded_integration_test.go +++ b/internal/config/store_postgres_execution_guarded_integration_test.go @@ -192,3 +192,31 @@ func TestRecoverReapedExecution_OnlyFailedRowsAndOnlySuccessStatuses(t *testing. require.Error(t, store.RecoverReapedExecution(ctx, bad)) assert.Equal(t, "failed", storedStatus(ctx, t, store, bad.ExecutionID)) } + +// The clear is only meaningful against a row that actually carries the proof: +// a failed+true row recovered to completed, or guard-saved to a non-failed +// status, must end with no_submission = false. +func TestGuardedSaves_ClearAProofThatWasActuallySet(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + seed := func() *PurchaseExecution { + exec := newNoSubmissionExec(ctx, t, store, "running") + exec.Status = "failed" + exec.NoSubmission = true + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + require.True(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "precondition: proof is set") + return exec + } + + rec := seed() + rec.Status = "completed" + rec.NoSubmission = false + require.NoError(t, store.RecoverReapedExecution(ctx, rec)) + assert.False(t, storedNoSubmission(ctx, t, store, rec.ExecutionID), "recovery to completed clears the proof") + + g := seed() + g.Status = "approved" + g.NoSubmission = false + require.NoError(t, store.SaveExecutionIfStatus(ctx, g, []string{"failed"})) + assert.False(t, storedNoSubmission(ctx, t, store, g.ExecutionID), "a non-failed guarded save clears the proof") +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index ba8c1852..c24eccc9 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -369,7 +369,7 @@ func (m *MockConfigStore) SaveExecutionIfStatus(ctx context.Context, exec *confi func (m *MockConfigStore) SaveExecutionIfStatusTx(ctx context.Context, tx pgx.Tx, exec *config.PurchaseExecution, expected []string) error { m.record("SaveExecutionIfStatusTx", ctx, tx, exec, expected) if !isExpected(&m.Mock, "SaveExecutionIfStatusTx") { - return m.SaveExecutionIfStatus(ctx, exec, expected) + return m.SavePurchaseExecutionTx(ctx, tx, exec) } args := m.Called(ctx, tx, exec, expected) return args.Error(0) diff --git a/internal/purchase/approvals.go b/internal/purchase/approvals.go index c574d1cb..ae48cac8 100644 --- a/internal/purchase/approvals.go +++ b/internal/purchase/approvals.go @@ -357,7 +357,7 @@ func (m *Manager) RunPlannedPurchaseNow(ctx context.Context, executionID, actor // transitionApproveAndExecute is the shared body behind ApproveAndExecute // and RunPlannedPurchaseNow: enforce the 4-eyes gate, atomically CAS the row -// from one of fromStatuses to "approved", stamp ApprovedBy, then run the +// from one of fromStatuses straight to "running" (the claim), stamp ApprovedBy, then run the // purchase synchronously. Parameterizing on fromStatuses lets both entry // points share one implementation of the gate and the CAS instead of the // gate being reimplemented (or skipped) at a second call site. @@ -389,13 +389,17 @@ func (m *Manager) transitionApproveAndExecute(ctx context.Context, executionID, // approvals (stamped onto transitioned_by); it is nil for token/SQS/system // flows so transitioned_by = NULL on those hops. The human-readable actor // email is recorded separately onto approved_by (below). - updated, err := m.config.TransitionExecutionStatus(ctx, executionID, fromStatuses, "approved", transitionedBy) + // One CAS straight to "running": winning it is the claim, so no other + // executor (SQS, cron, re-drive, scheduler) can run this row, and there is + // no window where the row sits "approved" with a stale copy in flight + // (issue #829). A lost CAS surfaces ErrExecutionNotInExpectedStatus. + updated, err := m.claimRunning(ctx, executionID, fromStatuses, transitionedBy) if err != nil { logging.Errorf("purchase[%s]: transitionApproveAndExecute status transition failed after %s: %v", executionID, time.Since(t0), err) return nil, "", fmt.Errorf("approve: %w", err) } - logging.Infof("purchase[%s]: status transitioned to approved in %s", executionID, time.Since(t0)) + logging.Infof("purchase[%s]: approved and claimed running in %s", executionID, time.Since(t0)) if skipReason != "" { now := time.Now() @@ -409,7 +413,7 @@ func (m *Manager) transitionApproveAndExecute(ctx context.Context, executionID, updated.ApprovedBy = &a } if actor != "" || skipReason != "" { - if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil { + 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. diff --git a/internal/purchase/approvals_test.go b/internal/purchase/approvals_test.go index 82292375..5df9312b 100644 --- a/internal/purchase/approvals_test.go +++ b/internal/purchase/approvals_test.go @@ -99,7 +99,7 @@ func TestManager_ApproveExecution_Success(t *testing.T) { } store.On("GetExecutionByID", ctx, "exec-123").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-456") _, err := manager.ApproveExecution(ctx, "exec-123", "valid-token", "") @@ -128,7 +128,7 @@ func TestManager_ApproveExecution_StampsApprovedBy(t *testing.T) { } store.On("GetExecutionByID", ctx, "exec-123").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-456") _, err := manager.ApproveExecution(ctx, "exec-123", "valid-token", "operator@example.com") @@ -163,7 +163,7 @@ func TestManager_ApproveExecution_NotifiedStatus(t *testing.T) { } store.On("GetExecutionByID", ctx, "exec-123").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-456") _, err := manager.ApproveExecution(ctx, "exec-123", "valid-token", "") @@ -252,7 +252,7 @@ func TestManager_ApproveExecution_TransitionFails(t *testing.T) { ApprovalToken: config.HashApprovalToken("valid-token"), } store.On("GetExecutionByID", ctx, "exec-123").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-123", approveFromStatuses, "running", (*string)(nil)). Return(nil, errors.New(`execution exec-123 cannot transition from "canceled" to "approved"`)) _, err := manager.ApproveExecution(ctx, "exec-123", "valid-token", "") @@ -281,7 +281,7 @@ func TestManager_ApproveAndExecute_EmptyPlanID(t *testing.T) { Status: "approved", ApprovalToken: "tok", } - store.On("TransitionExecutionStatus", ctx, "exec-direct-1", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-direct-1", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) // No GetPurchasePlan / GetPlanAccounts / UpdatePurchasePlan calls — see AssertNotCalled below. sender.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil) store.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) @@ -326,7 +326,7 @@ func TestManager_ApproveAndExecute_SkipsTokenCheck(t *testing.T) { StepNumber: 1, Recommendations: approvalTestRecs(), } - store.On("TransitionExecutionStatus", ctx, "exec-456", approveFromStatuses, "approved", + store.On("TransitionExecutionStatus", ctx, "exec-456", approveFromStatuses, "running", mock.MatchedBy(func(actor *string) bool { return actor != nil && *actor == actorUUID })).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-789") @@ -360,7 +360,7 @@ func TestManager_RunPlannedPurchaseNow_ExecutesFromPaused(t *testing.T) { StepNumber: 1, Recommendations: approvalTestRecs(), } - store.On("TransitionExecutionStatus", ctx, "exec-run-now", []string{"pending", "notified", "paused"}, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-run-now", []string{"pending", "notified", "paused"}, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-run-now") final, revocationToken, err := manager.RunPlannedPurchaseNow(ctx, "exec-run-now", "operator@example.com", nil) @@ -387,7 +387,7 @@ func TestManager_RunPlannedPurchaseNow_LostCASReturnsError(t *testing.T) { ctx := context.Background() manager, store, _ := newApproveManager(t) - store.On("TransitionExecutionStatus", ctx, "exec-race", []string{"pending", "notified", "paused"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-race", []string{"pending", "notified", "paused"}, "running", (*string)(nil)). Return(nil, config.ErrExecutionNotInExpectedStatus) final, _, err := manager.RunPlannedPurchaseNow(ctx, "exec-race", "operator@example.com", nil) @@ -412,7 +412,7 @@ func TestManager_DirectExecute_StampsAuditFieldsAfterClaim(t *testing.T) { StepNumber: 1, Recommendations: approvalTestRecs(), } - store.On("TransitionExecutionStatus", ctx, "exec-direct-stamp", approveFromStatuses, "approved", + store.On("TransitionExecutionStatus", ctx, "exec-direct-stamp", approveFromStatuses, "running", mock.MatchedBy(func(actor *string) bool { return actor != nil && *actor == actorUUID })).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-direct-stamp") @@ -433,7 +433,7 @@ func TestManager_DirectExecute_RefusedLeavesNoExecutedAt(t *testing.T) { ctx := context.Background() manager, store, _ := newApproveManager(t) - store.On("TransitionExecutionStatus", ctx, "exec-direct-lost", approveFromStatuses, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-direct-lost", approveFromStatuses, "running", (*string)(nil)). Return(nil, config.ErrExecutionNotInExpectedStatus) final, _, err := manager.DirectExecute(ctx, "exec-direct-lost", "operator@example.com", nil) @@ -457,7 +457,7 @@ func TestManager_ApproveAndExecute_ReturnsRowOnExecutionFailure(t *testing.T) { Status: "approved", Recommendations: approvalTestRecs(), } - store.On("TransitionExecutionStatus", ctx, "exec-fail-after-claim", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-fail-after-claim", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) store.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) final, token, err := manager.ApproveAndExecute(ctx, "exec-fail-after-claim", "operator@example.com", nil) @@ -548,7 +548,7 @@ func TestManager_ApproveAndExecute_FourEyesOn_AllowsDifferentApprover(t *testing store.On("GetGlobalConfig", ctx).Return(fourEyesCfgOnForManager(), nil) store.On("GetExecutionByID", ctx, "exec-direct-diff").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-direct-diff", approveFromStatuses, "approved", &approverUUID).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-direct-diff", approveFromStatuses, "running", &approverUUID).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-fourEyes") _, _, err := manager.ApproveAndExecute(ctx, "exec-direct-diff", approverEmail, &approverUUID) @@ -645,7 +645,7 @@ func TestManager_ApproveAndExecute_FourEyesOff_AllowsSelfApprove(t *testing.T) { creatorID := "user-creator" updated := &config.PurchaseExecution{ExecutionID: "exec-mode-off", PlanID: "plan-fourEyes", Status: "approved", StepNumber: 1, Recommendations: approvalTestRecs()} - store.On("TransitionExecutionStatus", ctx, "exec-mode-off", approveFromStatuses, "approved", &creatorID).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-mode-off", approveFromStatuses, "running", &creatorID).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-fourEyes") _, _, err := manager.ApproveAndExecute(ctx, "exec-mode-off", "creator@example.com", &creatorID) @@ -716,7 +716,7 @@ func TestManager_ApproveExecution_FourEyesOn_DifferentActor_Allowed(t *testing.T store.On("GetExecutionByID", ctx, "exec-sqs-diff").Return(execution, nil) store.On("GetGlobalConfig", ctx).Return(fourEyesCfgOnForManager(), nil) store.On("GetUserEmailByID", ctx, creatorID).Return("creator@example.com", nil) - store.On("TransitionExecutionStatus", ctx, "exec-sqs-diff", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-sqs-diff", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-fourEyes") _, err := manager.ApproveExecution(ctx, "exec-sqs-diff", "valid-token", "approver@example.com") @@ -1055,7 +1055,7 @@ func TestManager_ApproveExecution_ValidTokenWithinTTL(t *testing.T) { Recommendations: approvalTestRecs(), } store.On("GetExecutionByID", ctx, "exec-live").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-live", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-live", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-live") _, err := manager.ApproveExecution(ctx, "exec-live", "valid-token", "") @@ -1086,7 +1086,7 @@ func TestManager_ApproveExecution_NilExpiresAt_LegacyRow(t *testing.T) { Recommendations: approvalTestRecs(), } store.On("GetExecutionByID", ctx, "exec-legacy").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-legacy", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-legacy", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-legacy") _, err := manager.ApproveExecution(ctx, "exec-legacy", "valid-token", "") @@ -1148,7 +1148,7 @@ func TestManager_ApproveExecution_AWSOrphanFallsThrough(t *testing.T) { Recommendations: approvalTestRecs(), } store.On("GetExecutionByID", ctx, "exec-aws-ambient").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-aws-ambient", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-aws-ambient", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-aws") _, err := manager.ApproveExecution(ctx, "exec-aws-ambient", "valid-token", "") @@ -1268,7 +1268,7 @@ func TestApproveExecution_MintsRevocationToken(t *testing.T) { // returned and executeAndFinalize continued operating on -- in place; it // does no fetch of its own (see rotateApprovalToken's doc comment for why). store.On("GetExecutionByID", ctx, "exec-rotate").Return(execution, nil) - store.On("TransitionExecutionStatus", ctx, "exec-rotate", approveFromStatuses, "approved", (*string)(nil)).Return(updated, nil) + store.On("TransitionExecutionStatus", ctx, "exec-rotate", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) stubExecuteChain(t, store, sender, "plan-rotate") _, err := manager.ApproveExecution(ctx, "exec-rotate", "pre-rotate-token", "") diff --git a/internal/purchase/armed_redrive_test.go b/internal/purchase/armed_redrive_test.go index 79f61edf..193ef5af 100644 --- a/internal/purchase/armed_redrive_test.go +++ b/internal/purchase/armed_redrive_test.go @@ -127,10 +127,10 @@ func armedHarness(t *testing.T, serviceType common.ServiceType) (*Manager, *Mock // expectClaim wires the CAS the executor paths perform before executing, so the // execution reaches executeAndFinalize the way production does. -func expectClaim(store *MockConfigStore, exec *config.PurchaseExecution, from []string, to string) { +func expectClaim(store *MockConfigStore, exec *config.PurchaseExecution, from []string) { claimed := *exec - claimed.Status = to - store.On("TransitionExecutionStatus", mock.Anything, exec.ExecutionID, from, to, (*string)(nil)). + claimed.Status = "running" + store.On("TransitionExecutionStatus", mock.Anything, exec.ExecutionID, from, "running", (*string)(nil)). Return(&claimed, nil).Once() } @@ -142,7 +142,7 @@ func TestApproveAndExecuteRefusesArmedAzureSavingsPlanRetry(t *testing.T) { exec := armedExecution("azure", "savingsplans", 1, "pending") mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll) store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, []string{"pending", "notified"}, "approved") + expectClaim(store, exec, []string{"pending", "notified"}) _, _, err := mgr.ApproveAndExecute(context.Background(), exec.ExecutionID, "operator@example.com", nil) @@ -159,7 +159,7 @@ func TestClaimAndExecuteRefusesArmedAzureSavingsPlanRetry(t *testing.T) { exec := armedExecution("azure", "savingsplans", 2, "approved") mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll) store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, exec, []string{"approved", "pending", "notified"}) body, err := json.Marshal(AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: exec.ExecutionID}) require.NoError(t, err) @@ -183,7 +183,7 @@ func TestFireScheduledDelayedPurchasesRefusesArmedAzureSavingsPlanRetry(t *testi mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll) store.On("GetScheduledExecutionsDue", mock.Anything). Return([]config.PurchaseExecution{*exec}, nil).Once() - expectClaim(store, exec, []string{"scheduled"}, "approved") + expectClaim(store, exec, []string{"scheduled"}) result, err := mgr.FireScheduledDelayedPurchases(context.Background()) @@ -206,7 +206,7 @@ func TestFirstAzureSavingsPlanPurchaseStillExecutes(t *testing.T) { mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll) store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}).Return(exec.Recommendations, nil).Once() store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, exec, []string{"approved", "pending", "notified"}) body, err := json.Marshal(AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: exec.ExecutionID}) require.NoError(t, err) @@ -225,7 +225,7 @@ func TestSafeProviderRetrySuccessorStillExecutes(t *testing.T) { exec := armedExecution("aws", "rds", 3, "approved") mgr, store, rec := armedHarness(t, common.ServiceRDS) store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, exec, []string{"approved", "pending", "notified"}) body, err := json.Marshal(AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: exec.ExecutionID}) require.NoError(t, err) @@ -250,7 +250,7 @@ func TestArmedRowIsDefusedNotLeftArmed(t *testing.T) { return nil } store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, exec, []string{"approved", "pending", "notified"}) body, err := json.Marshal(AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: exec.ExecutionID}) require.NoError(t, err) diff --git a/internal/purchase/aws_no_token_redrive_test.go b/internal/purchase/aws_no_token_redrive_test.go index d187a832..90b9a2b6 100644 --- a/internal/purchase/aws_no_token_redrive_test.go +++ b/internal/purchase/aws_no_token_redrive_test.go @@ -103,7 +103,7 @@ func TestAWSNoTokenRetrySuccessorIsRefusedAndFirstBuyStillExecutes(t *testing.T) return nil } store.On("GetExecutionByID", mock.Anything, retry.ExecutionID).Return(retry, nil).Maybe() - expectClaim(store, retry, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, retry, []string{"approved", "pending", "notified"}) body, err := json.Marshal(AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: retry.ExecutionID}) require.NoError(t, err) _ = mgr.ProcessMessage(context.Background(), string(body)) @@ -117,7 +117,7 @@ func TestAWSNoTokenRetrySuccessorIsRefusedAndFirstBuyStillExecutes(t *testing.T) first := armedExecution("aws", service, 0, "approved") mgr, store, rec = armedHarness(t, svcType) store.On("GetExecutionByID", mock.Anything, first.ExecutionID).Return(first, nil).Maybe() - expectClaim(store, first, []string{"approved", "pending", "notified"}, "running") + expectClaim(store, first, []string{"approved", "pending", "notified"}) require.NoError(t, mgr.ProcessMessage(context.Background(), string(body))) assert.Equal(t, 1, rec.count(), "a first %s purchase must still go through; the guard is for retries only", service) }) diff --git a/internal/purchase/azure_pricing_integration_test.go b/internal/purchase/azure_pricing_integration_test.go index 85d98d65..dc5b5f73 100644 --- a/internal/purchase/azure_pricing_integration_test.go +++ b/internal/purchase/azure_pricing_integration_test.go @@ -60,7 +60,7 @@ func TestApproveAzurePricingRequiresMigrationAndRecollection(t *testing.T) { exec.Recommendations = []config.RecommendationRecord{rec} mgr, store, recorder := armedHarness(t, common.ServiceCompute) mgr.config = &azurePricingPostgresStore{MockConfigStore: store, pricing: pricing} - expectClaim(store, exec, []string{"pending", "notified"}, "approved") + expectClaim(store, exec, []string{"pending", "notified"}) var saved *config.PurchaseExecution store.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error { copy := *e diff --git a/internal/purchase/azure_pricing_test.go b/internal/purchase/azure_pricing_test.go index dc016fbe..85b913db 100644 --- a/internal/purchase/azure_pricing_test.go +++ b/internal/purchase/azure_pricing_test.go @@ -95,7 +95,7 @@ func TestAzurePricingRefusalEntryPoints(t *testing.T) { } return nil } - expectClaim(store, exec, []string{"pending", "notified"}, "approved") + expectClaim(store, exec, []string{"pending", "notified"}) var err error if entry == "approve" { _, _, err = mgr.ApproveAndExecute(context.Background(), exec.ExecutionID, "operator@example.com", nil) diff --git a/internal/purchase/claim.go b/internal/purchase/claim.go new file mode 100644 index 00000000..bad5ab08 --- /dev/null +++ b/internal/purchase/claim.go @@ -0,0 +1,22 @@ +package purchase + +import ( + "context" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// executorOwnedStatuses is the only status an executor may write its terminal +// result or audit stamps over: it owns the row while it holds the "running" +// claim and not after the reaper, a cancel or another writer moved it on. +var executorOwnedStatuses = []string{"running"} + +// claimRunning is the single claim primitive: an atomic CAS from one of from +// to "running". Winning it is the right to call the provider; every executor +// (approve, scheduler, SQS, cron, re-drive) claims before any provider call. +// A lost CAS returns an error satisfying +// errors.Is(err, config.ErrExecutionNotInExpectedStatus) or config.ErrNotFound. +// actor is the transitioned_by user UUID, nil for system-initiated claims. +func (m *Manager) claimRunning(ctx context.Context, executionID string, from []string, actor *string) (*config.PurchaseExecution, error) { + return m.config.TransitionExecutionStatus(ctx, executionID, from, "running", actor) +} diff --git a/internal/purchase/claim_guard_test.go b/internal/purchase/claim_guard_test.go new file mode 100644 index 00000000..9f6e74b9 --- /dev/null +++ b/internal/purchase/claim_guard_test.go @@ -0,0 +1,117 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func lostGuard() error { + return fmt.Errorf("%w: execution x is %q", config.ErrExecutionNotInExpectedStatus, "failed") +} + +func purchasedExec(status string) *config.PurchaseExecution { + return &config.PurchaseExecution{ + ExecutionID: "exec-x", Status: status, + Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "rds", Purchased: true}}, + } +} + +// These assert on the guarded methods themselves: the mock falls back to +// SavePurchaseExecution when no guarded expectation is set, which would make a +// regression back to the unguarded save invisible. +func TestSaveOrRecoverTerminal_GuardsOnRunning(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + m := &Manager{config: store} + exec := purchasedExec("completed") + store.On("SaveExecutionIfStatus", ctx, exec, []string{"running"}).Return(nil).Once() + + require.NoError(t, m.saveOrRecoverTerminal(ctx, exec, nil)) + store.AssertCalled(t, "SaveExecutionIfStatus", ctx, exec, []string{"running"}) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) + store.AssertNotCalled(t, "RecoverReapedExecution", mock.Anything, mock.Anything) +} + +func TestSaveOrRecoverTerminal_ReapedButPurchasedIsRecovered(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + m := &Manager{config: store} + exec := purchasedExec("partially_completed") + store.On("SaveExecutionIfStatus", ctx, exec, []string{"running"}).Return(lostGuard()) + store.On("RecoverReapedExecution", ctx, exec).Return(nil) + + require.NoError(t, m.saveOrRecoverTerminal(ctx, exec, nil)) + assert.Contains(t, exec.Error, recoveredAfterReapNote) + store.AssertCalled(t, "RecoverReapedExecution", ctx, exec) +} + +func TestSaveOrRecoverTerminal_RetryLinkedFailsLoudWithDoubleBuyFlag(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + m := &Manager{config: store} + exec := purchasedExec("completed") + store.On("SaveExecutionIfStatus", ctx, exec, []string{"running"}).Return(lostGuard()) + store.On("RecoverReapedExecution", ctx, exec).Return(fmt.Errorf("%w: x", config.ErrExecutionRetryLinked)) + + err := m.saveOrRecoverTerminal(ctx, exec, nil) + require.ErrorIs(t, err, config.ErrExecutionRetryLinked) + assert.Contains(t, err.Error(), "POSSIBLE DOUBLE BUY") +} + +func TestSaveOrRecoverTerminal_NothingPurchasedIsNeverOverwritten(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + m := &Manager{config: store} + exec := &config.PurchaseExecution{ExecutionID: "exec-x", Status: "failed"} + store.On("SaveExecutionIfStatus", ctx, exec, []string{"running"}).Return(lostGuard()) + + err := m.saveOrRecoverTerminal(ctx, exec, errors.New("boom")) + require.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + store.AssertNotCalled(t, "RecoverReapedExecution", mock.Anything, mock.Anything) +} + +// The reaper's annotation is guarded on failed; losing it to a recovered row is +// benign and the row still counts as reaped. +func TestReapOne_AnnotationLostToRecoveryIsBenign(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + row := stuckExec("exec-R", "running") + failed := row + failed.Status = failedStatus + store.On("TransitionExecutionStatus", ctx, "exec-R", stuckStatuses, failedStatus, (*string)(nil)).Return(&failed, nil) + store.On("SaveExecutionIfStatus", ctx, mock.Anything, []string{failedStatus}).Return(lostGuard()) + + result := &ReapResult{} + newReaperManager(store).reapOne(ctx, &row, 10*time.Minute, time.Now(), result) + assert.Equal(t, 1, result.Reaped) + assert.Zero(t, result.Errored) + store.AssertCalled(t, "SaveExecutionIfStatus", ctx, mock.Anything, []string{failedStatus}) +} + +// Approve claims straight to running and stamps the audit fields through the +// running-guarded save. +func TestApprove_ClaimsRunningAndStampsGuarded(t *testing.T) { + ctx := context.Background() + manager, store, sender := newApproveManager(t) + updated := &config.PurchaseExecution{ExecutionID: "exec-g", PlanID: "plan-g", Status: "running", StepNumber: 1, Recommendations: approvalTestRecs()} + store.On("GetExecutionByID", mock.Anything, "exec-g").Return(&config.PurchaseExecution{ExecutionID: "exec-g", PlanID: "plan-g", Status: "pending"}, nil).Maybe() + store.On("TransitionExecutionStatus", ctx, "exec-g", approveFromStatuses, "running", (*string)(nil)).Return(updated, nil) + stubExecuteChain(t, store, sender, "plan-g") + store.On("SaveExecutionIfStatus", ctx, mock.Anything, []string{"running"}).Return(nil) + + _, _, err := manager.ApproveAndExecute(ctx, "exec-g", "approver@example.invalid", 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) +} diff --git a/internal/purchase/claim_pin_test.go b/internal/purchase/claim_pin_test.go index 6bf41fe2..faf99749 100644 --- a/internal/purchase/claim_pin_test.go +++ b/internal/purchase/claim_pin_test.go @@ -105,7 +105,7 @@ func TestClaimPin_PostClaimCapRecheckFailsBeforeSubmission(t *testing.T) { claimed.Status = "running" store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), false).Return(&claimed, nil) var saved *config.PurchaseExecution - store.On("SavePurchaseExecutionTx", ctx, mock.Anything, mock.Anything). + store.On("SaveExecutionIfStatusTx", ctx, mock.Anything, mock.Anything, []string{"running"}). Run(func(args mock.Arguments) { saved = args.Get(2).(*config.PurchaseExecution) }).Return(nil) store.On("DeleteSuppressionsByExecutionTx", ctx, mock.Anything, "exec-gate").Return(nil) diff --git a/internal/purchase/coverage_extra_test.go b/internal/purchase/coverage_extra_test.go index 5e9e33b7..71ec647e 100644 --- a/internal/purchase/coverage_extra_test.go +++ b/internal/purchase/coverage_extra_test.go @@ -403,7 +403,7 @@ func TestProcessMessage_ApproveHappyPath(t *testing.T) { mockStore.On("GetExecutionByID", ctx, "exec-appv").Return(exec, nil).Times(2) mockStore.On("GetCloudAccount", ctx, accountID).Return(account, nil) // Atomic approve transition (issue #372 fix). - mockStore.On("TransitionExecutionStatus", ctx, "exec-appv", []string{"pending", "notified"}, "approved", (*string)(nil)).Return(approved, nil) + mockStore.On("TransitionExecutionStatus", ctx, "exec-appv", []string{"pending", "notified"}, "running", (*string)(nil)).Return(approved, nil) // Synchronous execute chain: GetPurchasePlan is called with the // approved execution's PlanID — pinning the non-empty id here means a // regression that drops it (e.g. passes "" or exec.ExecutionID by @@ -516,7 +516,7 @@ func TestProcessMessage_ApproveFourEyesOn_DifferentApproverSucceeds(t *testing.T mockStore.On("GetCloudAccount", ctx, accountID).Return(account, nil) mockStore.On("GetGlobalConfig", ctx).Return(fourEyesCfgOnForManager(), nil) mockStore.On("GetUserEmailByID", ctx, creatorID).Return("owner@example.com", nil) - mockStore.On("TransitionExecutionStatus", ctx, "exec-appv-diff", []string{"pending", "notified"}, "approved", (*string)(nil)).Return(approved, nil) + mockStore.On("TransitionExecutionStatus", ctx, "exec-appv-diff", []string{"pending", "notified"}, "running", (*string)(nil)).Return(approved, nil) plan := &config.PurchasePlan{ID: planID, Name: "test-plan"} mockStore.On("GetPurchasePlan", ctx, planID).Return(plan, nil) mockEmail.On("SendPurchaseConfirmation", ctx, mock.Anything).Return(nil) diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 8e70f179..24106ce3 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -253,7 +253,7 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if execErr != nil { logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr) } - if err := m.saveTerminalExecution(ctx, exec, execErr); err != nil { + if err := m.saveOrRecoverTerminal(ctx, exec, execErr); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) // Wrap with ErrAuditLoss regardless of whether executePurchase itself // failed. When execErr != nil (provider/partial error), finalizeExecution @@ -306,11 +306,14 @@ func (m *Manager) saveTerminalExecution(ctx context.Context, exec *config.Purcha // outcome-unknown one. exec.NoSubmission = true } + // Guarded on "running": the executor owns the row only while it holds the + // claim, so a row the reaper failed or an operator canceled in the meantime + // is never silently overwritten (issue #829). if exec.PlanID == "" || !provenNotSubmitted { - return m.config.SavePurchaseExecution(ctx, exec) + return m.config.SaveExecutionIfStatus(ctx, exec, executorOwnedStatuses) } return m.config.WithTx(ctx, func(tx pgx.Tx) error { - if err := m.config.SavePurchaseExecutionTx(ctx, tx, exec); err != nil { + if err := m.config.SaveExecutionIfStatusTx(ctx, tx, exec, executorOwnedStatuses); err != nil { return err } return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, exec.ExecutionID) @@ -547,7 +550,7 @@ func (m *Manager) claimAndRedrive(ctx context.Context, exec *config.PurchaseExec // on the same approved row. The CAS transitions "approved" -> "running"; // only the winner proceeds. // System-initiated: recovery sweep passes nil so transitioned_by = NULL. - claimed, claimErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved"}, "running", nil) + claimed, claimErr := m.claimRunning(ctx, exec.ExecutionID, []string{"approved"}, nil) if claimErr != nil { // ErrNotFound: row vanished between SELECT and CAS - benign race. // ErrExecutionNotInExpectedStatus: another sweep or the original run diff --git a/internal/purchase/notified_run_now_test.go b/internal/purchase/notified_run_now_test.go index 23b08007..35b017de 100644 --- a/internal/purchase/notified_run_now_test.go +++ b/internal/purchase/notified_run_now_test.go @@ -25,7 +25,7 @@ func TestRunNow_NotifiedRowReachesArmedRedriveGuard(t *testing.T) { exec := armedExecution("azure", "savingsplans", 1, "notified") mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll) store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, runNowFrom, "approved") + expectClaim(store, exec, runNowFrom) _, _, err := mgr.RunPlannedPurchaseNow(context.Background(), exec.ExecutionID, "operator@example.com", nil) @@ -43,7 +43,7 @@ func TestRunNow_NotifiedRowReachesStaleAzurePricingRefusal(t *testing.T) { store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}). Return([]config.RecommendationRecord{repriced}, nil).Maybe() store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe() - expectClaim(store, exec, runNowFrom, "approved") + expectClaim(store, exec, runNowFrom) _, _, err := mgr.RunPlannedPurchaseNow(context.Background(), exec.ExecutionID, "operator@example.com", nil) diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index 6eec5f57..c8889e9c 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -68,6 +68,9 @@ type rampStepFixture struct { mu sync.Mutex purchased []string // one entry per commitment that reached the fake cloud + + // onPurchase, when set, runs inside each fake provider call (issue #829 tests). + onPurchase func() } // newRampStepFixture builds the fixture. accountA is credential-resolvable from @@ -154,6 +157,9 @@ func newRampStepFixture(ctx context.Context, t *testing.T) *rampStepFixture { f.mu.Lock() f.purchased = append(f.purchased, opts.IdempotencyToken) f.mu.Unlock() + if f.onPurchase != nil { + f.onPurchase() + } }). Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil).Maybe() diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go index 4706be6a..e844c297 100644 --- a/internal/purchase/reaper.go +++ b/internal/purchase/reaper.go @@ -226,7 +226,15 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r } transitioned.Error = fmt.Sprintf("reaped after %dm in %s state — executor did not complete%s", ageMinutes, prevStatus, safeMsg) - if saveErr := m.config.SavePurchaseExecution(ctx, transitioned); saveErr != nil { + if saveErr := m.config.SaveExecutionIfStatus(ctx, transitioned, []string{failedStatus}); saveErr != nil { + if errors.Is(saveErr, config.ErrExecutionNotInExpectedStatus) { + // The executor finished late and recovered the row (completed), so + // the annotation must not overwrite it. The reap itself succeeded. + logging.Infof("purchase reaper: execution %s left failed before the reap annotation; leaving it as is: %v", + exec.ExecutionID, saveErr) + result.Reaped++ + return + } logging.Errorf("purchase reaper: failed to persist canonical error for execution %s (already flipped to failed): %v", exec.ExecutionID, saveErr) result.Errored++ diff --git a/internal/purchase/recover_reaped.go b/internal/purchase/recover_reaped.go new file mode 100644 index 00000000..8ee98503 --- /dev/null +++ b/internal/purchase/recover_reaped.go @@ -0,0 +1,48 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// recoveredAfterReapNote is the executor's own audit note for a row it had to +// recover; it does not depend on the reaper's annotation, which may land after +// or never. +const recoveredAfterReapNote = "recovered after reap; provider purchase succeeded" + +// saveOrRecoverTerminal persists the finalized row guarded on "running". When +// the guard fails because the reaper already flipped the row to failed while the +// provider call went on to succeed, the outcome is real money: it recovers the +// row to completed/partially_completed (unless a retry successor is already +// linked) so the ramp can advance and the commitment is not re-bought. Every +// other guard failure is returned as-is for the caller to wrap as audit loss. +func (m *Manager) saveOrRecoverTerminal(ctx context.Context, exec *config.PurchaseExecution, execErr error) error { + err := m.saveTerminalExecution(ctx, exec, execErr) + if err == nil || !errors.Is(err, config.ErrExecutionNotInExpectedStatus) || !purchasedOutcome(exec) { + return err + } + exec.Error = appendErrNote(exec.Error, recoveredAfterReapNote) + recErr := m.config.RecoverReapedExecution(ctx, exec) + switch { + case recErr == nil: + logging.Warnf("purchase[%s]: terminal save found the row reaped; recovered to %s", exec.ExecutionID, exec.Status) + return nil + case errors.Is(recErr, config.ErrExecutionRetryLinked): + return fmt.Errorf("POSSIBLE DOUBLE BUY, operator review needed: purchase succeeded but the reaped row already has a retry successor: %w", recErr) + default: + return fmt.Errorf("%w; recovery failed: %w", err, recErr) + } +} + +// purchasedOutcome reports whether the finalized row records real purchases. +// finalizeExecution yields completed only from a nil execErr and +// partially_completed only from a partial sentinel, so either status with +// recommendations means something was bought. It cannot test Purchased on the +// recs: a multi-account root never has them set, only its per-account copies do. +func purchasedOutcome(exec *config.PurchaseExecution) bool { + return (exec.Status == "completed" || exec.Status == "partially_completed") && len(exec.Recommendations) > 0 +} diff --git a/internal/purchase/scheduled_fire.go b/internal/purchase/scheduled_fire.go index 6dfc5275..4340593f 100644 --- a/internal/purchase/scheduled_fire.go +++ b/internal/purchase/scheduled_fire.go @@ -74,11 +74,11 @@ func (m *Manager) FireScheduledDelayedPurchases(ctx context.Context) (*FireResul // (true, false) on success, (false, true) when the CAS was lost to a // concurrent revoke, or (false, false) on a real error. func (m *Manager) fireOneDue(ctx context.Context, exec *config.PurchaseExecution) (fired, raceLost bool) { - // CAS: scheduled -> approved. If this fails with ErrExecutionNotInExpectedStatus + // CAS: scheduled -> running (the executor claim). If this fails with ErrExecutionNotInExpectedStatus // the revoke handler already transitioned the row to "canceled" — that is // not an error, just a CAS race loss. // Scheduler-initiated fire: no human session UUID, so transitioned_by = NULL. - updated, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"scheduled"}, "approved", nil) + updated, err := m.claimRunning(ctx, exec.ExecutionID, []string{"scheduled"}, nil) if err != nil { if errors.Is(err, config.ErrExecutionNotInExpectedStatus) || errors.Is(err, config.ErrNotFound) { logging.Infof("fireOneDue[%s]: CAS lost (execution already transitioned by another actor)", exec.ExecutionID) @@ -91,7 +91,7 @@ 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.SavePurchaseExecution(ctx, updated); saveErr != nil { + 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) diff --git a/internal/purchase/scheduled_fire_test.go b/internal/purchase/scheduled_fire_test.go index 16a96b40..8e2bb717 100644 --- a/internal/purchase/scheduled_fire_test.go +++ b/internal/purchase/scheduled_fire_test.go @@ -82,7 +82,7 @@ func TestFireScheduledDelayedPurchases_CASLostToRevokeClassifiedAsRaceLost(t *te row := dueExec("exec-revoked") store.On("GetScheduledExecutionsDue", ctx). Return([]config.PurchaseExecution{row}, nil) - store.On("TransitionExecutionStatus", ctx, "exec-revoked", []string{"scheduled"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-revoked", []string{"scheduled"}, "running", (*string)(nil)). Return(nil, fmt.Errorf("%w: execution exec-revoked cannot transition from %q to %q", config.ErrExecutionNotInExpectedStatus, "cancelled", "approved")) @@ -109,7 +109,7 @@ func TestFireScheduledDelayedPurchases_RowVanishedTreatedAsRaceLost(t *testing.T row := dueExec("exec-gone") store.On("GetScheduledExecutionsDue", ctx). Return([]config.PurchaseExecution{row}, nil) - store.On("TransitionExecutionStatus", ctx, "exec-gone", []string{"scheduled"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-gone", []string{"scheduled"}, "running", (*string)(nil)). Return(nil, fmt.Errorf("%w: execution exec-gone", config.ErrNotFound)) mgr := newFireManager(store) @@ -133,7 +133,7 @@ func TestFireScheduledDelayedPurchases_HardDBErrorClassifiedAsErrored(t *testing row := dueExec("exec-dberr") store.On("GetScheduledExecutionsDue", ctx). Return([]config.PurchaseExecution{row}, nil) - store.On("TransitionExecutionStatus", ctx, "exec-dberr", []string{"scheduled"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-dberr", []string{"scheduled"}, "running", (*string)(nil)). Return(nil, fmt.Errorf("connection reset by peer")) mgr := newFireManager(store) @@ -192,7 +192,7 @@ func TestFireScheduledDelayedPurchases_EndToEnd(t *testing.T) { approved := row approved.Status = "approved" - store.On("TransitionExecutionStatus", ctx, "exec-scheduled-fire", []string{"scheduled"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-scheduled-fire", []string{"scheduled"}, "running", (*string)(nil)). Return(&approved, nil) store.On("GetPurchasePlan", ctx, "plan-1").Return(plan, nil) @@ -203,11 +203,13 @@ func TestFireScheduledDelayedPurchases_EndToEnd(t *testing.T) { store.On("CompletePlanStep", ctx, "plan-1", 1).Return(nil) var savedFinal *config.PurchaseExecution - store.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error { - c := *e - savedFinal = &c - return nil - } + // The audit stamp and the terminal save are both guarded on running; an + // unguarded SavePurchaseExecution would be an unexpected call here. + store.On("SaveExecutionIfStatus", ctx, mock.Anything, []string{"running"}). + Run(func(args mock.Arguments) { + c := *args.Get(1).(*config.PurchaseExecution) + savedFinal = &c + }).Return(nil) store.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil) mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) @@ -270,7 +272,7 @@ func TestFireScheduledDelayedPurchases_AuditGapAfterCASStillFires(t *testing.T) approved := row approved.Status = "approved" - store.On("TransitionExecutionStatus", ctx, "exec-auditgap-fire", []string{"scheduled"}, "approved", (*string)(nil)). + store.On("TransitionExecutionStatus", ctx, "exec-auditgap-fire", []string{"scheduled"}, "running", (*string)(nil)). Return(&approved, nil) store.On("GetPurchasePlan", ctx, "plan-1").Return(plan, nil) diff --git a/internal/purchase/single_claim_integration_test.go b/internal/purchase/single_claim_integration_test.go new file mode 100644 index 00000000..44746326 --- /dev/null +++ b/internal/purchase/single_claim_integration_test.go @@ -0,0 +1,211 @@ +//go:build integration + +package purchase + +// Real PostgreSQL plus a counting fake provider (labeled fixture; nothing here +// reaches a cloud API). Issue #829: every executor claims the row "running" +// before any provider call, so one row is bought at most once. + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func claimRaceRow(ctx context.Context, t *testing.T, f *rampStepFixture, status string) *config.PurchaseExecution { + t.Helper() + f.repairAccounts(ctx, t) + row := &config.PurchaseExecution{ + PlanID: f.plan.ID, ExecutionID: uuid.New().String(), Status: status, StepNumber: 3, + ScheduledDate: time.Now(), CloudAccountID: &f.accounts[0].ID, + Recommendations: []config.RecommendationRecord{withAccount(rampStepRecommendation()[0], f.accounts[0].ID)}, + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, row)) + return row +} + +func rowStatus(ctx context.Context, t *testing.T, f *rampStepFixture, id string) *config.PurchaseExecution { + t.Helper() + row, err := f.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + return row +} + +// Both orderings of the approve/claimAndExecute race: the second executor loses +// the claim and never reaches the provider. Before the fix the approve path ran +// an "approved" row unclaimed, so the second ordering bought it twice. +func TestSingleClaim_BothOrderingsBuyOnce(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + global, err := f.store.GetGlobalConfig(ctx) + require.NoError(t, err) + require.True(t, global.AutoPurchaseEnabled) + + t.Run("claimAndExecute then approve", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, "notified") + before := f.purchaseCount() + claimed, execErr := f.manager.claimAndExecute(ctx, rowStatusCopy(row)) + require.NoError(t, execErr) + require.True(t, claimed) + _, _, err := f.manager.ApproveAndExecute(ctx, row.ExecutionID, "approver@example.invalid", nil) + require.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + assert.Equal(t, 1, f.purchaseCount()-before) + assert.Equal(t, "completed", rowStatus(ctx, t, f, row.ExecutionID).Status) + }) + + t.Run("approve then claimAndExecute", func(t *testing.T) { + row := claimRaceRow(ctx, t, f, "notified") + before := f.purchaseCount() + _, _, err := f.manager.ApproveAndExecute(ctx, row.ExecutionID, "approver@example.invalid", nil) + require.NoError(t, err) + claimed, execErr := f.manager.claimAndExecute(ctx, rowStatusCopy(row)) + require.NoError(t, execErr) + assert.False(t, claimed, "the approved-and-run row must not be claimable again") + assert.Equal(t, 1, f.purchaseCount()-before) + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, "completed", got.Status) + require.NotNil(t, got.ApprovedBy, "the audit stamp must survive and not revert the status") + }) +} + +func rowStatusCopy(r *config.PurchaseExecution) *config.PurchaseExecution { c := *r; return &c } + +// N goroutines, every executor that may claim the starting status: exactly one +// provider call and one terminal write. +func TestSingleClaim_ConcurrentExecutorsBuyOnce(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + const n = 4 + cases := map[string][]func(row *config.PurchaseExecution){} + approve := func(row *config.PurchaseExecution) { + _, _, _ = f.manager.ApproveAndExecute(ctx, row.ExecutionID, "a@example.invalid", nil) + } + direct := func(row *config.PurchaseExecution) { + _, _, _ = f.manager.DirectExecute(ctx, row.ExecutionID, "a@example.invalid", nil) + } + cron := func(row *config.PurchaseExecution) { _, _ = f.manager.claimAndExecute(ctx, rowStatusCopy(row)) } + redrive := func(row *config.PurchaseExecution) { _, _ = f.manager.claimAndRedrive(ctx, rowStatusCopy(row)) } + fire := func(row *config.PurchaseExecution) { f.manager.fireOneDue(ctx, rowStatusCopy(row)) } + cases["notified"] = []func(*config.PurchaseExecution){approve, direct, cron} + cases["approved"] = []func(*config.PurchaseExecution){cron, redrive} + cases["scheduled"] = []func(*config.PurchaseExecution){fire} + + for status, executors := range cases { + t.Run(status, func(t *testing.T) { + row := claimRaceRow(ctx, t, f, status) + before := f.purchaseCount() + var wg sync.WaitGroup + for i := 0; i < n; i++ { + for _, run := range executors { + wg.Add(1) + go func() { defer wg.Done(); run(row) }() + } + } + wg.Wait() + assert.Equal(t, 1, f.purchaseCount()-before, "exactly one provider call for one row") + assert.Equal(t, "completed", rowStatus(ctx, t, f, row.ExecutionID).Status) + }) + } +} + +// reapingStore lets the test hold the reaper's annotation until the executor +// has finished, producing the order: reaper CAS -> executor recovery -> reaper +// annotation. +type reapingStore struct { + *config.PostgresStore + release chan struct{} +} + +func (r *reapingStore) SaveExecutionIfStatus(ctx context.Context, e *config.PurchaseExecution, expected []string) error { + if len(expected) == 1 && expected[0] == "failed" { + <-r.release + } + return r.PostgresStore.SaveExecutionIfStatus(ctx, e, expected) +} + +// The reaper fails the row mid-run, the provider call still succeeds: the +// executor recovers the row to completed with its own note, advances the ramp, +// and the reaper's late annotation neither overwrites it nor counts as Errored. +// The fan-out case is the realistic one: the root row carries no purchased recs +// of its own, only its per-account copies do. +func TestSingleClaim_ReaperInterleavingRecoversPurchasedRow(t *testing.T) { + for _, tc := range []struct { + name string + purchases int + newRow func(ctx context.Context, t *testing.T, f *rampStepFixture) *config.PurchaseExecution + }{ + {"single account", 1, func(ctx context.Context, t *testing.T, f *rampStepFixture) *config.PurchaseExecution { + return claimRaceRow(ctx, t, f, "notified") + }}, + {"multi-account root", 3, func(ctx context.Context, t *testing.T, f *rampStepFixture) *config.PurchaseExecution { + f.repairAccounts(ctx, t) + return f.saveRootExecution(ctx, t, 3) + }}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + row := tc.newRow(ctx, t, f) + stepBefore := f.currentStep(ctx, t) + rs := &reapingStore{PostgresStore: f.store, release: make(chan struct{})} + f.manager.config = rs + + result := &ReapResult{} + reaped := make(chan struct{}) + var fired atomic.Bool + f.onPurchase = func() { // fires inside the first provider call + if fired.Swap(true) { + return + } + stuck := rowStatusCopy(row) + stuck.Status = "running" + go func() { defer close(reaped); f.manager.reapOne(ctx, stuck, 10*time.Minute, time.Now(), result) }() + require.Eventually(t, func() bool { return rowStatus(ctx, t, f, row.ExecutionID).Status == "failed" }, 5*time.Second, 10*time.Millisecond) + } + + _, _, err := f.manager.ApproveAndExecute(ctx, row.ExecutionID, "a@example.invalid", nil) + require.NoError(t, err) + close(rs.release) + <-reaped + + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, "completed", got.Status) + assert.Contains(t, got.Error, recoveredAfterReapNote) + assert.Equal(t, tc.purchases, f.purchaseCount()) + assert.Equal(t, 1, result.Reaped) + assert.Zero(t, result.Errored) + assert.Equal(t, stepBefore+1, f.currentStep(ctx, t), "recovery must still advance the ramp") + }) + } +} + +// Deterministic replay of the #829 race: while the approve path is inside its +// provider call, another executor (cron/SQS) tries to claim the same row. On the +// old code the row sat "approved" and the second executor bought it again. +func TestSingleClaim_ExecutorArrivingMidApproveCannotBuyAgain(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + row := claimRaceRow(ctx, t, f, "notified") + var fired atomic.Bool + var midClaimed bool + f.onPurchase = func() { + if fired.Swap(true) { + return + } + c := rowStatusCopy(row) + c.Status = "approved" + midClaimed, _ = f.manager.claimAndExecute(ctx, c) + } + _, _, err := f.manager.ApproveAndExecute(ctx, row.ExecutionID, "a@example.invalid", nil) + require.NoError(t, err) + assert.False(t, midClaimed, "the row is running, so a second executor loses the claim") + assert.Equal(t, 1, f.purchaseCount()) + assert.Equal(t, "completed", rowStatus(ctx, t, f, row.ExecutionID).Status) +}