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 c814fa038..d2b3414ae 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -162,8 +162,18 @@ 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 { - execErr = fmt.Errorf("audit loss: %w", err) + // 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) } } } @@ -175,22 +185,161 @@ 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 +} + +// 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) + // 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 +} + +// 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 { + // 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 + } + // 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 + // 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 +349,31 @@ 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) - continue + + // 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 != "" { + counted, driveErr := m.claimAndRedrive(ctx, exec) + if driveErr != nil { + return recovered, driveErr } - return recovered, fmt.Errorf("failed to transition stranded execution %s to failed: %w", exec.ExecutionID, txErr) + if counted { + 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..5a5aea4d7 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" @@ -318,12 +319,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 +336,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 +366,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 +400,228 @@ 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. +// +// 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) + 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-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}, + }, + } + runningRow := stranded + runningRow.Status = "running" + + 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) + // 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) + 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")) + // 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 +// 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() @@ -430,3 +649,281 @@ 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_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 +// 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_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 +// 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) +}