From 533f8cc812ec1de08b3df9cdcfd281695c7beca3 Mon Sep 17 00:00:00 2001 From: cyrus Date: Thu, 30 Jul 2026 22:37:14 +0530 Subject: [PATCH] fix(flows): give `flows_resume` the run-lifecycle safety `flows_run` already had (R-M1/M2/M3/M5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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. --- src/openhuman/flows/ops.rs | 270 +++++++++++++++++++++++----- src/openhuman/flows/ops_tests.rs | 236 ++++++++++++++++++++++-- src/openhuman/flows/run_registry.rs | 112 ++++++++++-- src/openhuman/flows/store.rs | 143 ++++++++++++--- src/openhuman/flows/store_tests.rs | 67 +++++++ 5 files changed, 735 insertions(+), 93 deletions(-) diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 1607a0b48a..44bcd4139e 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -4707,6 +4707,20 @@ async fn run_flow_body( let config: &Config = config_arc.as_ref(); let flow_id: &str = flow_id.as_str(); + // B42 drop-guard, armed BEFORE the first `.await` in this body (R-M5). + // + // The caller has already inserted the `running` row, so every await from + // here on is a window in which dropping this future would strand that row. + // The guard used to be constructed ~150 lines below, immediately around the + // engine call — which left the inference-readiness preflight directly below + // (a real network probe on a cache miss) unguarded: a client disconnect or + // an aborted detached task during that probe dropped the future before any + // finalizer existed, and the row stayed a perpetual `running` spinner until + // the NEXT process boot sweep (the in-process one had already run). Arming + // it here covers the whole awaiting region; every settled path below still + // disarms it after its own terminal write. + let finalizer = RunRowFinalizer::new(config_arc.clone(), &thread_id, flow_id); + // B45 run-time preflight (design correction — see the "Inference-readiness // check" module doc above): an `agent` node needs a working LLM provider // to run at all, but that is no longer enforced as an author-time gate — @@ -4752,6 +4766,7 @@ async fn run_flow_body( &[], Some(&msg), ); + finalizer.disarm(); return Err(msg); } @@ -4773,6 +4788,7 @@ async fn run_flow_body( &[], Some(&msg), ); + finalizer.disarm(); return Err(msg); } }; @@ -4797,6 +4813,7 @@ async fn run_flow_body( &[], Some(&msg), ); + finalizer.disarm(); return Err(msg); } }; @@ -4867,11 +4884,8 @@ async fn run_flow_body( ); let timed = tokio::time::timeout(std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), run); tokio::pin!(timed); - // B42 drop-guard: armed for the whole awaiting region below. If this future - // is dropped before any terminal write (harness abort, turn end, runtime - // shutdown, panic), its `Drop` reconciles the orphaned `running` row to - // `interrupted`. Every settled path disarms it after its own terminal write. - let finalizer = RunRowFinalizer::new(config_arc.clone(), &thread_id, flow_id); + // (The B42 drop-guard is armed near the top of this fn, before the first + // `.await` — see `finalizer` there.) // Race the run against a cancellation signal (issue G4). `biased` checks the // cancel arm first so a `flows_cancel_run` that lands right as the run // settles still wins deterministically. @@ -5093,11 +5107,64 @@ pub async fn flows_resume( } let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?; let config_arc = Arc::new(config.clone()); - let caps = - crate::openhuman::tinyflows::build_capabilities(config_arc, format!("flow:{flow_id}")); + let caps = crate::openhuman::tinyflows::build_capabilities( + config_arc.clone(), + format!("flow:{flow_id}"), + ); let checkpointer = crate::openhuman::tinyflows::open_flow_checkpointer(config).map_err(|e| e.to_string())?; + // Run-lifecycle parity with `flows_run` (R-M1). A resume executes the flow's + // real approved side effects for up to `FLOW_RUN_TIMEOUT_SECS`, so it needs + // the same three guards the run path has had since B41/B42 — it had none: + // + // 1. `run_registry::register` — without an entry, `flows_cancel_run` saw + // `is_in_flight == false`, took its "parked/stale" branch, wrote a + // terminal `cancelled` row and dropped the checkpoint out from under + // this still-executing resume. Registering makes the cancel take the + // signalled branch, which this fn now honours in the `select!` below. + // 2. `mark_run_resuming` — flips the row off `pending_approval` so the + // parked-run TTL sweep stops matching a resume that is actively + // running. + // 3. `RunRowFinalizer` — if this future is dropped mid-await (client + // disconnect during the long await), the row is reconciled to + // `interrupted` instead of being stranded at its old status. + // + // Register BEFORE the status flip for the same reason `flows_run` registers + // before inserting its row: never let a cancel observe a live-looking row + // that no registered run owns. + let (cancel_token, _run_guard) = run_registry::register(thread_id); + match store::mark_run_resuming(config, thread_id) { + Ok(true) => {} + Ok(false) => { + // The guarded flip matched nothing: the run was cancelled or + // TTL-expired between the status check above and here. Refuse + // rather than executing approved side effects for a run that is no + // longer live. + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + "[flows] flows_resume: run left 'pending_approval' before the resume could claim it — refusing" + ); + return Err(format!( + "no paused run to resume: run '{thread_id}' was cancelled or expired before the \ + resume could start" + )); + } + Err(e) => { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + error = %e, + "[flows] flows_resume: failed to mark run as resuming" + ); + return Err(e.to_string()); + } + } + let finalizer = RunRowFinalizer::new(config_arc, thread_id, flow_id); + tracing::debug!( target: "flows", flow_id = %flow_id, @@ -5148,50 +5215,93 @@ pub async fn flows_resume( ), ); - let journaled = match tokio::time::timeout( - std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), - run, - ) - .await - { - Ok(Ok(journaled)) => journaled, - Ok(Err(e)) => { - let _ = store::record_run(config, flow_id, "failed"); - let observed = current_persisted_steps(config, thread_id); - finish_flow_run_row( - config, - thread_id, - flow_id, - "failed", - &observed, - &[], - Some(&e.to_string()), + // Terminal-write helper for the two failure arms. Row FIRST, then the + // best-effort summary — see the settle path below for why the order matters. + let record_failed = |msg: &str| { + let observed = current_persisted_steps(config, thread_id); + finish_flow_run_row( + config, + thread_id, + flow_id, + "failed", + &observed, + &[], + Some(msg), + ); + if let Err(e) = store::record_run(config, flow_id, "failed") { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + error = %e, + "[flows] flows_resume: failed to record run summary (run row already finalized)" ); - tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, error = %e, "[flows] flows_resume: run failed"); - return Err(e.to_string()); } - Err(_elapsed) => { - let msg = format!("flow resume timed out after {FLOW_RUN_TIMEOUT_SECS}s"); - let _ = store::record_run(config, flow_id, "failed"); + }; + + let timed = tokio::time::timeout(std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), run); + tokio::pin!(timed); + // Race the resume against a cancellation signal, exactly as `run_flow_body` + // does. `biased` checks the cancel arm first so a `flows_cancel_run` landing + // as the resume settles still wins deterministically. + let journaled = tokio::select! { + biased; + _ = cancel_token.cancelled() => { + tracing::info!(target: "flows", flow_id = %flow_id, %thread_id, "[flows] flows_resume: cancelled mid-resume"); let observed = current_persisted_steps(config, thread_id); finish_flow_run_row( config, thread_id, flow_id, - "failed", + "cancelled", &observed, &[], - Some(&msg), + Some("run cancelled"), ); - tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_resume: run timed out"); - return Err(msg); + finalizer.disarm(); + if let Err(e) = store::record_run(config, flow_id, "cancelled") { + tracing::warn!(target: "flows", flow_id = %flow_id, error = %e, "[flows] flows_resume: failed to record cancelled run"); + } + drop_checkpoint(config, thread_id).await; + return Ok(RpcOutcome::single_log( + json!({ + "output": Value::Null, + "pending_approvals": Vec::::new(), + "thread_id": thread_id, + "cancelled": true, + }), + format!("flow resume cancelled: {thread_id}"), + )); } + result = &mut timed => match result { + Ok(Ok(journaled)) => journaled, + Ok(Err(e)) => { + record_failed(&e.to_string()); + finalizer.disarm(); + tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, error = %e, "[flows] flows_resume: run failed"); + return Err(e.to_string()); + } + Err(_elapsed) => { + let msg = format!("flow resume timed out after {FLOW_RUN_TIMEOUT_SECS}s"); + record_failed(&msg); + finalizer.disarm(); + tracing::warn!(target: "flows", flow_id = %flow_id, %thread_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_resume: run timed out"); + return Err(msg); + } + }, }; let outcome = journaled.outcome; let settled = settle_steps(config, thread_id, &outcome.output); let (status, error) = finalize_terminal_status(&settled, &outcome.pending_approvals); - store::record_run(config, flow_id, status).map_err(|e| e.to_string())?; + // Finalize the run row (and disarm the drop-guard) BEFORE the flow-summary + // write, matching `flows_run` (R-M3). This used to be inverted here, with + // `record_run` propagating via `?`: a concurrent flow delete made the + // summary write fail and returned early, leaving the row stranded at + // `pending_approval` even though the engine had completed and its side + // effects had fired — which the TTL sweep would later relabel `cancelled`. + // The row's terminal state is the correctness-critical write; the summary is + // best-effort observability. finish_flow_run_row( config, thread_id, @@ -5201,6 +5311,17 @@ pub async fn flows_resume( &outcome.pending_approvals, error.as_deref(), ); + finalizer.disarm(); + if let Err(e) = store::record_run(config, flow_id, status) { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + status, + error = %e, + "[flows] flows_resume: failed to record run summary (run row already finalized)" + ); + } export_run_to_langfuse( config, &flow.name, @@ -5325,6 +5446,25 @@ pub async fn sweep_expired_parked_runs(config: &Config) -> usize { if let Err(e) = store::record_run(config, flow_id, "cancelled") { tracing::warn!(target: "flows", run_id, flow_id, error = %e, "[flows] TTL sweep: failed to update flow summary for expired run"); } + // Announce the terminal transition (R-m4). `expire_parked_runs` writes + // the row directly rather than going through `finish_flow_run_row`, so + // without this the sweep was the one terminal path that emitted no + // `FlowRunFinished` — the boot sweep already publishes its own. Purely + // event-driven consumers (the runs rail) would otherwise not observe a + // TTL-expired run settle until their next poll. + tracing::debug!( + target: "flows", + run_id, + flow_id, + "[flows] TTL sweep: publishing FlowRunFinished for expired parked run" + ); + crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::FlowRunFinished { + flow_id: flow_id.to_string(), + run_id: run_id.to_string(), + status: "cancelled".to_string(), + }, + ); drop_checkpoint(config, run_id).await; } if !swept.is_empty() { @@ -5472,11 +5612,19 @@ pub async fn flows_cancel_run(config: &Config, run_id: &str) -> Result Result, -) { +) -> bool { let finished_at = Utc::now().to_rfc3339(); - if let Err(e) = store::finish_flow_run( + match store::finish_flow_run( config, thread_id, status, @@ -5581,7 +5745,26 @@ fn finish_flow_run_row( pending_approvals, error, ) { - tracing::warn!(target: "flows", thread_id, status, error = %e, "[flows] failed to persist flow run finish"); + Err(e) => { + tracing::warn!(target: "flows", thread_id, status, error = %e, "[flows] failed to persist flow run finish"); + return false; + } + // The guarded UPDATE (R-M2) matched nothing: the row had already + // settled to a terminal status before this write. Whoever settled it + // first also published `FlowRunFinished`, so publishing again here + // would emit a second terminal event for one run. Report the no-op + // instead of pretending the write landed. + Ok(false) => { + tracing::warn!( + target: "flows", + flow_id, + thread_id, + attempted_status = status, + "[flows] finish_flow_run_row: row already terminal — refusing to overwrite a settled run" + ); + return false; + } + Ok(true) => {} } // `status` can be `"pending_approval"` here (see `finalize_terminal_status`) @@ -5604,7 +5787,7 @@ fn finish_flow_run_row( status, "[flows] finish_flow_run_row: run paused for approval — not a finish, skipping FlowRunFinished" ); - return; + return true; } tracing::debug!( @@ -5619,6 +5802,7 @@ fn finish_flow_run_row( run_id: thread_id.to_string(), status: status.to_string(), }); + true } /// Reconstructs a lean per-node step list from a settled run's diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 4bdbff3a1f..c3afce1d42 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -2651,16 +2651,11 @@ async fn flows_cancel_run_of_a_completed_with_warnings_run_errors() { // Force the settled row to the warning status directly — an end-to-end // null-binding graph isn't needed to exercise this guard. - store::finish_flow_run( - &config, - &thread_id, - "completed_with_warnings", - &chrono::Utc::now().to_rfc3339(), - &[], - &[], - None, - ) - .unwrap(); + // Fixture-only forcing write: the run above already settled `completed`, so + // `finish_flow_run`'s liveness guard (correctly) refuses a terminal → + // terminal transition. Staging a row at an arbitrary terminal status is a + // test concern, not a production one. + store::force_run_status_for_test(&config, &thread_id, "completed_with_warnings", None).unwrap(); let err = flows_cancel_run(&config, &thread_id) .await @@ -2690,13 +2685,13 @@ async fn flows_cancel_run_of_an_interrupted_run_errors() { let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); // Force the settled row to `interrupted` directly. - store::finish_flow_run( + // Fixture-only forcing write — see the sibling test above: the run has + // already settled, and `finish_flow_run` now (correctly) refuses a + // terminal -> terminal transition. + store::force_run_status_for_test( &config, &thread_id, "interrupted", - &chrono::Utc::now().to_rfc3339(), - &[], - &[], Some("interrupted mid-flight"), ) .unwrap(); @@ -7502,3 +7497,216 @@ async fn approval_manifest_discloses_agent_ref_nodes_only() { Some("harness") ); } + +// ───────────────────────────────────────────────────────────────────────────── +// Run-lifecycle parity for `flows_resume` + guarded terminal writes +// (R-M1 / R-M2 / R-M3 / R-M5 / R-m4). +// +// `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`. These pin the mechanisms that +// close that gap. + +/// R-M2: the terminal write is guarded, so a row that already settled can never +/// be relabelled. Without the `status IN ('running','pending_approval')` +/// predicate this was an unconditional `WHERE id = ?`. +#[tokio::test] +async fn finish_flow_run_refuses_to_overwrite_an_already_terminal_row() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "guarded-finish".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + let run_id = "run-guarded-1"; + let now = Utc::now().to_rfc3339(); + store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); + + // First terminal write wins. + let first = store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None).unwrap(); + assert!(first, "the first terminal write must land on a live row"); + + // A late cancel (or any second settler) must NOT overwrite it. + let second = + store::finish_flow_run(&config, run_id, "cancelled", &now, &[], &[], Some("late")).unwrap(); + assert!( + !second, + "a terminal row must not be overwritten by a second settler" + ); + + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!( + row.status, "completed", + "the run's real outcome must survive a losing concurrent cancel" + ); +} + +/// R-M2 end-to-end: `flows_cancel_run` reads the status and consults the +/// registry as two separate observations. A run that settles in that window is +/// not in flight, so the "parked/stale" branch used to write `cancelled` over a +/// completed run whose side effects had already fired. +#[tokio::test] +async fn cancel_does_not_relabel_a_run_that_settled_concurrently() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "cancel-toctou".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + let run_id = "run-toctou-1"; + let now = Utc::now().to_rfc3339(); + store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); + // The run settles on its own (real side effects fired) and deregisters — + // exactly the state `flows_cancel_run` can observe one instant too late. + store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None).unwrap(); + + let result = flows_cancel_run(&config, run_id).await; + assert!( + result.is_err(), + "cancelling an already-settled run must report the conflict, not silently rewrite it" + ); + + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!( + row.status, "completed", + "a completed run must never be recorded as cancelled" + ); +} + +/// R-M1 (store half): claiming a parked run for a resume is a guarded flip, so +/// a run cancelled or TTL-expired in the meantime can never be revived. +#[tokio::test] +async fn mark_run_resuming_claims_only_a_parked_row() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "resume-claim".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + let run_id = "run-claim-1"; + let now = Utc::now().to_rfc3339(); + store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); + // Park it. + store::finish_flow_run( + &config, + run_id, + "pending_approval", + &now, + &[], + &["gate".to_string()], + None, + ) + .unwrap(); + + assert!( + store::mark_run_resuming(&config, run_id).unwrap(), + "a parked run must be claimable for resume" + ); + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!(row.status, "running"); + + // Claiming twice must not succeed — the second resume would execute the + // same approved side effects again. + assert!( + !store::mark_run_resuming(&config, run_id).unwrap(), + "a run already claimed (or cancelled/expired) must not be claimable again" + ); +} + +/// R-M1 (the race that mattered): a run approved just before its TTL used to be +/// swept to `cancelled` — and have its durable checkpoint dropped — WHILE the +/// resume was actively executing approved outbound nodes, because the row sat +/// at `pending_approval` for the whole resume. Claiming it as `running` moves it +/// out of the sweep's predicate. +#[tokio::test] +async fn ttl_sweep_cannot_expire_a_run_a_resume_has_claimed() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "resume-vs-ttl".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + // A run parked well past the TTL — the sweep would expire it right now. + let stale = (Utc::now() - chrono::Duration::seconds(FLOW_PARKED_TTL_SECS * 4)).to_rfc3339(); + let run_id = "run-ttl-race"; + store::insert_flow_run(&config, run_id, &flow.id, run_id, &stale).unwrap(); + store::finish_flow_run( + &config, + run_id, + "pending_approval", + &stale, + &[], + &["gate".to_string()], + None, + ) + .unwrap(); + + // The user approves in the nick of time and the resume claims the run. + assert!(store::mark_run_resuming(&config, run_id).unwrap()); + + // Any read-path sweep that now fires must leave the in-flight resume alone. + sweep_expired_parked_runs(&config).await; + + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!( + row.status, "running", + "a claimed resume must survive the parked-run TTL sweep — expiring it would drop the \ + checkpoint out from under a run that is executing real side effects" + ); +} + +/// A genuinely stale parked run (never claimed) must still be swept — the guard +/// above must not have disabled the TTL sweep wholesale. +#[tokio::test] +async fn ttl_sweep_still_expires_an_unclaimed_parked_run() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "ttl-still-works".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + let stale = (Utc::now() - chrono::Duration::seconds(FLOW_PARKED_TTL_SECS * 4)).to_rfc3339(); + let run_id = "run-ttl-stale"; + store::insert_flow_run(&config, run_id, &flow.id, run_id, &stale).unwrap(); + store::finish_flow_run( + &config, + run_id, + "pending_approval", + &stale, + &[], + &["gate".to_string()], + None, + ) + .unwrap(); + + let swept = sweep_expired_parked_runs(&config).await; + assert_eq!(swept, 1, "an unclaimed stale parked run must still expire"); + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!(row.status, "cancelled"); +} diff --git a/src/openhuman/flows/run_registry.rs b/src/openhuman/flows/run_registry.rs index b9aecc9920..0479ab34da 100644 --- a/src/openhuman/flows/run_registry.rs +++ b/src/openhuman/flows/run_registry.rs @@ -15,32 +15,59 @@ //! can never leave a stale token wedged in the map. use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{LazyLock, Mutex}; use tokio_util::sync::CancellationToken; -/// The live in-flight runs: `run_id` → its cancellation token. -static IN_FLIGHT: LazyLock>> = +/// Monotonic registration id, so a [`RunGuard`] can prove an entry is still +/// *its own* before removing it. See [`RunGuard::drop`]. +static NEXT_REGISTRATION: AtomicU64 = AtomicU64::new(1); + +/// The live in-flight runs: `run_id` → (registration id, cancellation token). +static IN_FLIGHT: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); /// Registers `run_id` as in-flight and returns both a clone of its /// cancellation token (to `select!` on) and a [`RunGuard`] that deregisters it /// on drop. Hold the guard for the whole run. /// -/// A duplicate `run_id` (should not happen — thread ids are UUID-suffixed) -/// replaces the prior token; the returned guard still removes exactly this -/// `run_id` on drop. +/// A duplicate `run_id` replaces the prior token, and each registration carries +/// a unique id so a guard only ever removes the entry it installed. +/// +/// This used to be documented as impossible ("thread ids are UUID-suffixed"), +/// which held while only `flows_run` / `flows_run_detached` registered — both +/// mint a fresh UUID `thread_id` per call, so two concurrent registrations could +/// never collide on a key. `flows_resume` is the first caller to register +/// against a **stable, pre-existing** id (the parked run's own), and nothing +/// serializes two concurrent resumes of the same run — a client double-submit or +/// a retry-on-timeout is enough. With removal keyed only by `run_id`, the loser +/// of that race would deregister the *winner's* live token on its way out, and a +/// later `flows_cancel_run` would then see `is_in_flight == false` for a run that +/// is genuinely executing, take its "parked/stale" branch, and drop the +/// checkpoint out from under it — the exact bug class this registry exists to +/// prevent. pub(crate) fn register(run_id: &str) -> (CancellationToken, RunGuard) { let token = CancellationToken::new(); - IN_FLIGHT + let registration = NEXT_REGISTRATION.fetch_add(1, Ordering::Relaxed); + 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())); + if displaced.is_some() { + tracing::warn!( + target: "flows", + run_id, + registration, + "[flows] run_registry: duplicate registration for a run id already in flight — the displaced guard will no longer deregister this entry" + ); + } + tracing::debug!(target: "flows", run_id, registration, "[flows] run_registry: registered in-flight run"); ( token, RunGuard { run_id: run_id.to_string(), + registration, }, ) } @@ -52,7 +79,7 @@ pub(crate) fn register(run_id: &str) -> (CancellationToken, RunGuard) { pub(crate) fn cancel(run_id: &str) -> bool { let guard = IN_FLIGHT.lock().unwrap_or_else(|e| e.into_inner()); match guard.get(run_id) { - Some(token) => { + Some((_registration, token)) => { token.cancel(); tracing::info!(target: "flows", run_id, "[flows] run_registry: signalled in-flight run to cancel"); true @@ -79,20 +106,38 @@ pub(crate) fn is_in_flight(run_id: &str) -> bool { /// RAII guard that removes a run's entry from the in-flight registry on drop. pub(crate) struct RunGuard { run_id: String, + registration: u64, } impl Drop for RunGuard { + /// Removes this run's entry — but only if it is still the entry THIS guard + /// installed. A plain `remove(&run_id)` would let the loser of a duplicate + /// registration deregister the winner's live token (see [`register`]). fn drop(&mut self) { - IN_FLIGHT - .lock() - .unwrap_or_else(|e| e.into_inner()) - .remove(&self.run_id); - tracing::debug!(target: "flows", run_id = %self.run_id, "[flows] run_registry: deregistered run"); + let mut map = IN_FLIGHT.lock().unwrap_or_else(|e| e.into_inner()); + match map.get(&self.run_id) { + Some((registration, _)) if *registration == self.registration => { + map.remove(&self.run_id); + tracing::debug!(target: "flows", run_id = %self.run_id, registration = self.registration, "[flows] run_registry: deregistered run"); + } + Some(_) => { + tracing::debug!( + target: "flows", + run_id = %self.run_id, + registration = self.registration, + "[flows] run_registry: entry belongs to a newer registration — leaving it in place" + ); + } + None => {} + } } } #[cfg(test)] mod tests { + // NOTE: `duplicate_registration_*` below pin the identity-based removal — + // see `register`'s doc for why `flows_resume` made this reachable. + use super::*; #[test] @@ -117,4 +162,43 @@ mod tests { fn cancel_of_unknown_run_is_false() { assert!(!cancel("flow:never-registered:run-x")); } + + /// The loser of a duplicate registration must NOT deregister the winner's + /// live token. Before removal compared registration identity, the loser's + /// guard dropping would clear the map entry the winner still relies on, and + /// a later `cancel` would report "not in flight" for a genuinely running + /// run — letting the caller drop its checkpoint mid-execution. + #[test] + fn a_displaced_guard_does_not_deregister_the_live_registration() { + let run_id = "dup-resume-run"; + let (_winner_token, winner_guard) = register(run_id); + // A second concurrent resume of the SAME run id registers on top. + let (_later_token, later_guard) = register(run_id); + + // The first (now displaced) guard drops, e.g. because it lost the + // guarded `mark_run_resuming` race and returned early. + drop(winner_guard); + + assert!( + is_in_flight(run_id), + "the newer, still-live registration must survive a displaced guard's drop" + ); + assert!(cancel(run_id), "the live run must still be cancellable"); + + drop(later_guard); + assert!( + !is_in_flight(run_id), + "the owning guard must still deregister its own entry" + ); + } + + /// The ordinary single-registration path is unchanged. + #[test] + fn the_owning_guard_still_deregisters_its_own_entry() { + let run_id = "solo-run"; + let (_token, guard) = register(run_id); + assert!(is_in_flight(run_id)); + drop(guard); + assert!(!is_in_flight(run_id)); + } } diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs index 0712aaac2c..bc7b9f69fb 100644 --- a/src/openhuman/flows/store.rs +++ b/src/openhuman/flows/store.rs @@ -723,6 +723,17 @@ fn prune_flow_runs_conn(conn: &Connection, flow_id: &str, keep: usize) -> Result /// Called once a `flows_run` / `flows_resume` invocation settles — including /// the timeout / capability-error paths, so a row never gets stuck at /// `"running"` when the process is still up. +/// +/// **Guarded write (R-M2).** The `UPDATE` only matches a row that is still +/// live — `status IN ('running','pending_approval')` — mirroring the same +/// re-check [`expire_parked_runs`] and [`mark_run_interrupted`] already do. +/// Without it this was an unconditional `WHERE id = ?`, so a caller that read a +/// non-terminal status and then lost a race could overwrite a row that had +/// meanwhile settled: `flows_cancel_run` reads `running`, the live run finishes +/// `completed` and deregisters, `run_registry::cancel` returns `false`, and the +/// "not in flight" branch then relabels a fully-completed run (whose real side +/// effects fired) as `cancelled`. Returns whether a row was actually updated so +/// callers can log the no-op instead of silently believing the write landed. pub fn finish_flow_run( config: &Config, id: &str, @@ -731,18 +742,20 @@ pub fn finish_flow_run( steps: &[FlowRunStep], pending_approvals: &[String], error: Option<&str>, -) -> Result<()> { +) -> Result { let steps_json = serde_json::to_string(steps).context("Failed to serialize flow run steps")?; let pending_json = serde_json::to_string(pending_approvals) .context("Failed to serialize flow run pending approvals")?; with_connection(config, |conn| { - conn.execute( - "UPDATE flow_runs SET status = ?1, finished_at = ?2, steps_json = ?3, \ - pending_approvals_json = ?4, error = ?5 WHERE id = ?6", - params![status, finished_at, steps_json, pending_json, error, id], - ) - .context("Failed to finish flow run")?; - Ok(()) + let updated = conn + .execute( + "UPDATE flow_runs SET status = ?1, finished_at = ?2, steps_json = ?3, \ + pending_approvals_json = ?4, error = ?5 \ + WHERE id = ?6 AND status IN ('running', 'pending_approval')", + params![status, finished_at, steps_json, pending_json, error, id], + ) + .context("Failed to finish flow run")?; + Ok(updated > 0) }) } @@ -796,9 +809,25 @@ pub fn upsert_flow_run_step(config: &Config, run_id: &str, step: &FlowRunStep) - /// (`COALESCE(finished_at, started_at)` — a run's `finished_at` is stamped when /// it pauses at a gate) is strictly older than `cutoff` (an RFC3339 instant), /// transitioning it to a terminal `"cancelled"` status stamped `now` with -/// `error_msg`. Returns the `(run_id, flow_id)` of each swept run so the caller -/// can update the flow summary + drop the durable checkpoint (issue G4 — -/// parked-run TTL). +/// `error_msg`. Returns the `(run_id, flow_id)` of the runs **actually flipped** +/// so the caller can update the flow summary, publish `FlowRunFinished`, and +/// drop the durable checkpoint (issue G4 — parked-run TTL) for real settles +/// only. +/// +/// **Candidates are not sweeps.** The `SELECT` and each row's guarded `UPDATE` +/// are separate statements on an autocommit connection (`with_connection` opens +/// a fresh connection per call, not a transaction spanning this function), so a +/// concurrent `mark_run_resuming` on another connection can land in between: the +/// row was `pending_approval` at `SELECT` time and no longer is when its own +/// `UPDATE` runs. The per-row `WHERE status = 'pending_approval'` re-check keeps +/// that row's data safe — but returning the unfiltered candidate list would let +/// the caller act on a run it never actually expired: dropping the checkpoint out +/// from under a resume that just claimed it, and publishing a terminal +/// `FlowRunFinished` for a run still executing. That false event is the worse +/// half, because the frontend de-dupes terminal events by `${flow_id}:${run_id}` +/// — so the run's real completion would later be discarded as an alias replay, +/// leaving a successful run displayed as cancelled. Only rows whose `UPDATE` +/// reports `changed > 0` are returned. /// /// RFC3339 timestamps produced by `chrono::Utc::…to_rfc3339()` all carry the /// same `+00:00` offset, so a lexicographic `<` is a valid chronological @@ -821,20 +850,32 @@ pub fn expire_parked_runs( .collect::>()?; drop(stmt); - for (run_id, _flow_id) in &stale { + let mut swept = Vec::with_capacity(stale.len()); + for (run_id, flow_id) in stale { // Re-check the status in the WHERE so a run resumed/cancelled - // between the SELECT and here is not clobbered. - conn.execute( - "UPDATE flow_runs SET status = 'cancelled', finished_at = ?1, error = ?2 \ - WHERE id = ?3 AND status = 'pending_approval'", - params![now, error_msg, run_id], - ) - .context("Failed to expire parked flow run")?; + // between the SELECT and here is not clobbered, and keep only the + // rows this sweep genuinely flipped — see the fn doc. + let changed = conn + .execute( + "UPDATE flow_runs SET status = 'cancelled', finished_at = ?1, error = ?2 \ + WHERE id = ?3 AND status = 'pending_approval'", + params![now, error_msg, &run_id], + ) + .context("Failed to expire parked flow run")?; + if changed > 0 { + swept.push((run_id, flow_id)); + } else { + tracing::debug!( + target: "flows", + run_id = %run_id, + "[flows] TTL sweep: run left 'pending_approval' concurrently — not expiring it" + ); + } } - if !stale.is_empty() { - tracing::info!(target: "flows", swept = stale.len(), "[flows] expired parked pending_approval runs past TTL"); + if !swept.is_empty() { + tracing::info!(target: "flows", swept = swept.len(), "[flows] expired parked pending_approval runs past TTL"); } - Ok(stale) + Ok(swept) }) } @@ -871,6 +912,64 @@ pub fn list_running_run_ids( }) } +/// Test-only unconditional status write, bypassing the +/// [`finish_flow_run`] liveness guard. +/// +/// Production code must never do a terminal → terminal transition — that is the +/// corruption [`finish_flow_run`]'s `status IN ('running','pending_approval')` +/// predicate exists to prevent. But a couple of tests legitimately need to +/// *stage* a row at an arbitrary terminal status (`completed_with_warnings`, +/// `interrupted`) to exercise the guards that read it, and they previously did +/// so by calling `finish_flow_run` twice — which the guard now correctly +/// refuses. Staging is a fixture concern, so it gets a fixture-only door rather +/// than a weaker production write. +#[cfg(test)] +pub fn force_run_status_for_test( + config: &Config, + id: &str, + status: &str, + error: Option<&str>, +) -> Result<()> { + with_connection(config, |conn| { + conn.execute( + "UPDATE flow_runs SET status = ?1, error = ?2 WHERE id = ?3", + params![status, error, id], + ) + .context("Failed to force flow run status (test fixture)")?; + Ok(()) + }) +} + +/// Flips a parked `'pending_approval'` row to `'running'` for the duration of a +/// [`crate::openhuman::flows::ops::flows_resume`], guarded by a +/// `status = 'pending_approval'` predicate so a run cancelled or expired +/// concurrently is never revived. Returns `true` when a row was actually +/// flipped. +/// +/// Without this flip the row stays `pending_approval` for the whole (up to +/// `FLOW_RUN_TIMEOUT_SECS`) resume, so +/// [`expire_parked_runs`]' TTL sweep still matches it: a run approved just +/// before its TTL would be relabelled `cancelled` and have its durable +/// checkpoint dropped **while the resume was actively executing approved +/// outbound nodes** (R-M1). Marking it `running` moves it out of the sweep's +/// predicate and into the same lifecycle state a `flows_run` occupies, which is +/// also what the boot orphan sweep already knows how to reconcile. +pub fn mark_run_resuming(config: &Config, id: &str) -> Result { + with_connection(config, |conn| { + let changed = conn + .execute( + "UPDATE flow_runs SET status = 'running', finished_at = NULL, error = NULL \ + WHERE id = ?1 AND status = 'pending_approval'", + params![id], + ) + .context("Failed to mark parked flow run as resuming")?; + if changed > 0 { + tracing::debug!(target: "flows", run_id = id, "[flows] marked parked run 'running' for the duration of the resume"); + } + Ok(changed > 0) + }) +} + /// Reconciles a single orphaned `'running'` run row to a terminal /// `'interrupted'` status stamped `now` (RFC3339) with `reason`, guarded by a /// `status = 'running'` predicate so a run that settled or was resumed diff --git a/src/openhuman/flows/store_tests.rs b/src/openhuman/flows/store_tests.rs index b4d50a2638..c49b2dc1af 100644 --- a/src/openhuman/flows/store_tests.rs +++ b/src/openhuman/flows/store_tests.rs @@ -928,3 +928,70 @@ fn mark_run_interrupted_is_a_noop_for_a_terminal_row() { assert_eq!(row.status, "completed"); assert!(row.error.is_none()); } + +/// `expire_parked_runs` must return only the runs it ACTUALLY flipped, not the +/// candidates its `SELECT` saw. +/// +/// The `SELECT` and each row's guarded `UPDATE` are separate statements on an +/// autocommit connection, so a concurrent `mark_run_resuming` can claim a row in +/// between. The per-row `WHERE status = 'pending_approval'` keeps that row safe, +/// but returning the unfiltered candidate list would let the caller act on a run +/// it never expired — dropping the checkpoint out from under a live resume and +/// publishing a terminal `FlowRunFinished` for a run still executing. That false +/// event is the worse half: the frontend de-dupes terminal events per +/// `flow_id:run_id`, so the run's real completion would later be discarded. +#[test] +fn expire_parked_runs_returns_only_rows_it_actually_flipped() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "ttl".to_string(), trigger_graph(), false, true).unwrap(); + + let stale_at = "2000-01-01T00:00:00+00:00"; + for id in ["claimed-run", "genuinely-stale-run"] { + insert_flow_run(&config, id, &flow.id, id, stale_at).unwrap(); + finish_flow_run( + &config, + id, + "pending_approval", + stale_at, + &[], + &["gate".to_string()], + None, + ) + .unwrap(); + } + + // Simulate the race: one candidate is claimed by a resume after the sweep's + // SELECT would have seen it, but before its UPDATE lands. + assert!(mark_run_resuming(&config, "claimed-run").unwrap()); + + let swept = expire_parked_runs( + &config, + "2099-01-01T00:00:00+00:00", + "2026-01-01T00:00:00+00:00", + "expired", + ) + .unwrap(); + + let swept_ids: Vec<&str> = swept.iter().map(|(id, _)| id.as_str()).collect(); + assert_eq!( + swept_ids, + vec!["genuinely-stale-run"], + "only the row whose guarded UPDATE matched may be reported as swept" + ); + assert_eq!( + get_flow_run(&config, "claimed-run") + .unwrap() + .unwrap() + .status, + "running", + "the claimed run must keep executing, untouched by the sweep" + ); + assert_eq!( + get_flow_run(&config, "genuinely-stale-run") + .unwrap() + .unwrap() + .status, + "cancelled" + ); +}