From 6621970ea2f4ee655a857db0326c4eba7111e38f Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 4 Jun 2026 21:05:28 +0200 Subject: [PATCH] fix(purchases): dispatch AWS-scoped stranded-approval recovery in reap task (closes #632) PR #728 added RecoverStrandedApprovals (issue #632 Option 5: idempotent AWS re-drive of executions stranded in "approved" by an interrupted sync run) but never wired it into the runtime. The only scheduled stuck-purchase sweep was ReapStuckExecutions (#678), which unconditionally fails stranded rows for visibility, so the re-drive-to-completion path was dead code: in production a stranded AWS approval was always failed at the reap threshold rather than idempotently re-driven. Wire RecoverStrandedApprovals into the reap_stuck_purchases scheduled task, running it before the reaper so AWS-only strands (15m staleApprovedThreshold, CAS-claimed, idempotency-token guarded) complete rather than being failed, while the reaper remains the durable safety net for anything still stuck. A recovery-sweep error is logged but not propagated so it can never block the reaper. Add RecoverStrandedApprovals to PurchaseManagerInterface and the test mock, plus a regression test asserting the recovery-then-reaper ordering and that a recovery error does not stop the reaper. --- internal/server/handler.go | 32 ++++++++++++++---- internal/server/handler_test.go | 59 +++++++++++++++++++++++++++++++++ internal/server/interfaces.go | 9 +++++ internal/testutil/mocks.go | 8 +++++ 4 files changed, 101 insertions(+), 7 deletions(-) diff --git a/internal/server/handler.go b/internal/server/handler.go index 8b0b7773c..86b8256f1 100644 --- a/internal/server/handler.go +++ b/internal/server/handler.go @@ -172,17 +172,35 @@ func (app *Application) handleCleanupExpiredRecords(ctx context.Context) (map[st return result, nil } -// handleReapStuckPurchases sweeps purchase_executions stuck in -// approved/running longer than the configured threshold and flips them -// to "failed" via the existing TransitionExecutionStatus CAS. See -// internal/purchase/reaper.go + issue #678 for the full rationale and -// safety properties. +// handleReapStuckPurchases recovers and reaps stranded purchase executions. // -// The threshold is read fresh from the PURCHASE_APPROVED_REAP_AFTER env -// var on every invocation (not cached at startup) so an ops tune via +// It runs two sweeps in order: +// 1. RecoverStrandedApprovals (issue #632): for executions stranded in +// "approved" past the internal staleApprovedThreshold, idempotently +// re-drives AWS-only rows to completion so an interrupted sync run +// actually finishes, and safe-fails mixed/Azure/GCP/legacy rows. +// This must run BEFORE the reaper so AWS strands get a chance to +// complete rather than being unconditionally failed. +// 2. ReapStuckExecutions (issue #678): flips anything still stuck in +// approved/running past the configured threshold to "failed" via the +// existing TransitionExecutionStatus CAS, so no row can be permanently +// stranded. See internal/purchase/reaper.go for the safety properties. +// +// The reap threshold is read fresh from the PURCHASE_APPROVED_REAP_AFTER +// env var on every invocation (not cached at startup) so an ops tune via // Lambda env-var rotation takes effect on the next sweep without a // redeploy. +// +// A recovery-sweep error is logged but NOT propagated: it must not block +// the reaper, which is the durable safety net that guarantees stranded +// rows always reach a terminal, retry-able state. func (app *Application) handleReapStuckPurchases(ctx context.Context) (*purchase.ReapResult, error) { + if recovered, recErr := app.Purchase.RecoverStrandedApprovals(ctx); recErr != nil { + log.Printf("Stranded-approval recovery sweep error (continuing to reaper): %v", recErr) + } else if recovered > 0 { + log.Printf("Stranded-approval recovery sweep recovered %d execution(s)", recovered) + } + reapAfter := purchase.ParseReapAfterFromEnv() log.Printf("Reaping stuck purchase executions (threshold: %s)...", reapAfter) result, err := app.Purchase.ReapStuckExecutions(ctx, reapAfter) diff --git a/internal/server/handler_test.go b/internal/server/handler_test.go index aa58c8411..d094e0fed 100644 --- a/internal/server/handler_test.go +++ b/internal/server/handler_test.go @@ -152,6 +152,65 @@ func TestHandleScheduledTask(t *testing.T) { } } +// TestHandleReapStuckPurchasesRunsRecoveryFirst is the regression test for +// issue #632: the AWS-scoped re-drive sweep (RecoverStrandedApprovals, added +// by PR #728) was built and unit-tested but never dispatched at runtime, so +// stranded "approved" executions were only ever failed by the reaper instead +// of being re-driven to completion. This asserts that the reap_stuck_purchases +// task now invokes RecoverStrandedApprovals BEFORE ReapStuckExecutions, and +// that a recovery error does not block the reaper (the durable safety net). +func TestHandleReapStuckPurchasesRunsRecoveryFirst(t *testing.T) { + t.Run("recovery runs before reaper", func(t *testing.T) { + t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") + ctx := testutil.TestContext(t) + + var order []string + mockPurchase := &testutil.MockPurchaseManager{ + RecoverStrandedApprovalsFunc: func(ctx context.Context) (int, error) { + order = append(order, "recover") + return 1, nil + }, + ReapStuckExecutionsFunc: func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { + order = append(order, "reap") + return &purchase.ReapResult{}, nil + }, + } + + app := &Application{Scheduler: &testutil.MockScheduler{}, Purchase: mockPurchase} + _, err := app.HandleScheduledTask(ctx, TaskReapStuckPurchases) + testutil.AssertNoError(t, err) + + if len(order) != 2 || order[0] != "recover" || order[1] != "reap" { + t.Fatalf("expected recovery then reaper, got %v", order) + } + }) + + t.Run("recovery error does not block reaper", func(t *testing.T) { + t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") + ctx := testutil.TestContext(t) + + reaped := false + mockPurchase := &testutil.MockPurchaseManager{ + RecoverStrandedApprovalsFunc: func(ctx context.Context) (int, error) { + return 0, errors.New("recovery sweep db error") + }, + ReapStuckExecutionsFunc: func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { + reaped = true + return &purchase.ReapResult{}, nil + }, + } + + app := &Application{Scheduler: &testutil.MockScheduler{}, Purchase: mockPurchase} + _, err := app.HandleScheduledTask(ctx, TaskReapStuckPurchases) + // Recovery failure is logged, not propagated; the reaper still runs + // and the task succeeds on the reaper's result. + testutil.AssertNoError(t, err) + if !reaped { + t.Fatal("reaper must still run when recovery sweep errors") + } + }) +} + func TestTaskLockID(t *testing.T) { t.Run("deterministic", func(t *testing.T) { id1 := taskLockID(TaskCollectRecommendations) diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index a374c210f..376771ff0 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -28,6 +28,15 @@ type PurchaseManagerInterface interface { ApproveExecution(ctx context.Context, execID, token, actor string) error ApproveAndExecute(ctx context.Context, execID, actor string) error CancelExecution(ctx context.Context, execID, token, actor string) error + // RecoverStrandedApprovals sweeps purchase_executions stranded in the + // "approved" status past the internal staleApprovedThreshold and, for + // AWS-only executions, idempotently re-drives them to completion (so an + // interrupted sync run actually finishes) before the reaper would fail + // them; mixed/Azure/GCP/legacy rows are safe-failed for visibility. + // Returns the number of rows acted on. Wired into the + // "reap_stuck_purchases" scheduled task ahead of ReapStuckExecutions. + // See issue #632. + RecoverStrandedApprovals(ctx context.Context) (int, error) // ReapStuckExecutions sweeps purchase_executions stuck in // approved/running longer than reapAfter and flips them to "failed" // via the existing TransitionExecutionStatus CAS. Wired into the diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index 4850d0844..d7d47d78f 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -45,6 +45,7 @@ type MockPurchaseManager struct { ApproveExecutionFunc func(ctx context.Context, execID, token, actor string) error ApproveAndExecuteFunc func(ctx context.Context, execID, actor string) error CancelExecutionFunc func(ctx context.Context, execID, token, actor string) error + RecoverStrandedApprovalsFunc func(ctx context.Context) (int, error) ReapStuckExecutionsFunc func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) } @@ -90,6 +91,13 @@ func (m *MockPurchaseManager) CancelExecution(ctx context.Context, execID, token return nil } +func (m *MockPurchaseManager) RecoverStrandedApprovals(ctx context.Context) (int, error) { + if m.RecoverStrandedApprovalsFunc != nil { + return m.RecoverStrandedApprovalsFunc(ctx) + } + return 0, nil +} + func (m *MockPurchaseManager) ReapStuckExecutions(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { if m.ReapStuckExecutionsFunc != nil { return m.ReapStuckExecutionsFunc(ctx, reapAfter)