From 63f37e0e6d707c1af683359ddeb96f2bfc1da2f1 Mon Sep 17 00:00:00 2001 From: Dan Date: Tue, 26 May 2026 21:25:10 -0400 Subject: [PATCH 1/3] docs(dispatch,control): document concurrency invariants from F-CONC audit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three audit findings (#494 F-CONC-4, #498 F-CONC-8, #490 F-CONC-11) were re-validated against current code and confirmed as correct-by- construction; what was missing was an in-source explanation of why. This commit captures those invariants where a future refactor would otherwise reintroduce the patterns the audit flagged. - `cancel_and_find_parent` (signals.rs): document the `subleads` → `worker_layer_index` → `workers.cancels` lock-acquisition order as a project invariant, naming every site that currently respects it (#494). - `register_sublead` (state.rs): explain why insert → install_watcher → cascade_to is load-bearing — the proposed reorder from the audit would have CAUSED a real bug because `install_cascade_cancel_watcher` is fire-once on root cancel. The eager cascade at step 3 is what closes the post-watcher-fired window for late-registered subleads (#498). - `control/server.rs` broadcast site: annotate the op-reply broadcast as intentional, complementing the existing PR-Q routing block. The audit re-validation confirmed `OpUnknownState.current_state` carries only the coarse enum already published via `WorkerStatus`, so the "subscriber visibility" surface is not a leak (#490). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/pitboss-cli/src/control/server.rs | 8 +++++++ crates/pitboss-cli/src/dispatch/signals.rs | 9 ++++++++ crates/pitboss-cli/src/dispatch/state.rs | 26 ++++++++++++++++++++++ 3 files changed, 43 insertions(+) diff --git a/crates/pitboss-cli/src/control/server.rs b/crates/pitboss-cli/src/control/server.rs index b7d24e2..1eb6506 100644 --- a/crates/pitboss-cli/src/control/server.rs +++ b/crates/pitboss-cli/src/control/server.rs @@ -654,6 +654,14 @@ async fn serve_connection( } } else { let reply = dispatch_op(&state, op).await; + // F-CONC-11 #490: broadcasting op replies to every + // subscriber (including the SPA's SSE bridge) is + // intentional — see the PR-Q routing block above for + // the design rationale and the `OpUnknownState` / + // current-state-leak analysis. Audit re-validation + // confirmed `current_state` is the same coarse enum + // already published via `WorkerStatus`, so the + // "subscriber visibility" surface is not a leak. state .root .broadcast_control_event(EventEnvelope { diff --git a/crates/pitboss-cli/src/dispatch/signals.rs b/crates/pitboss-cli/src/dispatch/signals.rs index ef7fb73..02c193e 100644 --- a/crates/pitboss-cli/src/dispatch/signals.rs +++ b/crates/pitboss-cli/src/dispatch/signals.rs @@ -172,6 +172,15 @@ pub async fn cancel_actor_with_reason( /// under a held `state.subleads` read lock — O(N) per cancel call, /// blocking new sublead registration. The `worker_layer_index` lookup /// mirrors `resolve_layer_for_caller`'s routing for KV ops (#150 audit). +/// +/// **Lock-acquisition order (project invariant):** `state.subleads` → +/// `state.worker_layer_index` → `sub_layer.workers.cancels`. Every +/// callsite that holds any pair of these locks acquires them in this +/// order. Inverting it under a `subleads.write()` would deadlock; today +/// no such path exists because every `subleads.write()` callsite uses a +/// single-statement guard (`register_sublead`, `reconcile_terminated_sublead`). +/// Mirror of the convention documented in `hierarchical.rs::await_workers_drained`. +/// (F-CONC-4 #494) async fn cancel_and_find_parent( state: &Arc, target: &str, diff --git a/crates/pitboss-cli/src/dispatch/state.rs b/crates/pitboss-cli/src/dispatch/state.rs index 269c671..f4394e0 100644 --- a/crates/pitboss-cli/src/dispatch/state.rs +++ b/crates/pitboss-cli/src/dispatch/state.rs @@ -566,6 +566,32 @@ impl DispatchState { /// `dispatch/sublead.rs:340-383` so the watcher install + eager /// cascade can never drift out of order. Pinned by the integration /// tests in `tests/cancel_cascade_flows.rs`. + /// + /// **Step order is load-bearing — do not reorder (F-CONC-8 #498):** + /// + /// - `insert` MUST happen before `cascade_to`. The cascade-watcher + /// installed by `install_cascade_cancel_watcher` walks + /// `cascade_to_subleads`, which reads `self.subleads`; if the + /// sublead were not yet inserted, a concurrent root-cancel could + /// miss this sub-layer and the eager `cascade_to` at step 3 would + /// be the only signal that ever reaches it. + /// - `install_sublead_cancel_watcher` MUST happen before the eager + /// `cascade_to`. The per-sub-layer watcher is fire-once on the + /// sub-layer's own `CancelToken`; if `cascade_to` runs first and + /// already toggles drain/terminate, the watcher started afterwards + /// still observes the canceled state and fans out to workers as + /// intended (the watcher's `select!` over `await_drain`/`await_terminate` + /// resolves immediately for an already-canceled token). The + /// ordering exists so the watcher is *registered* on the canonical + /// pre-cancel state for the (more common) non-canceled path. + /// + /// Between steps 1 and 3 there is a benign window where a concurrent + /// root cancellation's `cascade_to_subleads` walk could fire and + /// no-op (root not yet canceled when walked) — the eager `cascade_to` + /// at step 3 closes this window. The sublead subprocess is not + /// launched until after `register_sublead` returns + /// (`dispatch/sublead.rs::spawn_sublead_session`), so no worker + /// observes the transient state. pub async fn register_sublead(&self, sublead_id: String, sub_layer: Arc) { self.subleads .write() From 8c58ea2901b8daec5eb0205e5b51f9d332ee3a65 Mon Sep 17 00:00:00 2001 From: Dan Date: Tue, 26 May 2026 21:25:35 -0400 Subject: [PATCH 2/3] feat(dispatch): opt-in drain-mode auto-escalation timeout Closes the F-CONC-7 (#497) operator-surprise gap: after a single Ctrl-C the watcher would drain and then loop forever waiting for a second Ctrl-C. If an operator walked away and a worker hung with no timeout, the run would never wind down. Adds `PITBOSS_CTRLC_DRAIN_ESCALATE_SECS=` as an opt-in env var. When set to a positive integer, the watcher races a sleep timer against a follow-up Ctrl-C after the 5-second second-SIGINT window expires; whichever fires first escalates to terminate via `TerminateReason::OperatorCtrlC` (matching the existing double-Ctrl-C path). Default behaviour is preserved: with the env var unset, empty, zero, or unparseable, the watcher continues to drain forever (the pre-fix contract). Operators who want auto-cleanup can opt in. A `#[serial(env)]` unit test pins the env-parsing contract (none for unset/empty/zero/unparseable, Some(Duration) for positive integers with surrounding whitespace tolerated). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/pitboss-cli/src/dispatch/signals.rs | 97 +++++++++++++++++++++- 1 file changed, 95 insertions(+), 2 deletions(-) diff --git a/crates/pitboss-cli/src/dispatch/signals.rs b/crates/pitboss-cli/src/dispatch/signals.rs index 02c193e..2d447cc 100644 --- a/crates/pitboss-cli/src/dispatch/signals.rs +++ b/crates/pitboss-cli/src/dispatch/signals.rs @@ -68,6 +68,12 @@ fn send_group_signal(pid: u32, sig: libc::c_int, name: &'static str) -> Result<( /// 2nd SIGINT within window → terminate /// After the window, re-armed: a single later SIGINT is treated as a fresh first. /// +/// **Drain-mode auto-escalation (opt-in, F-CONC-7 #497):** set the env +/// var `PITBOSS_CTRLC_DRAIN_ESCALATE_SECS=` (positive integer seconds) +/// to make the watcher escalate to terminate after `n` seconds of drain +/// without a second Ctrl-C. Default behaviour (env unset / unparseable / +/// zero) is to remain in drain mode indefinitely. +/// /// **Idempotent.** Calling more than once in the same process is a no-op /// after the first install — pitboss has two callsites today (flat-mode /// and hierarchical entry points) and a future third callsite would @@ -102,14 +108,61 @@ pub fn install_ctrl_c_watcher(cancel: CancelToken) { return; } _ => { - tracing::info!("drain window expired; continuing in drain mode"); - // Loop again: if another Ctrl-C arrives later, start a new window. + // F-CONC-7 #497: optionally auto-escalate to terminate + // after a configurable drain-mode timeout, so operators + // who walk away after a single Ctrl-C don't leave the + // run hanging behind a stuck worker. Opt-in via env var + // to preserve the "drain forever" default behaviour. + if let Some(escalate_after) = drain_escalate_timeout() { + tracing::warn!( + "drain window expired; auto-escalating to terminate in {:?} \ + (PITBOSS_CTRLC_DRAIN_ESCALATE_SECS) unless another Ctrl-C arrives", + escalate_after, + ); + tokio::select! { + () = tokio::time::sleep(escalate_after) => { + cancel.terminate_with_reason( + pitboss_core::store::TerminateReason::OperatorCtrlC, + ); + tracing::warn!( + "drain-mode timeout reached — terminating subprocesses", + ); + return; + } + res = tokio::signal::ctrl_c() => { + if res.is_ok() { + cancel.terminate_with_reason( + pitboss_core::store::TerminateReason::OperatorCtrlC, + ); + tracing::warn!( + "received Ctrl-C during drain-mode timeout — terminating", + ); + } + return; + } + } + } else { + tracing::info!("drain window expired; continuing in drain mode"); + // Loop again: if another Ctrl-C arrives later, start a new window. + } } } } }); } +/// Read the optional drain-mode auto-escalation timeout from +/// `PITBOSS_CTRLC_DRAIN_ESCALATE_SECS`. Returns `None` if the env var is +/// unset, empty, or unparseable as a positive integer — keeping the +/// pre-F-CONC-7 "drain forever" semantics as the default. +fn drain_escalate_timeout() -> Option { + std::env::var("PITBOSS_CTRLC_DRAIN_ESCALATE_SECS") + .ok() + .and_then(|s| s.trim().parse::().ok()) + .filter(|n| *n > 0) + .map(Duration::from_secs) +} + // ── Kill-with-reason cascade (Task 4.5) ───────────────────────────────────── /// Cancel any actor in the tree (worker or sub-lead) and optionally deliver @@ -306,6 +359,7 @@ pub fn install_cascade_cancel_watcher(state: Arc) { #[cfg(test)] mod tests { use super::*; + use serial_test::serial; #[test] fn freeze_rejects_pid_zero() { @@ -319,6 +373,45 @@ mod tests { assert!(err.to_string().contains("pid is 0")); } + /// F-CONC-7 #497: the opt-in drain-mode auto-escalation timeout + /// returns `None` when the env var is unset (preserving the + /// pre-fix "drain forever" default) and `Some(_)` only for a + /// positive integer. + #[test] + #[serial(env)] + fn drain_escalate_timeout_is_opt_in() { + // SAFETY: `#[serial(env)]` serializes against other env-touching + // tests so we don't race readers in tracing/reqwest/locale init. + let key = "PITBOSS_CTRLC_DRAIN_ESCALATE_SECS"; + std::env::remove_var(key); + assert_eq!(drain_escalate_timeout(), None, "unset → no escalation"); + + std::env::set_var(key, ""); + assert_eq!(drain_escalate_timeout(), None, "empty → no escalation"); + + std::env::set_var(key, "0"); + assert_eq!(drain_escalate_timeout(), None, "zero → no escalation"); + + std::env::set_var(key, "not-a-number"); + assert_eq!( + drain_escalate_timeout(), + None, + "unparseable → no escalation" + ); + + std::env::set_var(key, "300"); + assert_eq!(drain_escalate_timeout(), Some(Duration::from_secs(300))); + + std::env::set_var(key, " 60 "); + assert_eq!( + drain_escalate_timeout(), + Some(Duration::from_secs(60)), + "surrounding whitespace tolerated" + ); + + std::env::remove_var(key); + } + /// End-to-end: spawn a sleeping child, SIGSTOP it, confirm /// /proc reports stopped state, SIGCONT, confirm runnable, /// then clean up. Linux-only because /proc isn't available on From 1cf345bf2bbb6163b71e26d64bf1123d05029e8a Mon Sep 17 00:00:00 2001 From: Dan Date: Tue, 26 May 2026 21:25:47 -0400 Subject: [PATCH 3/3] perf(dispatch): bump done_tx broadcast capacity from 64 to 1024 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F-CONC-12 #491: the per-`WorkerRegistry` broadcast channel that fans worker-completion events out to `wait_for_*` subscribers was sized at 64 — fine for the historical single-layer flat mode, but post depth-2 every root subscriber sees every worker's completion across the whole tree. In a mass-cancel scenario across a saturated tree, the slowest subscriber can lag the channel and fall back to `refresh_in_flight_from_maps`, which holds nested read locks on `subleads` + per-sublead `workers.states` across the recovery scan. Bumping to 1024 absorbs the storm without changing the fallback's correctness guarantee — the `Lagged` → `refresh_in_flight_from_maps` path remains in place as belt-and-braces, not the expected fast path. The capacity change is ~8 KiB additional memory per `WorkerRegistry` (tokio broadcast allocates the ring eagerly). Doc-comment updated to explain the sizing rationale so a future refactor that questions the value has the context. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/pitboss-cli/src/dispatch/layer.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/pitboss-cli/src/dispatch/layer.rs b/crates/pitboss-cli/src/dispatch/layer.rs index c8610fa..c97f1a6 100644 --- a/crates/pitboss-cli/src/dispatch/layer.rs +++ b/crates/pitboss-cli/src/dispatch/layer.rs @@ -109,12 +109,22 @@ pub struct WorkerRegistry { } impl WorkerRegistry { - /// Construct an empty registry with a fresh 64-slot `done_tx` channel. - /// The capacity matches the pre-split `LayerState::new` default; the - /// channel is mostly there for `wait_for_worker` consumers and a + /// Construct an empty registry with a fresh 1024-slot `done_tx` channel. + /// The channel is mostly there for `wait_for_worker` consumers and a /// `Lagged` receiver re-syncs from the `states` map. + /// + /// **Capacity (F-CONC-12 #491):** sized to absorb a mass-cancel storm + /// across a fully-saturated depth-2 tree (root + N sub-leads, each + /// with their own worker pool) without the slowest `wait_for_*` + /// subscriber lagging into the recovery rescan. The `Lagged` → + /// `refresh_in_flight_from_maps` path in + /// `hierarchical.rs::await_workers_drained` remains correct and is + /// kept as belt-and-braces, not the expected fast path. The pre-F-CONC-12 + /// capacity of 64 matched the historical single-layer default, but + /// post depth-2 every root subscriber sees every worker's completion + /// across the whole tree. pub fn new() -> Self { - let (done_tx, _) = broadcast::channel(64); + let (done_tx, _) = broadcast::channel(1024); Self { states: RwLock::new(HashMap::new()), cancels: RwLock::new(HashMap::new()),