From 40b0756cf92d641c090ca00cc904aed010229f8a Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 26 May 2026 01:53:22 +0200 Subject: [PATCH 1/5] fix(purchases): AWS-scoped auto-re-drive in RecoverStrandedApprovals (partial #632) Refs #632 (full Azure inclusion blocked by #721 -> #639). AWS-only strands now re-drive idempotently; mixed/Azure strands fall through to safe-fail-and-Retry. - Add allRecsAWS predicate: true when every rec has Provider=="aws" or Provider=="" (empty = legacy pre-multi-cloud rows). Returns false for an empty Recommendations slice so zero-rec executions safe-fail. - Add AWS-only + non-empty ExecutionID branch in RecoverStrandedApprovals: calls executeAndFinalize directly, reusing the existing idempotency tokens derived via DeriveIdempotencyToken(exec.ExecutionID, i) at execution.go:428. AWS dedupes on the same token, so a second call is safe even if the first created a commitment. - Mixed/Azure/GCP or legacy rows without a stable ExecutionID continue to use the atomic TransitionExecutionStatus->failed path. - Update existing FailsStrandedRow test to use an Azure rec so it stays on the safe-fail path (test intent preserved, provider changed). - Add 4 new tests: AWSOnlyRedrives, MixedAWSAzureSafeFails, PureAzureSafeFails, LegacyNoExecutionIDSafeFails. --- internal/purchase/manager.go | 143 ++++++++++++++----- internal/purchase/manager_test.go | 222 ++++++++++++++++++++++++++++-- 2 files changed, 318 insertions(+), 47 deletions(-) diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index c814fa038..551bad369 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -175,22 +175,92 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE return execErr } +// allRecsAWS reports whether every recommendation in the execution targets AWS. +// Empty provider ("") is treated as AWS because pre-multi-cloud rows predate the +// provider field and are all AWS. An execution with no recommendations returns +// false so it falls through to the safe-fail path (nothing to re-drive anyway). +// +// Azure and GCP recs are excluded: Azure re-drive idempotency is blocked by issue +// #721 (#639), and GCP commitments are currently flagged unsupported (#640). Only +// call executeAndFinalize on an execution when this predicate returns true AND the +// execution has a non-empty ExecutionID (needed for DeriveIdempotencyToken). +func allRecsAWS(exec *config.PurchaseExecution) bool { + if len(exec.Recommendations) == 0 { + return false + } + for _, rec := range exec.Recommendations { + if rec.Provider != "" && rec.Provider != "aws" { + return false + } + } + return true +} + +// safeFail atomically transitions a stranded execution to "failed" and stamps a +// recovery error on it. It is extracted from RecoverStrandedApprovals to keep +// that function's cyclomatic complexity within the gocyclo:10 limit. +// +// Returns (true, nil) when the row was successfully transitioned to "failed". +// Returns (false, nil) when TransitionExecutionStatus fails but the row has +// already left "approved" (benign race - the original run completed late; +// not counted as a recovery since no action was taken here). +// Returns (false, err) when a real store failure occurs. +func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) (bool, error) { + logging.Errorf("Recovering stranded approved execution %s (approved but never finalized; failing it for visibility)", exec.ExecutionID) + + updated, txErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved"}, "failed") + if txErr != nil { + // Distinguish benign races (row already left the "approved" + // state - concurrent sweep handled it, or the original run + // finished after the LIST snapshot) from real store + // failures (DB unreachable, query syntax error). A real + // store failure must fail the sweep so a transient DB + // outage does not silently under-recover. We probe the + // current row state via GetExecutionByID: a clean read + // with Status != "approved" confirms the race; any other + // outcome (read error, still-approved row) is a real + // failure worth propagating. + current, getErr := m.config.GetExecutionByID(ctx, exec.ExecutionID) + if getErr == nil && current != nil && current.Status != "approved" { + logging.Warnf("Skipping recovery of %s (already transitioned out of approved): %v", exec.ExecutionID, txErr) + return false, nil + } + return false, fmt.Errorf("failed to transition stranded execution %s to failed: %w", exec.ExecutionID, txErr) + } + + 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). Verify on the cloud provider that no commitment was created, then Retry." + if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil { + // 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". + logging.Errorf("AUDIT GAP: failed to stamp recovery error on %s: %v", exec.ExecutionID, saveErr) + } + return true, nil +} + // RecoverStrandedApprovals finds executions stuck in the "approved" status past -// staleApprovedThreshold and drives them into a terminal "failed" state so an -// approved row can never sit permanently stranded with no owner (issue #632). +// staleApprovedThreshold and either re-drives them idempotently (AWS-only +// executions with a durable ExecutionID) or drives them into a terminal "failed" +// state (mixed/Azure/GCP or legacy rows without a stable ExecutionID). // -// It deliberately does NOT re-run the purchase: there is no idempotency token on -// commitment creation (EC2 PurchaseReservedInstancesOffering sets no ClientToken; -// CreateSavingsPlan has none), so an automatic re-drive of a row that was -// interrupted *after* AWS created the commitment but *before* the row persisted -// would double-purchase. Failing the row makes it visible in the History view -// (which surfaces failed rows) and Retry-able by an operator who has confirmed -// the AWS-side state, instead of requiring a manual DB edit. +// AWS-only path (issue #632 Option 5): all AWS executors derive a deterministic +// per-rec idempotency token via common.DeriveIdempotencyToken(exec.ExecutionID, i) +// at purchase time (execution.go:428). Re-driving with the same ExecutionID +// produces the same token, so AWS dedupes the second call and no double-purchase +// occurs. The row transitions from "approved" directly to "completed" (or +// "failed"/"partially_completed" on a genuine error), bypassing the manual Retry +// step required by the old safe-fail path. // -// The transition is atomic: TransitionExecutionStatus only flips rows still in -// "approved", so if the original run finally completes between the stale SELECT -// and this UPDATE, the transition is a no-op and the genuine "completed" status -// is preserved. +// Safe-fail path (mixed/Azure/GCP/legacy): Azure two-step idempotency is blocked +// by issue #721; GCP is flagged unsupported (#640). Executions without a stable +// ExecutionID cannot derive tokens safely. These fall through to the original +// behaviour: the row is atomically transitioned to "failed" so it surfaces in +// History and can be Retry-ed by an operator after confirming the cloud-side state. +// +// The transition in the safe-fail path is atomic: TransitionExecutionStatus only +// flips rows still in "approved", so if the original run finally completes between +// the stale SELECT and this UPDATE, the transition is a no-op and the genuine +// "completed" status is preserved. func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) { stranded, err := m.config.GetStaleApprovedExecutions(ctx, staleApprovedThreshold) if err != nil { @@ -200,36 +270,33 @@ func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) { recovered := 0 for i := range stranded { exec := &stranded[i] - logging.Errorf("Recovering stranded approved execution %s (approved but never finalized; failing it for visibility)", exec.ExecutionID) - - updated, txErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved"}, "failed") - if txErr != nil { - // Distinguish benign races (row already left the "approved" - // state — concurrent sweep handled it, or the original run - // finished after the LIST snapshot) from real store - // failures (DB unreachable, query syntax error). A real - // store failure must fail the sweep so a transient DB - // outage does not silently under-recover. We probe the - // current row state via GetExecutionByID: a clean read - // with Status != "approved" confirms the race; any other - // outcome (read error, still-approved row) is a real - // failure worth propagating. - current, getErr := m.config.GetExecutionByID(ctx, exec.ExecutionID) - if getErr == nil && current != nil && current.Status != "approved" { - logging.Warnf("Skipping recovery of %s (already transitioned out of approved): %v", exec.ExecutionID, txErr) + + // AWS-only re-drive path (issue #632 Option 5): all AWS executors honour + // opts.IdempotencyToken via DeriveIdempotencyToken(exec.ExecutionID, i), + // so a second call with the same ExecutionID is a safe no-op on the AWS + // side. The ExecutionID must be non-empty to derive a unique token; an + // empty ID would map every legacy row to the same token set. + if allRecsAWS(exec) && exec.ExecutionID != "" { + logging.Infof("Recovering stranded AWS-only execution %s via idempotent re-drive (issue #632)", exec.ExecutionID) + if driveErr := m.executeAndFinalize(ctx, exec); driveErr != nil { + logging.Errorf("Re-drive of stranded execution %s failed: %v", exec.ExecutionID, driveErr) + // A re-drive error is not fatal to the sweep: executeAndFinalize + // already called finalizeExecution + SavePurchaseExecution, so + // the row is in a terminal state. Log and continue. continue } - return recovered, fmt.Errorf("failed to transition stranded execution %s to failed: %w", exec.ExecutionID, txErr) + recovered++ + continue } - 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). Verify on the cloud provider that no commitment was created, then Retry." - if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil { - // 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". - logging.Errorf("AUDIT GAP: failed to stamp recovery error on %s: %v", exec.ExecutionID, saveErr) + // Safe-fail path for mixed/Azure/GCP/legacy executions. + counted, failErr := m.safeFail(ctx, exec) + if failErr != nil { + return recovered, failErr + } + if counted { + recovered++ } - recovered++ } return recovered, nil diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index c70fd37d8..53c10210e 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -318,12 +318,12 @@ func TestManager_ProcessScheduledPurchases_ExecutionFails(t *testing.T) { } // TestManager_RecoverStrandedApprovals_FailsStrandedRow is the regression test -// for issue #632: an execution flipped to "approved" whose synchronous purchase -// run was interrupted (Lambda timeout / cold-start eviction / panic) before it -// finalized must NOT stay permanently "approved". The recovery sweep drives it -// into a terminal "failed" state with a clear error and — crucially — does NOT -// re-run the purchase (no provider/service-client calls), so there is no -// double-purchase even though commitment creation has no idempotency token. +// for issue #632 safe-fail path: an Azure execution flipped to "approved" whose +// synchronous purchase run was interrupted before it finalized must NOT stay +// permanently "approved". Azure re-drive idempotency is blocked by issue #721, so +// the recovery sweep drives the row into a terminal "failed" state with a clear +// error and does NOT re-run the purchase (no provider/service-client calls), +// eliminating any double-purchase risk for non-AWS providers. func TestManager_RecoverStrandedApprovals_FailsStrandedRow(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) @@ -335,7 +335,8 @@ func TestManager_RecoverStrandedApprovals_FailsStrandedRow(t *testing.T) { PlanID: "plan-456", Status: "approved", Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 500.0, Selected: true, Purchased: false}, + // Azure provider: falls through to safe-fail path (issue #721 blocks re-drive). + {Provider: "azure", Service: "reservations", ResourceType: "Standard_D4s_v3", Region: "eastus", Count: 1, UpfrontCost: 500.0, Selected: true, Purchased: false}, }, } failedRow := stranded @@ -364,7 +365,7 @@ func TestManager_RecoverStrandedApprovals_FailsStrandedRow(t *testing.T) { assert.Equal(t, 1, recovered) require.NotNil(t, saved) - assert.Equal(t, "failed", saved.Status, "stranded approved row must become terminally failed, never stay approved") + assert.Equal(t, "failed", saved.Status, "stranded Azure row must become terminally failed, never stay approved") assert.NotEmpty(t, saved.Error, "the failed row must carry a clear, operator-readable error") assert.Contains(t, saved.Error, "interrupted") assert.False(t, saved.Recommendations[0].Purchased, "recovery must not mark anything purchased") @@ -398,11 +399,214 @@ func TestManager_RecoverStrandedApprovals_FreshRowUntouched(t *testing.T) { mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } +// TestManager_RecoverStrandedApprovals_AWSOnlyRedrives is the regression test for +// issue #632 Option 5: a stranded AWS-only execution with a durable ExecutionID is +// re-driven via executeAndFinalize rather than failed. All AWS executors honour +// opts.IdempotencyToken via DeriveIdempotencyToken(exec.ExecutionID, i), so the +// second call is a safe no-op on the AWS side and the row transitions directly to +// "completed" without requiring a manual Retry. +func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockEmail := new(MockEmailSender) + mockSTS := new(MockSTSClient) + mockFactory := new(MockProviderFactory) + mockProvider := new(MockProvider) + mockServiceClient := new(MockServiceClient) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-aws-stranded", + PlanID: "plan-aws-456", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200.0, Selected: true, Purchased: false}, + }, + } + + plan := &config.PurchasePlan{ + ID: "plan-aws-456", + Name: "AWS Test Plan", + RampSchedule: config.RampSchedule{ + CurrentStep: 0, + TotalSteps: 4, + }, + } + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + mockStore.On("GetPurchasePlan", ctx, "plan-aws-456").Return(plan, nil).Twice() + mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil) + mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) + var saved *config.PurchaseExecution + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). + Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }). + Return(nil) + mockStore.On("UpdatePurchasePlan", ctx, mock.AnythingOfType("*config.PurchasePlan")).Return(nil) + mockSTS.On("GetCallerIdentity", ctx, mock.AnythingOfType("*sts.GetCallerIdentityInput")).Return(&sts.GetCallerIdentityOutput{ + Account: aws.String("123456789012"), + }, nil) + + // executeSinglePurchase wraps ctx in a per-rec WithTimeout before calling + // CreateAndValidateProvider, so we must match any context, not ctx itself. + mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil) + mockProvider.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(mockServiceClient, nil) + mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")).Return(common.PurchaseResult{ + Success: true, + CommitmentID: "ri-idempotent-12345", + }, nil) + + manager := &Manager{ + config: mockStore, + email: mockEmail, + stsClient: mockSTS, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err) + assert.Equal(t, 1, recovered, "AWS-only strand must be counted as recovered after re-drive") + + require.NotNil(t, saved) + assert.Equal(t, "completed", saved.Status, "successfully re-driven AWS execution must be completed, not failed") + + // The provider was reached: the re-drive called PurchaseCommitment exactly once. + mockServiceClient.AssertCalled(t, "PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")) + // TransitionExecutionStatus to "failed" must NOT have been called - we re-drove, not failed. + mockStore.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + mockStore.AssertExpectations(t) + mockEmail.AssertExpectations(t) + mockSTS.AssertExpectations(t) +} + +// TestManager_RecoverStrandedApprovals_MixedAWSAzureSafeFails verifies that a +// stranded execution containing both AWS and Azure recommendations falls through +// to the safe-fail path rather than being re-driven. Azure re-drive idempotency +// is blocked by issue #721; a mixed execution must never be auto-re-driven. +func TestManager_RecoverStrandedApprovals_MixedAWSAzureSafeFails(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockFactory := new(MockProviderFactory) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-mixed", + PlanID: "plan-mixed", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 100.0, Selected: true}, + {Provider: "azure", Service: "reservations", ResourceType: "Standard_D4s_v3", Region: "eastus", Count: 1, UpfrontCost: 100.0, Selected: true}, + }, + } + failedRow := stranded + failedRow.Status = "failed" + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + mockStore.On("TransitionExecutionStatus", ctx, "exec-mixed", []string{"approved"}, "failed"). + Return(&failedRow, nil) + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) + + manager := &Manager{ + config: mockStore, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err) + assert.Equal(t, 1, recovered) + + // No provider call: mixed execution falls through to safe-fail, not re-driven. + mockFactory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + mockStore.AssertExpectations(t) +} + +// TestManager_RecoverStrandedApprovals_PureAzureSafeFails verifies that a stranded +// execution whose every recommendation targets Azure falls through to the safe-fail +// path and is never re-driven (issue #721 guard against future regression). +func TestManager_RecoverStrandedApprovals_PureAzureSafeFails(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockFactory := new(MockProviderFactory) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-azure", + PlanID: "plan-azure", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "azure", Service: "reservations", ResourceType: "Standard_D4s_v3", Region: "eastus", Count: 2, UpfrontCost: 400.0, Selected: true}, + }, + } + failedRow := stranded + failedRow.Status = "failed" + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + mockStore.On("TransitionExecutionStatus", ctx, "exec-azure", []string{"approved"}, "failed"). + Return(&failedRow, nil) + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) + + manager := &Manager{ + config: mockStore, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err) + assert.Equal(t, 1, recovered) + + // Safe-fail: Azure-only execution must never reach the provider. + mockFactory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + mockStore.AssertExpectations(t) +} + +// TestManager_RecoverStrandedApprovals_LegacyNoExecutionIDSafeFails verifies that +// a stranded AWS execution with an empty ExecutionID (a pre-UUID legacy row) falls +// through to the safe-fail path. DeriveIdempotencyToken("", i) would produce the +// same token for every such row, making an auto-re-drive of legacy rows unsafe. +func TestManager_RecoverStrandedApprovals_LegacyNoExecutionIDSafeFails(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockFactory := new(MockProviderFactory) + + stranded := config.PurchaseExecution{ + ExecutionID: "", // legacy row: no stable ID for token derivation + PlanID: "plan-legacy", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200.0, Selected: true}, + }, + } + failedRow := stranded + failedRow.Status = "failed" + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + mockStore.On("TransitionExecutionStatus", ctx, "", []string{"approved"}, "failed"). + Return(&failedRow, nil) + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) + + manager := &Manager{ + config: mockStore, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err) + assert.Equal(t, 1, recovered) + + // No provider call: legacy execution must not be re-driven without a stable token. + mockFactory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + mockStore.AssertExpectations(t) +} + // TestManager_RecoverStrandedApprovals_LateCompletionNotClobbered covers the // race where the original interrupted run actually finalizes between the stale // SELECT and the recovery UPDATE. TransitionExecutionStatus's atomic // WHERE status='approved' returns an error (the row is no longer approved), so -// the sweep skips it — the genuine "completed" status is preserved and the row +// the sweep skips it - the genuine "completed" status is preserved and the row // is not re-saved as failed. func TestManager_RecoverStrandedApprovals_LateCompletionNotClobbered(t *testing.T) { ctx := context.Background() From d0addd8eb54b9034881468e8434824ddb380a294 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 26 May 2026 07:04:12 +0200 Subject: [PATCH 2/5] fix(purchase): add CAS claim before AWS re-drive and treat ErrNotFound as benign in safe-fail (CR #728) Finding 1: RecoverStrandedApprovals previously called executeAndFinalize on AWS-only stranded rows without claiming ownership first. Two concurrent sweeps (or a late original completion racing the re-drive) could both call executeAndFinalize on the same approved row, risking a double-purchase. Fix: extract claimAndRedrive helper that calls TransitionExecutionStatus ("approved" -> "running") as an atomic CAS claim before the re-drive. Only the winner of the CAS proceeds; ErrNotFound and ErrExecutionNotInExpectedStatus are both treated as benign race-losses (log and skip). Real DB errors propagate. Extracting the helper also brings RecoverStrandedApprovals back within gocyclo:10. Finding 2: safeFail's TransitionExecutionStatus error path called GetExecutionByID even when the error was config.ErrNotFound, which indicates the row has already been deleted, making the follow-up probe redundant. Fix: add an early branch that returns (false, nil) immediately on ErrNotFound, skipping the unnecessary round-trip. Both fixes reuse the established TransitionExecutionStatus CAS primitive. Tests: SafeFail_ErrNotFoundIsBenign asserts no GetExecutionByID call on ErrNotFound; AWSClaimLost_NoRedrive asserts executeAndFinalize is not reached when the CAS claim is lost. Existing AWSOnlyRedrives test updated to expect the new CAS claim call (approved -> running). --- internal/purchase/manager.go | 65 +++++++++++++++--- internal/purchase/manager_test.go | 107 ++++++++++++++++++++++++++++-- 2 files changed, 159 insertions(+), 13 deletions(-) diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 551bad369..6e6956897 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -196,6 +196,48 @@ func allRecsAWS(exec *config.PurchaseExecution) bool { return true } +// claimAndRedrive atomically claims a stranded AWS-only execution (by +// transitioning its status from "approved" to "running") and then re-drives it +// via executeAndFinalize. It is extracted from RecoverStrandedApprovals to keep +// that function's cyclomatic complexity within the gocyclo:10 limit. +// +// Returns (true, nil) when the claim was won and the re-drive completed (row is +// now in a terminal state). A re-drive error is not fatal -- executeAndFinalize +// already stamped a terminal status -- so it returns (false, nil) on drive +// failure too (the failed row is still visible in History). +// Returns (false, nil) when the CAS claim is lost to a concurrent sweep or the +// row vanishes mid-flight (both are benign races). +// Returns (false, err) only on a genuine DB error during the claim step. +func (m *Manager) claimAndRedrive(ctx context.Context, exec *config.PurchaseExecution) (bool, error) { + // Atomically claim ownership before re-driving to prevent two concurrent + // sweeps (or a late original completion) from both calling executeAndFinalize + // on the same approved row. The CAS transitions "approved" -> "running"; + // only the winner proceeds. + claimed, claimErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved"}, "running") + if claimErr != nil { + // ErrNotFound: row vanished between SELECT and CAS - benign race. + // ErrExecutionNotInExpectedStatus: another sweep or the original run + // already claimed/completed this row - also benign. + if errors.Is(claimErr, config.ErrNotFound) || errors.Is(claimErr, config.ErrExecutionNotInExpectedStatus) { + logging.Warnf("Skipping re-drive of %s (CAS claim lost to concurrent worker): %v", exec.ExecutionID, claimErr) + return false, nil + } + return false, fmt.Errorf("failed to claim execution %s for re-drive: %w", exec.ExecutionID, claimErr) + } + // Update the local struct to reflect the committed DB state so + // finalizeExecution starts from "running" rather than "approved". + exec.Status = claimed.Status + logging.Infof("Recovering stranded AWS-only execution %s via idempotent re-drive (issue #632)", exec.ExecutionID) + if driveErr := m.executeAndFinalize(ctx, exec); driveErr != nil { + logging.Errorf("Re-drive of stranded execution %s failed: %v", exec.ExecutionID, driveErr) + // A re-drive error is not fatal to the sweep: executeAndFinalize already + // called finalizeExecution + SavePurchaseExecution, so the row is in a + // terminal state. Log and continue without counting as recovered. + return false, nil + } + return true, nil +} + // safeFail atomically transitions a stranded execution to "failed" and stamps a // recovery error on it. It is extracted from RecoverStrandedApprovals to keep // that function's cyclomatic complexity within the gocyclo:10 limit. @@ -210,6 +252,15 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) updated, txErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved"}, "failed") if txErr != nil { + // ErrNotFound means the row vanished between the stale SELECT and + // this CAS attempt (e.g. deleted by an operator or a concurrent + // sweep already claimed and deleted it). That is a benign race-loss: + // there is nothing left to fail, and the caller should not be + // charged with an error. + if errors.Is(txErr, config.ErrNotFound) { + logging.Warnf("Skipping recovery of %s (row no longer exists, benign race-loss): %v", exec.ExecutionID, txErr) + return false, nil + } // Distinguish benign races (row already left the "approved" // state - concurrent sweep handled it, or the original run // finished after the LIST snapshot) from real store @@ -277,15 +328,13 @@ func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) { // side. The ExecutionID must be non-empty to derive a unique token; an // empty ID would map every legacy row to the same token set. if allRecsAWS(exec) && exec.ExecutionID != "" { - logging.Infof("Recovering stranded AWS-only execution %s via idempotent re-drive (issue #632)", exec.ExecutionID) - if driveErr := m.executeAndFinalize(ctx, exec); driveErr != nil { - logging.Errorf("Re-drive of stranded execution %s failed: %v", exec.ExecutionID, driveErr) - // A re-drive error is not fatal to the sweep: executeAndFinalize - // already called finalizeExecution + SavePurchaseExecution, so - // the row is in a terminal state. Log and continue. - continue + counted, driveErr := m.claimAndRedrive(ctx, exec) + if driveErr != nil { + return recovered, driveErr + } + if counted { + recovered++ } - recovered++ continue } diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index 53c10210e..a1724c731 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -3,6 +3,7 @@ package purchase import ( "context" "errors" + "fmt" "testing" "time" @@ -405,6 +406,10 @@ func TestManager_RecoverStrandedApprovals_FreshRowUntouched(t *testing.T) { // opts.IdempotencyToken via DeriveIdempotencyToken(exec.ExecutionID, i), so the // second call is a safe no-op on the AWS side and the row transitions directly to // "completed" without requiring a manual Retry. +// +// The CAS claim (approved -> running) is expected before the re-drive call; only +// the winner of this CAS proceeds to executeAndFinalize, preventing concurrent +// sweeps from double-purchasing. func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) @@ -414,6 +419,13 @@ func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { mockProvider := new(MockProvider) mockServiceClient := new(MockServiceClient) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + t.Cleanup(func() { mockEmail.AssertExpectations(t) }) + t.Cleanup(func() { mockSTS.AssertExpectations(t) }) + t.Cleanup(func() { mockFactory.AssertExpectations(t) }) + t.Cleanup(func() { mockProvider.AssertExpectations(t) }) + t.Cleanup(func() { mockServiceClient.AssertExpectations(t) }) + stranded := config.PurchaseExecution{ ExecutionID: "exec-aws-stranded", PlanID: "plan-aws-456", @@ -422,6 +434,8 @@ func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200.0, Selected: true, Purchased: false}, }, } + runningRow := stranded + runningRow.Status = "running" plan := &config.PurchasePlan{ ID: "plan-aws-456", @@ -434,6 +448,9 @@ func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). Return([]config.PurchaseExecution{stranded}, nil) + // CAS claim: approved -> running. The re-drive proceeds only after winning this. + mockStore.On("TransitionExecutionStatus", ctx, "exec-aws-stranded", []string{"approved"}, "running"). + Return(&runningRow, nil) mockStore.On("GetPurchasePlan", ctx, "plan-aws-456").Return(plan, nil).Twice() mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil) mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) @@ -472,11 +489,9 @@ func TestManager_RecoverStrandedApprovals_AWSOnlyRedrives(t *testing.T) { // The provider was reached: the re-drive called PurchaseCommitment exactly once. mockServiceClient.AssertCalled(t, "PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")) - // TransitionExecutionStatus to "failed" must NOT have been called - we re-drove, not failed. - mockStore.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything) - mockStore.AssertExpectations(t) - mockEmail.AssertExpectations(t) - mockSTS.AssertExpectations(t) + // The CAS claim (approved -> running) was called; "failed" transition was not. + mockStore.AssertCalled(t, "TransitionExecutionStatus", ctx, "exec-aws-stranded", []string{"approved"}, "running") + mockStore.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, "failed") } // TestManager_RecoverStrandedApprovals_MixedAWSAzureSafeFails verifies that a @@ -634,3 +649,85 @@ func TestManager_RecoverStrandedApprovals_LateCompletionNotClobbered(t *testing. mockStore.AssertExpectations(t) mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } + +// TestManager_RecoverStrandedApprovals_SafeFail_ErrNotFoundIsBenign verifies +// that when TransitionExecutionStatus returns config.ErrNotFound (row deleted +// between the stale SELECT and the CAS) the safe-fail path returns (false, nil) +// without calling GetExecutionByID. This avoids a pointless read on a row that +// no longer exists and treats the disappearance as a benign race-loss. +func TestManager_RecoverStrandedApprovals_SafeFail_ErrNotFoundIsBenign(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-vanished", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + // Azure provider falls through to safe-fail path. + {Provider: "azure", Service: "reservations", ResourceType: "Standard_D4s_v3", Region: "eastus", Count: 1, UpfrontCost: 400.0, Selected: true}, + }, + } + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + // TransitionExecutionStatus wraps "row vanished" as config.ErrNotFound. + mockStore.On("TransitionExecutionStatus", ctx, "exec-vanished", []string{"approved"}, "failed"). + Return(nil, fmt.Errorf("%w: execution exec-vanished", config.ErrNotFound)) + + manager := &Manager{config: mockStore, dashboardURL: "https://dashboard.example.com"} + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err, "ErrNotFound from CAS must not propagate as a sweep error") + assert.Equal(t, 0, recovered, "a vanished row contributes nothing to the recovery count") + + // GetExecutionByID must NOT be called: ErrNotFound already tells us the row + // is gone; the probe would be a redundant round-trip and the early branch + // avoids it. + mockStore.AssertNotCalled(t, "GetExecutionByID", mock.Anything, mock.Anything) + mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +// TestManager_RecoverStrandedApprovals_AWSClaimLost_NoRedrive verifies that when +// the CAS claim (approved -> running) fails with ErrExecutionNotInExpectedStatus +// the AWS re-drive path does NOT call executeAndFinalize. Only the sweep that +// wins the CAS should ever re-drive; the loser must skip silently. This prevents +// two overlapping sweeps from both calling executeAndFinalize and potentially +// double-purchasing. +func TestManager_RecoverStrandedApprovals_AWSClaimLost_NoRedrive(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockFactory := new(MockProviderFactory) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + t.Cleanup(func() { mockFactory.AssertExpectations(t) }) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-aws-claimed", + PlanID: "plan-aws-claimed", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300.0, Selected: true}, + }, + } + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + // A concurrent sweep has already claimed the row (status changed to "running"), + // so our CAS (approved -> running) is rejected as ErrExecutionNotInExpectedStatus. + mockStore.On("TransitionExecutionStatus", ctx, "exec-aws-claimed", []string{"approved"}, "running"). + Return(nil, fmt.Errorf("%w: execution exec-aws-claimed cannot transition from \"running\" to \"running\"", config.ErrExecutionNotInExpectedStatus)) + + manager := &Manager{ + config: mockStore, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err, "losing the CAS claim must not propagate as a sweep error") + assert.Equal(t, 0, recovered, "a row claimed by a concurrent sweep must not be counted by the loser") + + // The loser must never reach the provider - that would be a double-purchase. + mockFactory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} From d74d26ab89f0106e39fe881ff5b147f364e26a88 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 26 May 2026 07:15:13 +0200 Subject: [PATCH 3/5] fix(purchase): propagate persistence failures from executeAndFinalize (CR #728 round 2) When SavePurchaseExecution failed after the CAS-to-running CAS succeeded, claimAndRedrive returned (false, nil), silently dropping the stranded-in- running row. Add config.ErrAuditLoss sentinel and wrap the save error with it in executeAndFinalize; claimAndRedrive now propagates that sentinel so the sweep surfaces the error rather than losing the row. Benign provider/rec errors still return (false, nil) as before -- those paths already result in a persisted terminal status via finalizeExecution. --- internal/config/errors.go | 11 ++++- internal/purchase/manager.go | 20 ++++++-- internal/purchase/manager_test.go | 82 +++++++++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 5 deletions(-) diff --git a/internal/config/errors.go b/internal/config/errors.go index ff2a2de04..cc11eed8d 100644 --- a/internal/config/errors.go +++ b/internal/config/errors.go @@ -7,8 +7,17 @@ var ErrNotFound = errors.New("not found") // ErrExecutionNotInExpectedStatus is returned by TransitionExecutionStatus // when the target execution exists but its current status is not in the -// allowed `fromStatuses` set — i.e. the atomic CAS rejected because some +// allowed `fromStatuses` set -- i.e. the atomic CAS rejected because some // other writer transitioned the row first (e.g. the real executor finished // between the reaper's SELECT and CAS). Callers can use errors.Is to // distinguish this legitimate race-loss from a hard DB error. var ErrExecutionNotInExpectedStatus = errors.New("execution not in expected status") + +// ErrAuditLoss is returned (wrapped) by executeAndFinalize when the purchase +// run itself completed but the subsequent SavePurchaseExecution call failed. +// The execution is already "running" (per the CAS in claimAndRedrive) but its +// final state was never persisted -- the row is stranded in "running" until +// the next recovery sweep. Callers that silence all drive errors (e.g. +// claimAndRedrive) must propagate this sentinel so the sweep surfaces the +// persistence failure rather than silently dropping the stranded row. +var ErrAuditLoss = errors.New("audit loss: execution persistence failed after purchase") diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 6e6956897..a07f0ae2f 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -163,7 +163,11 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if err := m.config.SavePurchaseExecution(ctx, exec); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) if execErr == nil { - execErr = fmt.Errorf("audit loss: %w", err) + // Wrap with ErrAuditLoss so callers (e.g. claimAndRedrive) can + // distinguish a persistence failure -- where the row is left + // stranded in "running" -- from a benign provider/rec error + // where finalizeExecution already committed a terminal status. + execErr = fmt.Errorf("%w: %w", config.ErrAuditLoss, err) } } } @@ -230,9 +234,17 @@ func (m *Manager) claimAndRedrive(ctx context.Context, exec *config.PurchaseExec logging.Infof("Recovering stranded AWS-only execution %s via idempotent re-drive (issue #632)", exec.ExecutionID) if driveErr := m.executeAndFinalize(ctx, exec); driveErr != nil { logging.Errorf("Re-drive of stranded execution %s failed: %v", exec.ExecutionID, driveErr) - // A re-drive error is not fatal to the sweep: executeAndFinalize already - // called finalizeExecution + SavePurchaseExecution, so the row is in a - // terminal state. Log and continue without counting as recovered. + // Persistence failures (ErrAuditLoss) are non-benign: the row was CAS-ed + // to "running" but SavePurchaseExecution failed, so no terminal status was + // persisted. Propagate so the sweep surfaces the error rather than silently + // dropping a row that is now stranded in "running". + if errors.Is(driveErr, config.ErrAuditLoss) { + return false, fmt.Errorf("persistence failure re-driving execution %s (row stranded in running): %w", exec.ExecutionID, driveErr) + } + // Benign provider/rec errors: finalizeExecution already stamped a terminal + // status (failed/partially_completed) and SavePurchaseExecution succeeded. + // The row is in a terminal state; log and continue without counting as + // recovered. return false, nil } return true, nil diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index a1724c731..2dbd95aa0 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -688,6 +688,88 @@ func TestManager_RecoverStrandedApprovals_SafeFail_ErrNotFoundIsBenign(t *testin mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } +// TestManager_RecoverStrandedApprovals_AWSRedrive_PersistenceFailurePropagates is +// the regression test for the CR #728 round-2 finding: when executeAndFinalize's +// SavePurchaseExecution fails AFTER the CAS-to-running succeeded, claimAndRedrive +// must NOT return (false, nil). The row is now "running" in the DB with no +// terminal state persisted -- silently dropping the error would strand it there +// indefinitely. The sweep must surface the error (ErrAuditLoss) so the caller +// can stop processing and the next tick retries the recovery. +func TestManager_RecoverStrandedApprovals_AWSRedrive_PersistenceFailurePropagates(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockEmail := new(MockEmailSender) + mockSTS := new(MockSTSClient) + mockFactory := new(MockProviderFactory) + mockProvider := new(MockProvider) + mockServiceClient := new(MockServiceClient) + + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + t.Cleanup(func() { mockEmail.AssertExpectations(t) }) + t.Cleanup(func() { mockSTS.AssertExpectations(t) }) + t.Cleanup(func() { mockFactory.AssertExpectations(t) }) + t.Cleanup(func() { mockProvider.AssertExpectations(t) }) + t.Cleanup(func() { mockServiceClient.AssertExpectations(t) }) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-aws-persist-fail", + PlanID: "plan-aws-persist-456", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200.0, Selected: true, Purchased: false}, + }, + } + runningRow := stranded + runningRow.Status = "running" + + plan := &config.PurchasePlan{ + ID: "plan-aws-persist-456", + Name: "AWS Persist-Fail Test Plan", + RampSchedule: config.RampSchedule{ + CurrentStep: 0, + TotalSteps: 4, + }, + } + + saveErr := errors.New("DB connection lost") + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + // CAS claim: approved -> running. This succeeds -- the row is now "running". + mockStore.On("TransitionExecutionStatus", ctx, "exec-aws-persist-fail", []string{"approved"}, "running"). + Return(&runningRow, nil) + mockStore.On("GetPurchasePlan", ctx, "plan-aws-persist-456").Return(plan, nil).Once() + mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil) + mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) + // SavePurchaseExecution fails AFTER the CAS-to-running succeeded. + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). + Return(saveErr) + + mockSTS.On("GetCallerIdentity", ctx, mock.AnythingOfType("*sts.GetCallerIdentityInput")).Return(&sts.GetCallerIdentityOutput{ + Account: aws.String("123456789012"), + }, nil) + mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil) + mockProvider.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(mockServiceClient, nil) + mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")).Return(common.PurchaseResult{ + Success: true, + CommitmentID: "ri-persist-fail-12345", + }, nil) + + manager := &Manager{ + config: mockStore, + email: mockEmail, + stsClient: mockSTS, + providerFactory: mockFactory, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + // The persistence failure must surface as an error, not be silently dropped. + require.Error(t, err, "SavePurchaseExecution failure after CAS-to-running must not be silently dropped") + assert.ErrorIs(t, err, config.ErrAuditLoss, "error must wrap ErrAuditLoss so callers can classify it") + assert.Equal(t, 0, recovered, "a row that failed to persist must not be counted as recovered") +} + // TestManager_RecoverStrandedApprovals_AWSClaimLost_NoRedrive verifies that when // the CAS claim (approved -> running) fails with ErrExecutionNotInExpectedStatus // the AWS re-drive path does NOT call executeAndFinalize. Only the sweep that From 15308ee3ac502a09e9d86ee20605b87ec43c27e5 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 26 May 2026 08:06:43 +0200 Subject: [PATCH 4/5] fix(purchase): wrap ErrAuditLoss on every terminal save failure, not only on success (CR #728 round 3) The previous guard (`if execErr == nil`) only wrapped SavePurchaseExecution failures as ErrAuditLoss when executePurchase had itself succeeded. When executePurchase returned a provider error or partialPurchaseError AND SavePurchaseExecution then failed, the save failure was silently discarded: the row remained stranded in "running" with no terminal status persisted, and claimAndRedrive returned (false, nil), hiding the problem. Fix: always wrap SavePurchaseExecution failure as ErrAuditLoss. When execErr != nil, chain the original error as the innermost %w so errors.As/errors.Is callers can still reach it for diagnosis. When execErr == nil, keep the existing two-verb wrap (ErrAuditLoss + save err). Regression test: RecoverStrandedApprovals_AWSRedrive_ExecAndPersistBothFail simulates GetPurchasePlan failing (execErr != nil) followed by SavePurchaseExecution also failing. Asserts that the sweep surfaces a non-nil error, that errors.Is(err, config.ErrAuditLoss) holds, and that the original execution error remains reachable via errors.Is. --- internal/purchase/manager.go | 16 ++++--- internal/purchase/manager_test.go | 71 +++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 5 deletions(-) diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index a07f0ae2f..35b9f96b7 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -162,11 +162,17 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if !wasMultiAccount { if err := m.config.SavePurchaseExecution(ctx, exec); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) - if execErr == nil { - // Wrap with ErrAuditLoss so callers (e.g. claimAndRedrive) can - // distinguish a persistence failure -- where the row is left - // stranded in "running" -- from a benign provider/rec error - // where finalizeExecution already committed a terminal status. + // Wrap with ErrAuditLoss regardless of whether executePurchase itself + // failed. When execErr != nil (provider/partial error), finalizeExecution + // stamped a terminal status on the in-memory exec struct, but if + // SavePurchaseExecution then failed the DB row is still in "running" -- + // exactly the stranded-row scenario ErrAuditLoss signals. Preserve the + // original execErr as the innermost %w so errors.As/errors.Is can still + // reach it from callers (e.g. claimAndRedrive checking ErrAuditLoss). + if execErr != nil { + execErr = fmt.Errorf("%w: terminal save failed (%v); original execution error: %w", + config.ErrAuditLoss, err, execErr) + } else { execErr = fmt.Errorf("%w: %w", config.ErrAuditLoss, err) } } diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index 2dbd95aa0..31cdefba4 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -770,6 +770,77 @@ func TestManager_RecoverStrandedApprovals_AWSRedrive_PersistenceFailurePropagate assert.Equal(t, 0, recovered, "a row that failed to persist must not be counted as recovered") } +// TestManager_RecoverStrandedApprovals_AWSRedrive_ExecAndPersistBothFail is the +// regression test for CR #728 round-3: when executePurchase returns a non-nil +// error (e.g. partialPurchaseError or a plain provider error) AND +// SavePurchaseExecution then also fails, the old guard (execErr == nil) silently +// dropped the save failure. The row was left stranded in "running" because no +// terminal status was persisted. +// +// After the fix, any SavePurchaseExecution failure is wrapped as ErrAuditLoss +// regardless of whether executePurchase itself returned an error. The test +// verifies: +// - RecoverStrandedApprovals returns a non-nil error (not silently dropped) +// - errors.Is(err, config.ErrAuditLoss) is true (sentinel is present) +// - the original execution error (plan lookup failure) is still reachable via +// errors.As / errors.Unwrap so callers can inspect the root cause +func TestManager_RecoverStrandedApprovals_AWSRedrive_ExecAndPersistBothFail(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockEmail := new(MockEmailSender) + + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + t.Cleanup(func() { mockEmail.AssertExpectations(t) }) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-aws-both-fail", + PlanID: "plan-aws-both-fail", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200.0, Selected: true, Purchased: false}, + }, + } + runningRow := stranded + runningRow.Status = "running" + + // planErr is the error returned by executePurchase (simulates a DB blip + // during plan lookup that makes the purchase fail). + planErr := errors.New("plan DB read timeout") + saveErr := errors.New("terminal save DB connection lost") + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + // CAS claim succeeds: the row is now "running" with no terminal state yet. + mockStore.On("TransitionExecutionStatus", ctx, "exec-aws-both-fail", []string{"approved"}, "running"). + Return(&runningRow, nil) + // executePurchase calls GetPurchasePlan; make it fail so execErr != nil. + mockStore.On("GetPurchasePlan", ctx, "plan-aws-both-fail").Return(nil, planErr).Once() + // SavePurchaseExecution also fails: the row remains "running" in the DB. + mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). + Return(saveErr) + + manager := &Manager{ + config: mockStore, + email: mockEmail, + dashboardURL: "https://dashboard.example.com", + } + + recovered, err := manager.RecoverStrandedApprovals(ctx) + + // The combined failure must surface -- neither the exec error nor the save + // error may be silently swallowed. + require.Error(t, err, "both exec and save failure must not be silently dropped") + assert.ErrorIs(t, err, config.ErrAuditLoss, + "error must wrap ErrAuditLoss so claimAndRedrive surfaces it to the sweep") + + // The original execution error must remain reachable so operators can + // diagnose the root cause without reading logs. + assert.ErrorIs(t, err, planErr, + "original execution error (plan lookup failure) must be reachable via errors.Is") + + assert.Equal(t, 0, recovered, "a row with both exec and save failures must not be counted as recovered") +} + // TestManager_RecoverStrandedApprovals_AWSClaimLost_NoRedrive verifies that when // the CAS claim (approved -> running) fails with ErrExecutionNotInExpectedStatus // the AWS re-drive path does NOT call executeAndFinalize. Only the sweep that From 418c20bc33e40153ae2d576d8aac525aa925af6d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 26 May 2026 13:34:44 +0200 Subject: [PATCH 5/5] fix(purchase): short-circuit ErrExecutionNotInExpectedStatus in safeFail before read-back (CR #728 round 4) TransitionExecutionStatus already wraps the benign CAS-loss case with config.ErrExecutionNotInExpectedStatus when the row exists but its status has moved out of "approved". Re-reading via GetExecutionByID after receiving this sentinel is unnecessary: the sentinel itself already confirms the row left the target state. The probe can turn the race into a hard sweep error if the second read flakes or the row disappears between the CAS rejection and the read. Add a short-circuit returning (false, nil) for ErrExecutionNotInExpectedStatus immediately after the existing ErrNotFound short-circuit, mirroring the pattern already used by claimAndRedrive (manager.go:231) and the purchase reaper (reaper.go:186), both of which treat this sentinel as terminally benign. Regression test: TestManager_RecoverStrandedApprovals_SafeFail_ErrExecutionNotInExpectedStatusIsBenign asserts GetExecutionByID is never called and the sweep returns (0, nil). --- internal/purchase/manager.go | 10 +++++++ internal/purchase/manager_test.go | 43 +++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 35b9f96b7..d2b3414ae 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -279,6 +279,16 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) logging.Warnf("Skipping recovery of %s (row no longer exists, benign race-loss): %v", exec.ExecutionID, txErr) return false, nil } + // ErrExecutionNotInExpectedStatus means the row exists but its + // status has already moved out of "approved" (a concurrent sweep or + // the original run won the CAS race). Treat identically to the + // ErrNotFound case: nothing left to do here, no re-read needed. + // This mirrors claimAndRedrive and reaper.go, which both treat this + // sentinel as terminally benign. + if errors.Is(txErr, config.ErrExecutionNotInExpectedStatus) { + logging.Warnf("Skipping recovery of %s (row already left approved state, benign CAS race-loss): %v", exec.ExecutionID, txErr) + return false, nil + } // Distinguish benign races (row already left the "approved" // state - concurrent sweep handled it, or the original run // finished after the LIST snapshot) from real store diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index 31cdefba4..5a5aea4d7 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -688,6 +688,49 @@ func TestManager_RecoverStrandedApprovals_SafeFail_ErrNotFoundIsBenign(t *testin mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } +// TestManager_RecoverStrandedApprovals_SafeFail_ErrExecutionNotInExpectedStatusIsBenign +// verifies that when TransitionExecutionStatus returns +// config.ErrExecutionNotInExpectedStatus (the row exists but its status has +// already moved out of "approved" -- a concurrent sweep or the original run +// won the CAS) safeFail returns (false, nil) WITHOUT calling GetExecutionByID. +// Re-reading in this case is unnecessary: the sentinel already tells us the row +// is in some non-approved state. Probing can also turn the benign race into a +// hard sweep error if the second read flakes or the row disappears between the +// CAS rejection and the probe. This matches how claimAndRedrive and reaper.go +// already treat ErrExecutionNotInExpectedStatus as terminally benign. +func TestManager_RecoverStrandedApprovals_SafeFail_ErrExecutionNotInExpectedStatusIsBenign(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + + stranded := config.PurchaseExecution{ + ExecutionID: "exec-already-transitioned", + Status: "approved", + Recommendations: []config.RecommendationRecord{ + // Azure provider routes to safe-fail (not claimAndRedrive). + {Provider: "azure", Service: "reservations", ResourceType: "Standard_D4s_v3", Region: "eastus", Count: 1, UpfrontCost: 400.0, Selected: true}, + }, + } + + mockStore.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold). + Return([]config.PurchaseExecution{stranded}, nil) + // TransitionExecutionStatus signals the row is in a non-approved state. + mockStore.On("TransitionExecutionStatus", ctx, "exec-already-transitioned", []string{"approved"}, "failed"). + Return(nil, fmt.Errorf("%w: execution exec-already-transitioned status is completed", config.ErrExecutionNotInExpectedStatus)) + + manager := &Manager{config: mockStore, dashboardURL: "https://dashboard.example.com"} + + recovered, err := manager.RecoverStrandedApprovals(ctx) + require.NoError(t, err, "ErrExecutionNotInExpectedStatus from CAS must not propagate as a sweep error") + assert.Equal(t, 0, recovered, "a row already out of approved contributes nothing to the recovery count") + + // GetExecutionByID must NOT be called: ErrExecutionNotInExpectedStatus already + // tells us the row has moved on; the probe is a redundant round-trip and the + // early branch avoids it (mirroring the ErrNotFound short-circuit above it). + mockStore.AssertNotCalled(t, "GetExecutionByID", mock.Anything, mock.Anything) + mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + // TestManager_RecoverStrandedApprovals_AWSRedrive_PersistenceFailurePropagates is // the regression test for the CR #728 round-2 finding: when executeAndFinalize's // SavePurchaseExecution fails AFTER the CAS-to-running succeeded, claimAndRedrive