From 13726429791e82fccc43f95886d752393161313d Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 19 Oct 2026 11:46:41 +0100 Subject: [PATCH] fix(purchase): re-drive the claimed row, not the pre-claim snapshot RecoverStrandedApprovals picks the in-place re-drive path from a SELECT, then claimAndRedrive claims the row (approved -> running) and bought the snapshot it was handed, copying only the claimed status onto it. An edit or a concurrent writer landing between the SELECT and the CAS (new recommendations, amount or account) was bought unchecked, and the terminal save then wrote the stale snapshot back over the row. claimAndRedrive now carries the whole claimed row and re-runs the re-drive gate (redrivable: every rec safe to re-drive, durable ExecutionID) on it before any provider call. A claimed row that no longer qualifies is failed the way safeFail fails a listed row in that state: the same recovery message, guarded on "running", and no no_submission stamp or suppression release, since the interrupted original run may already have bought. A failed terminal save surfaces as ErrAuditLoss. Closes #1037 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/manager.go | 37 ++-- .../redrive_claimed_row_integration_test.go | 153 ++++++++++++++ internal/purchase/redrive_claimed_row_test.go | 197 ++++++++++++++++++ internal/purchase/stranded_redrive.go | 51 +++++ 4 files changed, 425 insertions(+), 13 deletions(-) create mode 100644 internal/purchase/redrive_claimed_row_integration_test.go create mode 100644 internal/purchase/redrive_claimed_row_test.go create mode 100644 internal/purchase/stranded_redrive.go diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index dbd9579e4..c32fcd7e7 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -563,13 +563,24 @@ func awsPurchaseHasNoDuplicateGuard(service string) bool { // executeAndFinalize. It is extracted from RecoverStrandedApprovals to keep that // function's cyclomatic complexity within the gocyclo:10 limit. // +// It re-drives the row the claim returned, never the caller's pre-claim +// snapshot (issue #1037). The sweep chose the re-drive path from a SELECT, and +// an edit or a concurrent writer can change the row's recommendations, amount +// or account before the claim lands. The re-drive gate (redrivable) is therefore +// re-run on the claimed row; when the claimed row no longer qualifies it is +// failed exactly as safeFail would have failed it had the sweep read it in +// that state (failClaimedRedrive), and nothing is bought. +// // 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). +// now in a terminal state), or when the claimed row was failed because it no +// longer qualifies for a re-drive. 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. +// Returns (false, err) on a genuine DB error during the claim step, or when a +// terminal status could not be persisted (ErrAuditLoss). 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 @@ -587,9 +598,13 @@ func (m *Manager) claimAndRedrive(ctx context.Context, exec *config.PurchaseExec } 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 + // Carry the whole committed row, not just its status: the purchase below + // must buy exactly the row this claim owns. The snapshot the sweep listed + // may predate an edit that landed before the CAS (issue #1037). + *exec = *claimed + if !redrivable(exec) { + return m.failClaimedRedrive(ctx, exec) + } logging.Infof("Recovering stranded execution %s via idempotent re-drive (issue #639)", exec.ExecutionID) if driveErr := m.executeAndFinalize(ctx, exec); driveErr != nil { logging.Errorf("Re-drive of stranded execution %s failed: %v", exec.ExecutionID, driveErr) @@ -661,11 +676,7 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) return false, fmt.Errorf("failed to transition stranded execution %s to failed: %w", exec.ExecutionID, txErr) } - next := "Verify on the cloud provider that no commitment was created, then Retry." - if RedriveRefusalReason(exec) != "" { - next = "Retry is refused for this purchase because a repeat could buy a second commitment. Verify on the cloud provider whether the commitment was created; if it was not, start a new purchase." - } - updated.Error = "execution was approved but its purchase run was interrupted before completing and never finalized; failed by the recovery sweep so it is not silently stuck (issue #632). " + next + updated.Error = strandedRecoveryError(exec) 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 @@ -722,7 +733,7 @@ func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) { // provider side. The ExecutionID must be non-empty so the lineage key // (or its ExecutionID fallback for legacy rows) is unique; an empty ID // would map every legacy row to the same token set. - if allRecsSafeToRedrive(exec) && exec.ExecutionID != "" { + if redrivable(exec) { counted, driveErr := m.claimAndRedrive(ctx, exec) if driveErr != nil { return recovered, driveErr diff --git a/internal/purchase/redrive_claimed_row_integration_test.go b/internal/purchase/redrive_claimed_row_integration_test.go new file mode 100644 index 000000000..071d30d3f --- /dev/null +++ b/internal/purchase/redrive_claimed_row_integration_test.go @@ -0,0 +1,153 @@ +//go:build integration + +package purchase + +// Issue #1037 on real PostgreSQL: a write lands between the stranded-approval +// sweep's SELECT and its approved -> running claim. The re-drive must buy the +// row the claim returned. Real store, real Manager, counting fake provider +// (labeled fixture; nothing here reaches a cloud API). + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// editAfterListStore runs the real stale-approved SELECT, then holds every +// caller until afterList has run once all n sweeps have listed, so the edit +// lands after each sweep's snapshot and before any claim. +// +// The SELECT is called with a zero age instead of staleApprovedThreshold: the +// fixture row was written moments ago, and the 15-minute age filter is not +// what this test is about. Every other store call is the production one. +type editAfterListStore struct { + *config.PostgresStore + listed sync.WaitGroup + edited chan struct{} + afterList func() error + editErr error + once sync.Once +} + +func newEditAfterListStore(store *config.PostgresStore, sweeps int, afterList func() error) *editAfterListStore { + s := &editAfterListStore{PostgresStore: store, edited: make(chan struct{}), afterList: afterList} + s.listed.Add(sweeps) + return s +} + +func (s *editAfterListStore) GetStaleApprovedExecutions(ctx context.Context, _ time.Duration) ([]config.PurchaseExecution, error) { + rows, err := s.PostgresStore.GetStaleApprovedExecutions(ctx, 0) + s.listed.Done() + s.once.Do(func() { + go func() { + s.listed.Wait() + s.editErr = s.afterList() + close(s.edited) + }() + }) + <-s.edited + return rows, err +} + +// editRow rewrites the stored row's recommendations through the production +// full-row upsert, the same write an edit or a concurrent writer performs. It +// runs off the test goroutine, so it returns its error for the test to check. +func editRow(ctx context.Context, store *config.PostgresStore, id string, mutate func(*config.RecommendationRecord)) error { + row, err := store.GetExecutionByID(ctx, id) + if err != nil { + return err + } + if row.Status != "approved" { + return fmt.Errorf("row %s is %q before the edit, want approved", id, row.Status) + } + mutate(&row.Recommendations[0]) + row.TotalUpfrontCost = row.Recommendations[0].UpfrontCost + return store.SavePurchaseExecution(ctx, row) +} + +func (f *rampStepFixture) boughtCounts() []int { + f.mu.Lock() + defer f.mu.Unlock() + return append([]int(nil), f.counts...) +} + +// The row is repriced from 1 to 5 instances after four concurrent sweeps have +// listed it. Exactly one sweep wins the claim and buys the claimed 5; before +// the fix the winner bought the listed 1 and saved that snapshot back. +func TestRecoverStrandedApprovals_ConcurrentSweepsBuyClaimedRowOnce(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + row := claimRaceRow(ctx, t, f, "approved") + require.Equal(t, 1, row.Recommendations[0].Count) + + const sweeps = 4 + store := newEditAfterListStore(f.store, sweeps, func() error { + return editRow(ctx, f.store, row.ExecutionID, func(r *config.RecommendationRecord) { + r.Count = 5 + r.UpfrontCost = 1500 + }) + }) + f.manager.config = store + + var wg sync.WaitGroup + recovered := make([]int, sweeps) + errs := make([]error, sweeps) + for i := 0; i < sweeps; i++ { + wg.Add(1) + go func() { + defer wg.Done() + recovered[i], errs[i] = f.manager.RecoverStrandedApprovals(ctx) + }() + } + wg.Wait() + require.NoError(t, store.editErr) + + total := 0 + for i := range recovered { + require.NoError(t, errs[i]) + total += recovered[i] + } + assert.Equal(t, 1, total, "exactly one sweep wins the claim") + assert.Equal(t, []int{5}, f.boughtCounts(), "one provider call, for the claimed row's 5 instances") + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, "completed", got.Status) + require.Len(t, got.Recommendations, 1) + assert.Equal(t, 5, got.Recommendations[0].Count, "the terminal save must not write the listed snapshot back") + assert.InDelta(t, 1500, got.TotalUpfrontCost, 0.001) +} + +// The listed row was a safe RDS re-drive; the edit turns it into EC2, which has +// no provider duplicate guard. Nothing may be bought: the claimed row is failed +// as the safe-fail path would fail it, without the no_submission stamp. +func TestRecoverStrandedApprovals_ClaimedRowTurnedUnsafeIsFailedNotBought(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + row := claimRaceRow(ctx, t, f, "approved") + + store := newEditAfterListStore(f.store, 1, func() error { + return editRow(ctx, f.store, row.ExecutionID, func(r *config.RecommendationRecord) { + r.Service = "ec2" + r.ResourceType = "m5.large" + }) + }) + f.manager.config = store + + recovered, err := f.manager.RecoverStrandedApprovals(ctx) + require.NoError(t, store.editErr) + + require.NoError(t, err) + assert.Equal(t, 1, recovered) + assert.Equal(t, 0, f.purchaseCount(), "neither the listed RDS rec nor the claimed EC2 rec may be bought") + got := rowStatus(ctx, t, f, row.ExecutionID) + assert.Equal(t, "failed", got.Status) + assert.False(t, got.NoSubmission, "the interrupted original run may have bought: no_submission must stay false") + assert.Contains(t, got.Error, "start a new purchase") + require.Len(t, got.Recommendations, 1) + assert.Equal(t, "ec2", got.Recommendations[0].Service, "the failed row keeps the claimed recommendations") +} diff --git a/internal/purchase/redrive_claimed_row_test.go b/internal/purchase/redrive_claimed_row_test.go new file mode 100644 index 000000000..9ce40c3c3 --- /dev/null +++ b/internal/purchase/redrive_claimed_row_test.go @@ -0,0 +1,197 @@ +package purchase + +// Issue #1037: the stranded-approval re-drive must buy the row its claim +// returned, not the snapshot RecoverStrandedApprovals listed before the claim. +// Mock store and mock provider (labeled fixtures; nothing here reaches a cloud +// API). The Postgres version of the race is in +// redrive_claimed_row_integration_test.go. + +import ( + "context" + "errors" + "testing" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sts" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +const redriveExecID = "exec-redrive-1037" + +// redriveSnapshot is the row as the sweep's SELECT saw it: a stranded approved +// AWS RDS purchase of one instance, safe to re-drive. +func redriveSnapshot() config.PurchaseExecution { + return config.PurchaseExecution{ + ExecutionID: redriveExecID, + PlanID: "plan-redrive-1037", + Status: "approved", + StepNumber: 1, + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "rds", ResourceType: "db.r5.large", Region: "us-east-1", Count: 1, UpfrontCost: 200, Selected: true}, + }, + TotalUpfrontCost: 200, + } +} + +// refuseProviderCalls makes any provider construction fail AND be recorded, so +// AssertNotCalled can see it. Without an expectation the mock panics inside the +// fan-out's recover, and the attempt would never reach the call log. +func refuseProviderCalls(factory *MockProviderFactory) { + factory.On("CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything). + Return(nil, errors.New("the provider must not be reached")).Maybe() +} + +// expectRedriveClaim lists snapshot and makes the approved -> running claim +// return claimed, modeling a write that landed between the SELECT and the CAS. +func expectRedriveClaim(ctx context.Context, store *MockConfigStore, snapshot, claimed config.PurchaseExecution) { + claimed.Status = "running" + store.On("GetStaleApprovedExecutions", ctx, staleApprovedThreshold).Return([]config.PurchaseExecution{snapshot}, nil) + store.On("TransitionExecutionStatus", ctx, redriveExecID, []string{"approved"}, "running", (*string)(nil)).Return(&claimed, nil).Once() +} + +// The claimed row was repriced (1 -> 7 instances, $200 -> $1400) after the +// sweep listed it. The provider must be asked for the claimed 7, and the +// terminal row must carry the claimed recommendations. Before the fix the +// re-drive bought the listed 1 and then saved the stale snapshot over the row. +func TestRecoverStrandedApprovals_RedrivesClaimedRowNotSnapshot(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + emailSender := new(MockEmailSender) + stsClient := new(MockSTSClient) + factory := new(MockProviderFactory) + prov := new(MockProvider) + client := new(MockServiceClient) + t.Cleanup(func() { store.AssertExpectations(t) }) + t.Cleanup(func() { client.AssertExpectations(t) }) + + snapshot := redriveSnapshot() + claimed := redriveSnapshot() + claimed.Recommendations[0].Count = 7 + claimed.Recommendations[0].UpfrontCost = 1400 + claimed.TotalUpfrontCost = 1400 + expectRedriveClaim(ctx, store, snapshot, claimed) + + store.On("GetPurchasePlan", mock.Anything, snapshot.PlanID).Return(&config.PurchasePlan{ + ID: snapshot.PlanID, RampSchedule: config.RampSchedule{TotalSteps: 4}, + }, nil).Once() + store.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil) + store.On("CompletePlanStep", mock.Anything, snapshot.PlanID, 1).Return(nil) + var saved *config.PurchaseExecution + store.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")). + Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }).Return(nil) + emailSender.On("SendPurchaseConfirmation", mock.Anything, mock.AnythingOfType("email.NotificationData")).Return(nil) + stsClient.On("GetCallerIdentity", mock.Anything, mock.AnythingOfType("*sts.GetCallerIdentityInput")). + Return(&sts.GetCallerIdentityOutput{Account: aws.String("123456789012")}, nil) + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, common.ServiceRDS, "us-east-1").Return(client, nil) + var boughtCounts []int + client.On("PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")). + Run(func(args mock.Arguments) { + boughtCounts = append(boughtCounts, args.Get(1).(common.Recommendation).Count) + }). + Return(common.PurchaseResult{Success: true, CommitmentID: "ri-redrive-1037"}, nil) + + mgr := &Manager{config: store, email: emailSender, stsClient: stsClient, providerFactory: factory, dashboardURL: "https://dashboard.example.com"} + recovered, err := mgr.RecoverStrandedApprovals(ctx) + + require.NoError(t, err) + assert.Equal(t, 1, recovered) + assert.Equal(t, []int{7}, boughtCounts, "the re-drive must buy the claimed row's 7 instances, not the listed snapshot's 1") + require.NotNil(t, saved) + assert.Equal(t, "completed", saved.Status) + require.Len(t, saved.Recommendations, 1) + assert.Equal(t, 7, saved.Recommendations[0].Count, "the terminal save must not write the stale snapshot back over the claimed row") + assert.InDelta(t, 1400, saved.TotalUpfrontCost, 0.001) +} + +// The listed snapshot was safe to re-drive, but by the time the claim landed +// the row no longer qualifies: its rec became EC2 (no provider duplicate +// guard) or it lost its recommendations. Nothing may be bought, and the row is +// failed the way safeFail fails a listed row in that state: no no_submission +// stamp, because the interrupted original run may already have bought. +func TestRecoverStrandedApprovals_ClaimedRowNoLongerRedrivableIsFailed(t *testing.T) { + cases := []struct { + name string + mutate func(*config.PurchaseExecution) + wantError string + }{ + { + name: "recommendation changed to EC2", + mutate: func(e *config.PurchaseExecution) { + e.Recommendations[0].Service = "ec2" + e.Recommendations[0].ResourceType = "m5.large" + }, + wantError: "start a new purchase", + }, + { + name: "recommendations emptied", + mutate: func(e *config.PurchaseExecution) { e.Recommendations = nil }, + wantError: "then Retry", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + factory := new(MockProviderFactory) + refuseProviderCalls(factory) + t.Cleanup(func() { store.AssertExpectations(t) }) + + snapshot := redriveSnapshot() + claimed := redriveSnapshot() + tc.mutate(&claimed) + expectRedriveClaim(ctx, store, snapshot, claimed) + var saved *config.PurchaseExecution + var savedFrom []string + store.On("SaveExecutionIfStatus", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution"), mock.Anything). + Run(func(args mock.Arguments) { + saved = args.Get(1).(*config.PurchaseExecution) + savedFrom = args.Get(2).([]string) + }).Return(nil).Once() + + mgr := &Manager{config: store, providerFactory: factory, dashboardURL: "https://dashboard.example.com"} + recovered, err := mgr.RecoverStrandedApprovals(ctx) + + require.NoError(t, err) + assert.Equal(t, 1, recovered, "a failed stranded row counts as recovered, as on the safeFail path") + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + require.NotNil(t, saved) + assert.Equal(t, []string{"running"}, savedFrom, "the terminal save must be guarded on the claim this run holds") + assert.Equal(t, "failed", saved.Status) + assert.False(t, saved.NoSubmission, "the original run may have bought: no_submission must not be stamped") + assert.Contains(t, saved.Error, "interrupted before completing") + assert.Contains(t, saved.Error, tc.wantError) + assert.Equal(t, claimed.Recommendations, saved.Recommendations, "the failed row keeps the claimed recommendations") + }) + } +} + +// A terminal save failure after failing the claimed row leaves it in +// "running"; the sweep must surface that as ErrAuditLoss, not count it. +func TestRecoverStrandedApprovals_ClaimedRowFailSaveErrorIsAuditLoss(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + factory := new(MockProviderFactory) + refuseProviderCalls(factory) + t.Cleanup(func() { store.AssertExpectations(t) }) + + claimed := redriveSnapshot() + claimed.Recommendations = nil + expectRedriveClaim(ctx, store, redriveSnapshot(), claimed) + saveErr := errors.New("db connection lost") + store.On("SaveExecutionIfStatus", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution"), []string{"running"}). + Return(saveErr).Once() + + mgr := &Manager{config: store, providerFactory: factory} + recovered, err := mgr.RecoverStrandedApprovals(ctx) + + require.Error(t, err) + assert.ErrorIs(t, err, config.ErrAuditLoss) + assert.ErrorIs(t, err, saveErr) + assert.Equal(t, 0, recovered) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) +} diff --git a/internal/purchase/stranded_redrive.go b/internal/purchase/stranded_redrive.go new file mode 100644 index 000000000..0f999dd84 --- /dev/null +++ b/internal/purchase/stranded_redrive.go @@ -0,0 +1,51 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// strandedRecoveryError is the operator-facing error a stranded approved row is +// failed with when the sweep does not re-drive it. The interrupted run may +// already have bought, so it always tells the operator to check the provider. +func strandedRecoveryError(exec *config.PurchaseExecution) string { + next := "Verify on the cloud provider that no commitment was created, then Retry." + if RedriveRefusalReason(exec) != "" { + next = "Retry is refused for this purchase because a repeat could buy a second commitment. Verify on the cloud provider whether the commitment was created; if it was not, start a new purchase." + } + return "execution was approved but its purchase run was interrupted before completing and never finalized; failed by the recovery sweep so it is not silently stuck (issue #632). " + next +} + +// redrivable reports whether RecoverStrandedApprovals may re-drive exec in +// place: every recommendation is safe to re-drive and the row has a durable +// ExecutionID for the idempotency lineage. The sweep checks it on the listed +// snapshot to pick a path, and claimAndRedrive checks it again on the claimed +// row before buying (issue #1037). +func redrivable(exec *config.PurchaseExecution) bool { + return allRecsSafeToRedrive(exec) && exec.ExecutionID != "" +} + +// failClaimedRedrive fails a stranded row claimAndRedrive has claimed +// ("running") but whose claimed state no longer qualifies for a re-drive, for +// example because its recommendations were edited to an EC2 one between the +// sweep's SELECT and the claim. The outcome is the one safeFail gives a row +// listed in that state: failed with the recovery explanation. It is NOT a +// beforeSubmission refusal: the interrupted original run may already have +// bought, so no_submission is not stamped and plan suppressions are kept. +// Returns (true, nil) when the row was failed, matching safeFail's count, and +// an ErrAuditLoss error when the terminal save fails (the row would then sit +// in "running" for the reaper). +func (m *Manager) failClaimedRedrive(ctx context.Context, exec *config.PurchaseExecution) (bool, error) { + cause := errors.New(strandedRecoveryError(exec)) + logging.Errorf("Recovering stranded approved execution %s: the claimed row no longer qualifies for an in-place re-drive; failing it for visibility", exec.ExecutionID) + m.finalizeExecution(exec, cause) + if err := m.saveTerminalExecution(ctx, exec, cause); err != nil { + return false, fmt.Errorf("%w: failed to save stranded execution %s as failed after claiming it (row stranded in running): %w", + config.ErrAuditLoss, exec.ExecutionID, err) + } + return true, nil +}