From d1ee3d8cc5273f951d8a9ff805f069303bb75b7f Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 9 Oct 2026 14:28:43 +0200 Subject: [PATCH] hive-jobq: treat a panicking job runner as a failed node The future `Scheduler::claim_next` hands back completed its node only after `run`'s future returned. A panic while running (the workspace unwinds) skipped that completion, so the node stayed `Running` and its resource units stayed held until the process restarted; later nodes needing the same resource never started. The run future is now polled under `catch_unwind`. A panic completes the node as `Outcome::Failed("runner panicked: ")` with nothing grown, through the same `complete_growing` path as an ordinary failure: `AFTER_OK` dependents are skipped, pending children cascade, and the node's units are released. The node's future is boxed: wrapping it inline pushed the futures swarm-controller and hive-c0re spawn over clippy's `large_futures` limit. Closes #5058 --- hive-jobq/README.md | 3 +- hive-jobq/src/scheduler.rs | 79 +++++++++++++++++++++++++++++++++++++- 2 files changed, 79 insertions(+), 3 deletions(-) diff --git a/hive-jobq/README.md b/hive-jobq/README.md index a396080d..352c9abc 100644 --- a/hive-jobq/README.md +++ b/hive-jobq/README.md @@ -59,7 +59,8 @@ finished_at, error }`. All fields public; derives serde for the wire (acquiring resources atomically), `complete(id, outcome)` reports a finished node's result and rolls terminality up the parent chain, releasing grants once a subtree is done. `Outcome::{Done, Failed(String)}` — the failure reason - rides `Failed` onto the node's `error`. + rides `Failed` onto the node's `error`. A runner that panics completes its node as + `Failed` with the panic message, which releases its grants like any failure. - **`ResourceTable`** — per-name capacities; unconfigured names default to capacity 1. diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 3d4ae466..1cb219f7 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -29,10 +29,14 @@ //! its whole subtree are terminal. (Early release once no subtree node needs it //! is a deferred optimization — unsafe under dynamically-appended subnodes.) +use std::any::Any; use std::collections::HashMap; use std::future::Future; use std::hash::Hash; +use std::panic::{self, AssertUnwindSafe}; +use std::pin::pin; use std::sync::{Arc, Mutex}; +use std::task::Poll; use crate::builder::{BuildError, JobBuilder, NodeGuid}; use crate::resources::ResourceTable; @@ -189,7 +193,8 @@ impl Scheduler { /// The output carries the insert result rather than swallowing it — this /// crate has no logger, so a malformed grown job is reported to the caller, /// who is the one that can log it. The node is completed either way: its - /// own work already happened. + /// own work already happened. A panic in `run` completes it as an ordinary + /// [`Outcome::Failed`] carrying the panic message, releasing its units. pub fn claim_next( sched: &Arc>, run: F, @@ -207,7 +212,18 @@ impl Scheduler { }; let sched = Arc::clone(sched); Some(async move { - let (grown, outcome) = run(id, payload, JobBuilder::new()).await; + // Boxed: inline, the node's future would sit beside `run` and + // `payload` in this one's state, and the caller spawns this one. + let (grown, outcome) = catch_panic(Box::pin(async move { + run(id, payload, JobBuilder::new()).await + })) + .await + .unwrap_or_else(|msg| { + ( + JobBuilder::new(), + Outcome::Failed(format!("runner panicked: {msg}")), + ) + }); let grew = sched .lock() .expect("jobq scheduler mutex poisoned") @@ -641,6 +657,30 @@ impl Scheduler { } } +/// Drive `fut` to completion, returning `Err` with the panic message if any +/// poll of it panics. +async fn catch_panic(fut: impl Future) -> Result { + let mut fut = pin!(fut); + std::future::poll_fn(|cx| { + match panic::catch_unwind(AssertUnwindSafe(|| fut.as_mut().poll(cx))) { + Ok(Poll::Ready(out)) => Poll::Ready(Ok(out)), + Ok(Poll::Pending) => Poll::Pending, + Err(payload) => Poll::Ready(Err(panic_message(payload.as_ref()))), + } + }) + .await +} + +/// The message a panic was raised with, for the two payload types `panic!` +/// produces. +fn panic_message(payload: &(dyn Any + Send)) -> String { + payload + .downcast_ref::<&str>() + .map(|s| (*s).to_owned()) + .or_else(|| payload.downcast_ref::().cloned()) + .unwrap_or_else(|| "non-string panic payload".to_owned()) +} + #[cfg(test)] mod tests { use super::*; @@ -752,6 +792,41 @@ mod tests { assert_eq!(s.graph().node(n).unwrap().state, State::Failed); } + /// A runner that panics completes its node as `Failed`, like any other + /// failure: the node is terminal, its `AFTER_OK` dependent is ruled out, + /// and its resource unit is free for the next node that needs it. + #[test] + fn a_panicking_runner_fails_its_node_and_releases_its_resources() { + let mut s = scheduler_with_slots(1); + let a = s.append("a", res_dep("build-slot"), None).expect("a"); + let dependent = s.append("dep", vec![after_ok(a)], None).expect("dep"); + let b = s.append("b", res_dep("build-slot"), None).expect("b"); + let sched = Arc::new(Mutex::new(s)); + + let run = Scheduler::claim_next(&sched, |_, _, _| async { panic!("boom") }) + .expect("a is runnable"); + let mut cx = std::task::Context::from_waker(std::task::Waker::noop()); + let Poll::Ready((id, grew)) = pin!(run).poll(&mut cx) else { + panic!("a panicking runner completes on its first poll"); + }; + assert_eq!(id, a); + grew.expect("a failed node grows nothing"); + + let s = sched.lock().expect("not poisoned"); + let node = s.graph().node(a).unwrap(); + assert_eq!(node.state, State::Failed); + assert_eq!(node.error.as_deref(), Some("runner panicked: boom")); + assert_eq!(s.graph().node(dependent).unwrap().state, State::Skipped); + assert_eq!(avail(&s, "build-slot"), 1); + drop(s); + + let next = + Scheduler::claim_next(&sched, |_, _, builder| async { (builder, Outcome::Done) }); + assert!(next.is_some(), "b can claim the released slot"); + let s = sched.lock().expect("not poisoned"); + assert_eq!(s.graph().node(b).unwrap().state, State::Running); + } + /// A contended resource goes to the oldest waiter. /// /// [`Scheduler::claim_one`] scans [`Graph::nodes`] — insertion order — and