From 6310b4de718f8cf9941ba88787e28618c2090e8d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 22 May 2026 16:57:25 +0200 Subject: [PATCH 1/5] feat(config): add ListStuckExecutions store method MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selects purchase_executions rows whose status is in a caller-supplied set AND whose updated_at is older than a caller-supplied interval. Newest-stuck-first, capped at MaxListLimit per call. Mirrors the existing GetStaleProcessingExchanges pattern for RI exchanges. Used by the reaper sweep (issue #678) to find executions stuck in approved/running because the synchronous executor failed mid-flight (Lambda timeout, OOM, network hang) without flipping the row to a terminal state. Local SELECT only — no provider-side mutation. Adds the method to StoreInterface and to every test-side mock store that satisfies it (purchase, api, analytics, scheduler, server health). Adds pgxmock coverage for the happy path, the empty-statuses short-circuit, and a query-error path. --- internal/analytics/collector_test.go | 4 + internal/api/mocks_test.go | 8 ++ internal/config/interfaces.go | 7 ++ internal/config/store_postgres.go | 44 +++++++++++ .../config/store_postgres_pgxmock_test.go | 78 +++++++++++++++++++ internal/purchase/mocks_test.go | 8 ++ internal/scheduler/scheduler_test.go | 8 ++ internal/server/test_helpers_test.go | 4 + 8 files changed, 161 insertions(+) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 1741b5e5d..ce12bd0a1 100644 --- a/internal/analytics/collector_test.go +++ b/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 } diff --git a/internal/api/mocks_test.go b/internal/api/mocks_test.go index a3762ba86..e1edcf3d3 100644 --- a/internal/api/mocks_test.go +++ b/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) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index e81f98b4d..ac3cc2931 100644 --- a/internal/config/interfaces.go +++ b/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 diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index eb348ada6..40e666604 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -910,6 +910,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 := ` diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index 08dc89e13..263935624 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/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 { diff --git a/internal/purchase/mocks_test.go b/internal/purchase/mocks_test.go index 63aaa35bd..8b0199876 100644 --- a/internal/purchase/mocks_test.go +++ b/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) diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 3bd167896..afb7b03bd 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -216,6 +216,14 @@ func (m *MockConfigStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, return args.Bool(0), args.String(1), args.Error(2) } +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) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 917e485ab..033e27790 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -115,6 +115,10 @@ func (m *mockConfigStoreForHealth) CancelExecutionAtomic(ctx context.Context, tx return false, "", nil } +func (m *mockConfigStoreForHealth) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStoreForHealth) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error { return nil } From 141a5a10228913d373cd4efccfff08c156ca8a81 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 22 May 2026 17:00:50 +0200 Subject: [PATCH 2/5] feat(purchase): add ReapStuckExecutions sweep + per-row CAS to failed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New Manager.ReapStuckExecutions(ctx, reapAfter): one sweep that finds executions stuck in approved/running longer than reapAfter and atomically transitions them to "failed" via the existing TransitionExecutionStatus CAS. Each successful transition also writes a canonical, human-readable error string so the History UI (#621) shows operators why the row was reaped and confirms it's safe to retry. Safety properties: - Local-status-only: never touches provider commitments. If the real executor did manage to create a commitment before dying, the retry hits the idempotency path (#636/#638/#652) which surfaces a duplicate-reservation error cleanly. - CAS-protected: if the real executor wakes up and finishes between the SELECT and the transition, the CAS rejects; logged at INFO, not surfaced as an error. The real executor wins the race. - Per-row error-isolation: a failure on row N never blocks N+1..K. Adds ParseReapAfterFromEnv to read PURCHASE_APPROVED_REAP_AFTER and fall back to a 10m default on missing-or-malformed env (with a WARN log so ops can spot a typo). Never panics — a misconfigured env var must not crash the other scheduled tasks sharing the Lambda. Tests cover: - stale approved row → flipped to failed with canonical message - stale running row → same - younger-than-threshold rows are filtered out by the SELECT (no TransitionExecutionStatus / SavePurchaseExecution calls happen) - terminal-status rows are never passed to the SELECT (regression guard against accidentally widening stuckStatuses) - CAS race lost → logged + counted, not surfaced as sweep error - CAS (nil, nil) defensive path treated as race-lost - 3-stuck-row integration with per-row CAS expectations - SELECT error → propagated as sweep error - per-row save error after successful CAS → Reaped++ AND Errored++ - env-var: unset, valid, invalid, non-default unit (2h30m) --- internal/purchase/reaper.go | 200 ++++++++++++++++++++ internal/purchase/reaper_test.go | 302 +++++++++++++++++++++++++++++++ 2 files changed, 502 insertions(+) create mode 100644 internal/purchase/reaper.go create mode 100644 internal/purchase/reaper_test.go diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go new file mode 100644 index 000000000..516ebf611 --- /dev/null +++ b/internal/purchase/reaper.go @@ -0,0 +1,200 @@ +package purchase + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/logging" +) + +// 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 +// the originating state machine. +// +// Kept as a package-level var (not a const slice) so tests can read it +// without re-typing the literal, while staying immutable in callers. +var stuckStatuses = []string{"approved", "running"} + +// failedStatus is the terminal status the reaper sets via the atomic CAS in +// TransitionExecutionStatus. Named as a constant so a future rename (e.g. +// 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. +const DefaultReapAfter = 10 * time.Minute + +// reapAfterEnvVar is the env var read by ParseReapAfterFromEnv. Centralised +// so the wiring code, the tests, and future ops documentation all reference +// the same name. +const reapAfterEnvVar = "PURCHASE_APPROVED_REAP_AFTER" + +// ReapResult summarises one sweep of ReapStuckExecutions. Returned for the +// scheduled-task handler to log + surface in CloudWatch / metrics. +type ReapResult struct { + // Found is the number of rows the SELECT returned (i.e. stuck rows the + // sweep saw — pre-CAS). + Found int `json:"found"` + // Reaped is the number of rows that successfully transitioned to + // "failed" via the atomic CAS. Reaped <= Found; the gap is rows the + // real executor finished between the SELECT and the CAS (CAS race + // rejected the reap — correct behaviour, not an error). + Reaped int `json:"reaped"` + // RaceLost is the number of rows where the CAS rejected the reap + // because the row's status changed between the SELECT and the + // transition (the real executor woke up and beat us). Logged at + // INFO; not surfaced as an error. + RaceLost int `json:"race_lost"` + // Errored is the number of rows where the persistence step failed + // (CAS itself errored or the canonical-error follow-up save failed). + // These are real ops issues worth surfacing. + Errored int `json:"errored"` +} + +// ParseReapAfterFromEnv reads PURCHASE_APPROVED_REAP_AFTER from the +// environment and parses it via time.ParseDuration. Falls back to +// DefaultReapAfter on either an absent env var or a parse failure — and +// logs a WARN on the parse failure so ops can spot a typo without the +// reaper silently running at the default. Never panics: a misconfigured +// env var must not crash the Lambda's other scheduled tasks. +func ParseReapAfterFromEnv() time.Duration { + raw := os.Getenv(reapAfterEnvVar) + if raw == "" { + return DefaultReapAfter + } + d, err := time.ParseDuration(raw) + if err != nil { + logging.Warnf("purchase reaper: failed to parse %s=%q as duration: %v — using default %s", + reapAfterEnvVar, raw, err, DefaultReapAfter) + return DefaultReapAfter + } + return d +} + +// 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. +// +// Safety properties: +// - Local-status-only: the reaper never touches provider commitments. If +// the real executor did get a commitment created on the provider before +// dying, the row is still flipped to failed locally — the operator's +// retry hits the idempotency path (#636/#638/#652) which surfaces a +// duplicate-reservation error and short-circuits cleanly. +// - CAS-protected: if the real executor wakes up and finishes between the +// SELECT and the CAS, TransitionExecutionStatus returns an error +// ("cannot transition from completed/failed"); we log at INFO and move +// on. The real executor wins the race. +// - 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{Found: len(stuck)} + if len(stuck) == 0 { + logging.Debugf("purchase reaper: no stuck executions (threshold %s)", reapAfter) + return result, nil + } + + now := time.Now() + for _, exec := range stuck { + 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) + + return result, nil +} + +// reapOne handles a single stuck row: atomic CAS to failed, then a +// best-effort persistence of the canonical error message so History can +// show the operator why the row was reaped. Updates the shared ReapResult +// counters in place. Extracted from ReapStuckExecutions so the per-row +// path stays under the gocyclo threshold. +// +// `now` is passed in (rather than read inline) so age math is consistent +// across all rows in one sweep — useful for log-spam-free re-runs and for +// deterministic tests that inject a fixed clock. +func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, reapAfter time.Duration, now time.Time, result *ReapResult) { + prevStatus := exec.Status + + // Best-effort age estimate. The store does not currently return + // updated_at on the PurchaseExecution struct (issue #678 may add it + // later), so we lower-bound the age at reapAfter — the SELECT + // guarantees updated_at < NOW() - reapAfter, so the real age is at + // least reapAfter. Rounded to whole minutes for the canonical + // message; the WARN log carries the same lower bound. + age := reapAfter + ageMinutes := int(age.Round(time.Minute) / time.Minute) + if ageMinutes < 1 { + ageMinutes = 1 + } + + logging.Warnf("purchase reaper: reaping execution %s (status=%s, age>=%dm)", exec.ExecutionID, prevStatus, ageMinutes) + + transitioned, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, stuckStatuses, failedStatus) + if err != nil { + // CAS rejection is the "real executor beat us" path. We can't + // distinguish it from a hard DB error without inspecting the + // error string, which is brittle. Treat both as race_lost: the + // row is no longer in approved/running, so there is nothing for + // the reaper to do regardless of why. A persistent DB outage + // will surface via repeated Errored counts on the canonical- + // error save below, which IS treated as a real error. + logging.Infof("purchase reaper: CAS race lost for execution %s (status changed from %q before transition): %v", + exec.ExecutionID, prevStatus, err) + result.RaceLost++ + return + } + if transitioned == nil { + // Defensive: the store contract returns either (record, nil) or + // (nil, err). Treat (nil, nil) as a race-lost too — same caller + // intent, no canonical-error save to attempt. + logging.Infof("purchase reaper: CAS returned nil row for execution %s — treating as race lost", exec.ExecutionID) + result.RaceLost++ + return + } + + // Persist the canonical error message. The CAS already flipped status + // to "failed"; this follow-up SavePurchaseExecution writes the human- + // readable error string so History shows why the row was reaped. + // Best-effort: if this save fails (network blip, etc.) the row is + // still in "failed" — operator just sees a generic failure without + // the reaper attribution. Bump Errored so ops can track it. + transitioned.Error = fmt.Sprintf("reaped after %dm in %s state — executor did not complete; safe to retry", + ageMinutes, prevStatus) + if saveErr := m.config.SavePurchaseExecution(ctx, transitioned); saveErr != nil { + logging.Errorf("purchase reaper: failed to persist canonical error for execution %s (already flipped to failed): %v", + exec.ExecutionID, saveErr) + result.Errored++ + // Still count as Reaped — the status flip succeeded, only the + // error-message annotation failed. Otherwise Reaped would + // undercount real recoveries. + result.Reaped++ + _ = now // kept in signature for future deterministic-clock injection + return + } + + result.Reaped++ + _ = now +} diff --git a/internal/purchase/reaper_test.go b/internal/purchase/reaper_test.go new file mode 100644 index 000000000..3ad38e3e6 --- /dev/null +++ b/internal/purchase/reaper_test.go @@ -0,0 +1,302 @@ +package purchase + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/LeanerCloud/CUDly/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// newReaperManager builds a Manager wired with just enough mocks for the +// reaper sweep. Other deps (email, providers) are unused by ReapStuck +// and intentionally left nil — the reaper is local-status-only by design. +func newReaperManager(store *MockConfigStore) *Manager { + return &Manager{ + config: store, + } +} + +// stuckExec builds a representative stuck-approved execution. Tests +// override Status / ExecutionID as needed. The age claim (`updated_at` +// being old) is implicit — the SELECT in ListStuckExecutions is what +// enforces it, and the test mocks return whatever the test wants. +func stuckExec(id, status string) config.PurchaseExecution { + return config.PurchaseExecution{ + PlanID: "plan-1", + ExecutionID: id, + Status: status, + ApprovalToken: "tok-" + id, + ScheduledDate: time.Now().Add(-1 * time.Hour), + } +} + +func TestReapStuckExecutions_StaleApprovedFlippedToFailed(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-A", "approved") + transitioned := row + transitioned.Status = failedStatus + + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-A", stuckStatuses, failedStatus). + Return(&transitioned, nil) + store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + // Canonical error message + previous-status attribution. + return e.ExecutionID == "exec-A" && + e.Status == failedStatus && + strings.Contains(e.Error, "reaped after") && + strings.Contains(e.Error, "approved") && + strings.Contains(e.Error, "safe to retry") + })).Return(nil) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 1, result.Found) + assert.Equal(t, 1, result.Reaped) + assert.Equal(t, 0, result.RaceLost) + assert.Equal(t, 0, result.Errored) + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_StaleRunningFlippedToFailed(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-B", "running") + transitioned := row + transitioned.Status = failedStatus + + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-B", stuckStatuses, failedStatus). + Return(&transitioned, nil) + store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + // "running" attribution path + return e.ExecutionID == "exec-B" && + e.Status == failedStatus && + strings.Contains(e.Error, "running") + })).Return(nil) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 1, result.Reaped) + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_YoungerThanThresholdNotTouched(t *testing.T) { + // The reaper's age filtering is enforced by the SELECT in + // ListStuckExecutions — so "younger than reapAfter" means "the store + // returns an empty slice". The reaper must then return early with + // Found=0 and never call TransitionExecutionStatus or + // SavePurchaseExecution. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{}, nil) + // No TransitionExecutionStatus / SavePurchaseExecution expectations + // — mock.AssertExpectations will fail if either is called. + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 0, result.Found) + assert.Equal(t, 0, result.Reaped) + store.AssertExpectations(t) + store.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +func TestReapStuckExecutions_TerminalStatusNotTouched(t *testing.T) { + // The store filters out terminal statuses via the WHERE clause; the + // reaper never sees completed/failed/cancelled rows. Verify the + // reaper passes only stuckStatuses (approved/running) to the store + // — never the terminal set. This regression-guards a future change + // that accidentally widens stuckStatuses. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + store.On("ListStuckExecutions", ctx, mock.MatchedBy(func(statuses []string) bool { + // Must contain "approved" and "running" and NOT contain any + // terminal status. + seen := map[string]bool{} + for _, s := range statuses { + seen[s] = true + } + return seen["approved"] && seen["running"] && + !seen["completed"] && !seen["failed"] && !seen["cancelled"] && !seen["pending"] && !seen["notified"] + }), reapAfter). + Return([]config.PurchaseExecution{}, nil) + + mgr := newReaperManager(store) + _, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_CASRaceLostNoError(t *testing.T) { + // CAS race: the SELECT returns a stuck row, but between SELECT and + // CAS, the real executor flips the row to completed. The CAS rejects + // with "cannot transition from X". The reaper must log and move on + // without erroring the sweep. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-race", "approved") + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-race", stuckStatuses, failedStatus). + Return(nil, errors.New("execution exec-race cannot transition from \"completed\" to \"failed\"")) + // No SavePurchaseExecution expectation — we lost the race, the real + // executor's status flip stands. + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) // sweep itself succeeds even on per-row race + assert.Equal(t, 1, result.Found) + assert.Equal(t, 0, result.Reaped) + assert.Equal(t, 1, result.RaceLost) + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +func TestReapStuckExecutions_CASReturnsNilNilTreatedAsRaceLost(t *testing.T) { + // Defensive coverage: if the store contract is violated and returns + // (nil, nil), the reaper treats it as a race-lost rather than NPEing + // on the follow-up save. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-nilnil", "running") + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-nilnil", stuckStatuses, failedStatus). + Return(nil, nil) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 1, result.RaceLost) + assert.Equal(t, 0, result.Reaped) + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_ThreeStuckRowsAllReaped(t *testing.T) { + // Integration-style: 3 stuck rows → 3 CAS calls → 3 SavePurchaseExecution + // writes with the correct execution_id passed through each path. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 15 * time.Minute + + rows := []config.PurchaseExecution{ + stuckExec("exec-1", "approved"), + stuckExec("exec-2", "running"), + stuckExec("exec-3", "approved"), + } + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter).Return(rows, nil) + + for _, r := range rows { + flipped := r + flipped.Status = failedStatus + store.On("TransitionExecutionStatus", ctx, r.ExecutionID, stuckStatuses, failedStatus). + Return(&flipped, nil).Once() + store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool { + return e.ExecutionID == r.ExecutionID && e.Status == failedStatus + })).Return(nil).Once() + } + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 3, result.Found) + assert.Equal(t, 3, result.Reaped) + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_ListStoreError(t *testing.T) { + // A store-level failure on the SELECT propagates as a sweep-level + // error so the caller knows to log + alert. Per-row errors do NOT + // propagate (see CASRaceLost test) — this is the wholesale-failure + // path. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return(nil, errors.New("db down")) + + mgr := newReaperManager(store) + _, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to list stuck executions") + store.AssertExpectations(t) +} + +func TestReapStuckExecutions_SaveErrorAfterCASStillCountsAsReaped(t *testing.T) { + // The CAS already flipped the row to failed; the follow-up + // canonical-error save is best-effort. A save failure must not + // double-count as a race-lost (the row IS reaped, status-wise), + // but Errored should bump so ops can spot the persistence flake. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-saveflake", "approved") + flipped := row + flipped.Status = failedStatus + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-saveflake", stuckStatuses, failedStatus). + Return(&flipped, nil) + store.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). + Return(errors.New("write conflict")) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 1, result.Reaped) + assert.Equal(t, 1, result.Errored) + store.AssertExpectations(t) +} + +func TestParseReapAfterFromEnv_DefaultWhenUnset(t *testing.T) { + t.Setenv(reapAfterEnvVar, "") + got := ParseReapAfterFromEnv() + assert.Equal(t, DefaultReapAfter, got) +} + +func TestParseReapAfterFromEnv_ValidDuration(t *testing.T) { + t.Setenv(reapAfterEnvVar, "7m") + got := ParseReapAfterFromEnv() + assert.Equal(t, 7*time.Minute, got) +} + +func TestParseReapAfterFromEnv_InvalidFallsBackToDefault(t *testing.T) { + t.Setenv(reapAfterEnvVar, "not-a-duration") + got := ParseReapAfterFromEnv() + assert.Equal(t, DefaultReapAfter, got, "malformed env value must fall back to default, not crash") +} + +func TestParseReapAfterFromEnv_NonStandardButValidGoDuration(t *testing.T) { + // Sanity check Go's parser accepts non-minute units — ops may + // reasonably set e.g. "2h" for a relaxed cadence. + t.Setenv(reapAfterEnvVar, "2h30m") + got := ParseReapAfterFromEnv() + assert.Equal(t, 2*time.Hour+30*time.Minute, got) +} From b4d12a18968f2a9239eb9d7682792e0688df30ff Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 22 May 2026 17:16:33 +0200 Subject: [PATCH 3/5] feat(purchase): wire periodic reaper invocation (closes #678) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the ReapStuckExecutions sweep (added in the previous commit) into the existing scheduled-task pipeline: - New ScheduledTaskType "reap_stuck_purchases" registered alongside the other periodic tasks (cleanup, analytics_refresh, ri_exchange_reshape). Adds a dedicated handleReapStuckPurchases dispatcher that reads the threshold via purchase.ParseReapAfterFromEnv on every invocation so an ops-side env-var tune via PURCHASE_APPROVED_REAP_AFTER takes effect on the next sweep without a redeploy. - Adds ReapStuckExecutions to PurchaseManagerInterface so the handler contract is symmetric with the Manager + the testutil mock can satisfy it. MockPurchaseManager gains ReapStuckExecutionsFunc following the existing per-method-Func mock convention. - New EventBridge rule "${stack_name}-reap-stuck-purchases" on rate(5 minutes), targeting the main Lambda with {"action":"reap_stuck_purchases"}. Cadence is intentionally more frequent than the 10m default threshold so a stuck row is reaped within ~1 threshold-window. The reaper itself is CAS-protected (TransitionExecutionStatus from approved/running → failed) so an over-run is safe — the real executor wins the race. - New terraform vars: enable_reap_stuck_purchases_schedule (bool, default true), reap_stuck_purchases_schedule (string, default "rate(5 minutes)"), purchase_approved_reap_after (string, default "" → use in-code DefaultReapAfter). Empty-string default for the threshold avoids pinning a Lambda env value when ops hasn't taken a position. - ParseScheduledEvent learns the "reap_stuck_purchases" action so the EventBridge → Lambda payload round-trips through the dispatcher. Tests cover the dispatcher success path (asserts the default 10m threshold is passed when the env var is unset), the store-error propagation path, the lock-ID uniqueness guard, and the ParseScheduledEvent action mapping. --- internal/server/handler.go | 34 ++++++++++++ internal/server/handler_test.go | 32 +++++++++++ internal/server/interfaces.go | 6 +++ internal/testutil/mocks.go | 9 ++++ terraform/modules/compute/aws/lambda/main.tf | 54 +++++++++++++++++++ .../modules/compute/aws/lambda/variables.tf | 18 +++++++ 6 files changed, 153 insertions(+) diff --git a/internal/server/handler.go b/internal/server/handler.go index 2bb78e5e8..8b0b7773c 100644 --- a/internal/server/handler.go +++ b/internal/server/handler.go @@ -27,6 +27,13 @@ const ( TaskCleanupExpiredRecords ScheduledTaskType = "cleanup" TaskRefreshAnalytics ScheduledTaskType = "analytics_refresh" TaskRIExchangeReshape ScheduledTaskType = "ri_exchange_reshape" + // TaskReapStuckPurchases sweeps purchase_executions stuck in + // approved/running longer than PURCHASE_APPROVED_REAP_AFTER and flips + // them to "failed" via the existing TransitionExecutionStatus CAS. + // Backstop for synchronous-executor crashes (Lambda timeout, OOM, + // network hang) that leave rows orphaned in an in-flight state. + // See internal/purchase/reaper.go + issue #678. + TaskReapStuckPurchases ScheduledTaskType = "reap_stuck_purchases" ) // HandleScheduledTask processes a scheduled task by type. @@ -70,6 +77,8 @@ func (app *Application) dispatchTask(ctx context.Context, taskType ScheduledTask return app.handleRefreshAnalytics(ctx) case TaskRIExchangeReshape: return app.handleRIExchangeReshape(ctx) + case TaskReapStuckPurchases: + return app.handleReapStuckPurchases(ctx) default: return nil, fmt.Errorf("unknown scheduled task type: %s", taskType) } @@ -163,6 +172,29 @@ 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. +// +// 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 +// Lambda env-var rotation takes effect on the next sweep without a +// redeploy. +func (app *Application) handleReapStuckPurchases(ctx context.Context) (*purchase.ReapResult, error) { + reapAfter := purchase.ParseReapAfterFromEnv() + log.Printf("Reaping stuck purchase executions (threshold: %s)...", reapAfter) + result, err := app.Purchase.ReapStuckExecutions(ctx, reapAfter) + if err != nil { + log.Printf("Failed to reap stuck purchase executions: %v", err) + return nil, err + } + log.Printf("Reap sweep complete: found=%d reaped=%d race_lost=%d errored=%d", + result.Found, result.Reaped, result.RaceLost, result.Errored) + return result, nil +} + // handleRefreshAnalytics refreshes materialized views and analytics data func (app *Application) handleRefreshAnalytics(ctx context.Context) (map[string]any, error) { log.Println("Refreshing analytics...") @@ -231,6 +263,8 @@ func ParseScheduledEvent(rawEvent json.RawMessage) (ScheduledTaskType, error) { return TaskRefreshAnalytics, nil case "ri_exchange_reshape": return TaskRIExchangeReshape, nil + case "reap_stuck_purchases": + return TaskReapStuckPurchases, nil default: return "", fmt.Errorf("unknown scheduled task action: %q", event.Action) } diff --git a/internal/server/handler_test.go b/internal/server/handler_test.go index cdfcafda5..179c4eff9 100644 --- a/internal/server/handler_test.go +++ b/internal/server/handler_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/LeanerCloud/CUDly/internal/purchase" "github.com/LeanerCloud/CUDly/internal/scheduler" @@ -86,6 +87,31 @@ func TestHandleScheduledTask(t *testing.T) { setupMocks: func(s *testutil.MockScheduler, p *testutil.MockPurchaseManager) {}, expectError: false, }, + { + name: "reap_stuck_purchases success", + taskType: TaskReapStuckPurchases, + setupMocks: func(s *testutil.MockScheduler, p *testutil.MockPurchaseManager) { + p.ReapStuckExecutionsFunc = func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { + // The wiring uses ParseReapAfterFromEnv; default is + // 10 min when env is unset (which it is in tests). + if reapAfter != 10*time.Minute { + return nil, errors.New("expected default 10m threshold when env unset") + } + return &purchase.ReapResult{Found: 2, Reaped: 2}, nil + } + }, + expectError: false, + }, + { + name: "reap_stuck_purchases propagates store error", + taskType: TaskReapStuckPurchases, + setupMocks: func(s *testutil.MockScheduler, p *testutil.MockPurchaseManager) { + p.ReapStuckExecutionsFunc = func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { + return nil, errors.New("db down") + } + }, + expectError: true, + }, { name: "unknown task type", taskType: ScheduledTaskType("unknown"), @@ -139,6 +165,7 @@ func TestTaskLockID(t *testing.T) { TaskCleanupExpiredRecords, TaskRefreshAnalytics, TaskRIExchangeReshape, + TaskReapStuckPurchases, } seen := make(map[int64]ScheduledTaskType) for _, task := range tasks { @@ -311,6 +338,11 @@ func TestParseScheduledEvent(t *testing.T) { rawEvent: `{"action": "analytics_refresh"}`, expectedTask: TaskRefreshAnalytics, }, + { + name: "reap_stuck_purchases event", + rawEvent: `{"action": "reap_stuck_purchases"}`, + expectedTask: TaskReapStuckPurchases, + }, { name: "unknown action returns error", rawEvent: `{"action": "unknown"}`, diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index 4f627aacd..765e463a6 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -2,6 +2,7 @@ package server import ( "context" + "time" "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/purchase" @@ -22,6 +23,11 @@ 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 + // ReapStuckExecutions sweeps purchase_executions stuck in + // approved/running longer than reapAfter and flips them to "failed" + // via the existing TransitionExecutionStatus CAS. Wired into the + // "reap_stuck_purchases" scheduled task. See issue #678. + ReapStuckExecutions(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) } // AnalyticsStoreInterface defines the methods required for analytics storage diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index 8143f58d7..833f4cb90 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -2,6 +2,7 @@ package testutil import ( "context" + "time" "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/purchase" @@ -36,6 +37,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 + ReapStuckExecutionsFunc func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) } func (m *MockPurchaseManager) ProcessScheduledPurchases(ctx context.Context) (*purchase.ProcessResult, error) { @@ -79,3 +81,10 @@ func (m *MockPurchaseManager) CancelExecution(ctx context.Context, execID, token } return nil } + +func (m *MockPurchaseManager) ReapStuckExecutions(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) { + if m.ReapStuckExecutionsFunc != nil { + return m.ReapStuckExecutionsFunc(ctx, reapAfter) + } + return &purchase.ReapResult{}, nil +} diff --git a/terraform/modules/compute/aws/lambda/main.tf b/terraform/modules/compute/aws/lambda/main.tf index 96feb5ba4..6700a924c 100644 --- a/terraform/modules/compute/aws/lambda/main.tf +++ b/terraform/modules/compute/aws/lambda/main.tf @@ -83,6 +83,13 @@ resource "aws_lambda_function" "main" { # that aws_lambda_function.main.arn would create here. SCHEDULER_LAMBDA_ARN = "arn:${data.aws_partition.current.partition}:lambda:${var.region}:${data.aws_caller_identity.current.account_id}:function:${var.stack_name}-api" }, + # Stuck-purchase reaper threshold (#678). Only set when the + # operator overrides the default — empty string means "use the + # in-code DefaultReapAfter (10m)" so we don't pin a Lambda env + # value when ops hasn't taken a position. + var.purchase_approved_reap_after != "" ? { + PURCHASE_APPROVED_REAP_AFTER = var.purchase_approved_reap_after + } : {}, var.additional_env_vars ) } @@ -526,3 +533,50 @@ resource "aws_lambda_permission" "eventbridge_ri_exchange" { principal = "events.amazonaws.com" source_arn = aws_cloudwatch_event_rule.ri_exchange[0].arn } + +# ============================================== +# EventBridge Rule for Stuck-Purchase Reaper (#678) +# ============================================== +# +# Periodic sweep that flips purchase_executions stuck in approved/running +# (longer than PURCHASE_APPROVED_REAP_AFTER, default 10m) to "failed" via +# the existing TransitionExecutionStatus CAS. Backstop for synchronous- +# executor crashes (Lambda timeout, OOM, network hang) that orphan rows +# in an in-flight state with no automatic recovery. +# +# Schedule cadence must be more frequent than the reap-after threshold so +# a stuck row is reaped within ~1 threshold-window — default rate(5m) +# vs. 10m threshold gives ~2 sweeps of headroom. The reaper itself is +# CAS-protected so over-running is safe (real executor wins the race). + +resource "aws_cloudwatch_event_rule" "reap_stuck_purchases" { + count = var.enable_reap_stuck_purchases_schedule ? 1 : 0 + + name = "${var.stack_name}-reap-stuck-purchases" + description = "Trigger stuck-purchase reaper sweep (issue #678)" + schedule_expression = var.reap_stuck_purchases_schedule + + tags = var.tags +} + +resource "aws_cloudwatch_event_target" "reap_stuck_purchases" { + count = var.enable_reap_stuck_purchases_schedule ? 1 : 0 + + rule = aws_cloudwatch_event_rule.reap_stuck_purchases[0].name + target_id = "lambda" + arn = aws_lambda_function.main.arn + + input = jsonencode({ + action = "reap_stuck_purchases" + }) +} + +resource "aws_lambda_permission" "eventbridge_reap_stuck_purchases" { + count = var.enable_reap_stuck_purchases_schedule ? 1 : 0 + + statement_id = "AllowExecutionFromEventBridgeReapStuckPurchases" + action = "lambda:InvokeFunction" + function_name = aws_lambda_function.main.function_name + principal = "events.amazonaws.com" + source_arn = aws_cloudwatch_event_rule.reap_stuck_purchases[0].arn +} diff --git a/terraform/modules/compute/aws/lambda/variables.tf b/terraform/modules/compute/aws/lambda/variables.tf index 05f2bc89f..131c9c676 100644 --- a/terraform/modules/compute/aws/lambda/variables.tf +++ b/terraform/modules/compute/aws/lambda/variables.tf @@ -156,6 +156,24 @@ variable "ri_exchange_schedule" { default = "rate(6 hours)" } +variable "enable_reap_stuck_purchases_schedule" { + description = "Enable scheduled sweep of purchase_executions stuck in approved/running. Issue #678 backstop for synchronous-executor crashes." + type = bool + default = true +} + +variable "reap_stuck_purchases_schedule" { + description = "EventBridge schedule for the stuck-purchase reaper. Should be more frequent than PURCHASE_APPROVED_REAP_AFTER (default 10m) so a stuck row is reaped within ~1 threshold-window. Default rate(5 minutes) gives the 10m threshold ~2 sweeps of headroom." + type = string + default = "rate(5 minutes)" +} + +variable "purchase_approved_reap_after" { + description = "Threshold age for the stuck-purchase reaper. Any execution sitting in approved/running longer than this gets flipped to failed on the next sweep. Parsed via Go time.ParseDuration (e.g. \"10m\", \"15m\", \"1h\"). Empty string falls back to the in-code default (10m)." + type = string + default = "" +} + variable "additional_env_vars" { description = "Additional environment variables" type = map(string) From 570a2be1df09879922c89747b0ff003c7921bd5c Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 22 May 2026 18:10:57 +0200 Subject: [PATCH 4/5] fix(purchase/reaper): distinguish CAS race-loss from real DB errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reaper's per-row error handler bucketed every failure from TransitionExecutionStatus into RaceLost, including real DB outages (connection refused, query timeout, etc.). That masked persistent store failures: a downed DB would surface only as a quiet count inflation on RaceLost, with no errored signal for ops to alert on. Wrap the two legitimate race-loss outcomes from TransitionExecutionStatus in sentinel errors so callers can use errors.Is rather than brittle string matching: - ErrExecutionNotInExpectedStatus: row exists but its status moved out of the allowed set before the CAS (the real executor finished between SELECT and UPDATE — race lost). - ErrNotFound (reused for the "row vanished" case): row was deleted between SELECT and UPDATE — also a race outcome, nothing for the reaper to do. In the reaper, classify errors.Is(err, ErrExecutionNotInExpectedStatus) or errors.Is(err, ErrNotFound) as RaceLost (INFO log); everything else bumps Errored with an ERROR log so the ops signal is visible. Addresses A1 from CodeRabbit round-1 review on #681. Regression tests: - TestReapStuckExecutions_CASRaceLostNoError now wraps the sentinel (was raw errors.New); guards the "race-loss is INFO-only" path. - TestReapStuckExecutions_RowVanishedTreatedAsRaceLost covers the ErrNotFound branch (manual DELETE between SELECT and CAS). - TestReapStuckExecutions_HardDBErrorClassifiedAsErrored asserts a non-sentinel error bumps Errored, not RaceLost — the regression guard for the original A1 finding. The wrapped errors keep the original message substrings ("not found", "cannot transition from ...") so existing call sites that match on substrings (e.g. internal/purchase/manager.go's RecoverStrandedApprovals, internal/purchase/approvals_test.go) are unaffected. --- internal/config/errors.go | 8 ++++ internal/config/store_postgres.go | 10 ++++- internal/purchase/reaper.go | 33 +++++++++++---- internal/purchase/reaper_test.go | 68 +++++++++++++++++++++++++++++-- 4 files changed, 104 insertions(+), 15 deletions(-) diff --git a/internal/config/errors.go b/internal/config/errors.go index 8ce0cf278..ff2a2de04 100644 --- a/internal/config/errors.go +++ b/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") diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 40e666604..869f62658 100644 --- a/internal/config/store_postgres.go +++ b/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 diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go index 516ebf611..4ca7d032f 100644 --- a/internal/purchase/reaper.go +++ b/internal/purchase/reaper.go @@ -2,6 +2,7 @@ package purchase import ( "context" + "errors" "fmt" "os" "time" @@ -154,16 +155,30 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r transitioned, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, stuckStatuses, failedStatus) if err != nil { - // CAS rejection is the "real executor beat us" path. We can't - // distinguish it from a hard DB error without inspecting the - // error string, which is brittle. Treat both as race_lost: the - // row is no longer in approved/running, so there is nothing for - // the reaper to do regardless of why. A persistent DB outage - // will surface via repeated Errored counts on the canonical- - // error save below, which IS treated as a real error. - logging.Infof("purchase reaper: CAS race lost for execution %s (status changed from %q before transition): %v", + // Distinguish CAS race-loss (the real executor finished between + // our SELECT and CAS, so the row is no longer in + // approved/running) from a hard DB error (connection dropped, + // query syntax error, etc.). The store wraps both legitimate + // race outcomes in sentinel errors so we can use errors.Is + // rather than brittle string matching: + // - ErrExecutionNotInExpectedStatus: row exists but its + // status moved out of approved/running before the CAS + // (the real executor won the race — expected, log INFO). + // - ErrNotFound: row vanished between SELECT and CAS (very + // rare — e.g. a manual DELETE; still a race outcome the + // reaper has nothing to do about, log INFO). + // Anything else is a real ops issue (DB outage, etc.) and + // must bump Errored so it surfaces in metrics/alerts instead + // of being silently absorbed as "race lost". + if errors.Is(err, config.ErrExecutionNotInExpectedStatus) || errors.Is(err, config.ErrNotFound) { + logging.Infof("purchase reaper: CAS race lost for execution %s (status changed from %q before transition): %v", + exec.ExecutionID, prevStatus, err) + result.RaceLost++ + return + } + logging.Errorf("purchase reaper: transition failed for execution %s (status=%q): %v", exec.ExecutionID, prevStatus, err) - result.RaceLost++ + result.Errored++ return } if transitioned == nil { diff --git a/internal/purchase/reaper_test.go b/internal/purchase/reaper_test.go index 3ad38e3e6..90dd084ba 100644 --- a/internal/purchase/reaper_test.go +++ b/internal/purchase/reaper_test.go @@ -3,6 +3,7 @@ package purchase import ( "context" "errors" + "fmt" "strings" "testing" "time" @@ -150,9 +151,11 @@ func TestReapStuckExecutions_TerminalStatusNotTouched(t *testing.T) { func TestReapStuckExecutions_CASRaceLostNoError(t *testing.T) { // CAS race: the SELECT returns a stuck row, but between SELECT and - // CAS, the real executor flips the row to completed. The CAS rejects - // with "cannot transition from X". The reaper must log and move on - // without erroring the sweep. + // CAS, the real executor flips the row to completed. The store wraps + // the rejection in ErrExecutionNotInExpectedStatus so the reaper can + // use errors.Is to recognise the race outcome and move on without + // erroring the sweep — this regression-guards the A1 CR finding + // (must not classify all CAS errors as race-lost). ctx := context.Background() store := new(MockConfigStore) reapAfter := 10 * time.Minute @@ -161,7 +164,8 @@ func TestReapStuckExecutions_CASRaceLostNoError(t *testing.T) { store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). Return([]config.PurchaseExecution{row}, nil) store.On("TransitionExecutionStatus", ctx, "exec-race", stuckStatuses, failedStatus). - Return(nil, errors.New("execution exec-race cannot transition from \"completed\" to \"failed\"")) + Return(nil, fmt.Errorf("%w: execution exec-race cannot transition from %q to %q", + config.ErrExecutionNotInExpectedStatus, "completed", "failed")) // No SavePurchaseExecution expectation — we lost the race, the real // executor's status flip stands. @@ -171,6 +175,62 @@ func TestReapStuckExecutions_CASRaceLostNoError(t *testing.T) { assert.Equal(t, 1, result.Found) assert.Equal(t, 0, result.Reaped) assert.Equal(t, 1, result.RaceLost) + assert.Equal(t, 0, result.Errored) + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +func TestReapStuckExecutions_RowVanishedTreatedAsRaceLost(t *testing.T) { + // Defensive: between SELECT and CAS the row might be deleted (e.g. + // manual DBA intervention). The store wraps that case in + // config.ErrNotFound; the reaper must treat it as a race-lost (the + // row is no longer in approved/running, nothing for the reaper to do) + // rather than as a real DB error. Regression-guards the second half + // of the A1 CR finding's sentinel handling. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-gone", "approved") + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-gone", stuckStatuses, failedStatus). + Return(nil, fmt.Errorf("%w: execution exec-gone", config.ErrNotFound)) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) + assert.Equal(t, 1, result.Found) + assert.Equal(t, 0, result.Reaped) + assert.Equal(t, 1, result.RaceLost) + assert.Equal(t, 0, result.Errored) + store.AssertExpectations(t) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) +} + +func TestReapStuckExecutions_HardDBErrorClassifiedAsErrored(t *testing.T) { + // The A1 fix's whole point: a non-sentinel error from + // TransitionExecutionStatus (DB connection dropped, query syntax + // error, etc.) is NOT a race-loss — it must bump Errored so the ops + // signal is visible. Before A1, this was silently absorbed as + // RaceLost; this test is the regression guard. + ctx := context.Background() + store := new(MockConfigStore) + reapAfter := 10 * time.Minute + + row := stuckExec("exec-dbflake", "running") + store.On("ListStuckExecutions", ctx, stuckStatuses, reapAfter). + Return([]config.PurchaseExecution{row}, nil) + store.On("TransitionExecutionStatus", ctx, "exec-dbflake", stuckStatuses, failedStatus). + Return(nil, errors.New("connection refused")) + + mgr := newReaperManager(store) + result, err := mgr.ReapStuckExecutions(ctx, reapAfter) + require.NoError(t, err) // sweep itself still succeeds; per-row errors don't propagate + assert.Equal(t, 1, result.Found) + assert.Equal(t, 0, result.Reaped) + assert.Equal(t, 0, result.RaceLost, "real DB errors must NOT be classified as race-lost") + assert.Equal(t, 1, result.Errored, "real DB errors must bump Errored so ops can see the outage") store.AssertExpectations(t) store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } From 892e0c8584e171133ef5b349d2bc453721b457b7 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 22 May 2026 18:12:17 +0200 Subject: [PATCH 5/5] fix(purchase/reaper): reject non-positive reap durations + stabilize env tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ParseReapAfterFromEnv accepted "0s" and negative durations such as "-5m" because time.ParseDuration treats them as syntactically valid. Feeding either into ListStuckExecutions would be catastrophic: - 0s: the cutoff "updated_at < NOW() - 0s" matches every approved/running row regardless of age — the reaper would flip fresh, in-flight executions to failed. - -5m: the cutoff "updated_at < NOW() - (-5m)" == "updated_at < NOW() + 5m" matches rows from 5 minutes in the future, i.e. effectively every row. Same outcome. The store-side guard added in the original PR (ListStuckExecutions rejects olderThan <= 0) prevents the broken SELECT from executing, but a misconfigured env value would still cause every sweep to fail silently with a confusing store error rather than a WARN at the config-parse boundary. Reject non-positive durations at the env-parse layer with a WARN log + fallback to DefaultReapAfter so the misconfig is visible in the Lambda's startup logs. Addresses A2 + A3 (defense-in-depth) + A4 + A5 from CodeRabbit round-1 review on #681. Regression tests (A4): - TestParseReapAfterFromEnv_ZeroFallsBackToDefault — "0s" - TestParseReapAfterFromEnv_NegativeFallsBackToDefault — "-5m" - TestParseReapAfterFromEnv_GarbageFallsBackToDefault — explicit "garbage" case complementing the existing _Invalid* test, named per the CR finding so the regression intent is searchable. handler_test.go (A5): - TestHandleScheduledTask's table-driven loop now calls t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") inside the per-case sub-Run. The two reap_stuck_purchases cases assert reapAfter == 10*time.Minute (the default); without an explicit env clear, an ambient PURCHASE_APPROVED_REAP_AFTER in CI/dev would silently make them flaky. t.Setenv auto-restores the prior value at cleanup. The store-side olderThan <= 0 guard (A3) was already folded into the earlier rebased commit "feat(config): add ListStuckExecutions store method" so no additional store-side change is needed here — A3 is covered by the existing defense-in-depth, and this commit completes the env-side validation A2 calls for. --- internal/purchase/reaper.go | 21 +++++++++++++++++---- internal/purchase/reaper_test.go | 29 +++++++++++++++++++++++++++++ internal/server/handler_test.go | 8 ++++++++ 3 files changed, 54 insertions(+), 4 deletions(-) diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go index 4ca7d032f..64c589752 100644 --- a/internal/purchase/reaper.go +++ b/internal/purchase/reaper.go @@ -67,10 +67,18 @@ type ReapResult struct { // ParseReapAfterFromEnv reads PURCHASE_APPROVED_REAP_AFTER from the // environment and parses it via time.ParseDuration. Falls back to -// DefaultReapAfter on either an absent env var or a parse failure — and -// logs a WARN on the parse failure so ops can spot a typo without the -// reaper silently running at the default. Never panics: a misconfigured -// env var must not crash the Lambda's other scheduled tasks. +// DefaultReapAfter on either an absent env var, a parse failure, or a +// non-positive duration (0s / negative) — each variant logs a WARN so ops +// can spot a typo without the reaper silently running at the default. +// Never panics: a misconfigured env var must not crash the Lambda's other +// scheduled tasks. +// +// Non-positive values are explicitly rejected: time.ParseDuration accepts +// "0s" and "-5m" as valid, but feeding them into ListStuckExecutions would +// either match every row (0s) or invert the SELECT into "updated_at < +// NOW() + |d|" (negative) and reap fresh executions. The store has its +// own guard (defense-in-depth) but rejecting here keeps the misconfig +// visible in the WARN log rather than as a confusing store error. func ParseReapAfterFromEnv() time.Duration { raw := os.Getenv(reapAfterEnvVar) if raw == "" { @@ -82,6 +90,11 @@ func ParseReapAfterFromEnv() time.Duration { reapAfterEnvVar, raw, err, DefaultReapAfter) return DefaultReapAfter } + if d <= 0 { + logging.Warnf("purchase reaper: invalid %s=%q (must be > 0) — using default %s", + reapAfterEnvVar, raw, DefaultReapAfter) + return DefaultReapAfter + } return d } diff --git a/internal/purchase/reaper_test.go b/internal/purchase/reaper_test.go index 90dd084ba..05a5305aa 100644 --- a/internal/purchase/reaper_test.go +++ b/internal/purchase/reaper_test.go @@ -353,6 +353,35 @@ func TestParseReapAfterFromEnv_InvalidFallsBackToDefault(t *testing.T) { assert.Equal(t, DefaultReapAfter, got, "malformed env value must fall back to default, not crash") } +func TestParseReapAfterFromEnv_GarbageFallsBackToDefault(t *testing.T) { + // Explicit garbage-input case named per the A4 CR finding so the + // regression intent is searchable. Complements + // _InvalidFallsBackToDefault above. + t.Setenv(reapAfterEnvVar, "garbage") + got := ParseReapAfterFromEnv() + assert.Equal(t, DefaultReapAfter, got) +} + +func TestParseReapAfterFromEnv_ZeroFallsBackToDefault(t *testing.T) { + // "0s" is a syntactically valid Go duration but would make the SELECT + // match every approved/running row regardless of age — the reaper + // would flip in-flight executions. A2 rejects non-positive durations + // at parse time; this test is the regression guard. + t.Setenv(reapAfterEnvVar, "0s") + got := ParseReapAfterFromEnv() + assert.Equal(t, DefaultReapAfter, got, "0s must fall back to default — would otherwise reap fresh executions") +} + +func TestParseReapAfterFromEnv_NegativeFallsBackToDefault(t *testing.T) { + // "-5m" parses cleanly as a negative duration. Passed unchanged into + // the store, it would invert the cutoff: "updated_at < NOW() - (-5m)" + // == "updated_at < NOW() + 5m" — reaping rows from 5 minutes in the + // future, i.e. effectively every row. A2 rejects this at parse time. + t.Setenv(reapAfterEnvVar, "-5m") + got := ParseReapAfterFromEnv() + assert.Equal(t, DefaultReapAfter, got, "negative duration must fall back to default — would otherwise reap fresh executions") +} + func TestParseReapAfterFromEnv_NonStandardButValidGoDuration(t *testing.T) { // Sanity check Go's parser accepts non-minute units — ops may // reasonably set e.g. "2h" for a relaxed cadence. diff --git a/internal/server/handler_test.go b/internal/server/handler_test.go index 179c4eff9..aa58c8411 100644 --- a/internal/server/handler_test.go +++ b/internal/server/handler_test.go @@ -122,6 +122,14 @@ func TestHandleScheduledTask(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + // Clear PURCHASE_APPROVED_REAP_AFTER so the reap_stuck_purchases + // cases below see the deterministic default (10m) regardless of + // ambient env in CI/dev. The reap subtests assert reapAfter == + // 10*time.Minute in their setupMocks; without this, an + // inherited env value would silently make them flaky (A5 CR). + // t.Setenv automatically restores the prior value at cleanup. + t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") + ctx := testutil.TestContext(t) mockScheduler := &testutil.MockScheduler{}