fix(flows): give flows_resume the run-lifecycle safety flows_run already had - #5286
Conversation
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughFlow run lifecycle handling now uses guarded state transitions. Execution installs finalizers before awaits, resume claims pending approvals atomically, cancellation and expiry preserve concurrent terminal states, and duplicate completion writes are rejected. ChangesFlow run lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ResumeFlow
participant RunRegistry
participant FlowRunStore
participant FlowEngine
ResumeFlow->>RunRegistry: Register run
ResumeFlow->>FlowRunStore: Claim pending approval
FlowRunStore-->>ResumeFlow: Return claim result
ResumeFlow->>FlowEngine: Execute resumed flow
ResumeFlow->>FlowRunStore: Write guarded terminal status
FlowRunStore-->>ResumeFlow: Return update result
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
Comment |
0b7105f to
a28d1a3
Compare
…already had (R-M1/M2/M3/M5)
`flows_run` has had cancellation safety since B41/B42 — register-before-row, a
`RunRowFinalizer` drop-guard, and terminal writes ordered row-then-summary.
`flows_resume` had none of it, despite executing the flow's real approved side
effects for up to `FLOW_RUN_TIMEOUT_SECS`. The doc comment on `run_flow_body`
describes exactly the race resume still carried.
R-M2 — `store::finish_flow_run` is now guarded on
`status IN ('running','pending_approval')` and reports whether it landed, the
same re-check `expire_parked_runs` and `mark_run_interrupted` already do. It was
an unconditional `WHERE id = ?`, so `flows_cancel_run` — which reads the status
and consults the registry as two separate observations — could relabel a run
that settled in between: a completed run whose real side effects had fired was
recorded as `cancelled`. The cancel path now attempts the guarded row write
FIRST and treats it as the authority, only recording the summary and dropping
the checkpoint once it has won.
R-M1 — resume now registers in the run registry, claims its row via a new
guarded `mark_run_resuming` (`pending_approval` -> `running`), and honours the
cancellation token in a `biased` select. Without the registry entry a
`flows_cancel_run` took the "parked/stale" branch and dropped the checkpoint out
from under an executing resume; without the status flip, a run approved just
before its TTL was expired by the parked-run sweep mid-execution.
R-M3 — resume finalizes the run row BEFORE the flow-summary write and no longer
propagates a `record_run` failure with `?`. A flow deleted mid-resume used to
return early and strand the row at `pending_approval` even though the engine had
completed, which the TTL sweep would later relabel `cancelled`. Adds a
`RunRowFinalizer` so a dropped resume future reconciles instead of stranding.
R-M5 — the drop-guard in `run_flow_body` is armed before the first `.await`
rather than ~150 lines later, closing the window where a client disconnect
during the inference-readiness network probe stranded a `running` row until the
next process boot. A third early-return path was also missing its `disarm`.
R-m4 — the parked-run TTL sweep now publishes `FlowRunFinished`; it was the one
terminal path that emitted no event, so event-driven consumers only saw the
transition on their next poll.
Two existing tests staged a row at an arbitrary terminal status by calling
`finish_flow_run` twice, which the new guard correctly refuses. Staging is a
fixture concern, so it gets a `#[cfg(test)]` forcing helper rather than a weaker
production write.
554 flows tests pass.
|
Review follow-up: two real bugs found and fixed, both of which this PR made load-bearing. 1. 2. 557 flows tests pass, Also noted for follow-up, not changed here: |
a28d1a3 to
533f8cc
Compare
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
graycyrus has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
coderabbitai
Bot
added
bug
rust-core
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 533f8cc812
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| let displaced = IN_FLIGHT | ||
| .lock() | ||
| .unwrap_or_else(|e| e.into_inner()) | ||
| .insert(run_id.to_string(), token.clone()); | ||
| tracing::debug!(target: "flows", run_id, "[flows] run_registry: registered in-flight run"); | ||
| .insert(run_id.to_string(), (registration, token.clone())); |
There was a problem hiding this comment.
When two flows_resume requests for the same parked run race, both can pass the initial pending_approval read and call register(thread_id) before either flips the row; this insert lets the later request replace the earlier request's cancellation token. If the earlier request then wins mark_run_resuming and continues executing while the later request gets Ok(false) and drops its guard, that guard removes the only registry entry, so a subsequent flows_cancel_run observes is_in_flight == false for a live resume and can take the parked/stale branch, mark it cancelled, and drop its checkpoint mid-execution. The registry needs to fail/return the existing registration on duplicate rather than displacing it before the row claim is won.
Useful? React with 👍 / 👎.
1 participant
Summary
flows_resumethe run-lifecycle safetyflows_runhas had since B41/B42 — it had none of it, despite executing a flow's real approved side effects for up toFLOW_RUN_TIMEOUT_SECS..awaitinrun_flow_body, closing a window where a dropped future stranded arunningrow until the next process boot.FlowRunFinished— it was the one terminal path that emitted no event.Problem
Four independent defects, all in the same lifecycle area:
1.
flows_resumenever registered in the run registry. The row stayedpending_approvalfor the whole resume and no cancellation token existed, so:flows_cancel_runsawis_in_flight == false, took its "parked/stale" branch, wrote a terminalcancelledrow and dropped the durable checkpoint — while the resume kept executing real outbound nodes and finally overwrotecancelledwith its own status.sweep_expired_parked_runs, which matches onstatus = 'pending_approval'— a status the resume never changed.run_flow_body's own doc comment describes exactly this race; it was fixed for the detached run path and never applied to resume.2.
store::finish_flow_runwas an unconditionalUPDATE … WHERE id = ?.flows_cancel_runreads the status and consults the registry as two separate observations. A run that settles in that window is not in flight, so the not-in-flight branch relabelled a fully-completed run whose real side effects had fired ascancelled.3.
flows_resume's write order was inverted vsflows_run—record_run(...)?ran beforefinish_flow_run_row. A flow deleted mid-resume made the summary write fail and returned early, stranding the row atpending_approvaleven though the engine had completed; the TTL sweep later relabelled that completed runcancelled.flows_rundoes the opposite, with an explicit comment saying why.4. The
RunRowFinalizerwas constructed ~150 lines after the first.awaitinrun_flow_body, leaving the inference-readiness network probe unguarded: a client disconnect there dropped the future before any finalizer existed, and the row stayed a perpetualrunningspinner.Solution
finish_flow_runnow carriesstatus IN ('running','pending_approval')and returns whether it landed, mirroring the re-checkexpire_parked_runs/mark_run_interruptedalready do.flows_cancel_runattempts that write first and treats it as the authority: afalsemeans the run settled underneath us, so it reports the conflict and leaves the recorded outcome (and the checkpoint) alone.run_registry::registerbefore anything else, a new guardedstore::mark_run_resuming(pending_approval→running) that moves the row out of the TTL sweep's predicate, and the cancel token honoured in abiasedselect exactly asrun_flow_bodydoes.record_runfailures are logged, never propagated with?.disarmis also fixed.Note for reviewers: two pre-existing tests changed, and it is not a regression
Adding the guard failed
flows_cancel_run_of_a_completed_with_warnings_run_errorsandflows_cancel_run_of_an_interrupted_run_errors. Both staged a row at an arbitrary terminal status by callingfinish_flow_runtwice — precisely the terminal→terminal overwrite the guard exists to block. The guard was correct; the fixtures were leaning on an unguarded production write. Staging therefore gets a#[cfg(test)] force_run_status_for_testdoor rather than the production path being weakened.Submission Checklist
cargo test --lib openhuman::flows= 554 passed, 0 failedN/A: behaviour-only bug fix, no feature rows added/removed/renamed## Related—N/A: no matrix feature rows affectedN/A: no user-facing surface change; run/resume/cancel semantics only become more correctCloses #NNN—N/A: found by code review, no tracking issue filed yetImpact
mark_run_resumingreuses existing columns).runningfor its duration rather than stayingpending_approval. Verify the runs rail renders that sensibly — it is an existing status, andpending_approvals_jsonis not cleared by the flip.flows_runpath (finalizer placement + its terminal-write helper). The existingflows_runsuite is the regression net and stays green.Related
N/Aflows_resumegraph-hash pin (stale-approval-after-graph-swap) is stacked on this branch; store resilience (skip corrupt rows) and the detached-run RPC also stack on it.src/openhuman/flows/ops.rs+ops_tests.rswith the authorization-boundaries PR, andstore.rswith the contract-drift PR. Merge this one first; the others rebase cleanly on it.AI Authored PR Metadata (required for Codex/Linear PRs)
Linear Issue
Commit & Branch
fix/flows-resume-run-lifecycle0b7105fa7Validation Run
pnpm --filter openhuman-app format:check— N/A, no frontend files changedpnpm typecheck— N/A, no TypeScript changedGGML_NATIVE=OFF cargo test --lib openhuman::flows→ 554 passed, 0 failedGGML_NATIVE=OFF cargo check --manifest-path Cargo.tomlcleanapp/src-tauriuntouchedValidation Blocked
command:N/Aerror:N/Aimpact:N/ABehavior Changes
runningwhile it executes; cancelling an already-finished run now returns a clear error instead of silently relabelling it.Parity Contract
flows_run/flows_run_detachedsemantics are unchanged apart from the finalizer being armed earlier; every existing terminal-status transition (running/pending_approval→ terminal) still lands.expire_parked_runs, andmark_run_interruptedguards are untouched and still pinned by their existing tests.Duplicate / Superseded PR Handling
Summary by CodeRabbit