fix(purchase): reap purchase_executions stuck in approved/running >10m (closes #678) - #681
Merged
cristim merged 5 commits intoMay 22, 2026
Conversation
Contributor
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (15)
📝 WalkthroughWalkthroughThis PR implements a background "reaper" service that automatically rescues purchase executions orphaned in ChangesStuck Purchase Execution Reaper
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related issues
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Member
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (2)
internal/config/interfaces.go (1)
52-53: ⚡ Quick winFix ordering documentation contradiction.
Line 52 says “Newest-stuck-first” but
ORDER BY updated_at ASCis oldest-first. Please align the wording to avoid future behavioral drift.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/config/interfaces.go` around lines 52 - 53, The comment currently says “Newest-stuck-first” but the SQL shown uses ORDER BY updated_at ASC (oldest first); update the documentation comment to match the actual ordering by changing the phrase to “Oldest-stuck-first (ORDER BY updated_at ASC)” so the description and the ORDER BY (updated_at ASC) and related symbol MaxListLimit remain consistent.internal/config/store_postgres.go (1)
820-823: ⚡ Quick winCommented ordering does not match SQL ordering.
Lines 820-823 describe newest-first processing, but Line 842 is
ORDER BY updated_at ASC(oldest-first). Please make the comment match the query semantics.Also applies to: 842-843
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/config/store_postgres.go` around lines 820 - 823, The block comment above the sweep query currently claims "newest-first" but the SQL uses ORDER BY updated_at ASC (oldest-first); update the comment in internal/config/store_postgres.go to reflect that rows are returned oldest-first (so the oldest stuck rows are processed first) to match the SQL semantics, and likewise adjust the similar comment near the other occurrence (lines 842-843) so both comments accurately describe the ORDER BY updated_at ASC behavior rather than "newest-first".
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/config/store_postgres.go`:
- Around line 829-846: ListStuckExecutions builds an SQL interval from olderThan
without validating it, so zero/negative durations produce an overly-broad
cutoff; add a guard at the start of PostgresStore.ListStuckExecutions to check
that olderThan > 0 and return early (e.g., nil, nil or a clear error) if not,
before constructing intervalArg (the fmt.Sprintf call) and before calling
s.queryExecutions, to prevent selecting fresh executions.
In `@internal/purchase/reaper_test.go`:
- Around line 278-302: Add two regression tests that verify
ParseReapAfterFromEnv treats "0s" and negative durations as invalid and returns
DefaultReapAfter: create tests named e.g.
TestParseReapAfterFromEnv_ZeroFallsBackToDefault and
TestParseReapAfterFromEnv_NegativeFallsBackToDefault; in each use
t.Setenv(reapAfterEnvVar, "0s") and t.Setenv(reapAfterEnvVar, "-5m")
respectively, call ParseReapAfterFromEnv(), and assert.Equal(t,
DefaultReapAfter, got) to ensure zero and negative values fall back to the
default just like malformed input.
In `@internal/purchase/reaper.go`:
- Around line 78-85: The parsed duration may be zero or negative (e.g., "0s" or
"-5m") which causes immediate/incorrect reaping; after calling
time.ParseDuration(raw) in the parsing function, validate that the resulting d >
0 and if not call logging.Warnf (similar to the existing error path) indicating
the invalid non-positive value for reapAfterEnvVar and raw, then return
DefaultReapAfter instead of the non-positive duration; keep the existing
error-warning behavior for parse failures and ensure you reference the same
symbols (time.ParseDuration, d, raw, reapAfterEnvVar, DefaultReapAfter,
logging.Warnf).
- Around line 155-167: The current handler treats all errors from
m.config.TransitionExecutionStatus as a CAS "race lost"; instead, detect
CAS-specific rejection errors (e.g. compare with the CAS/conflict sentinel via
errors.Is or a type assertion on the returned error from
TransitionExecutionStatus) and only increment result.RaceLost and return for
that specific case; for any other error, increment result.Errored (or propagate
the error) and log it via logging.Infof/other logger so real DB/store outages
are reported. Use the existing identifiers m.config.TransitionExecutionStatus,
exec.ExecutionID, prevStatus, result.RaceLost and result.Errored when making the
change.
In `@internal/server/handler_test.go`:
- Around line 90-114: The two table-driven cases for TaskReapStuckPurchases rely
on PURCHASE_APPROVED_REAP_AFTER being unset and are flaky in CI; stabilize them
by explicitly clearing or setting that env var in the test (use
t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") or set to "10m") before invoking
the test logic so ParseReapAfterFromEnv yields the expected default; add the
t.Setenv call in the test body that runs these cases (the entries named
"reap_stuck_purchases success" and "reap_stuck_purchases propagates store
error") or inside their setupMocks invocation so the ReapStuckExecutionsFunc
assertions see the deterministic reapAfter value.
---
Nitpick comments:
In `@internal/config/interfaces.go`:
- Around line 52-53: The comment currently says “Newest-stuck-first” but the SQL
shown uses ORDER BY updated_at ASC (oldest first); update the documentation
comment to match the actual ordering by changing the phrase to
“Oldest-stuck-first (ORDER BY updated_at ASC)” so the description and the ORDER
BY (updated_at ASC) and related symbol MaxListLimit remain consistent.
In `@internal/config/store_postgres.go`:
- Around line 820-823: The block comment above the sweep query currently claims
"newest-first" but the SQL uses ORDER BY updated_at ASC (oldest-first); update
the comment in internal/config/store_postgres.go to reflect that rows are
returned oldest-first (so the oldest stuck rows are processed first) to match
the SQL semantics, and likewise adjust the similar comment near the other
occurrence (lines 842-843) so both comments accurately describe the ORDER BY
updated_at ASC behavior rather than "newest-first".
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ccb72ab2-6947-4bcd-b2ed-dafeb088c83d
📒 Files selected for processing (16)
internal/analytics/collector_test.gointernal/api/mocks_test.gointernal/config/interfaces.gointernal/config/store_postgres.gointernal/config/store_postgres_pgxmock_test.gointernal/purchase/mocks_test.gointernal/purchase/reaper.gointernal/purchase/reaper_test.gointernal/scheduler/scheduler_test.gointernal/server/handler.gointernal/server/handler_test.gointernal/server/interfaces.gointernal/server/test_helpers_test.gointernal/testutil/mocks.goterraform/modules/compute/aws/lambda/main.tfterraform/modules/compute/aws/lambda/variables.tf
Comment on lines
+829
to
+846
| func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []string, olderThan time.Duration) ([]PurchaseExecution, error) { | ||
| if len(statuses) == 0 { | ||
| return nil, nil | ||
| } | ||
| 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 | ||
| 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) |
Contributor
There was a problem hiding this comment.
Guard non-positive olderThan before building the SQL interval.
Line 845 allows 0 or negative durations to produce a dangerously broad cutoff. That can select fresh executions and trigger incorrect reaps.
💡 Proposed fix
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("olderThan must be > 0, got %s", olderThan)
+ }
query := `
SELECT plan_id, execution_id, status, step_number, scheduled_date,
notification_sent, approval_token, recommendations,
@@
ORDER BY updated_at ASC
LIMIT $3
`
- intervalArg := fmt.Sprintf("%d seconds", int(olderThan.Seconds()))
+ intervalArg := fmt.Sprintf("%d seconds", int64((olderThan+time.Second-1)/time.Second))
return s.queryExecutions(ctx, query, statuses, intervalArg, MaxListLimit)
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/config/store_postgres.go` around lines 829 - 846,
ListStuckExecutions builds an SQL interval from olderThan without validating it,
so zero/negative durations produce an overly-broad cutoff; add a guard at the
start of PostgresStore.ListStuckExecutions to check that olderThan > 0 and
return early (e.g., nil, nil or a clear error) if not, before constructing
intervalArg (the fmt.Sprintf call) and before calling s.queryExecutions, to
prevent selecting fresh executions.
Comment on lines
+278
to
+302
| 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) | ||
| } |
Contributor
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Add regression tests for 0s and negative env durations.
Given the reaper’s impact, add explicit tests that PURCHASE_APPROVED_REAP_AFTER=0s and negative values fall back to DefaultReapAfter (same as malformed input).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/purchase/reaper_test.go` around lines 278 - 302, Add two regression
tests that verify ParseReapAfterFromEnv treats "0s" and negative durations as
invalid and returns DefaultReapAfter: create tests named e.g.
TestParseReapAfterFromEnv_ZeroFallsBackToDefault and
TestParseReapAfterFromEnv_NegativeFallsBackToDefault; in each use
t.Setenv(reapAfterEnvVar, "0s") and t.Setenv(reapAfterEnvVar, "-5m")
respectively, call ParseReapAfterFromEnv(), and assert.Equal(t,
DefaultReapAfter, got) to ensure zero and negative values fall back to the
default just like malformed input.
Comment on lines
+78
to
+85
| 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 | ||
| } |
Contributor
There was a problem hiding this comment.
Reject non-positive reap durations in env parsing.
time.ParseDuration accepts values like 0s and -5m. Passing those through can cause immediate/incorrect reaping of active executions. Validate d > 0 and fall back to default with a warning.
💡 Proposed fix
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
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| 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 | |
| } | |
| 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 | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/purchase/reaper.go` around lines 78 - 85, The parsed duration may be
zero or negative (e.g., "0s" or "-5m") which causes immediate/incorrect reaping;
after calling time.ParseDuration(raw) in the parsing function, validate that the
resulting d > 0 and if not call logging.Warnf (similar to the existing error
path) indicating the invalid non-positive value for reapAfterEnvVar and raw,
then return DefaultReapAfter instead of the non-positive duration; keep the
existing error-warning behavior for parse failures and ensure you reference the
same symbols (time.ParseDuration, d, raw, reapAfterEnvVar, DefaultReapAfter,
logging.Warnf).
Comment on lines
+155
to
+167
| 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 |
Contributor
There was a problem hiding this comment.
Do not classify all CAS errors as race-lost.
Line 156 currently buckets every transition failure into RaceLost, including real DB/store errors. That masks outages and underreports Errored.
💡 Proposed fix
+import "strings"
@@
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++
+ errMsg := err.Error()
+ if strings.Contains(errMsg, "cannot transition") || strings.Contains(errMsg, "not found") {
+ 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: %v", exec.ExecutionID, err)
+ result.Errored++
return
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| 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 | |
| transitioned, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, stuckStatuses, failedStatus) | |
| if err != nil { | |
| errMsg := err.Error() | |
| if strings.Contains(errMsg, "cannot transition") || strings.Contains(errMsg, "not found") { | |
| 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: %v", exec.ExecutionID, err) | |
| result.Errored++ | |
| return | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/purchase/reaper.go` around lines 155 - 167, The current handler
treats all errors from m.config.TransitionExecutionStatus as a CAS "race lost";
instead, detect CAS-specific rejection errors (e.g. compare with the
CAS/conflict sentinel via errors.Is or a type assertion on the returned error
from TransitionExecutionStatus) and only increment result.RaceLost and return
for that specific case; for any other error, increment result.Errored (or
propagate the error) and log it via logging.Infof/other logger so real DB/store
outages are reported. Use the existing identifiers
m.config.TransitionExecutionStatus, exec.ExecutionID, prevStatus,
result.RaceLost and result.Errored when making the change.
Comment on lines
+90
to
+114
| { | ||
| 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, | ||
| }, |
Contributor
There was a problem hiding this comment.
Stabilize env-dependent reaper tests with explicit t.Setenv.
These cases rely on PURCHASE_APPROVED_REAP_AFTER being unset (see Line 95 comment), which can be false in CI/runtime environments and make the test flaky.
✅ Suggested fix
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
+ t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "")
ctx := testutil.TestContext(t)
mockScheduler := &testutil.MockScheduler{}
mockPurchase := &testutil.MockPurchaseManager{}
tt.setupMocks(mockScheduler, mockPurchase)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/server/handler_test.go` around lines 90 - 114, The two table-driven
cases for TaskReapStuckPurchases rely on PURCHASE_APPROVED_REAP_AFTER being
unset and are flaky in CI; stabilize them by explicitly clearing or setting that
env var in the test (use t.Setenv("PURCHASE_APPROVED_REAP_AFTER", "") or set to
"10m") before invoking the test logic so ParseReapAfterFromEnv yields the
expected default; add the t.Setenv call in the test body that runs these cases
(the entries named "reap_stuck_purchases success" and "reap_stuck_purchases
propagates store error") or inside their setupMocks invocation so the
ReapStuckExecutionsFunc assertions see the deterministic reapAfter value.
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.
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)
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.
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.
…env tests
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.
cristim
force-pushed
the
fix/issue-678-stuck-approved-reaper
branch
from
May 22, 2026 16:13
c7bf86e to
892e0c8
Compare
Member
Author
|
Rebased onto latest A1 (Major, reaper.go) — A2 (Major, reaper.go) — A3 (Major, store_postgres.go): Already folded into the rebased A4 (Major, reaper_test.go) — A5 (Minor, handler_test.go) — Verification: |
Member
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
cristim
added a commit
that referenced
this pull request
May 25, 2026
…K retries Introduce providers/aws/internal/purchasecfg.NewConfig which returns a copy of the caller's aws.Config with RetryMaxAttempts=2 and http.Client.Timeout=15s. Worst-case wall clock: 2 * 15s = 30s, well inside the 300s Lambda budget. Recommendation-collection clients (recommendations.NewClient) are explicitly NOT touched; they call Cost Explorer and need the higher default retry budget. Adds five unit tests covering: retry cap, HTTP timeout, immutability of the base config, region preservation, and the wall-clock bound invariant. Closes #683. Refs #667 #668 #632 #681.
cristim
added a commit
that referenced
this pull request
May 25, 2026
Raise lambda_timeout from 60s to 300s in github-dev, github-staging, and github-prod tfvars. The SDK retry tightening (purchasecfg, max 30s wall clock) combined with the increased Lambda budget means transient AWS API slowness either completes within 30s or surfaces a clear error to the user -- no more silent context deadline exceeded from the Lambda itself expiring mid-retry. Lambda Function URL has no upstream timeout ceiling, so 300s is safe. Closes #683. Refs #667 #681.
cristim
added a commit
that referenced
this pull request
May 25, 2026
… 300s (closes #683) (#684) * feat(providers/aws): add purchasecfg helper to bound purchase-path SDK retries Introduce providers/aws/internal/purchasecfg.NewConfig which returns a copy of the caller's aws.Config with RetryMaxAttempts=2 and http.Client.Timeout=15s. Worst-case wall clock: 2 * 15s = 30s, well inside the 300s Lambda budget. Recommendation-collection clients (recommendations.NewClient) are explicitly NOT touched; they call Cost Explorer and need the higher default retry budget. Adds five unit tests covering: retry cap, HTTP timeout, immutability of the base config, region preservation, and the wall-clock bound invariant. Closes #683. Refs #667 #668 #632 #681. * fix(purchases): wire purchasecfg into all 7 purchase-path SDK constructors Switch EC2, RDS, ElastiCache, MemoryDB, OpenSearch, Redshift, and SavingsPlans NewClient calls to use purchasecfg.NewConfig(cfg) so every purchase-path SDK client gets RetryMaxAttempts=2 and http.Client.Timeout=15s. OpenSearch and Redshift also pass the tightened config to their STS client (used for ARN construction / tagging on the purchase path). Recommendation-collection clients (recommendations.NewClient) are not touched; they use their own constructor and deliberately keep the SDK default retry budget for Cost Explorer calls. All 694 tests pass. go build ./... clean. Closes #683. Refs #667. * fix(terraform): bump Lambda timeout 60 -> 300s in all three env tfvars Raise lambda_timeout from 60s to 300s in github-dev, github-staging, and github-prod tfvars. The SDK retry tightening (purchasecfg, max 30s wall clock) combined with the increased Lambda budget means transient AWS API slowness either completes within 30s or surfaces a clear error to the user -- no more silent context deadline exceeded from the Lambda itself expiring mid-retry. Lambda Function URL has no upstream timeout ceiling, so 300s is safe. Closes #683. Refs #667 #681. * feat(purchase): cap per-rec execution at 30s via context.WithTimeout Each recommendation in processPurchaseRecommendations now runs under a 30s per-rec deadline derived from context.WithTimeout. This matches the purchasecfg hard limits (2 retries * 15s HTTP timeout) so a single hung rec (SDK retry storm, STS hang, or pagination hang) cannot exhaust the Lambda budget and starve the remaining recs in the same execution. On timeout, a purchase[<id>]: per-rec deadline 30s exceeded log line fires so CloudWatch pinpoints which rec hung and what the elapsed time was, without requiring a second failure to reproduce. Closes part of issue #683. * feat(purchase/api): add approve handler entry/exit diagnostic logging Adds purchase[<exec-id>]: tagged timing logs at every stage of the approval path so CloudWatch can pinpoint exactly where a future hang occurs: - approvePurchaseViaSession: entry (auth=session) + elapsed on exit - ApproveExecution: entry (auth=token) + elapsed on exit for both the preflight-reject path and the full execute path - ApproveAndExecute: entry + status-transition elapsed + execute elapsed Actor emails are masked to ***@Domain before logging (PII rule). Part of issue #683 diagnostic logging extension. * feat(purchase): add timing logs to resolveAccountProvider and sub-resolvers Adds purchase[resolveAccountProvider/resolveAWSProvider/resolveAzureProvider/ resolveGCPProvider]: entry + elapsed-on-exit logs so CloudWatch can show whether a hang is in credential resolution (STS AssumeRole, OIDC token exchange) vs the cloud API call itself. Each log includes provider, account ID, auth mode, and elapsed time. Error paths also log before returning so a failed resolution is visible in the trace even if the caller swallows the error in an aggregator. Part of issue #683 diagnostic logging extension. * feat(providers): add STS GetCallerIdentity and Azure subscriptions timing logs Instruments the two network calls that ValidateCredentials makes during provider construction so CloudWatch distinguishes a credential-validation hang from the subsequent offering-search or purchase hang: - providers/aws/provider.go: purchase[STS]: GetCallerIdentity starting / returned in <elapsed> / failed after <elapsed> - providers/azure/provider.go: purchase[Azure]: subscriptions.NewListPager/ NextPage starting / returned in <elapsed> / failed after <elapsed> Both use the purchase[...]: prefix for grep-ability. Profile/subscription context is logged but no tokens, key material, or full identities. Part of issue #683 diagnostic logging extension. * feat(providers/aws/services): add per-page timing logs to findOfferingID in all 7 AWS service clients Adds purchase[<exec-id>]: tagged per-pagination-page timing logs to the findOfferingID loop in each of the 7 AWS purchase-path service clients (EC2, RDS, ElastiCache, MemoryDB, OpenSearch, Redshift, SavingsPlans). Each log line records: - Entry: service, instance/node type, term, payment option - Per page: page number, offering count returned, page elapsed - Hit: page number where match was found, total elapsed - Exhausted: total pages scanned, total elapsed (no-match path) The exec-id is threaded from PurchaseCommitment(opts.ExecutionID) into findOfferingID so all lines share the same purchase[<uuid>]: prefix and a single CloudWatch Logs Insights query can reconstruct the full pagination trace for any hanging execution. ValidateOffering and GetOfferingDetails callers pass "" (rendered as "no-exec") since they run outside the purchase flow. Part of issue #683 diagnostic logging extension. * feat(execution): add per-goroutine entry/exit timing in FanOutWithConcurrency Adds purchase[fan-out]: entry + elapsed-on-exit log lines to each goroutine launched by FanOutWithConcurrency. Both the account fan-out (multi-account execution) and the per-rec fan-out (processPurchaseRec- ommendations) go through FanOutWithConcurrency, so one addition covers both call sites. Logs: - purchase[fan-out]: goroutine starting for account=<id> - purchase[fan-out]: goroutine for account=<id> completed in <elapsed> - purchase[fan-out]: goroutine for account=<id> failed after <elapsed>: <err> Together with the per-rec ctx.WithTimeout and findOfferingID page logs, a CloudWatch Logs Insights query on purchase[ now reconstructs the full timing tree for a stuck execution: fan-out goroutine start -> credential resolution -> STS -> service client -> findOfferingID pagination -> PurchaseCommitment. Part of issue #683 diagnostic logging extension. * fix(purchase/tests): fix per-rec context.WithTimeout breaking mock matchers Move context.WithTimeout from the fan-out closure (processPurchaseRecommendations) into executeSinglePurchase where it belongs -- the budget governs the cloud API call chain, not the fan-out bookkeeping. Update all GetServiceClient and PurchaseCommitment mock.On() setups in execution_test.go, coverage_extra_test.go, and manager_test.go to match on deadline-carrying contexts via: mock.MatchedBy(func(c context.Context) bool { _, ok := c.Deadline(); return ok }) This positively asserts the 30s per-rec timeout is propagated to every cloud SDK call rather than silently accepting any context (mock.Anything). Also fix the savingsplans/client_test.go build error: findOfferingID gained an execID string parameter in commit 5 but the direct test call was not updated (only the PurchaseCommitment path was updated). All 180 purchase package tests and 308 AWS service tests now pass. Part of issue #683 diagnostic logging extension. * fix(purchase): address CodeRabbit review on PR #684 Three changes per CR pass-1 triage: 1. Actionable (execution.go): distinguish context.DeadlineExceeded from context.Canceled in the post-PurchaseCommitment error log. The previous branch logged "per-rec deadline 30s exceeded" for any non-nil recCtx.Err(), including parent cancellations -- mislabeling Canceled as a timeout makes diagnostic triage harder. Logic extracted into logRecCtxErr helper to keep executeSinglePurchase cyclomatic complexity within the gocyclo limit. 2. Nitpick (execution_test.go): replace mock.Anything with a deadline-aware mock.MatchedBy on the first arg of all 12 CreateAndValidateProvider expectations. The 30s per-rec budget is set before the factory call, so all three mocked stages (factory, GetServiceClient, PurchaseCommitment) now assert deadline propagation end-to-end. 3. Nitpick (purchasecfg/config.go + config_test.go): preserve the caller's *http.Client when one is provided in base config. NewConfig now shallow- copies the client and only overrides Timeout, keeping any custom Transport, Jar, or CheckRedirect. Falls back to a fresh client when base.HTTPClient is nil or a non-*http.Client implementation. Adds TestNewConfig_PreservesCustomTransport to pin the contract. * test(purchase): tighten ctx matcher to assert per-rec 30s budget (CR #684) CodeRabbit on PR #684 (review 4348606803) noted that the 53 mock matchers using `_, ok := c.Deadline(); return ok` only assert *some* deadline is set -- they would still pass if `executeSinglePurchase` silently reused the parent's longer deadline instead of its own `context.WithTimeout(ctx, 30*time.Second)` wrap. Tighten the contract: introduce `hasPerRecDeadline(max time.Duration)` in `execution_test.go` and replace all 53 sites across `execution_test.go`, `coverage_extra_test.go`, and `manager_test.go` with `mock.MatchedBy(hasPerRecDeadline(30*time.Second))`. The matcher now positively asserts the remaining deadline is in (0, 30s], so removing or loosening the wrap fails the test. * refactor(aws/services): extract findOfferingID helpers to fit gocyclo budget memorydb: extract scanMemoryDBOfferingPage (per-page offering scan with mismatch guard) and isLastMemoryDBPage (terminal-page check) from findOfferingID, mirroring the helpers already in the ec2 package. ec2: extract buildEC2QueryFromRec (details type-assertion + payment-option conversion + query assembly) from findOfferingID into a standalone method. Both findOfferingID functions now sit at cyclomatic complexity ≤10. No behaviour change -- pure extraction verified by 72 passing unit tests.
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds a backstop reaper for
purchase_executionsrows orphaned inapproved/runningstate when the synchronous executor dies mid-flight (Lambda timeout, OOM, network hang, container kill). The reaper runs as a scheduled task every 5 minutes, finds rows older thanPURCHASE_APPROVED_REAP_AFTER(default10m), and atomically transitions them tofailedvia the existingTransitionExecutionStatusCAS — never touching the provider commitment. Closes #678.Root cause
Manager.ApproveAndExecute(internal/purchase/approvals.go) persistsStatus = "approved"BEFORE invokingexecuteAndFinalize. The terminal flip tocompleted/failedonly happens insidefinalizeExecutionAFTERexecutePurchasereturns. Any interruption between those two points (timeout / OOM / network hang / context cancellation) orphans the row indefinitely — no further state machinery exists to advance it.Fix shape
Purely additive — no changes to
ApproveAndExecuteor the executor itself.internal/config: newListStuckExecutions(ctx, statuses, olderThan)store method (StoreInterface+store_postgres.go) + pgxmock coverage. Mirrors the existingGetStaleProcessingExchangesshape used by the RI-exchange reshape sweep.internal/purchase/reaper.go: newManager.ReapStuckExecutions(ctx, reapAfter)method that does the SELECT + per-row CAS loop. CAS losses are counted asRaceLost(not surfaced as errors — real executor won the race). Each successful CAS persists the canonical error"reaped after Xm in <prevStatus> state — executor did not complete; safe to retry"so bug(purchases): approved AWS purchase vanishes from History — 'approved'/'running' execution statuses hidden by the History list (P0) #621's History UI shows operators why the row was reaped.internal/purchase/reaper.go: newParseReapAfterFromEnv()readsPURCHASE_APPROVED_REAP_AFTER, falls back toDefaultReapAfter(10m) on missing or malformed env with aWARNlog. Never panics.internal/server/handler.go: newTaskReapStuckPurchasesscheduled-task type wired through the existingdispatchTask/ParseScheduledEventpipeline, alongsidecleanup/analytics_refresh/ri_exchange_reshape.terraform/modules/compute/aws/lambda/: new EventBridge rule onrate(5 minutes)(configurable), guarded byenable_reap_stuck_purchases_schedule(default true). New optionalpurchase_approved_reap_afterLambda env var — empty string falls back to the in-code default.Safety properties
ReapResultfor ops logging.Test plan
go build ./...— cleango vet ./...— cleangofmt -l ./...— cleango test ./internal/purchase/... -count=1— 114 passedgo test ./internal/config/... -count=1— 487 passedgo test ./internal/server/... ./internal/testutil/... -count=1— 328 passedgo test ./... -count=1— 4280 passed across 37 packagesapprovedrow reaped, stalerunningrow reaped, fresh row not touched (SELECT filters), terminal-status rows never queried (regression guard onstuckStatuses), CAS race lost (real executor wins),(nil, nil)defensive path, 3-stuck-row integration with per-row CAS expectations, SELECT error propagation, per-row save error after CAS success (counted as Reaped + Errored), env var unset / valid / invalid / non-default unit.Commits
feat(config): add ListStuckExecutions store method(766414c)feat(purchase): add ReapStuckExecutions sweep + per-row CAS to failed(40984d0)feat(purchase): wire periodic reaper invocation (closes #678)(c7bf86e)Deploy note
After merge + deploy, stuck-
approvedrows currently sitting on the cluster will be reaped within the next sweep cycle (≤5 min for new arrivals; first sweep after deploy for existing orphans, all of which already exceed the 10m threshold) and surfaced in History per #621. Operator can re-approve or cancel; the idempotency path (#636/#638/#652) prevents double-buys on retry.Summary by CodeRabbit
New Features
PURCHASE_APPROVED_REAP_AFTERenvironment variableChores