fix(purchase): let stranded-approval recovery run before the reaper fails approved rows

internal/purchase/manager.go

@@ -86,14 +86,11 @@ type ProcessResult struct {
 // 15min (matching the RI-exchange stale-sweep threshold in pkg/exchange)
 // leaves a wide safety margin.
 //
-// Note: the purchase.Reaper (reaper.go) uses DefaultReapAfter (10m) to cover
-// both "approved" and "running" rows via an atomic CAS. These two thresholds
-// serve different sweep paths and the 15m vs 10m difference is intentional:
-// the reaper is CAS-protected and can safely reap "running" rows (the real
-// executor wins the CAS race if it is still alive), while this legacy sweep
-// only targets "approved" rows from the cron path and is deliberately
-// conservative (05-N3). A future consolidation should align both under a
-// single env-configurable threshold.
+// Note: the stuck-purchase reaper (reaper.go) fails "running" rows after
+// DefaultReapAfter (10m) but leaves "approved" rows alone until
+// minApprovedReapAfter, which is derived from this constant plus one process
+// sweep interval, so this recovery always gets to re-drive a stranded
+// approval before the reaper fails it (issue #894).
 const staleApprovedThreshold = 15 * time.Minute
 
 // NotificationResult holds the result of sending notifications.

internal/purchase/reaper.go

@@ -12,12 +12,13 @@ import (
 )
 
 // stuckStatuses are the in-flight execution statuses that the reaper will
-// flip to "failed" when the row has been sitting in one of them for longer
-// than the configured reapAfter duration. These are the only intermediate
-// states the synchronous executor passes through between the user clicking
-// approve and the row landing in a terminal state (completed/failed) —
-// pending/notified executions are advanced by the scheduler tick on its
-// own cadence, not by us. See ApproveExecution + executeAndFinalize for
+// flip to "failed" when the row has been sitting in one of them for too long:
+// reapAfter for "running", approvedReapAfter(reapAfter) for "approved" (see
+// reapPasses). These are the only intermediate states the synchronous
+// executor passes through between the user clicking approve and the row
+// landing in a terminal state (completed/failed) — pending/notified
+// executions are advanced by the scheduler tick on its own cadence, not by
+// us. See ApproveExecution + executeAndFinalize for
 // the originating state machine.
 //
 // Kept as a package-level var (not a const slice) so tests can read it
@@ -29,13 +30,13 @@ var stuckStatuses = []string{"approved", "running"}
 // migrating to a typed status enum) shows up at the one call site.
 const failedStatus = "failed"
 
-// DefaultReapAfter is the default age at which an approved/running execution
-// is considered stuck. The issue (#678) calls for "configurable via env var,
-// default 10m" — the threshold is intentionally conservative: the happy-path
-// executor completes in <60s; the longest legitimately-slow paths
-// (multi-account fan-out, provider rate-limit backoff) settle in <3min. 10
-// minutes leaves multiple multiples of headroom so the reaper never fights
-// a real executor.
+// DefaultReapAfter is the default age at which a running execution is
+// considered stuck (approved rows use approvedReapAfter). The issue (#678)
+// calls for "configurable via env var, default 10m" — the threshold is
+// intentionally conservative: the happy-path executor completes in <60s; the
+// longest legitimately-slow paths (multi-account fan-out, provider rate-limit
+// backoff) settle in <3min. 10 minutes leaves multiple multiples of headroom
+// so the reaper never fights a real executor.
 const DefaultReapAfter = 10 * time.Minute
 
 // reapAfterEnvVar is the env var read by ParseReapAfterFromEnv. Centralized
@@ -98,12 +99,59 @@ func ParseReapAfterFromEnv() time.Duration {
 	return d
 }
 
+// processSweepInterval is the default cadence of the process_scheduled_purchases
+// sweep, which is what runs RecoverStrandedApprovals (terraform:
+// process_scheduled_purchases_schedule defaults to "rate(1 hour)").
+const processSweepInterval = time.Hour
+
+// processSweepSlack covers the process sweep's own run time and EventBridge
+// delivery jitter, so a recovery run that starts on schedule can still claim
+// the row before the reaper's next tick.
+const processSweepSlack = 5 * time.Minute
+
+// minApprovedReapAfter is the earliest age at which the reaper may fail an
+// "approved" row (issue #894). RecoverStrandedApprovals only sees approved rows
+// older than staleApprovedThreshold, and it runs once per process sweep, so a
+// stranded approval first becomes recoverable somewhere in
+// (staleApprovedThreshold, staleApprovedThreshold + processSweepInterval]. The
+// reaper must wait past that window, or it fails the row first and the
+// idempotent re-drive never runs. "running" rows are not affected: recovery
+// never re-drives them, so the reaper stays their only backstop.
+const minApprovedReapAfter = staleApprovedThreshold + processSweepInterval + processSweepSlack
+
+// approvedReapAfter returns the age threshold the reaper applies to "approved"
+// rows: the configured reapAfter, raised to minApprovedReapAfter when lower.
+func approvedReapAfter(reapAfter time.Duration) time.Duration {
+	return max(reapAfter, minApprovedReapAfter)
+}
+
+// reapPass is one SELECT of the sweep: the statuses it lists and the age past
+// which those rows count as stuck.
+type reapPass struct {
+	statuses []string
+	after    time.Duration
+}
+
+// reapPasses splits the sweep by status so "approved" rows get the later
+// threshold that leaves room for RecoverStrandedApprovals (issue #894).
+func reapPasses(reapAfter time.Duration) []reapPass {
+	return []reapPass{
+		{statuses: []string{"running"}, after: reapAfter},
+		{statuses: []string{"approved"}, after: approvedReapAfter(reapAfter)},
+	}
+}
+
 // ReapStuckExecutions runs one sweep that finds purchase_executions stuck in
-// approved/running longer than reapAfter and atomically transitions them to
-// "failed" via the existing TransitionExecutionStatus CAS. Each successful
-// transition also persists a canonical error message so the History UI
-// (issue #621) can show why the row was reaped and the operator knows it's
-// safe to retry.
+// running longer than reapAfter, or in approved longer than
+// approvedReapAfter(reapAfter), and atomically transitions them to "failed" via
+// the existing TransitionExecutionStatus CAS. Each successful transition also
+// persists a canonical error message so the History UI (issue #621) can show
+// why the row was reaped and the operator knows it's safe to retry.
+//
+// Approved rows get the later threshold so the process sweep's
+// RecoverStrandedApprovals can re-drive them first (issue #894); the reaper
+// only fails the ones recovery never picked up, e.g. on deployments that do
+// not schedule the process sweep.
 //
 // Safety properties:
 //   - Local-status-only: the reaper never touches provider commitments. If
@@ -118,25 +166,27 @@ func ParseReapAfterFromEnv() time.Duration {
 //   - Per-row error-isolation: a failure on row N never blocks rows
 //     N+1..K. Counts are aggregated in ReapResult for the caller to log.
 func (m *Manager) ReapStuckExecutions(ctx context.Context, reapAfter time.Duration) (*ReapResult, error) {
-	stuck, err := m.config.ListStuckExecutions(ctx, stuckStatuses, reapAfter)
-	if err != nil {
-		return nil, fmt.Errorf("failed to list stuck executions: %w", err)
+	result := &ReapResult{}
+	now := time.Now()
+	for _, pass := range reapPasses(reapAfter) {
+		stuck, err := m.config.ListStuckExecutions(ctx, pass.statuses, pass.after)
+		if err != nil {
+			return nil, fmt.Errorf("failed to list stuck executions: %w", err)
+		}
+		result.Found += len(stuck)
+		for i := range stuck {
+			m.reapOne(ctx, &stuck[i], pass.after, now, result)
+		}
 	}
 
-	result := &ReapResult{Found: len(stuck)}
-	if len(stuck) == 0 {
-		logging.Debugf("purchase reaper: no stuck executions (threshold %s)", reapAfter)
+	if result.Found == 0 {
+		logging.Debugf("purchase reaper: no stuck executions (threshold running=%s approved=%s)",
+			reapAfter, approvedReapAfter(reapAfter))
 		return result, nil
 	}
 
-	now := time.Now()
-	for _rvc := range stuck {
-		exec := stuck[_rvc]
-		m.reapOne(ctx, &exec, reapAfter, now, result)
-	}
-
-	logging.Infof("purchase reaper sweep complete: found=%d reaped=%d race_lost=%d errored=%d (threshold %s)",
-		result.Found, result.Reaped, result.RaceLost, result.Errored, reapAfter)
+	logging.Infof("purchase reaper sweep complete: found=%d reaped=%d race_lost=%d errored=%d (threshold running=%s approved=%s)",
+		result.Found, result.Reaped, result.RaceLost, result.Errored, reapAfter, approvedReapAfter(reapAfter))
 
 	return result, nil
 }

internal/purchase/reaper_ordering_test.go

@@ -0,0 +1,123 @@
+package purchase
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// defaultProcessSchedule is the process_scheduled_purchases sweep cadence the
+// terraform modules default to ("rate(1 hour)"). RecoverStrandedApprovals runs
+// on that sweep, so a stranded approval may wait this long past
+// staleApprovedThreshold before recovery sees it.
+const defaultProcessSchedule = time.Hour
+
+// listedThresholds runs one reaper sweep against a store that returns no rows
+// and records the age threshold each ListStuckExecutions call used per status.
+func listedThresholds(t *testing.T, reapAfter time.Duration) map[string]time.Duration {
+	t.Helper()
+	ctx := context.Background()
+	store := new(MockConfigStore)
+	got := map[string]time.Duration{}
+	store.On("ListStuckExecutions", ctx, mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			for _, s := range args.Get(1).([]string) {
+				got[s] = args.Get(2).(time.Duration)
+			}
+		}).
+		Return([]config.PurchaseExecution{}, nil)
+
+	_, err := newReaperManager(store).ReapStuckExecutions(ctx, reapAfter)
+	require.NoError(t, err)
+	return got
+}
+
+// Issue #894: with the default thresholds the reaper must not list approved
+// rows until RecoverStrandedApprovals has had a full process sweep past
+// staleApprovedThreshold to re-drive them. Before the fix approved rows were
+// listed at reapAfter (10m) and failed before recovery could see them (15m).
+func TestReapStuckExecutions_ApprovedRowsWaitForStrandedRecovery(t *testing.T) {
+	got := listedThresholds(t, DefaultReapAfter)
+
+	require.Contains(t, got, "approved", "the reaper must still back-stop approved rows")
+	assert.Greater(t, got["approved"], staleApprovedThreshold+defaultProcessSchedule,
+		"approved rows must outlive the stranded-approval recovery window")
+	require.Contains(t, got, "running")
+	assert.Equal(t, DefaultReapAfter, got["running"],
+		"running rows are not re-driven by recovery and keep the configured threshold")
+}
+
+// A reapAfter configured below the recovery window (PURCHASE_APPROVED_REAP_AFTER)
+// must not reopen the race for approved rows.
+func TestReapStuckExecutions_ShortReapAfterDoesNotBeatRecovery(t *testing.T) {
+	got := listedThresholds(t, 2*time.Minute)
+
+	assert.Greater(t, got["approved"], staleApprovedThreshold+defaultProcessSchedule)
+	assert.Equal(t, 2*time.Minute, got["running"])
+}
+
+// A reapAfter configured above the recovery window is honored as is for both
+// statuses: the floor only ever delays the approved pass, never hastens it.
+func TestReapStuckExecutions_LongReapAfterIsHonoredForApproved(t *testing.T) {
+	long := 3 * time.Hour
+	got := listedThresholds(t, long)
+
+	assert.Equal(t, long, got["approved"])
+	assert.Equal(t, long, got["running"])
+}
+
+// The approved and running passes together must cover exactly stuckStatuses,
+// so splitting the sweep neither drops a status nor widens it.
+func TestReapStuckExecutions_PassesCoverStuckStatuses(t *testing.T) {
+	got := listedThresholds(t, DefaultReapAfter)
+
+	listed := slices.Sorted(maps.Keys(got))
+	want := slices.Clone(stuckStatuses)
+	slices.Sort(want)
+	assert.Equal(t, want, listed)
+}
+
+// The per-row reap annotation reports the threshold of the pass that listed
+// the row, so an approved row is not described as stuck for only 10 minutes.
+func TestReapStuckExecutions_ApprovedAnnotationUsesApprovedThreshold(t *testing.T) {
+	ctx := context.Background()
+	store := new(MockConfigStore)
+	row := stuckExec("exec-old-approved", "approved")
+	flipped := row
+	flipped.Status = failedStatus
+	expectStuckLists(ctx, store, DefaultReapAfter, row)
+	store.On("TransitionExecutionStatus", ctx, row.ExecutionID, stuckStatuses, failedStatus, (*string)(nil)).
+		Return(&flipped, nil)
+	var saved *config.PurchaseExecution
+	store.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).
+		Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }).Return(nil)
+
+	result, err := newReaperManager(store).ReapStuckExecutions(ctx, DefaultReapAfter)
+
+	require.NoError(t, err)
+	assert.Equal(t, 1, result.Reaped)
+	require.NotNil(t, saved)
+	wantMinutes := int(approvedReapAfter(DefaultReapAfter) / time.Minute)
+	assert.Contains(t, saved.Error, fmt.Sprintf("reaped after %dm in approved state", wantMinutes))
+	store.AssertExpectations(t)
+}
+
+// expectStuckLists stubs both reaper passes: running rows listed at reapAfter
+// and approved rows at approvedReapAfter(reapAfter). Rows are routed to the
+// pass matching their status.
+func expectStuckLists(ctx context.Context, store *MockConfigStore, reapAfter time.Duration, rows ...config.PurchaseExecution) {
+	byStatus := map[string][]config.PurchaseExecution{"running": {}, "approved": {}}
+	for _, r := range rows {
+		byStatus[r.Status] = append(byStatus[r.Status], r)
+	}
+	store.On("ListStuckExecutions", ctx, []string{"running"}, reapAfter).Return(byStatus["running"], nil)
+	store.On("ListStuckExecutions", ctx, []string{"approved"}, approvedReapAfter(reapAfter)).Return(byStatus["approved"], nil)
+}