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/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/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..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 @@ -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 := ` 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/purchase/reaper.go b/internal/purchase/reaper.go new file mode 100644 index 000000000..64c589752 --- /dev/null +++ b/internal/purchase/reaper.go @@ -0,0 +1,228 @@ +package purchase + +import ( + "context" + "errors" + "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, 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 == "" { + 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 + } + if d <= 0 { + logging.Warnf("purchase reaper: invalid %s=%q (must be > 0) — using default %s", + reapAfterEnvVar, raw, 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 { + // 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.Errored++ + 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..05a5305aa --- /dev/null +++ b/internal/purchase/reaper_test.go @@ -0,0 +1,391 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + "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 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 + + 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, 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. + + 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) + 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) +} + +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_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. + t.Setenv(reapAfterEnvVar, "2h30m") + got := ParseReapAfterFromEnv() + assert.Equal(t, 2*time.Hour+30*time.Minute, got) +} 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/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..aa58c8411 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"), @@ -96,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{} @@ -139,6 +173,7 @@ func TestTaskLockID(t *testing.T) { TaskCleanupExpiredRecords, TaskRefreshAnalytics, TaskRIExchangeReshape, + TaskReapStuckPurchases, } seen := make(map[int64]ScheduledTaskType) for _, task := range tasks { @@ -311,6 +346,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/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 } 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)