fix(approvals): await resume_run so channel is ready before response returns
ci / gates (push) Successful in 33s
ci / frontend (push) Successful in 1m33s
ci / rust (push) Failing after 3m25s
ci / e2e (push) Has been skipped
ci / publish (push) Has been skipped

The `decide()` helper (used by both approve and reject) wrapped
`runtime.resume_run(ready).await` inside `tokio::spawn`, so the handler
returned 200 to the client before the resume had even started. When a
client (or test) immediately reconnected to /api/gateway with resumeFrom,
the run's broadcast channel didn't exist yet — `runtime.subscribe(run_id)`
returned None, the journal was still empty (no new events yet), and the
SSE stream closed with zero events. Symptom: approvals_api tests flaky in
CI (`unwrap on None` at line 282 of reject_over_http_executes_nothing,
sometimes `missing step_finished` on the accept path).

resume_run's *own* awaited portion is only the setup — claim_resume,
checkpoint load, and open_channel. It internally spawns the long-running
multi-step work. Awaiting it inline means we wait milliseconds for setup,
then return. By the time the client reconnects, the channel exists, so
subscribe() finds it and live-tail works. The sweeper still handles the
crash-between-decision-and-resume case for durability.
This commit is contained in:
Omar Sobh
2026-07-05 09:40:20 -07:00
parent 48c6adfa1b
commit 202e8535f0
+8 -6
View File
@@ -56,17 +56,19 @@ async fn decide(
workspace_approval(&state, &user, id).await?;
let approval = approvals::decide(&state.pool, id, user.user_id, decision).await?;
// Kick the resume immediately; the sweeper remains the durable
// fallback if this process dies first.
let runtime = state.runtime.clone();
// Kick the resume before returning. resume_run's awaited portion is only
// the setup (claim + checkpoint load + open the broadcast channel); it
// spawns the actual multi-step work internally, so this doesn't block the
// client on the run itself — but it does guarantee the channel exists by
// the time the client reconnects via /api/gateway, closing the observable
// race where subscribe(run_id) would return None. The sweeper remains the
// durable fallback if this process dies before setup completes.
let ready = ResumeReady {
run_id: approval.run_id,
approval_id: approval.id,
approved: decision == Decision::Approve,
};
tokio::spawn(async move {
let _ = runtime.resume_run(ready).await;
});
let _ = state.runtime.resume_run(ready).await;
Ok(Json(
json!({ "id": approval.id, "status": approval.status }),