From 7d7ff2a17ac3404701f8da9c7e1fedc34f485887 Mon Sep 17 00:00:00 2001 From: Nikolas Demiridis Date: Wed, 9 Sep 2026 23:59:45 +0300 Subject: [PATCH 1/3] fix(runtime): settle a parked run on its leader's disconnect, never on silence alone A parked run (waitingUser + turnSettled) used to be settled to `lost` by the inactivity sweep once it went quiet past the inactivity deadline (ADR-0027 wave 3). That conflated two different situations: a worker gone quiet while its leader is still connected and free to steer or finish it, and a leader that has itself vanished, leaving the run unfinishable. Silence looks identical from the runtime's side in both cases, but only the second is actually abandoned. Silence alone is never evidence a leader gave up -- only the leader's connection actually being gone is. Adds ipc::leader_registry::LeaderRegistry: a refcounted registry keyed by leader instance id, living in ipc::server::Shared beside the existing active_connections counter but populated from inside ipc::connection::handle (the negotiated instance id doesn't exist until the initialize handshake completes, which admit's spawn wrapper never sees). Refcounted rather than a set or a flag, because more than one physical connection can share one instance id (the crew-extension fallback used by any leader session with no real session id of its own), and plain addition resolves both connect/disconnect race orderings correctly without extra care. Each entry is {count, zero_since}, not a bare count. zero_since records the instant of the entry's most recent 1->0 transition and is cleared on any 0->1 -- closing a real bug in the simpler design: a leader that disconnects, reconnects, and disconnects again within one grace window can otherwise leave a stale timer (armed for the first disconnect) waking after the window and reading count == 0, correctly, but for the second disconnect, not the one it was spawned to watch. Recording the transition's instant lets a woken timer ask the actual question -- has THIS absence lasted the full window -- rather than merely whether the count happens to read zero right now. The registry's decrement is a drop guard (LeaderConnectionGuard), not a statement paired by hand at each exit from the dispatch loop: that loop's decrement needs the instance id, which only exists once execution is already inside it, so unlike active_connections it cannot be bookkept from the spawn wrapper, and a future early return would otherwise leak the registration silently, failing this feature closed with nothing to notice. The guard's on_gone callback fires exactly on the drop that causes the true 1->0 transition, arming the grace-window timer from inside Drop::drop itself, so arming is as structural as the decrement it rides on. The grace window is 60 seconds (LEADER_DISCONNECT_GRACE_WINDOW): short enough that an abandoned run does not tie up a concurrency slot indefinitely, long enough to absorb an ordinary restart or reconnect blip. Teardown reuses run_cancel's own mechanical shape rather than a second implementation of it: the guts of run_cancel are factored into a shared cancel_and_settle_run helper parameterized on the target RunState. run_cancel still asks for the literal "cancelled", checked against the requesting principal; the new settle_leader_gone path asks for RunState::unrendered_verdict() and passes no principal, bypassing the ownership check the same way the removed backstop did. Before settling, a WorkerTimeout { kind: LeaderGone } fact is journaled first -- the same report-before-decide shape the inactivity sweep already uses for Inactivity/Total. Daemon restart: at startup nothing has connected to the fresh registry yet, so every leader owning a non-terminal run is, from this process's own perspective, disconnected as of right now. OrchestrationService::seed_leader_registry_after_restart seeds each such owner with zero_since set to the daemon's own startup instant and arms the same grace-window timer a live disconnect would get. The clock deliberately starts at the daemon's own startup instant, never at a run's last-seen activity: a leader cannot reconnect to a daemon that is not running, so any time the daemon spent stopped is the daemon's own absence, not the leader's, and must never be charged against the window -- doing otherwise would settle live, reconnecting leaders' runs on every restart that followed a stop longer than the window. The disconnect grace window is now a ServerConfig field (leader_disconnect_grace) rather than a bare constant reference at each call site, defaulting to the production constant, so tests can shrink it instead of waiting out the real window. Removes the inactivity sweep's settle_abandoned_turn path entirely, along with its three dedicated tests. A parked run whose leader is still connected is never settled by the runtime now, however long it stays silent -- the inactivity fact is still journaled, for run/timeoutAck to react to, but it no longer drives a state change. lost returns to its original, narrower meaning: the evidence itself is missing or ambiguous, never "the leader went quiet." Adds ADR-0036 recording this decision, amending ADR-0027 (the backstop-settles-on-silence half of its Decision point 3 is superseded; the "never succeeded" half is inherited unamended) and noting the crew-extension shared-instance-id hazard the refcounting has to be safe under without resolving the pre-existing ownership ambiguity that id already carries elsewhere. Updates docs/architecture.md and docs/future-features.md, which described the now-removed backstop as current behavior. Tests: 9 unit tests on LeaderRegistry covering the refcount/zero_since mechanics directly (including the reconnect-within-window stale-zero case above); 2 unit tests on OrchestrationService::seed_leader_registry_after_restart against a real database (a parked run whose leader never reconnects is settled after the window; one that reconnects within it is left untouched and journals nothing) -- unit tests rather than integration tests because the mechanism under test is deliberately internal, never a surface a leader talks to; and 6 integration tests against the real IPC server covering a live disconnect's teardown, the shared settle helper, an in-window reconnect, cross-leader isolation, and the inactivity fact's continued journaling without settlement. bun run generate re-run for the new TimeoutKind::LeaderGone variant. --- crates/protocol/src/event.rs | 6 + crates/runtime/src/adapter/activity.rs | 6 + crates/runtime/src/domain/repository.rs | 125 +----- crates/runtime/src/ipc/connection.rs | 37 ++ crates/runtime/src/ipc/leader_registry.rs | 351 +++++++++++++++ crates/runtime/src/ipc/mod.rs | 10 + crates/runtime/src/ipc/server.rs | 41 ++ crates/runtime/src/lifecycle.rs | 18 + crates/runtime/src/service/orchestration.rs | 401 +++++++++++++++++- crates/runtime/src/service/query.rs | 59 +++ crates/runtime/src/timeout_sweep.rs | 120 ++---- crates/runtime/tests/orchestration_rpc.rs | 290 ++++++++++++- docs/adr/0027-turn-end-settles-a-run.md | 8 +- .../0036-leader-disconnect-grace-window.md | 160 +++++++ docs/adr/README.md | 3 +- docs/architecture.md | 9 +- docs/future-features.md | 5 +- packages/protocol-ts/schema/crew.schema.json | 17 +- .../protocol-ts/src/generated/TimeoutKind.ts | 2 +- 19 files changed, 1432 insertions(+), 236 deletions(-) create mode 100644 crates/runtime/src/ipc/leader_registry.rs create mode 100644 docs/adr/0036-leader-disconnect-grace-window.md diff --git a/crates/protocol/src/event.rs b/crates/protocol/src/event.rs index c22aef45..781738b9 100644 --- a/crates/protocol/src/event.rs +++ b/crates/protocol/src/event.rs @@ -346,6 +346,12 @@ pub enum AnsweredBy { pub enum TimeoutKind { Inactivity, Total, + /// The run's owning leader has held no live connection for the full + /// disconnect grace window. Distinct from `inactivity`/`total`, + /// which describe the worker's own liveness -- this describes the + /// leader's, and is the fact a grace-window teardown journals before + /// the run is settled. + LeaderGone, } /// Why a settled run resumed to `working`. The two causes map exactly diff --git a/crates/runtime/src/adapter/activity.rs b/crates/runtime/src/adapter/activity.rs index c1ae3e67..882d22b2 100644 --- a/crates/runtime/src/adapter/activity.rs +++ b/crates/runtime/src/adapter/activity.rs @@ -130,6 +130,12 @@ impl ActivityClock { match kind { TimeoutKind::Inactivity => entry.inactivity_journaled = true, TimeoutKind::Total => entry.total_journaled = true, + // This clock tracks worker/PTY liveness only -- a + // leader's connection state is `ipc::leader_registry`'s + // concern, tracked separately and never through here. + TimeoutKind::LeaderGone => unreachable!( + "ActivityClock has no leader-connection state; every real caller's `kind` comes from `due_timeouts`, which never produces LeaderGone" + ), } } } diff --git a/crates/runtime/src/domain/repository.rs b/crates/runtime/src/domain/repository.rs index 3e8f3402..7e25f088 100644 --- a/crates/runtime/src/domain/repository.rs +++ b/crates/runtime/src/domain/repository.rs @@ -1836,46 +1836,11 @@ impl<'c> DomainRepository<'c> { )?)) } - /// ADR-0027 wave 3's inactivity backstop: settles a run whose vendor - /// finished a turn that the leader then never closed. - /// - /// Settles to `lost`, never `succeeded`. The runtime observed only - /// silence, and ADR-0015's precedent is to name that uncertainty rather - /// than guess an outcome -- a success reached by nobody looking would be - /// a fabricated fact. The run's journaled result text stays readable - /// either way, since `run/result` reads any terminal state. - /// - /// Returns `None` unless the run is parked in `waitingUser` **and** - /// flagged `turnSettled`. That pairing is the whole safety property: a - /// `waitingUser` that means "the worker asked a question" or "an - /// approval is pending" must never be settled by a timeout, and neither - /// must a `waitingPeer` or a `paused` run. The read and the transition - /// share one transaction so a leader finishing the run concurrently - /// cannot be raced. - pub fn settle_abandoned_turn( - &mut self, - run_id: RunId, - ) -> Result, DomainError> { - let Some((state, turn_settled)): Option<(String, i64)> = self - .conn - .query_row( - "SELECT state, flags_turn_settled FROM runs WHERE run_id = ?1", - [run_id.to_string()], - |row| Ok((row.get(0)?, row.get(1)?)), - ) - .optional()? - else { - return Ok(None); - }; - if state != "waitingUser" || turn_settled == 0 { - return Ok(None); - } - let to = RunState::try_from("lost").map_err(|_| DomainError::NotFound { - kind: "run-state", - id: "lost".to_string(), - })?; - self.transition_run(run_id, &to, None).map(Some) - } + // `settle_abandoned_turn` (ADR-0027 wave 3's inactivity backstop, + // which settled a leader-abandoned turn to `lost` after inactivity + // alone) is removed: the maintainer ruled that silence is never + // evidence a leader gave up, only its connection actually being gone + // is. See ADR-0036 and `crate::ipc::leader_registry`. /// Updates a message's delivery state. Emits the matching `Message*` /// event. @@ -3706,86 +3671,6 @@ mod tests { assert_eq!(turns_used, 0, "answers must not consume turns"); } - /// The backstop settles a turn nobody closed -- to `lost`, because - /// silence is not evidence of success. - #[test] - fn the_backstop_settles_an_abandoned_settled_turn_as_lost() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working", "waitingUser"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to waitingUser"); - } - repo.set_run_flag(run_id, RunFlag::TurnSettled, true) - .expect("mark the wait as a finished turn"); - - let committed = repo - .settle_abandoned_turn(run_id) - .expect("settle") - .expect("an abandoned settled turn must be settled"); - match committed.envelope.event { - RuntimeEvent::RunEvent { ref state, .. } => assert_eq!(state.as_str(), "lost"), - ref other => panic!("expected a RunEvent, got {other:?}"), - } - // This layer only settles the state; clearing the marker is the - // sweep's job, so it is still set here. - assert!(repo.read_run_flags(run_id).unwrap().turn_settled); - } - - /// The safety property: a wait that is not a finished turn -- a - /// question, an approval -- must survive the backstop untouched. - #[test] - fn the_backstop_never_settles_a_wait_that_is_not_a_finished_turn() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working", "waitingUser"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to waitingUser"); - } - // No turnSettled flag: this run is waiting on a human, not holding - // a finished answer. - - assert!( - repo.settle_abandoned_turn(run_id) - .expect("settle") - .is_none(), - "a question wait must never be settled by a timeout" - ); - assert!( - !repo.read_run_flags(run_id).unwrap().turn_settled, - "sanity: the flag really is unset" - ); - } - - /// A run still working is not abandoned, however quiet it has been. - #[test] - fn the_backstop_never_settles_a_working_run() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to working"); - } - repo.set_run_flag(run_id, RunFlag::TurnSettled, true) - .expect("set the flag even though the state does not match"); - - assert!( - repo.settle_abandoned_turn(run_id) - .expect("settle") - .is_none(), - "the state and the flag must BOTH hold; the flag alone is not enough" - ); - } - #[test] fn timeout_facts_skip_terminal_runs_but_journal_for_live_ones() { let mut conn = open_test_db(); diff --git a/crates/runtime/src/ipc/connection.rs b/crates/runtime/src/ipc/connection.rs index 647c7cbd..6f6cc9c5 100644 --- a/crates/runtime/src/ipc/connection.rs +++ b/crates/runtime/src/ipc/connection.rs @@ -88,6 +88,43 @@ pub(crate) async fn handle(stream: UnixStream, ctx: ConnContext, shared: Arc 2 -> 1`, never touching +//! zero; a reconnect racing the other way (a fresh `0 -> 1` racing a +//! stale decrement already in flight) still resolves correctly, because +//! plain addition does not care which side's `+1`/`-1` is observed +//! first. +//! +//! A bare `count == 0` is not enough to decide "gone for the grace +//! window", though -- see [`Entry::zero_since`]'s own doc comment for +//! why, and [`LeaderRegistry::gone_for_at_least`] for the check that +//! actually answers it. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// How long a run's owning leader may hold no live connection at all +/// before the run is settled as unrendered (`RunState::unrendered_verdict`). +/// +/// The maintainer's ruling: a slot is held for at most a minute after a +/// leader genuinely leaves, which is short enough that an abandoned run +/// does not tie up a concurrency slot indefinitely, and long enough to +/// cover a quick OMP restart or reconnect -- the ordinary case this must +/// never fire for. A daemon restart is charged the same way (see +/// `LeaderRegistry::seed_disconnected_since`'s own doc comment for why +/// the clock starts at the daemon's own startup instant, never at the +/// run's last-seen time). +pub(crate) const LEADER_DISCONNECT_GRACE_WINDOW: Duration = Duration::from_secs(60); + +/// One instance id's live-connection bookkeeping. +#[derive(Default)] +struct Entry { + count: usize, + /// The instant `count` last dropped to zero, cleared the moment it + /// rises above zero again. A grace-window timer wakes long after it + /// was spawned and cannot tell, from `count == 0` alone, whether that + /// zero is the SAME one it was spawned to watch or a fresher one from + /// an intervening reconnect-then-disconnect cycle that happened to + /// also land on zero. Recording the instant of the transition (not + /// just the fact of it) is what lets the timer ask the actual + /// question: has the CURRENT absence lasted the full window, not + /// merely "is the count zero right now". + zero_since: Option, +} + +/// Refcounted live-connection registry, keyed by leader instance id. +/// Cloning shares the same underlying map (cheap, `Arc`-backed) -- every +/// clone observes and mutates the same state. +#[derive(Clone, Default)] +pub(crate) struct LeaderRegistry { + entries: Arc>>, +} + +impl LeaderRegistry { + pub(crate) fn new() -> Self { + Self::default() + } + + /// Registers one live connection for `instance_id`, returning a guard + /// that deregisters it on drop -- including on an early `return` out + /// of the connection's dispatch loop or a panic unwind, never only on + /// the loop's "normal" exit. See [`LeaderConnectionGuard`]'s own doc + /// comment for why that has to be structural. + /// + /// `on_gone` fires exactly when THIS drop is the one that brings the + /// count to zero (never on a decrement that leaves it above zero, and + /// never more than once per 1->0 transition) -- the caller's hook to + /// arm a grace-window timer. Fired from inside `Drop::drop` itself, + /// so arming is exactly as structural as the decrement it rides on: + /// a future early `return` that would otherwise skip an explicit + /// "now arm a timer" statement cannot skip this one, because there + /// is no separate statement to skip. This module knows nothing about + /// what the timer does or what "settling" means -- that stays in + /// `ipc::connection`/`service::orchestration`, which is what `on_gone` + /// is for. + pub(crate) fn register( + &self, + instance_id: String, + on_gone: impl Fn(String) + Send + Sync + 'static, + ) -> LeaderConnectionGuard { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let entry = entries.entry(instance_id.clone()).or_default(); + entry.count += 1; + entry.zero_since = None; + drop(entries); + LeaderConnectionGuard { + registry: self.clone(), + instance_id, + on_gone: Arc::new(on_gone), + } + } + + /// Decrements `instance_id`'s live count, returning `true` exactly + /// when this decrement is the one that brought it to zero (the + /// signal [`LeaderConnectionGuard::drop`] uses to fire `on_gone`). + fn deregister_reporting_zero_transition(&self, instance_id: &str) -> bool { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let Some(entry) = entries.get_mut(instance_id) else { + return false; + }; + entry.count = entry.count.saturating_sub(1); + if entry.count == 0 { + entry.zero_since = Some(Instant::now()); + true + } else { + false + } + } + + /// Seeds `instance_id` as having been at zero live connections since + /// `since`, WITHOUT creating a live connection -- the daemon-restart + /// case: at startup there are no connections at all yet, but a + /// parked run's owning leader was disconnected (from THIS daemon's + /// perspective) at the moment this process started, not whenever it + /// last happened to be observed. A leader cannot reconnect to a + /// daemon that is not running, so the clock for "how long has the + /// leader been gone" starts at startup, never at the run's own + /// last-seen time -- charging the daemon's own downtime against the + /// leader would settle live runs on every restart after a long stop, + /// which is exactly the destructive direction this feature exists to + /// avoid. + /// + /// A no-op if `instance_id` already has a live connection (count > 0) + /// by the time this runs -- seeding never clobbers a real connection + /// that beat it to registering. + pub(crate) fn seed_disconnected_since(&self, instance_id: String, since: Instant) { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let entry = entries.entry(instance_id).or_default(); + if entry.count == 0 { + entry.zero_since.get_or_insert(since); + } + } + + /// Whether `instance_id` has had zero live connections continuously + /// for at least `grace`, since the transition [`Self::gone_for_at_least`]'s + /// caller is actually asking about -- not merely whether the count + /// happens to read zero at the moment of the call. `zero_since` is + /// cleared by any intervening reconnect (see [`Self::register`]), so + /// a stale timer waking against a fresher zero correctly sees that + /// fresher zero hasn't aged enough yet, rather than settling early + /// against a zero that isn't the one it was spawned for. + pub(crate) fn gone_for_at_least(&self, instance_id: &str, grace: Duration) -> bool { + let entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + match entries.get(instance_id) { + Some(entry) => { + entry.count == 0 + && entry + .zero_since + .is_some_and(|since| since.elapsed() >= grace) + } + // No entry at all is the same fact as "gone": nothing has + // ever registered a live connection for this id. + None => true, + } + } +} + +/// Deregisters its connection's slot from a [`LeaderRegistry`] on drop. +/// +/// The existing `active_connections` counter (`server.rs`) gets this +/// "always decremented, however the connection ends" property for free, +/// because its own decrement lives in the SPAWN WRAPPER around the whole +/// connection future (`Server::admit`), which runs regardless of how +/// that future resolves. This bookkeeping instead has to live INSIDE the +/// connection's own future (`connection::handle`), because it needs the +/// negotiated instance id, which the spawn wrapper never sees -- the +/// `initialize` handshake that produces it happens deep inside `handle`, +/// not in `admit`. Living inside the future means every `break`/`return` +/// out of `handle`'s dispatch loop has to reach the decrement, and that +/// is a property of today's control flow, not of the type -- the next +/// early `return` added to that function would otherwise leak a +/// registration silently. A leaked registration means the count never +/// reaches zero, which means this leader is never considered gone: the +/// feature fails CLOSED, silently, and no test would notice unless it +/// happened to exercise that exact exit path. A drop guard makes the +/// invariant structural instead of conventional. +pub(crate) struct LeaderConnectionGuard { + registry: LeaderRegistry, + instance_id: String, + on_gone: Arc, +} + +impl Drop for LeaderConnectionGuard { + fn drop(&mut self) { + if self + .registry + .deregister_reporting_zero_transition(&self.instance_id) + { + (self.on_gone)(self.instance_id.clone()); + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use super::{LeaderConnectionGuard, LeaderRegistry}; + + const GRACE: Duration = Duration::from_millis(20); + + /// Registers with a no-op `on_gone`, for tests that only care about + /// the count/`zero_since` bookkeeping, not the callback. + fn reg(registry: &LeaderRegistry, id: &str) -> LeaderConnectionGuard { + registry.register(id.to_string(), |_| {}) + } + + #[test] + fn a_single_connection_is_not_gone_while_live() { + let registry = LeaderRegistry::new(); + let _guard = reg(®istry, "leader-1"); + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn dropping_the_only_connection_eventually_reads_gone() { + let registry = LeaderRegistry::new(); + let guard = reg(®istry, "leader-1"); + drop(guard); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "not yet -- the window hasn't elapsed" + ); + std::thread::sleep(GRACE * 2); + assert!(registry.gone_for_at_least("leader-1", GRACE)); + } + + /// A reconnect landing before the old connection's own teardown runs: + /// count goes 1 -> 2 -> 1, never touching zero. Never gone. + #[test] + fn a_reconnect_racing_the_previous_teardown_does_not_strand_the_leader() { + let registry = LeaderRegistry::new(); + let old = reg(®istry, "leader-1"); + let new = reg(®istry, "leader-1"); // reconnect arrives first + drop(old); // old connection's teardown lands second + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "one live connection (`new`) remains" + ); + std::thread::sleep(GRACE * 2); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "still live after the window elapses" + ); + drop(new); + } + + /// The race the other direction: the OLD connection's decrement is + /// the one to land second even though the reconnect happened first in + /// wall-clock terms -- still resolves correctly under commutative + /// arithmetic. + #[test] + fn a_reconnect_before_a_delayed_old_teardown_still_leaves_the_leader_live() { + let registry = LeaderRegistry::new(); + let old = reg(®istry, "leader-1"); + let new = reg(®istry, "leader-1"); + // old's teardown is delayed (simulated by ordering the drop last) + drop(old); + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + drop(new); + } + + #[test] + fn a_zero_that_reconnects_and_disconnects_again_within_the_window_resets_the_clock() { + let registry = LeaderRegistry::new(); + let first = reg(®istry, "leader-1"); + drop(first); // 1 -> 0, zero_since = now + std::thread::sleep(GRACE); // the original zero is now old enough... + let second = reg(®istry, "leader-1"); // ...but a reconnect arrives + drop(second); // 1 -> 0 again -- a FRESH zero_since + // The fresh zero has not aged the full window yet, even though + // enough wall-clock time has passed since the ORIGINAL zero that + // a stale timer checking only `count == 0` would have settled. + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn seeding_a_never_connected_instance_treats_it_as_gone_since_the_given_instant() { + let registry = LeaderRegistry::new(); + let since = std::time::Instant::now() - GRACE * 2; + registry.seed_disconnected_since("leader-1".to_string(), since); + assert!(registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn seeding_never_clobbers_a_connection_that_already_registered() { + let registry = LeaderRegistry::new(); + let _guard = reg(®istry, "leader-1"); + let since = std::time::Instant::now() - GRACE * 2; + registry.seed_disconnected_since("leader-1".to_string(), since); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "a live connection must never be treated as gone" + ); + } + + #[test] + fn an_unregistered_instance_id_reads_as_gone() { + let registry = LeaderRegistry::new(); + assert!(registry.gone_for_at_least("never-seen", GRACE)); + } + + #[test] + fn on_gone_fires_exactly_once_on_the_true_1_to_0_transition() { + let registry = LeaderRegistry::new(); + let fired: Arc>> = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&fired); + let a = registry.register("leader-1".to_string(), move |id| { + recorded.lock().unwrap().push(id) + }); + let recorded2 = Arc::clone(&fired); + let b = registry.register("leader-1".to_string(), move |id| { + recorded2.lock().unwrap().push(id) + }); + + drop(a); // 2 -> 1: not zero, on_gone must not fire + assert!( + fired.lock().unwrap().is_empty(), + "a decrement that leaves the count above zero must never fire on_gone" + ); + + drop(b); // 1 -> 0: the true transition + assert_eq!(*fired.lock().unwrap(), vec!["leader-1".to_string()]); + } +} diff --git a/crates/runtime/src/ipc/mod.rs b/crates/runtime/src/ipc/mod.rs index f772a2a5..7d9432de 100644 --- a/crates/runtime/src/ipc/mod.rs +++ b/crates/runtime/src/ipc/mod.rs @@ -14,6 +14,7 @@ //! guarantees that. mod connection; +pub(crate) mod leader_registry; mod server; use std::sync::Arc; @@ -208,6 +209,14 @@ pub struct ServerConfig { /// Effective retention policy for `retention/clean`. The daemon sets /// this from `crew.json`; absent test/embedded servers refuse the RPC. pub retention: Option, + /// How long a run's owning leader may hold no live connection before + /// its non-terminal runs are settled (see + /// [`leader_registry::LEADER_DISCONNECT_GRACE_WINDOW`]'s own doc + /// comment for the production value and its rationale). A field + /// rather than a bare constant reference at each call site so tests + /// can shrink it to a few milliseconds instead of waiting out the + /// real window. + pub leader_disconnect_grace: std::time::Duration, } impl Default for ServerConfig { @@ -229,6 +238,7 @@ impl Default for ServerConfig { activity_clock: None, pane_reopen: None, retention: None, + leader_disconnect_grace: leader_registry::LEADER_DISCONNECT_GRACE_WINDOW, } } } diff --git a/crates/runtime/src/ipc/server.rs b/crates/runtime/src/ipc/server.rs index 72152609..0271f0be 100644 --- a/crates/runtime/src/ipc/server.rs +++ b/crates/runtime/src/ipc/server.rs @@ -28,6 +28,17 @@ pub(crate) struct Shared { /// Number of connections currently admitted and being served. Used to /// decide whether the runtime is idle. pub(crate) active_connections: Arc, + /// Refcounted live-connection registry keyed by leader instance id -- + /// distinct from [`Self::active_connections`], which counts every + /// connection regardless of identity. Populated from inside + /// [`super::connection::handle`], not [`Server::admit`]: the + /// negotiated instance id doesn't exist until the `initialize` + /// handshake completes, deep inside `handle`, so unlike + /// `active_connections` this cannot be bookkept from the spawn + /// wrapper. See `leader_registry`'s own module doc for the refcount + /// rationale and the disconnect-grace-window mechanism this exists + /// for. + pub(crate) leader_registry: super::leader_registry::LeaderRegistry, /// Fired by an in-band `runtime/shutdown` request to trigger a graceful /// shutdown of the accept loop. pub(crate) shutdown: Arc, @@ -286,6 +297,7 @@ impl Server { started_at: Instant::now(), events_tx, active_connections: Arc::new(AtomicUsize::new(0)), + leader_registry: super::leader_registry::LeaderRegistry::new(), shutdown: Arc::new(Notify::new()), orchestration, coordination, @@ -373,6 +385,35 @@ impl Server { Arc::clone(&self.shared.violation) } + /// Routes every orchestration method to the domain repository -- + /// exposed (like [`Server::coordination_broker`]) so the daemon-restart + /// leader-grace-window seeding in `lifecycle.rs` can settle a run + /// through the SAME service instance every live disconnect uses, + /// never a second one. + #[must_use] + pub fn orchestration_service(&self) -> Arc { + Arc::clone(&self.shared.orchestration) + } + + /// The refcounted leader-connection registry -- exposed so + /// `lifecycle.rs` can seed it with every non-terminal run's owner at + /// startup (the daemon-restart case: nothing has connected yet, but a + /// leader may already be gone from this process's own perspective). + #[must_use] + pub(crate) fn leader_registry(&self) -> super::leader_registry::LeaderRegistry { + self.shared.leader_registry.clone() + } + + /// The configured disconnect grace window -- same value the live + /// per-connection teardown in `connection::handle` waits out, so the + /// restart-seeding sweep in `lifecycle.rs` never drifts from it (and + /// tests can shrink it via [`super::ServerConfig::leader_disconnect_grace`] + /// instead of waiting out the real production window). + #[must_use] + pub(crate) fn leader_disconnect_grace(&self) -> std::time::Duration { + self.shared.config.leader_disconnect_grace + } + /// Accepts and serves connections until `shutdown` resolves. /// /// Each accepted connection has its peer credentials read *before* any diff --git a/crates/runtime/src/lifecycle.rs b/crates/runtime/src/lifecycle.rs index 0d392b60..8a524c99 100644 --- a/crates/runtime/src/lifecycle.rs +++ b/crates/runtime/src/lifecycle.rs @@ -526,6 +526,24 @@ pub async fn serve(opts: &ServeOptions) -> Result<(), ServeError> { } }); + // Leader-disconnect grace window, daemon-restart case: at startup no + // connection has registered for ANY instance id yet, so every leader + // that owns a non-terminal run is -- from THIS process's own + // perspective -- disconnected as of right now. Seeding and arming is + // `OrchestrationService::seed_leader_registry_after_restart`'s own + // job (see its doc comment for the full "why the clock starts at + // startup, never at a run's last-seen activity" reasoning) -- + // extracted there, rather than inlined here, so a test can drive the + // restart case directly against a seeded database. + server + .orchestration_service() + .seed_leader_registry_after_restart( + &server.leader_registry(), + server.leader_disconnect_grace(), + Instant::now(), + ) + .await; + // The dashboard is an opt-in, read-only, localhost-only projection; a // bind failure (port taken) degrades to "no dashboard" rather than // failing a daemon that orchestrates fine without it. diff --git a/crates/runtime/src/service/orchestration.rs b/crates/runtime/src/service/orchestration.rs index b7185db1..64bb40d9 100644 --- a/crates/runtime/src/service/orchestration.rs +++ b/crates/runtime/src/service/orchestration.rs @@ -8,6 +8,7 @@ //! evidence, monotonic revision, connected-instance identity). use std::sync::Arc; +use std::time::{Duration, Instant}; use crew_protocol::{ ApprovalId, ApprovalRequest, CrewMethod, DeliveryState, EventEnvelope, IsolationKind, @@ -1529,14 +1530,40 @@ impl OrchestrationService { params: &Value, ) -> Result { let run_id = parse_run_id(params.get("runId"))?; - let project_id = self.project_id; let to = RunState::try_from("cancelled").expect("cancelled is valid"); - let principal_instance_id = principal.instance_id.clone(); + self.cancel_and_settle_run(run_id, to, Some(principal.instance_id.clone())) + .await + } + + /// The mechanical guts shared by two callers with different reasons + /// for the same shape of transition: `run_cancel` (an explicit + /// request, always `to: "cancelled"`, checked against the requesting + /// principal's ownership) and [`Self::settle_leader_gone`] (a + /// system-initiated settlement once a run's owning leader has been + /// gone for the full disconnect grace window, `to: + /// RunState::unrendered_verdict()`, no principal to check -- + /// `principal_instance_id: None` bypasses `transition_run`'s + /// ownership check the same way `settle_abandoned_turn` used to). + /// + /// Transitions and broadcasts first, then best-effort kills the + /// vendor subprocess (`CancelScope::Worker`) -- a kill failure never + /// fails the call (the transition already committed); it sets + /// `degradedControl` instead. Lease/slot release is not this + /// function's job: it happens downstream, evidence-driven, once the + /// adapter's own exit/cancel produces a `ProcessExited`/turn-boundary + /// event (see `adapter::registry::watch_slot`'s own doc comment). + async fn cancel_and_settle_run( + &self, + run_id: RunId, + to: RunState, + principal_instance_id: Option, + ) -> Result { + let project_id = self.project_id; let mut result = self .db .run_domain_op(Box::new(move |conn| { let mut repo = DomainRepository::new(conn, project_id); - repo.transition_run(run_id, &to, Some(&principal_instance_id)) + repo.transition_run(run_id, &to, principal_instance_id.as_deref()) .map(|c| embed_envelope(json!({ "sequence": c.sequence }), &c.envelope)) })) .await @@ -1549,10 +1576,10 @@ impl OrchestrationService { { // A real kill failure (an absent adapter is the clean // `CancelOutcome::NoRunningAdapter`, not an `Err`): the run is - // journaled `cancelled` but a vendor process may still be - // live. Make that visible to `run/get` and the monitor via - // `degradedControl`, mirroring the policy-violation - // path's treatment -- guarded write, journaled, broadcast. + // journaled but a vendor process may still be live. Make that + // visible to `run/get` and the monitor via `degradedControl`, + // mirroring the policy-violation path's treatment -- guarded + // write, journaled, broadcast. tracing::warn!(error = %err, run_id = %run_id, "failed to cancel running adapter subprocess"); match self .db @@ -1574,6 +1601,172 @@ impl OrchestrationService { Ok(result) } + + /// Settles every non-terminal run owned by `instance_id`, once its + /// grace-window timer (`ipc::leader_registry`) has confirmed it has + /// held no live connection for the full window. Called from exactly + /// two places, both re-checking `LeaderRegistry::gone_for_at_least` + /// themselves right before this runs: a live disconnect's own timer + /// (`ipc::connection::handle`'s `on_gone` callback), and the + /// daemon-startup seed for runs whose owner was never reconnected + /// after a restart ([`Self::seed_leader_registry_after_restart`]). + /// + /// `grace` is only for the journaled fact's `since_ms` -- the actual + /// decision already happened in the caller's `gone_for_at_least` + /// check; this never re-verifies it, and takes the caller's word for + /// how long the window it waited out actually was (the configured + /// value, which a test may have shrunk well below the production + /// [`crate::ipc::leader_registry::LEADER_DISCONNECT_GRACE_WINDOW`]). + /// + /// For each owned run: journals a `WorkerTimeout { kind: LeaderGone }` + /// fact FIRST (the same "report before deciding" shape the inactivity + /// sweep uses for `Inactivity`/`Total`), then transitions to + /// `RunState::unrendered_verdict()` -- never a literal `"cancelled"`, + /// so the one function that owns that word (`run.rs`) is the only + /// place it can ever change. A run that already went terminal between + /// the query and this loop (the leader finished it just in time) is + /// silently skipped: `transition_run`'s own guarded write refuses an + /// illegal edge, and that refusal is exactly correct here, not an + /// error to surface. + pub(crate) async fn settle_leader_gone(&self, instance_id: &str, grace: Duration) { + let project_id = self.project_id; + let run_ids: Vec = match self + .db + .run_domain_op(query::owned_nonterminal_run_ids_op( + instance_id.to_string(), + project_id, + )) + .await + { + Ok(value) => value + .as_array() + .map(|a| { + a.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + Err(err) => { + tracing::warn!(error = %err, instance_id, "failed to look up runs owned by a gone leader"); + return; + } + }; + + for run_id_str in run_ids { + let Ok(run_id) = RunId::parse(&run_id_str) else { + continue; + }; + + let mut timeout_fact = match self + .db + .run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + repo.record_worker_timeout_if_live( + run_id, + crew_protocol::TimeoutKind::LeaderGone, + u64::try_from(grace.as_millis()).unwrap_or(u64::MAX), + ) + .map(|committed| { + committed + .map(|c| embed_envelope(json!({ "sequence": c.sequence }), &c.envelope)) + .unwrap_or(Value::Null) + }) + })) + .await + { + Ok(value) => value, + Err(err) => { + tracing::warn!(error = %err, run_id = %run_id, "failed to journal a leaderGone timeout fact"); + Value::Null + } + }; + if !timeout_fact.is_null() { + self.broadcast(&mut timeout_fact); + } + + if let Err(err) = self + .cancel_and_settle_run(run_id, RunState::unrendered_verdict(), None) + .await + { + tracing::warn!( + error = %err.message, + run_id = %run_id, + instance_id, + "failed to settle a run whose leader is gone" + ); + } + } + } + + /// The daemon-restart half of the disconnect-grace-window feature: + /// at startup, nothing has connected to `leader_registry` yet, so + /// every leader that owns a non-terminal run is -- from THIS + /// process's own perspective -- disconnected as of `startup_instant`. + /// Seeds each such owner into the registry and arms the same + /// grace-window timer a live disconnect would, so a leader that + /// reconnects within the window clears it exactly as an ordinary + /// reconnect does (`LeaderRegistry::register`'s own `zero_since = + /// None`), and one that never returns gets its runs settled through + /// [`Self::settle_leader_gone`] -- the same path a live disconnect + /// uses -- rather than being stranded forever, which is the gap this + /// method exists to close. + /// + /// `startup_instant`, never a run's own last-seen activity: a leader + /// cannot reconnect to a daemon that is not running, so any time + /// this daemon spent stopped is the daemon's own absence, not the + /// leader's, and must never be charged against the grace window -- + /// doing so would settle live, reconnecting leaders' runs on every + /// restart that followed a stop longer than the window, which is + /// exactly the destructive direction this feature exists to avoid. + /// See [`crate::ipc::leader_registry::LeaderRegistry::seed_disconnected_since`]'s + /// own doc comment for the same reasoning at the mechanism that + /// enforces it. + /// + /// Extracted from `lifecycle::serve` as its own method (rather than + /// inlined at the one call site) so a test can drive it directly + /// against a seeded database, without spinning up a second full + /// daemon process to exercise a restart. + pub(crate) async fn seed_leader_registry_after_restart( + self: &Arc, + leader_registry: &crate::ipc::leader_registry::LeaderRegistry, + grace: Duration, + startup_instant: Instant, + ) { + let owners: Vec = match self + .db + .run_domain_op(query::distinct_owners_of_nonterminal_runs_op( + self.project_id, + )) + .await + { + Ok(value) => value + .as_array() + .map(|a| { + a.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + Err(err) => { + tracing::warn!(error = %err, "failed to list owners of non-terminal runs at startup"); + Vec::new() + } + }; + + for owner in owners { + leader_registry.seed_disconnected_since(owner.clone(), startup_instant); + let leader_registry = leader_registry.clone(); + let orchestration = Arc::clone(self); + tokio::spawn(async move { + tokio::time::sleep(grace).await; + if leader_registry.gone_for_at_least(&owner, grace) { + orchestration.settle_leader_gone(&owner, grace).await; + } + }); + } + } // ------------------------------------------------------- workspace /// Builds a materializer carrying this daemon's configured copy @@ -3491,3 +3684,197 @@ fn parse_message_kind(value: Option<&Value>) -> Result, + ) -> Arc { + let db = Arc::new(DatabaseHandle::start(db_path).await.expect("db starts")); + let violation = Arc::new(crate::policy::ViolationService::new( + db.clone(), + project_id, + events_tx.clone(), + None, + crate::config::NestedViolationAction::default(), + crate::security::redaction::Redactor::new(), + )); + let lease_service = Arc::new( + crate::workspace::LeaseService::open_in_memory(project_id).expect("lease service"), + ); + Arc::new(OrchestrationService::new( + db, + project_id, + None, + Arc::new(crate::approval::NoopApprovalCallback), + violation, + events_tx, + lease_service, + Arc::new(ArtifactStore::new()), + std::path::PathBuf::new(), + crate::config::crew::Limits::default().turn_budget_per_subtask, + Arc::new(crate::adapter::ActivityClock::new()), + )) + } + + /// Creates a task owned by `owner`, a worker, and a run submitted + /// (and left) at `queued` -- non-terminal, no adapter ever started -- + /// against `orchestration`'s own database. Returns the run id. + async fn seed_nonterminal_run(orchestration: &OrchestrationService, owner: &str) -> RunId { + let project_id = orchestration.project_id; + let db = Arc::clone(&orchestration.db); + let owner = owner.to_string(); + db.run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + let task_id = TaskId::new(); + repo.upsert_task( + task_id, + &TaskRef { + owner_client_instance_id: owner.clone(), + revision: 1, + }, + )?; + let worker_id = WorkerId::new(); + repo.create_worker(&Worker { + worker_id, + profile_ref: WorkerProfileRef { + id: worker_id, + fingerprint: "sha256:fake".into(), + adapter: "fake".into(), + model: "test".into(), + permission_envelope: serde_json::json!({}), + }, + parent_worker_id: None, + created_at: Timestamp::now(), + })?; + let run_id = RunId::new(); + repo.submit_run( + &Run { + run_id, + task_id, + worker_id, + state: RunState::try_from("queued").expect("queued is valid"), + flags: RunFlags::default(), + vendor_session_id: None, + started_at: None, + completed_at: None, + }, + None, + None, + )?; + Ok(serde_json::json!(run_id.to_string())) + })) + .await + .expect("seed run") + .as_str() + .and_then(|s| RunId::parse(s).ok()) + .expect("run id round-trips") + } + + async fn run_state(orchestration: &OrchestrationService, run_id: RunId) -> String { + orchestration + .db + .run_domain_op(query::run_get_op(run_id)) + .await + .expect("run_get")["state"] + .as_str() + .expect("state is a string") + .to_string() + } + + #[tokio::test] + async fn a_restart_settles_a_run_whose_leader_never_reconnects_after_the_window() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, mut events_rx) = broadcast::channel(16); + let orchestration = + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx).await; + let run_id = seed_nonterminal_run(&orchestration, "leader-gone").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(20); + // Nothing has connected to `leader_registry` since it was just + // constructed -- exactly the daemon-restart precondition: an + // empty in-memory registry sitting on top of a database that + // still has this run's owner recorded. + orchestration + .seed_leader_registry_after_restart(&leader_registry, grace, Instant::now()) + .await; + + // Past the window, with margin: the leader never reconnects. + tokio::time::sleep(grace * 5).await; + + assert_eq!( + run_state(&orchestration, run_id).await, + "cancelled", + "a leader that never returns after a restart must be settled via unrendered_verdict()" + ); + + let fact = events_rx.try_recv().expect("a fact was broadcast"); + match fact.event { + crew_protocol::RuntimeEvent::WorkerTimeout { + run_id: fact_run_id, + kind, + .. + } => { + assert_eq!(fact_run_id, run_id); + assert_eq!(kind, crew_protocol::TimeoutKind::LeaderGone); + } + other => panic!("expected a WorkerTimeout{{kind: LeaderGone}} fact, got {other:?}"), + } + } + + #[tokio::test] + async fn a_restart_leaves_a_run_untouched_when_its_leader_reconnects_within_the_window() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, mut events_rx) = broadcast::channel(16); + let orchestration = + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx).await; + let run_id = seed_nonterminal_run(&orchestration, "leader-returns").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(60); + orchestration + .seed_leader_registry_after_restart(&leader_registry, grace, Instant::now()) + .await; + + // The leader reconnects well within the window and stays connected + // (the guard is held for the rest of the test) -- an ordinary + // reconnect, which must clear `zero_since` exactly as a live + // disconnect's own reconnect does. + let _reconnected = leader_registry.register("leader-returns".to_string(), |_| {}); + + tokio::time::sleep(grace * 3).await; + + assert_eq!( + run_state(&orchestration, run_id).await, + "queued", + "a leader that reconnects within the window must leave its runs untouched" + ); + assert!( + events_rx.try_recv().is_err(), + "a reconnect within the window must journal nothing" + ); + } +} diff --git a/crates/runtime/src/service/query.rs b/crates/runtime/src/service/query.rs index a35880c1..242e87f1 100644 --- a/crates/runtime/src/service/query.rs +++ b/crates/runtime/src/service/query.rs @@ -404,6 +404,65 @@ pub fn owned_run_ids_op( }) } +/// The run states [`crew_protocol::RunState::is_terminal`] considers +/// terminal, duplicated here because this is raw SQL rather than a Rust +/// match -- kept as one named constant, not inlined at each use, so the +/// two queries below (and any future one) cannot drift from each other +/// even if they drift from `is_terminal` itself. If `is_terminal`'s own +/// set ever changes, this is the one place to update. +const TERMINAL_STATES_SQL: &str = "'succeeded','failed','cancelled','lost'"; + +/// Non-terminal run ids owned (via `tasks.owner_client_instance_id`) by +/// `owner_instance_id` -- the leader-disconnect-grace-window teardown's +/// own query: once a leader has been gone for the grace window, these +/// are exactly the runs it left behind that still need settling. Unlike +/// [`owned_run_ids_op`], this filters to non-terminal runs directly in +/// SQL rather than leaving the caller to filter afterward -- the caller +/// here (`settle_leader_gone`) has no other use for a terminal run's id. +pub fn owned_nonterminal_run_ids_op( + owner_instance_id: String, + project_id: ProjectId, +) -> DomainClosure { + Box::new(move |conn| { + let sql = format!( + "SELECT r.run_id FROM runs r JOIN tasks t ON r.task_id = t.task_id \ + WHERE t.project_id = ?1 AND t.owner_client_instance_id = ?2 \ + AND r.state NOT IN ({TERMINAL_STATES_SQL})" + ); + let mut stmt = conn.prepare(&sql)?; + let ids: Vec = stmt + .query_map( + rusqlite::params![project_id.to_string(), owner_instance_id], + |row| row.get(0), + )? + .filter_map(|r| r.ok()) + .collect(); + Ok(json!(ids)) + }) +} + +/// Every DISTINCT owner instance id with at least one non-terminal run, +/// project-wide -- the daemon-restart seed: at startup nothing has +/// connected yet, so every one of these owners is, from this process's +/// own perspective, disconnected as of right now (see +/// `ipc::leader_registry::LeaderRegistry::seed_disconnected_since`'s own +/// doc comment for why the clock starts at startup, not at the run's +/// last-seen time). +pub fn distinct_owners_of_nonterminal_runs_op(project_id: ProjectId) -> DomainClosure { + Box::new(move |conn| { + let sql = format!( + "SELECT DISTINCT t.owner_client_instance_id FROM runs r JOIN tasks t ON r.task_id = t.task_id \ + WHERE t.project_id = ?1 AND r.state NOT IN ({TERMINAL_STATES_SQL})" + ); + let mut stmt = conn.prepare(&sql)?; + let owners: Vec = stmt + .query_map(rusqlite::params![project_id.to_string()], |row| row.get(0))? + .filter_map(|r| r.ok()) + .collect(); + Ok(json!(owners)) + }) +} + /// Confirms `expected_instance_id` currently owns the task that owns /// `run_id`, for `workspace/acquire`'s ownership arbitration. /// diff --git a/crates/runtime/src/timeout_sweep.rs b/crates/runtime/src/timeout_sweep.rs index d8665798..b5a639d1 100644 --- a/crates/runtime/src/timeout_sweep.rs +++ b/crates/runtime/src/timeout_sweep.rs @@ -13,6 +13,17 @@ //! Every journal decision re-checks liveness inside the database actor //! closure (`record_worker_timeout_if_live`), so a run that settled //! between the clock snapshot and the write never receives a timeout fact. +//! +//! This sweep never transitions a run's state on its own -- a parked run +//! whose leader is still connected is never settled here, however long +//! it stays silent. It used to: an inactivity backstop settled a +//! leader-abandoned turn to `lost` (ADR-0027 wave 3). The maintainer +//! ruled that out -- silence alone is never evidence the leader gave up, +//! only the leader's connection actually being gone is -- so settlement +//! now happens exclusively through `crate::ipc::leader_registry`'s +//! disconnect grace window (see ADR-0036). `lost` keeps its narrower, +//! original meaning (ADR-0023): the evidence itself is missing or +//! ambiguous, not "the leader went quiet." use std::time::{Duration, Instant}; @@ -51,6 +62,15 @@ pub async fn sweep_once( let since_ms = match kind { TimeoutKind::Inactivity => millis_since(activity.last_activity, now), TimeoutKind::Total => millis_since(activity.started_at, now), + // `due_timeouts` (the only producer of `kind` here) only + // ever returns `Inactivity`/`Total` -- it has no notion of + // a leader's connection state at all. `LeaderGone` is + // journaled from the leader-grace-window teardown + // directly (`service::orchestration::settle_leader_gone`), + // never through this worker-liveness sweep. + TimeoutKind::LeaderGone => unreachable!( + "due_timeouts never produces LeaderGone -- it has no leader-connection state to observe" + ), }; let outcome = db .run_domain_op(Box::new({ @@ -81,19 +101,20 @@ pub async fn sweep_once( // Mark only after a successful commit: a failed append // leaves the flag unset so the next tick retries. clock.mark_journaled(&run_id, kind); - // ADR-0027 wave 3: the inactivity backstop. A TUI run - // parked by a finished turn that the leader never - // closed is settled here rather than sitting forever. - // Deliberately narrow -- `settle_abandoned_turn` - // requires BOTH `waitingUser` and `turnSettled`, so a - // question wait, an approval wait, `waitingPeer` and - // `paused` are all left alone, and the runtime's - // "journal, never decide" stance still holds for - // everything except the one case the leader provably - // walked away from. - if matches!(kind, TimeoutKind::Inactivity) { - settle_abandoned_turn(db, project_id, events_tx, clock, run_id).await; - } + // A parked run whose leader is still alive (connected, + // regardless of how long it stays silent) is never + // settled here -- only the fact above is journaled. + // Settlement now happens exclusively through the + // leader-disconnect grace window + // (`ipc::leader_registry`, + // `OrchestrationService::settle_leader_gone`): the + // inactivity backstop that used to settle a + // leader-abandoned turn to `lost` after this same + // `Inactivity` fact (ADR-0027 wave 3) was removed + // once the maintainer ruled that the runtime must + // never decide a leader gave up based on silence + // alone -- only on its connection actually being + // gone. See ADR-0036. } Ok(_) => { // The run is terminal or gone: drop its clock so this @@ -112,76 +133,3 @@ pub async fn sweep_once( } } } - -/// Settles a run parked by a finished turn the leader never closed, and -/// clears the marker that described that pause. -/// -/// Split out of [`sweep_once`] so the "did anything actually need -/// settling?" decision lives next to its own commit-and-broadcast, and so -/// the sweep's per-timeout loop stays readable. Every failure is logged and -/// dropped: the timeout fact itself is already durable, and the next tick -/// re-attempts anything that did not land. -async fn settle_abandoned_turn( - db: &DatabaseHandle, - project_id: ProjectId, - events_tx: &broadcast::Sender, - clock: &ActivityClock, - run_id: crew_protocol::RunId, -) { - let settled = db - .run_domain_op(Box::new(move |conn| { - Ok(DomainRepository::new(conn, project_id) - .settle_abandoned_turn(run_id)? - .map(|committed| { - embed_envelope( - json!({ "sequence": committed.sequence }), - &committed.envelope, - ) - }) - .unwrap_or(serde_json::Value::Null)) - })) - .await; - match settled { - Ok(mut value) if !value.is_null() => { - broadcast_committed(events_tx, &mut value); - tracing::info!( - run_id = %run_id, - "settled an abandoned turn as lost after inactivity" - ); - // The run is terminal now, so its clock has nothing left to - // measure. - clock.forget(&run_id); - // The marker described a pause that no longer exists; leaving - // it set would tell `run/get` an answer is still waiting to be - // collected on a run that has ended. - match db - .run_domain_op(Box::new(move |conn| { - DomainRepository::new(conn, project_id) - .set_run_flag(run_id, crate::domain::RunFlag::TurnSettled, false) - .map(|committed| { - embed_envelope( - json!({ "sequence": committed.sequence }), - &committed.envelope, - ) - }) - })) - .await - { - Ok(mut cleared) => { - broadcast_committed(events_tx, &mut cleared); - } - Err(err) => tracing::warn!( - run_id = %run_id, - error = %err, - "failed to clear turnSettled after settling an abandoned turn" - ), - } - } - Ok(_) => {} - Err(err) => tracing::warn!( - run_id = %run_id, - error = %err, - "failed to settle an abandoned turn" - ), - } -} diff --git a/crates/runtime/tests/orchestration_rpc.rs b/crates/runtime/tests/orchestration_rpc.rs index 18c84f46..eac615f6 100644 --- a/crates/runtime/tests/orchestration_rpc.rs +++ b/crates/runtime/tests/orchestration_rpc.rs @@ -26,7 +26,7 @@ use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::net::unix::OwnedWriteHalf; -use tokio::sync::oneshot; +use tokio::sync::{broadcast, oneshot}; use tokio::task::JoinHandle; #[path = "support/spawn_evidence_adapter.rs"] @@ -5929,14 +5929,16 @@ impl RunDriver for SettlingRunDriver { } } -/// Submits a run against `adapter: "fake"`, returning its id. Mirrors the -/// other suites' helper; ids start at 2 because `initialize` consumed 1. -async fn submit_fake_run(client: &mut Client) -> String { +/// Submits a run against `adapter: "fake"`, owned by `owner` (which must +/// match `client`'s own connected instance id -- `task/upsert` refuses a +/// mismatch), returning its id. Mirrors the other suites' helper; ids +/// start at 2 because `initialize` consumed 1. +async fn submit_fake_run(client: &mut Client, owner: &str) -> String { let task = client .call( 2, "task/upsert", - json!({ "ownerClientInstanceId": "omp-1", "revision": 1 }), + json!({ "ownerClientInstanceId": owner, "revision": 1 }), ) .await; let task_id = task["result"]["taskId"].as_str().unwrap().to_string(); @@ -5984,7 +5986,7 @@ async fn run_finish_settles_a_settled_turn_as_succeeded_and_tears_down_the_worke }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!( wait_for_state(&mut client, 5, &run_id, "waitingUser").await, "the seeded turn boundary must park the run first" @@ -6034,7 +6036,7 @@ async fn run_finish_can_settle_a_run_as_failed_when_the_leader_says_so() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let finish = client @@ -6060,7 +6062,7 @@ async fn run_finish_refuses_an_already_terminal_run() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let first = client @@ -6092,7 +6094,7 @@ async fn run_finish_refuses_an_unknown_outcome() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let finish = client @@ -6450,3 +6452,273 @@ async fn a_child_denial_reason_is_redacted_before_it_is_journaled() { .expect("a childWorkerRequestDenied event"); assert_reason_masked(denied); } + +// -------------------------------------------- leader-disconnect grace window + +/// The `workerTimeout` `kind` tag values journaled for `run_id` in a replay +/// response, in emission order. Mirrors `workspace_event_kinds_for_run`'s +/// shape for the sibling `workerTimeout` event type. +fn worker_timeout_kinds_for_run(replay: &Value, run_id: &str) -> Vec { + replay["result"] + .as_array() + .expect("events/replay returns an array") + .iter() + .map(|e| &e["event"]) + .filter(|e| e["type"] == "workerTimeout" && e["payload"]["runId"] == run_id) + .map(|e| e["payload"]["kind"].as_str().unwrap().to_string()) + .collect() +} + +/// A leader that disconnects entirely (drops its only connection) and never +/// comes back must have its parked run's vendor session torn down once the +/// grace window elapses -- the same `cancel_run(Worker)` call `run/cancel` +/// itself makes. +#[tokio::test] +async fn a_leader_disconnecting_stops_its_runs_workers() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + let mut client = omp_client(&harness, "leader-1").await; + let run_id = submit_fake_run(&mut client, "leader-1").await; + assert!( + wait_for_state(&mut client, 5, &run_id, "waitingUser").await, + "the seeded turn boundary must park the run first" + ); + + drop(client); // the leader disconnects entirely; nobody reconnects + + // Past the grace window, with margin for the timer's own wakeup. + tokio::time::sleep(Duration::from_millis(500)).await; + + let calls = driver.cancel_calls(); + assert_eq!( + calls.len(), + 1, + "the disconnect teardown must cancel the parked run's worker: {calls:?}" + ); + assert_eq!(calls[0].1, CancelScope::Worker); +} + +/// The same disconnect settles the run's own state through +/// `OrchestrationService::cancel_and_settle_run` -- the exact helper +/// `run/cancel` uses -- landing on `RunState::unrendered_verdict()` +/// (`"cancelled"`), never a literal decided at this call site. +#[tokio::test] +async fn a_disconnect_settles_owned_runs_via_the_shared_helper() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + let mut client = omp_client(&harness, "leader-2").await; + let run_id = submit_fake_run(&mut client, "leader-2").await; + assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); + + drop(client); + tokio::time::sleep(Duration::from_millis(500)).await; + + // `run/get` is not ownership-scoped -- a fresh, unrelated connection + // can read the settled state back. + let mut checker = omp_client(&harness, "checker").await; + let get = checker.call(2, "run/get", json!({ "runId": run_id })).await; + assert_eq!( + get["result"]["state"], "cancelled", + "the shared cancel_and_settle_run helper must settle the run: {get:?}" + ); +} + +/// A leader that disconnects and reconnects (same instance id) well within +/// the grace window must never have its run touched, and the grace-window +/// teardown must journal nothing -- an ordinary reconnect is indistinguishable +/// from a leader that was never gone. +#[tokio::test] +async fn a_reconnect_within_the_window_leaves_the_run_untouched_and_journals_nothing() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(150); + }) + .await; + let mut client = omp_client(&harness, "leader-3").await; + let run_id = submit_fake_run(&mut client, "leader-3").await; + assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); + + drop(client); // disconnects... + tokio::time::sleep(Duration::from_millis(30)).await; // ...well within the window... + let mut reconnected = omp_client(&harness, "leader-3").await; // ...and reconnects. + + // Past the ORIGINAL window, with margin. + tokio::time::sleep(Duration::from_millis(400)).await; + + assert!( + driver.cancel_calls().is_empty(), + "a reconnect within the window must never tear the worker down: {:?}", + driver.cancel_calls() + ); + + let get = reconnected + .call(5, "run/get", json!({ "runId": run_id })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "a reconnect within the window must leave the run untouched: {get:?}" + ); + + let replay = reconnected + .call(6, "events/replay", json!({ "afterSequence": 0 })) + .await; + assert!( + worker_timeout_kinds_for_run(&replay, &run_id) + .iter() + .all(|k| k != "leaderGone"), + "a reconnect within the window must journal nothing: {replay:?}" + ); +} + +/// A run owned by one leader is untouched by a different leader's +/// connection lifecycle entirely -- disconnecting `leader-4` must never +/// affect a run owned by `leader-5`, still live. +#[tokio::test] +async fn a_run_owned_by_another_live_client_is_not_touched() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + + let mut departing = omp_client(&harness, "leader-4").await; + let departing_run = submit_fake_run(&mut departing, "leader-4").await; + assert!(wait_for_state(&mut departing, 5, &departing_run, "waitingUser").await); + + let mut staying = omp_client(&harness, "leader-5").await; + let staying_run = submit_fake_run(&mut staying, "leader-5").await; + assert!(wait_for_state(&mut staying, 5, &staying_run, "waitingUser").await); + + drop(departing); // only leader-4 goes away + + tokio::time::sleep(Duration::from_millis(500)).await; + + let calls = driver.cancel_calls(); + assert_eq!( + calls + .iter() + .filter(|(id, _)| id.to_string() == departing_run) + .count(), + 1, + "the departed leader's run must be torn down: {calls:?}" + ); + assert_eq!( + calls + .iter() + .filter(|(id, _)| id.to_string() == staying_run) + .count(), + 0, + "a run owned by a still-connected leader must never be touched: {calls:?}" + ); + + let get = staying + .call(6, "run/get", json!({ "runId": staying_run })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "the still-connected leader's own run must be untouched: {get:?}" + ); +} + +/// A parked run whose leader is still connected must never be settled by +/// the inactivity sweep alone, however long it stays quiet -- only the +/// leader's connection actually being gone (`ipc::leader_registry`) may +/// settle it. This is the maintainer's ruling that removed +/// `settle_abandoned_turn` (see `timeout_sweep`'s own module doc). +#[tokio::test] +async fn a_parked_run_whose_leader_is_alive_is_never_settled_by_inactivity() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + }) + .await; + let mut client = omp_client(&harness, "leader-6").await; + let run_id_str = submit_fake_run(&mut client, "leader-6").await; + assert!(wait_for_state(&mut client, 5, &run_id_str, "waitingUser").await); + // `client` is never dropped: the leader stays connected for the whole test. + + let run_id = RunId::parse(&run_id_str).expect("run id round-trips"); + let clock = ActivityClock::new(); + // Stale activity: comfortably past even a millisecond-scale threshold. + clock.touch(&run_id, std::time::Instant::now() - Duration::from_secs(10)); + + let sweep_db = DatabaseHandle::start(harness.database.clone()) + .await + .expect("a second reader/writer handle onto the same WAL-mode database"); + let (events_tx, _events_rx) = broadcast::channel(16); + crew_runtime::timeout_sweep::sweep_once( + &sweep_db, + harness.project_id, + &events_tx, + &clock, + Duration::from_millis(1), + Duration::from_secs(999_999), + ) + .await; + + let get = client + .call(6, "run/get", json!({ "runId": run_id_str })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "inactivity alone must never settle a run whose leader is still connected: {get:?}" + ); + assert!( + driver.cancel_calls().is_empty(), + "no teardown may fire from inactivity alone: {:?}", + driver.cancel_calls() + ); +} + +/// The inactivity fact itself is unaffected by `settle_abandoned_turn`'s +/// removal: the sweep still journals `workerTimeout { kind: inactivity }` +/// exactly as before, for `run/timeoutAck` to react to -- only the +/// runtime-side settlement that used to follow it is gone. +#[tokio::test] +async fn an_inactivity_timeout_still_journals_the_fact() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + }) + .await; + let mut client = omp_client(&harness, "leader-7").await; + let run_id_str = submit_fake_run(&mut client, "leader-7").await; + assert!(wait_for_state(&mut client, 5, &run_id_str, "waitingUser").await); + + let run_id = RunId::parse(&run_id_str).expect("run id round-trips"); + let clock = ActivityClock::new(); + clock.touch(&run_id, std::time::Instant::now() - Duration::from_secs(10)); + + let sweep_db = DatabaseHandle::start(harness.database.clone()) + .await + .expect("a second reader/writer handle onto the same WAL-mode database"); + let (events_tx, _events_rx) = broadcast::channel(16); + crew_runtime::timeout_sweep::sweep_once( + &sweep_db, + harness.project_id, + &events_tx, + &clock, + Duration::from_millis(1), + Duration::from_secs(999_999), + ) + .await; + + let replay = client + .call(6, "events/replay", json!({ "afterSequence": 0 })) + .await; + assert_eq!( + worker_timeout_kinds_for_run(&replay, &run_id_str), + vec!["inactivity".to_string()], + "the inactivity fact must still be journaled: {replay:?}" + ); +} diff --git a/docs/adr/0027-turn-end-settles-a-run.md b/docs/adr/0027-turn-end-settles-a-run.md index ac4c72bd..c5c3ee89 100644 --- a/docs/adr/0027-turn-end-settles-a-run.md +++ b/docs/adr/0027-turn-end-settles-a-run.md @@ -1,6 +1,6 @@ # A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state -* Status: Accepted +* Status: Accepted; Decision point 3's backstop-settles-on-silence half superseded by [0036](0036-leader-disconnect-grace-window.md) -- its "never `succeeded`" half stands, unamended. * Date: 2026-08-29 * Amends: [0023](0023-run-state-edges-from-adapter-evidence.md) @@ -84,8 +84,10 @@ current bug under a nicer name — a leader that dies or loses interest leaves r Turn-end is journaled and broadcast as evidence, driving `working -> waitingUser` — an already-legal, already-modelled non-terminal state meaning "the runtime is not the blocker", which `next_hop` already routes through. `run/result` becomes readable for a run in `waitingUser` that has -at least one turn-end. The leader settles explicitly via `run/finish`; an inactivity backstop settles -an abandoned run without one. +at least one turn-end. The leader settles explicitly via `run/finish`; an inactivity backstop settled +an abandoned run without one until [0036](0036-leader-disconnect-grace-window.md) replaced that +backstop with a disconnect-grace-window teardown -- silence alone was never sufficient evidence that +a leader had given up, only its connection actually being gone is. Against it: it is two-phase rather than one change, and it borrows `waitingUser`, whose existing meaning in the monitor is "the worker asked a question" — so the two need to be distinguishable. diff --git a/docs/adr/0036-leader-disconnect-grace-window.md b/docs/adr/0036-leader-disconnect-grace-window.md new file mode 100644 index 00000000..f2221554 --- /dev/null +++ b/docs/adr/0036-leader-disconnect-grace-window.md @@ -0,0 +1,160 @@ +# A parked run is settled on its leader's connection actually being gone, never on silence alone + +* Status: Accepted +* Date: 2026-09-09 +* Amends: [0027](0027-turn-end-settles-a-run.md) + +## Context and Problem Statement + +ADR-0027's wave 3 gave the inactivity sweep a backstop: a run parked in `waitingUser` with +`turnSettled` set, quiet past the inactivity deadline, was settled to `lost`. That was written when +the only failure this daemon could observe was silence, and it conflated two genuinely different +situations under one signal: a worker gone quiet while its leader is still there, still watching, +still free to steer or finish it; and a leader that has itself vanished, leaving a parked run nobody +will ever finish or retry. Silence looks identical from the runtime's side in both cases, but only +the second one is actually abandoned. + +The maintainer ruled the backstop out on exactly that basis: silence alone is never evidence a leader +gave up, only the leader's own connection actually being gone is. Acting on that ruling required a +signal this daemon did not yet track — leader *connection* identity as its own liveness concern, +independent of the worker/vendor process activity `ActivityClock` already measures. It also surfaced +a second, un-mentioned failure mode once the first was fixed: an in-memory notion of "who is +connected" starts empty on every daemon restart regardless of what was true a moment before, so +naively treating "no live connection recorded yet" as "gone" would settle every parked run on every +restart — a regression far worse than the backstop this decision removes. + +## Decision Drivers + +* Never settle a run because it was quiet; settle it only because its owning leader is provably gone. +* Never strand a run forever either — an abandoned leader is not something a human will notice and + cancel by hand. +* Reuse `run_cancel`'s existing settle mechanism rather than invent a second one: the ordering and + broadcast invariants ADR-0020 requires were already solved once, and a second, parallel + implementation is a second place they could quietly diverge. +* A daemon restart is not a leader event. The daemon's own downtime must never be charged against a + leader's grace window, or a long-stopped daemon would settle every reconnecting leader's runs the + moment it came back up. +* More than one physical connection can legitimately share one leader instance id (the + `crew-extension` fallback id, used by any leader session with no real session id of its own), so + correctness cannot assume one instance id maps to one connection. + +## Considered Options + +* **Keep the inactivity backstop.** Rejected outright by the maintainer's ruling above — it conflates + worker silence with leader abandonment, which are not the same fact. +* **Track leader liveness as a boolean or a set of connected ids.** Fails the shared-instance-id case + (a second physical connection under the same id would overwrite or be indistinguishable from the + first) and both reconnect-race orderings (a fresh connect racing a stale disconnect can land either + order). +* **A bare refcount, re-checked at `count == 0` when a timer wakes.** Handles the shared-id and + race-ordering cases, but has a real bug: a leader that disconnects, reconnects, and disconnects + again — all within one grace window — can leave a *stale* timer (armed for the first disconnect) + waking after the window and reading `count == 0`, which is true, but for the *second* disconnect, + not the one the timer was spawned to watch. It would settle a leader that has only been gone a + fraction of the configured window. +* **A refcounted registry recording the instant of each true 1→0 transition.** Chosen — see below. + +## Decision Outcome + +Chosen option: a refcounted `LeaderRegistry`, keyed by leader instance id, living in +`ipc::server::Shared` beside the existing `active_connections` counter but populated from inside +`ipc::connection::handle` rather than `Server::admit` — the negotiated instance id does not exist +until the `initialize` handshake completes, deep inside `handle`, which `admit`'s spawn wrapper never +sees. + +Each entry is `{count, zero_since}`, not a bare count. `zero_since` records the instant of the +entry's most recent 1→0 transition and is cleared on any 0→1 — the property that closes the bare- +refcount bug above: a grace-window timer waking later can tell "the same zero it was spawned to +watch" apart from "a fresher zero from an intervening reconnect/disconnect cycle" (see +`ipc::leader_registry::Entry::zero_since`'s own doc comment for the worked timeline). The disconnect +grace window itself is 60 seconds (`LEADER_DISCONNECT_GRACE_WINDOW`): short enough that an abandoned +run does not tie up a concurrency slot indefinitely, long enough to absorb an ordinary restart or +reconnect blip — the case this must never fire for. The decrement that can bring an entry to zero is +a drop guard (`LeaderConnectionGuard`), not a statement paired by hand at each exit from `handle`'s +dispatch loop: that loop's decrement needs the instance id, which only exists once execution is +already inside it, so unlike `active_connections` it cannot be bookkept from the spawn wrapper, and a +future early `return` would otherwise leak the registration silently, failing this feature closed +with nothing to notice. The guard's own `on_gone` callback fires exactly on the drop that causes the +true 1→0 transition, arming the grace-window timer from inside `Drop::drop` itself, so arming is as +structural as the decrement it rides on. + +Teardown reuses `run_cancel`'s own mechanical shape rather than a second implementation of it, +factored into a shared `cancel_and_settle_run` helper parameterized on the target `RunState`: +`run_cancel` itself still asks for the literal `"cancelled"`, checked against the requesting +principal; the grace-window path asks for `RunState::unrendered_verdict()` and passes no principal, +bypassing the ownership check the same way the removed backstop did. `unrendered_verdict()` is the +one function that owns the word "cancelled" for an evidence-free settlement, so this path and the +process-exit-after-settled-turn path that already called it can never drift on what word they use. +Before settling, a `WorkerTimeout { kind: LeaderGone }` fact is journaled first — the same +report-before-decide shape the inactivity sweep already uses for `Inactivity`/`Total` — so the +journal always states the reason before the state changes because of it. + +Removed entirely: the inactivity sweep's `settle_abandoned_turn` path (ADR-0027 wave 3). A parked run +whose leader is still connected is never settled by the runtime now, however long it stays silent — +the inactivity fact is still journaled, for `run/timeoutAck` to react to, but it no longer drives a +state change on its own. `lost` returns to ADR-0023's original, narrower meaning: the evidence itself +is missing or ambiguous, never "the leader went quiet." + +**Daemon restart.** At startup, nothing has connected to the freshly constructed `LeaderRegistry` +yet, so every leader that owns a non-terminal run is — from this process's own perspective — +disconnected as of right now. Each such owner is seeded with `zero_since` set to the daemon's own +startup instant, and the same grace-window timer is armed for it as a live disconnect would get. A +leader that reconnects within the window clears it exactly as an ordinary reconnect does; one that +never returns gets its runs settled through the identical path a live disconnect uses. The clock +deliberately starts at the daemon's *own* startup instant, never at a run's last-seen activity: a +leader cannot reconnect to a daemon that is not running, so any time the daemon spent stopped is the +daemon's own absence, not the leader's, and must never be charged against the window — doing +otherwise would settle live, reconnecting leaders' runs on every restart that followed a stop longer +than the window, which is exactly the destructive direction this feature exists to avoid. + +**Shared-instance-id hazard.** The `"crew-extension"` fallback id — used by any leader session with +no real session id of its own — means more than one physical leader process can share a single +registry entry. Refcounting keeps the *mechanism* safe under that sharing: both connect/disconnect +orderings resolve correctly by construction (plain addition does not care which side's `+1`/`-1` is +observed first), and the entry only reads "gone" once every sharer has disconnected. It does not +resolve the pre-existing ownership ambiguity the shared id already carries elsewhere in this +protocol — which of several processes sharing the id actually "owns" a given run — and this decision +inherits that ambiguity rather than creating or fixing it. + +### Positive Consequences + +* A run is never settled on a worker's silence alone; only a leader's connection actually being gone, + for the full grace window, can do that — closing exactly the gap the maintainer's ruling named. +* An abandoned leader's runs are no longer stranded forever: they settle within one grace window of + the leader's last connection actually going away, restart included. +* The settle path shares its mechanical shape (and its ordering/broadcast guarantees) with + `run_cancel`, so a future change to that shape cannot fix one call site and silently miss the + other. +* `lost` regains ADR-0023's original single meaning, removing the two-meanings-under-one-flag + ambiguity ADR-0027 wave 3 introduced for it. + +### Negative Consequences + +* A leader whose connection drops for reasons other than actually leaving (a long GC pause, a flaky + local socket) and does not reconnect within 60 seconds has its parked runs settled the same as a + leader that genuinely gave up — the grace window trades a false negative (settling too early) for + eliminating the false positive (settling on silence) the backstop had; it does not eliminate every + failure mode. +* The in-memory registry adds real state to reason about across a restart, which is why the + restart-seeding half of this decision exists at all; a bug in that seeding has the same destructive + shape (settling live runs) as the failure mode this whole decision exists to prevent. +* The shared-instance-id ownership ambiguity named above is now load-bearing for a second mechanism + (grace-window teardown) beyond whatever already depended on it; a future fix to that ambiguity must + account for this decision's use of it too. + +## Links + +* Amends [0027](0027-turn-end-settles-a-run.md), whose Decision point 3 is split by this ADR: the + backstop-settles-on-silence half is superseded here; the "never `succeeded`" half is inherited + unamended (a grace-window settlement is `RunState::unrendered_verdict()`, never `succeeded`, + exactly as the backstop it replaces never was). +* Related: [ADR-0023](0023-run-state-edges-from-adapter-evidence.md) (`lost`'s original, narrower + meaning, restored here), [ADR-0016](0016-coordination-scope-tokens-bound-to-run-and-pid-ancestry.md) + (the same "reuse a settle mechanism rather than parallel it" reasoning, applied there to + coordination scope and here to run teardown), [ADR-0020](0020-per-mutation-event-broadcast-is-not-optional.md) + (the broadcast-in-the-same-call invariant this decision's teardown path inherits from + `cancel_and_settle_run`). +* Implementation: `crates/runtime/src/ipc/leader_registry.rs` (the registry and its drop guard), + `crates/runtime/src/service/orchestration.rs` (`settle_leader_gone`, + `seed_leader_registry_after_restart`, `cancel_and_settle_run`), `crates/runtime/src/timeout_sweep.rs` + (the backstop's removal). diff --git a/docs/adr/README.md b/docs/adr/README.md index 06e51be3..aeb0029f 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -69,7 +69,7 @@ rather than behind a number nobody can resolve. | [0024](0024-project-scoped-reads-are-open-ownership-gates-writes.md) | Project-scoped reads are open; ownership gates writes | Accepted | | [0025](0025-crew-v2-tui-control-plane.md) | Crew v2 TUI control plane | Accepted | | [0026](0026-headless-retirement.md) | Headless control plane retirement | Accepted | -| [0027](0027-turn-end-settles-a-run.md) | A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state | Accepted | +| [0027](0027-turn-end-settles-a-run.md) | A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state | Accepted, Decision point 3 partially superseded by [0036](0036-leader-disconnect-grace-window.md) | | [0028](0028-submit-prompt-is-journaled-redacted-run-intent.md) | The submit prompt is journaled, redacted, as durable run intent | Accepted | | [0029](0029-placement-follows-the-backend-embedded-deleted.md) | Display placement is the backend's natural form, and `Embedded` is deleted rather than deprecated | Accepted | | [0030](0030-paste-delivery-bounded-on-progress.md) | Paste delivery is bounded on progress, not on elapsed time | Accepted | @@ -77,6 +77,7 @@ rather than behind a number nobody can resolve. | [0032](0032-model-names-are-resolved-and-persisted-on-confirmation.md) | A worker's model name is resolved against the host's catalogue, and persisted only once confirmed | Accepted | | [0034](0034-dashboard-bearer-token-over-loopback.md) | A daemon-hosted web dashboard, authenticated by a per-run bearer token over loopback | Accepted | | [0035](0035-protocol-doc-comment-sigil-split.md) | Protocol doc comments split by sigil: `///` ships, `//` stays internal | Accepted | +| [0036](0036-leader-disconnect-grace-window.md) | A parked run is settled on its leader's connection actually being gone, never on silence alone | Accepted | ## When to add one diff --git a/docs/architecture.md b/docs/architecture.md index 0347360b..cd9738ac 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -766,9 +766,12 @@ sequenceDiagram A TUI vendor never exits, so `ProcessExited` alone would leave such a run non-terminal forever. `TurnEnded` is the vendor's own end-of-turn boundary, journaled as durable evidence and driving a **non-terminal** edge: the boundary says the *turn* ended, never that the *task* succeeded, so - only the leader closes a run (`run/finish`), with an inactivity backstop settling an abandoned one - to `lost`. A run parked this way carries the `turnSettled` flag, which is what lets a snapshot - reader tell "the answer is ready" from "the worker asked a question" — both are `waitingUser`. + only the leader closes a run (`run/finish`) — or, if the leader's own connection is gone for the + full disconnect grace window, the runtime settles it via `RunState::unrendered_verdict()` + ([ADR-0036](adr/0036-leader-disconnect-grace-window.md)); silence alone, with the leader still + connected, never settles a parked run. A run parked this way carries the `turnSettled` flag, which + is what lets a snapshot reader tell "the answer is ready" from "the worker asked a question" — both + are `waitingUser`. Two ordering properties hold. The state edge is committed durably before the signal it triggers, as it always was. And the concurrency slot is released on the **first of a turn boundary or a diff --git a/docs/future-features.md b/docs/future-features.md index 3c6c2fb0..997b98c4 100644 --- a/docs/future-features.md +++ b/docs/future-features.md @@ -258,8 +258,9 @@ fragile about input, opaque about completion, and unverifiable about delivery. E answer in the TUI path itself: prompts are delivered as one bracketed paste in paced chunks with a bounded write, so a vendor that stops reading fails loudly instead of truncating silently; a run reaches a settled state from the vendor's own end-of-turn boundary rather than only from process -exit, with `run/finish` for the leader and an inactivity backstop for an abandoned turn -([ADR-0027](adr/0027-turn-end-settles-a-run.md)); and post-submit verification compares what the +exit, with `run/finish` for the leader and a disconnect-grace-window teardown for a leader that is +actually gone, never merely quiet ([ADR-0027](adr/0027-turn-end-settles-a-run.md), +[ADR-0036](adr/0036-leader-disconnect-grace-window.md)); and post-submit verification compares what the vendor actually recorded against what was sent, so a truncating composer fails the run rather than producing a plausible-looking short answer. A headless plane would have to beat that, not merely match a PTY's old weaknesses. diff --git a/packages/protocol-ts/schema/crew.schema.json b/packages/protocol-ts/schema/crew.schema.json index b8dde0df..c850dfc4 100644 --- a/packages/protocol-ts/schema/crew.schema.json +++ b/packages/protocol-ts/schema/crew.schema.json @@ -3060,10 +3060,19 @@ }, "TimeoutKind": { "description": "Which liveness deadline a `workerTimeout` event reports.", - "type": "string", - "enum": [ - "inactivity", - "total" + "oneOf": [ + { + "type": "string", + "enum": [ + "inactivity", + "total" + ] + }, + { + "description": "The run's owning leader has held no live connection for the full\ndisconnect grace window. Distinct from `inactivity`/`total`,\nwhich describe the worker's own liveness -- this describes the\nleader's, and is the fact a grace-window teardown journals before\nthe run is settled.", + "type": "string", + "const": "leaderGone" + } ] }, "DisplayConfig": { diff --git a/packages/protocol-ts/src/generated/TimeoutKind.ts b/packages/protocol-ts/src/generated/TimeoutKind.ts index dec6231c..1b711b82 100644 --- a/packages/protocol-ts/src/generated/TimeoutKind.ts +++ b/packages/protocol-ts/src/generated/TimeoutKind.ts @@ -3,4 +3,4 @@ /** * Which liveness deadline a `workerTimeout` event reports. */ -export type TimeoutKind = "inactivity" | "total"; +export type TimeoutKind = "inactivity" | "total" | "leaderGone"; From 29f0ed4e2b0a65f2fe3dab26bc31444d8f42a0b2 Mon Sep 17 00:00:00 2001 From: Nikolas Demiridis Date: Thu, 10 Sep 2026 12:50:47 +0300 Subject: [PATCH 2/3] fix(runtime): re-check a leader's connection state before each run's settle, not only once settle_leader_gone took its caller's `gone_for_at_least` check as final: the actual decision happened before the DB query for the leader's owned runs, and was never revisited across the loop that settles each one. For a leader with several parked runs, that loop is a query plus two DB round trips per run -- not instantaneous. A leader that reconnects inside that window had every one of its runs settled anyway, after it was demonstrably back. "Abandoned" means the leader is gone now, having been gone the full window -- not merely that it was gone for the window at some earlier instant. A leader that has reconnected is not abandoned, and settling its work serves no purpose this feature exists for. settle_leader_gone now takes the LeaderRegistry and re-checks gone_for_at_least immediately before each run's own settle, inside the loop. The caller's own check stays as the cheap early-out that skips the query entirely in the common case; the new check is what actually protects a leader that comes back mid-loop, run by run, rather than only at the loop's start. Both call sites (a live disconnect's own timer, the daemon-restart seed) pass their registry through. Doc comment rewritten to state the rule instead of describing the old, narrower behavior. Proven with a test using a driver whose first cancel_run call blocks until released, rather than a guessed sleep: the reconnect is registered the instant the first run's settle is confirmed committed (its cancel_run has started), landing deterministically between the two runs' settles. The run reached first is still settled; the run reached after the reconnect is left untouched. Two more hardening changes from the same review: gone_for_at_least on an instance id with no registry entry at all now answers false, not true. Both real callers (a live connection's register, the restart seed's seed_disconnected_since) always create the entry before ever asking this question, so the branch is unreachable from either path today -- but the destructive answer must never be the default for a question nobody has established one for. Logs a warning when it happens; the corresponding test now asserts false and states why the branch is a guard, not a path. The LeaderGone arm in the inactivity sweep's since_ms computation, and in ActivityClock::mark_journaled, no longer panics on an invariant that should hold but runs inside a long-lived background task rather than a test. Both log an error and continue (the sweep skips journaling that one kind for that tick; the clock skips setting a flag), so a future change that violates the invariant in error degrades to a missed fact rather than taking the sweep down. --- crates/runtime/src/adapter/activity.rs | 16 +- crates/runtime/src/ipc/connection.rs | 2 +- crates/runtime/src/ipc/leader_registry.rs | 35 +++- crates/runtime/src/service/orchestration.rs | 213 ++++++++++++++++++-- crates/runtime/src/timeout_sweep.rs | 21 +- 5 files changed, 256 insertions(+), 31 deletions(-) diff --git a/crates/runtime/src/adapter/activity.rs b/crates/runtime/src/adapter/activity.rs index 882d22b2..3f1a38d4 100644 --- a/crates/runtime/src/adapter/activity.rs +++ b/crates/runtime/src/adapter/activity.rs @@ -133,9 +133,19 @@ impl ActivityClock { // This clock tracks worker/PTY liveness only -- a // leader's connection state is `ipc::leader_registry`'s // concern, tracked separately and never through here. - TimeoutKind::LeaderGone => unreachable!( - "ActivityClock has no leader-connection state; every real caller's `kind` comes from `due_timeouts`, which never produces LeaderGone" - ), + // Should be unreachable on that invariant (the sweep's + // own `since_ms` computation already skips this kind + // before ever calling here), but this is a `pub fn` a + // future caller could reach directly with a bad `kind` -- + // logging and doing nothing costs a call that flags + // nothing, never a panic in a long-lived background task. + TimeoutKind::LeaderGone => { + tracing::error!( + run_id = %run_id, + "mark_journaled called with LeaderGone -- ActivityClock has no \ + leader-connection state to flag; ignoring" + ); + } } } } diff --git a/crates/runtime/src/ipc/connection.rs b/crates/runtime/src/ipc/connection.rs index 6f6cc9c5..502ae55d 100644 --- a/crates/runtime/src/ipc/connection.rs +++ b/crates/runtime/src/ipc/connection.rs @@ -119,7 +119,7 @@ pub(crate) async fn handle(stream: UnixStream, ctx: ConnContext, shared: Arc= grace) } - // No entry at all is the same fact as "gone": nothing has - // ever registered a live connection for this id. - None => true, + // No entry at all means nobody has ever registered OR seeded + // this id -- an unestablished question, not evidence of + // absence. Both real callers (`register`'s connection path, + // `seed_disconnected_since`'s restart path) always create the + // entry before ever asking this question, so this branch is a + // guard against a future caller's misuse, never a path either + // takes today. The destructive answer (`true`, which a caller + // reads as license to settle) must never be the default for a + // question nobody has established an answer to -- `false` + // costs nothing on the paths that exist, and refuses to + // fabricate an instant, grace-free settlement for one that + // doesn't. + None => { + tracing::warn!( + instance_id, + "gone_for_at_least asked about an instance id with no registry entry at \ + all -- neither a live connection nor a restart seed ever created one; \ + answering false rather than settling on an unestablished question" + ); + false + } } } } @@ -320,10 +338,17 @@ mod tests { ); } + /// Both real callers (a live connection's `register`, restart's + /// `seed_disconnected_since`) always create the entry before ever + /// asking `gone_for_at_least` -- so an id with no entry at all is + /// only reachable by a future caller's misuse, never by either path + /// that exists today. `false` is the answer that refuses to settle + /// on that unestablished question, rather than fabricating an + /// instant, grace-free "gone" for it. #[test] - fn an_unregistered_instance_id_reads_as_gone() { + fn an_unregistered_instance_id_never_reads_as_gone() { let registry = LeaderRegistry::new(); - assert!(registry.gone_for_at_least("never-seen", GRACE)); + assert!(!registry.gone_for_at_least("never-seen", GRACE)); } #[test] diff --git a/crates/runtime/src/service/orchestration.rs b/crates/runtime/src/service/orchestration.rs index 64bb40d9..6a2c7766 100644 --- a/crates/runtime/src/service/orchestration.rs +++ b/crates/runtime/src/service/orchestration.rs @@ -1602,23 +1602,33 @@ impl OrchestrationService { Ok(result) } - /// Settles every non-terminal run owned by `instance_id`, once its - /// grace-window timer (`ipc::leader_registry`) has confirmed it has - /// held no live connection for the full window. Called from exactly - /// two places, both re-checking `LeaderRegistry::gone_for_at_least` - /// themselves right before this runs: a live disconnect's own timer - /// (`ipc::connection::handle`'s `on_gone` callback), and the - /// daemon-startup seed for runs whose owner was never reconnected + /// Settles every non-terminal run owned by `instance_id` that is + /// STILL abandoned, once its grace-window timer (`ipc::leader_registry`) + /// has confirmed it held no live connection for the full window. + /// Called from exactly two places, both making that same initial + /// check themselves right before calling this: a live disconnect's + /// own timer (`ipc::connection::handle`'s `on_gone` callback), and + /// the daemon-startup seed for runs whose owner was never reconnected /// after a restart ([`Self::seed_leader_registry_after_restart`]). /// - /// `grace` is only for the journaled fact's `since_ms` -- the actual - /// decision already happened in the caller's `gone_for_at_least` - /// check; this never re-verifies it, and takes the caller's word for - /// how long the window it waited out actually was (the configured - /// value, which a test may have shrunk well below the production + /// "Abandoned" means gone NOW, having been gone the full window -- + /// never merely "was gone for the window at some earlier instant". + /// The caller's own check is a cheap early-out (skips the query below + /// entirely in the common case: nothing owned, or the leader never + /// reconnects) but is not enough on its own -- the query plus two DB + /// round trips per owned run is not instantaneous, and a leader with + /// several parked runs can reconnect partway through this loop. So + /// `leader_registry.gone_for_at_least` is re-checked immediately + /// before EACH run's own settle, inside the loop below, not only + /// once at entry: a leader that comes back mid-loop leaves every run + /// from that point on untouched, exactly as if it had never been + /// gone. `grace` is also used for the journaled fact's `since_ms`, + /// taking the caller's word for how long the window it waited out + /// actually was (the configured value, which a test may have shrunk + /// well below the production /// [`crate::ipc::leader_registry::LEADER_DISCONNECT_GRACE_WINDOW`]). /// - /// For each owned run: journals a `WorkerTimeout { kind: LeaderGone }` + /// For each owned run still confirmed abandoned: journals a `WorkerTimeout { kind: LeaderGone }` /// fact FIRST (the same "report before deciding" shape the inactivity /// sweep uses for `Inactivity`/`Total`), then transitions to /// `RunState::unrendered_verdict()` -- never a literal `"cancelled"`, @@ -1628,7 +1638,12 @@ impl OrchestrationService { /// silently skipped: `transition_run`'s own guarded write refuses an /// illegal edge, and that refusal is exactly correct here, not an /// error to surface. - pub(crate) async fn settle_leader_gone(&self, instance_id: &str, grace: Duration) { + pub(crate) async fn settle_leader_gone( + &self, + instance_id: &str, + grace: Duration, + leader_registry: &crate::ipc::leader_registry::LeaderRegistry, + ) { let project_id = self.project_id; let run_ids: Vec = match self .db @@ -1658,6 +1673,27 @@ impl OrchestrationService { continue; }; + // Re-checked immediately before THIS run's settle, not only + // once by the caller before the query above: the query plus + // two DB round trips per run is not instantaneous, and a + // leader with several parked runs can reconnect partway + // through this loop. "Abandoned" means gone NOW, having been + // gone the full window -- never merely "was gone for the + // window at some earlier instant" the caller happened to + // observe. The caller's own check stays, as the cheap + // early-out that skips this query entirely in the common + // case (nothing owned, or the leader never reconnects); this + // one is what actually protects a leader that comes back + // mid-loop, run by run, rather than only at the start of it. + if !leader_registry.gone_for_at_least(instance_id, grace) { + tracing::info!( + instance_id, + run_id = %run_id, + "leader reconnected mid-settle; leaving this run untouched" + ); + continue; + } + let mut timeout_fact = match self .db .run_domain_op(Box::new(move |conn| { @@ -1762,7 +1798,9 @@ impl OrchestrationService { tokio::spawn(async move { tokio::time::sleep(grace).await; if leader_registry.gone_for_at_least(&owner, grace) { - orchestration.settle_leader_gone(&owner, grace).await; + orchestration + .settle_leader_gone(&owner, grace, &leader_registry) + .await; } }); } @@ -3709,6 +3747,7 @@ mod restart_seeding_tests { db_path: &std::path::Path, project_id: ProjectId, events_tx: broadcast::Sender, + run_driver: Option>, ) -> Arc { let db = Arc::new(DatabaseHandle::start(db_path).await.expect("db starts")); let violation = Arc::new(crate::policy::ViolationService::new( @@ -3725,7 +3764,7 @@ mod restart_seeding_tests { Arc::new(OrchestrationService::new( db, project_id, - None, + run_driver, Arc::new(crate::approval::NoopApprovalCallback), violation, events_tx, @@ -3808,7 +3847,7 @@ mod restart_seeding_tests { let project_id = ProjectId::new(); let (events_tx, mut events_rx) = broadcast::channel(16); let orchestration = - build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx).await; + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx, None).await; let run_id = seed_nonterminal_run(&orchestration, "leader-gone").await; let leader_registry = LeaderRegistry::new(); @@ -3850,7 +3889,7 @@ mod restart_seeding_tests { let project_id = ProjectId::new(); let (events_tx, mut events_rx) = broadcast::channel(16); let orchestration = - build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx).await; + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx, None).await; let run_id = seed_nonterminal_run(&orchestration, "leader-returns").await; let leader_registry = LeaderRegistry::new(); @@ -3877,4 +3916,142 @@ mod restart_seeding_tests { "a reconnect within the window must journal nothing" ); } + + /// A [`RunDriver`] whose `cancel_run` blocks on its FIRST call until + /// the test releases it, notifying the test the instant that call + /// starts. Gives a test deterministic control over exactly when + /// `settle_leader_gone`'s loop has finished processing one run -- + /// `cancel_and_settle_run` already commits the run's own + /// `transition_run` (and broadcasts it) *before* calling + /// `cancel_run`, so "the first `cancel_run` call has started" is + /// proof the first run is already settled in the database, without + /// guessing at timing. + struct BlockingOnFirstCancelDriver { + first_call_started: Arc, + release_first_call: Arc, + calls: std::sync::atomic::AtomicUsize, + } + + impl RunDriver for BlockingOnFirstCancelDriver { + fn active_run_count(&self) -> usize { + 0 + } + + fn start( + &self, + _ctx: crate::service::RunDriverContext, + ) -> crate::service::AdapterFuture<'static, Result<(), String>> { + Box::pin(async { Ok(()) }) + } + + fn send_follow_up( + &self, + _run_id: RunId, + _task_id: TaskId, + _worker_id: WorkerId, + _prompt: String, + _kind: crew_protocol::MessageKind, + ) -> crate::service::AdapterFuture<'static, Result<(), String>> { + Box::pin(async { Ok(()) }) + } + + fn running_adapter(&self, _run_id: RunId) -> Option> { + None + } + + fn cancel_run( + &self, + _run_id: RunId, + _scope: CancelScope, + ) -> crate::service::AdapterFuture<'static, Result> + { + let first_call_started = Arc::clone(&self.first_call_started); + let release_first_call = Arc::clone(&self.release_first_call); + let is_first = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0; + Box::pin(async move { + if is_first { + first_call_started.notify_one(); + release_first_call.notified().await; + } + Ok(crate::service::CancelOutcome::Cancelled) + }) + } + } + + /// The re-check staff's review named directly: `settle_leader_gone`'s + /// caller-side check (before the DB query for owned runs) is not + /// enough on its own for a leader with more than one parked run, + /// because the query plus per-run DB round trips are not + /// instantaneous. This proves the loop's OWN re-check (immediately + /// before each run's settle) actually stops the run reached AFTER a + /// reconnect, using a blocking driver rather than a guessed sleep to + /// land the reconnect deterministically between the two runs' + /// settles -- never racing on timing. + #[tokio::test] + async fn a_leader_reconnecting_mid_settle_leaves_the_later_run_untouched() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, _events_rx) = broadcast::channel(16); + let driver = Arc::new(BlockingOnFirstCancelDriver { + first_call_started: Arc::new(tokio::sync::Notify::new()), + release_first_call: Arc::new(tokio::sync::Notify::new()), + calls: std::sync::atomic::AtomicUsize::new(0), + }); + let orchestration = build_orchestration( + &dir.path().join("db.sqlite"), + project_id, + events_tx, + Some(Arc::clone(&driver) as Arc), + ) + .await; + + let first_run = seed_nonterminal_run(&orchestration, "leader-mid").await; + let second_run = seed_nonterminal_run(&orchestration, "leader-mid").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(20); + // Already gone for the full window from the start -- the + // caller-side check this loop's own re-check exists alongside. + leader_registry.seed_disconnected_since("leader-mid".to_string(), Instant::now() - grace); + + let orch = Arc::clone(&orchestration); + let registry_for_task = leader_registry.clone(); + let settle_task = tokio::spawn(async move { + orch.settle_leader_gone("leader-mid", grace, ®istry_for_task) + .await; + }); + + // Waits for proof the first run is already settled (its + // `cancel_run` call has started, which only happens after its + // own `transition_run` commit), not a guessed duration. + driver.first_call_started.notified().await; + + // The leader reconnects here -- strictly between the first run's + // settle (already committed) and the second's (not yet reached: + // the loop is still blocked inside the first run's `cancel_run`). + let _reconnected = leader_registry.register("leader-mid".to_string(), |_| {}); + driver.release_first_call.notify_one(); + + settle_task.await.expect("settle_leader_gone task"); + + // `owned_nonterminal_run_ids_op` carries no `ORDER BY`, so which + // of the two rows the loop reaches first is not a guarantee this + // test should depend on -- only that exactly one of them was + // already settled when the reconnect landed (the one whose + // `cancel_run` blocked the loop) and the other was reached only + // after, and so must survive. + let first_state = run_state(&orchestration, first_run).await; + let second_state = run_state(&orchestration, second_run).await; + let states = [first_state.as_str(), second_state.as_str()]; + assert_eq!( + states.iter().filter(|s| **s == "cancelled").count(), + 1, + "exactly the run reached before the reconnect must be settled: {states:?}" + ); + assert_eq!( + states.iter().filter(|s| **s == "queued").count(), + 1, + "exactly the run reached after the reconnect must be left untouched: {states:?}" + ); + } } diff --git a/crates/runtime/src/timeout_sweep.rs b/crates/runtime/src/timeout_sweep.rs index b5a639d1..d8b749b7 100644 --- a/crates/runtime/src/timeout_sweep.rs +++ b/crates/runtime/src/timeout_sweep.rs @@ -67,10 +67,23 @@ pub async fn sweep_once( // a leader's connection state at all. `LeaderGone` is // journaled from the leader-grace-window teardown // directly (`service::orchestration::settle_leader_gone`), - // never through this worker-liveness sweep. - TimeoutKind::LeaderGone => unreachable!( - "due_timeouts never produces LeaderGone -- it has no leader-connection state to observe" - ), + // never through this worker-liveness sweep. This arm + // should be unreachable on that invariant, but this loop + // runs inside a long-lived background task, not a test -- + // panicking here would silently take the whole sweep down + // rather than fail loudly where someone would notice. Log + // and skip this one kind instead, so a future producer + // added here in error degrades to a missed fact, not a + // dead daemon. + TimeoutKind::LeaderGone => { + tracing::error!( + run_id = %run_id, + "due_timeouts produced LeaderGone, which it should never be able to \ + observe -- skipping this kind rather than journaling a nonsensical \ + since_ms" + ); + continue; + } }; let outcome = db .run_domain_op(Box::new({ From 38cd5159de3b23a91d733576eb29fe34400fa768 Mon Sep 17 00:00:00 2001 From: "crew-bot[bot]" <322634903+crew-bot[bot]@users.noreply.github.com> Date: Thu, 10 Sep 2026 09:59:34 +0000 Subject: [PATCH 3/3] chore(extension): refresh bundle from linux-x64 CI Build: https://github.com/nikolasd/crew/actions/runs/34463623060 Source: 29f0ed4e2b0a65f2fe3dab26bc31444d8f42a0b2 --- packages/extension/dist/index.js | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/packages/extension/dist/index.js b/packages/extension/dist/index.js index c71c5d8a..d37d71c7 100644 --- a/packages/extension/dist/index.js +++ b/packages/extension/dist/index.js @@ -10179,10 +10179,19 @@ own (see that event's doc comment).`, }, TimeoutKind: { description: "Which liveness deadline a `workerTimeout` event reports.", - type: "string", - enum: [ - "inactivity", - "total" + oneOf: [ + { + type: "string", + enum: [ + "inactivity", + "total" + ] + }, + { + description: "The run's owning leader has held no live connection for the full\ndisconnect grace window. Distinct from `inactivity`/`total`,\nwhich describe the worker's own liveness -- this describes the\nleader's, and is the fact a grace-window teardown journals before\nthe run is settled.", + type: "string", + const: "leaderGone" + } ] }, DisplayConfig: {