fix(purchase): reap purchase_executions stuck in approved/running >10m (closes #678)

internal/analytics/collector_test.go

@@ -216,6 +216,10 @@ func (m *mockConfigStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx,
 	return false, "", nil
 }
 
+func (m *mockConfigStore) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]config.PurchaseExecution, error) {
+	return nil, nil
+}
+
 func (m *mockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error {
 	return nil
 }

internal/api/mocks_test.go

@@ -241,6 +241,14 @@ func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return args.Get(0).(*config.PurchaseExecution), args.Error(1)
 }
 
+func (m *MockConfigStore) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]config.PurchaseExecution, error) {
+	args := m.Called(ctx, statuses, olderThan)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).([]config.PurchaseExecution), args.Error(1)
+}
+
 func (m *MockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error {
 	args := m.Called(ctx, record)
 	return args.Error(0)

internal/config/errors.go

@@ -4,3 +4,11 @@ import "errors"
 
 // ErrNotFound is returned when a requested config-store row does not exist.
 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
+// 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")

internal/config/interfaces.go

@@ -72,6 +72,13 @@ type StoreInterface interface {
 	// Must be called inside a WithTx block so the suppression cleanup and
 	// the status flip commit atomically.
 	CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (cancelled bool, currentStatus string, err error)
+	// ListStuckExecutions returns executions in any of the given statuses
+	// whose updated_at is older than the given duration. Used by the
+	// reaper sweep (issue #678) to find rows stuck in approved/running
+	// after the synchronous executor failed mid-flight without flipping
+	// them to a terminal state. Oldest-stuck-first (ORDER BY updated_at
+	// ASC), capped at MaxListLimit per sweep.
+	ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]PurchaseExecution, error)
 
 	// Purchase history
 	SavePurchaseHistory(ctx context.Context, record *PurchaseHistoryRecord) error

internal/config/store_postgres.go

@@ -794,9 +794,15 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution
 	if len(records) == 0 {
 		existing, existErr := s.GetExecutionByID(ctx, executionID)
 		if existErr != nil || existing == nil {
-			return nil, fmt.Errorf("execution not found: %s", executionID)
+			// Wrap ErrNotFound so callers (e.g. the purchase reaper) can
+			// use errors.Is to distinguish "row vanished mid-flight" — a
+			// legitimate CAS race-loss — from a hard DB error.
+			return nil, fmt.Errorf("%w: execution %s", ErrNotFound, executionID)
 		}
-		return nil, fmt.Errorf("execution %s cannot transition from %q to %q", executionID, existing.Status, toStatus)
+		// Wrap ErrExecutionNotInExpectedStatus so callers can use
+		// errors.Is to recognise CAS rejection (status changed between
+		// SELECT and UPDATE) as race-lost rather than a real error.
+		return nil, fmt.Errorf("%w: execution %s cannot transition from %q to %q", ErrExecutionNotInExpectedStatus, executionID, existing.Status, toStatus)
 	}
 
 	return &records[0], nil
@@ -910,6 +916,50 @@ func (s *PostgresStore) GetStaleApprovedExecutions(ctx context.Context, olderTha
 	return s.queryExecutions(ctx, query, fmt.Sprintf("%d seconds", int(olderThan.Seconds())))
 }
 
+// ListStuckExecutions returns purchase executions whose Status is any of the
+// supplied values and whose updated_at is older than the given duration. Used
+// by the reaper sweep (issue #678) to find executions stuck in
+// approved/running long enough that the synchronous executor has clearly
+// failed without flipping the row to a terminal state.
+//
+// Returns rows oldest-first (ORDER BY updated_at ASC) so the longest-stuck
+// rows are processed first within a single sweep, capped at MaxListLimit so
+// an unbounded backlog doesn't blow up the Lambda's memory budget. The reaper
+// invokes the sweep periodically; a backlog larger than MaxListLimit just
+// gets drained across successive invocations.
+//
+// olderThan must be > 0; a zero/negative value would invert the WHERE clause
+// into "updated_at < NOW() + |olderThan|" and reap fresh rows. Defense-in-
+// depth: the caller (ParseReapAfterFromEnv) also rejects non-positive env
+// values.
+//
+// olderThan is passed as a Postgres interval (seconds) so the comparison
+// happens server-side against NOW() — keeping the cutoff in the DB clock
+// avoids any drift between the API process and the database.
+func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]PurchaseExecution, error) {
+	if len(statuses) == 0 {
+		return nil, nil
+	}
+	if olderThan <= 0 {
+		return nil, fmt.Errorf("ListStuckExecutions: olderThan must be > 0, got %s", olderThan)
+	}
+	query := `
+		SELECT plan_id, execution_id, status, step_number, scheduled_date,
+		       notification_sent, approval_token, recommendations,
+		       total_upfront_cost, estimated_savings, completed_at, error, expires_at,
+		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
+		       created_by_user_id, retry_execution_id, retry_attempt_n,
+		       approval_token_expires_at
+		FROM purchase_executions
+		WHERE status = ANY($1)
+		  AND updated_at < NOW() - $2::interval
+		ORDER BY updated_at ASC
+		LIMIT $3
+	`
+	intervalArg := fmt.Sprintf("%d seconds", int(olderThan.Seconds()))
+	return s.queryExecutions(ctx, query, statuses, intervalArg, MaxListLimit)
+}
+
 // GetPendingExecutions retrieves all pending purchase executions
 func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExecution, error) {
 	query := `

internal/config/store_postgres_pgxmock_test.go

@@ -1430,6 +1430,84 @@ func TestPGXMock_ListPendingExecutionIDsForAccount_Empty(t *testing.T) {
 	require.NoError(t, mock.ExpectationsWereMet())
 }
 
+// ─── ListStuckExecutions ─────────────────────────────────────────────────────
+
+// stuckExecRow builds a pgxmock row that matches the queryExecutions scan
+// order. Mirrors the inline row construction in TestPGXMock_GetExecutionByID_*
+// but factored out for the reaper sweep tests below which need 3 rows.
+func stuckExecRow(execID, status string, scheduled time.Time) []any {
+	recsJSON, _ := json.Marshal([]RecommendationRecord{})
+	return []any{
+		"plan-1", execID, status, 1, scheduled,
+		sql.NullTime{}, "tok-" + execID, recsJSON,
+		100.0, 200.0, sql.NullTime{}, "", sql.NullTime{},
+		nil, "cudly-web", nil, nil, 100,
+		nil, nil, 0,
+		sql.NullTime{},
+	}
+}
+
+func stuckExecCols() []string {
+	return []string{
+		"plan_id", "execution_id", "status", "step_number", "scheduled_date",
+		"notification_sent", "approval_token", "recommendations",
+		"total_upfront_cost", "estimated_savings", "completed_at", "error", "expires_at",
+		"cloud_account_id", "source", "approved_by", "cancelled_by", "capacity_percent",
+		"created_by_user_id", "retry_execution_id", "retry_attempt_n",
+		"approval_token_expires_at",
+	}
+}
+
+func TestPGXMock_ListStuckExecutions_ReturnsMultiple(t *testing.T) {
+	mock := newMock(t)
+	store := storeWith(mock)
+	ctx := context.Background()
+
+	now := time.Now().Truncate(time.Second)
+	rows := pgxmock.NewRows(stuckExecCols()).
+		AddRow(stuckExecRow("exec-1", "approved", now)...).
+		AddRow(stuckExecRow("exec-2", "running", now)...).
+		AddRow(stuckExecRow("exec-3", "approved", now)...)
+	mock.ExpectQuery("SELECT.*FROM purchase_executions.*status = ANY.*updated_at < NOW").
+		WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg(), pgxmock.AnyArg()).
+		WillReturnRows(rows)
+
+	execs, err := store.ListStuckExecutions(ctx, []string{"approved", "running"}, 10*time.Minute)
+	require.NoError(t, err)
+	assert.Len(t, execs, 3)
+	assert.Equal(t, "exec-1", execs[0].ExecutionID)
+	assert.Equal(t, "approved", execs[0].Status)
+	assert.Equal(t, "running", execs[1].Status)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestPGXMock_ListStuckExecutions_EmptyStatuses(t *testing.T) {
+	mock := newMock(t)
+	store := storeWith(mock)
+	ctx := context.Background()
+
+	// No statuses → caller wants nothing — short-circuit returns nil with no
+	// SQL roundtrip. pgxmock will fail if any expectation is unmet (we
+	// register none) so this also guards against an accidental query.
+	execs, err := store.ListStuckExecutions(ctx, nil, 10*time.Minute)
+	require.NoError(t, err)
+	assert.Nil(t, execs)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestPGXMock_ListStuckExecutions_QueryError(t *testing.T) {
+	mock := newMock(t)
+	store := storeWith(mock)
+	ctx := context.Background()
+
+	mock.ExpectQuery("SELECT").
+		WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg(), pgxmock.AnyArg()).
+		WillReturnError(errors.New("db down"))
+
+	_, err := store.ListStuckExecutions(ctx, []string{"approved"}, 10*time.Minute)
+	require.Error(t, err)
+}
+
 // ─── errNoRows helper ────────────────────────────────────────────────────────
 
 func errNoRows() error {

internal/purchase/mocks_test.go

@@ -315,6 +315,14 @@ func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return args.Get(0).(*config.PurchaseExecution), args.Error(1)
 }
 
+func (m *MockConfigStore) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]config.PurchaseExecution, error) {
+	args := m.Called(ctx, statuses, olderThan)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).([]config.PurchaseExecution), args.Error(1)
+}
+
 func (m *MockConfigStore) SavePurchaseHistory(ctx context.Context, record *config.PurchaseHistoryRecord) error {
 	args := m.Called(ctx, record)
 	return args.Error(0)