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: <msg>")` 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

@@ -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<R>`** — per-name capacities; unconfigured names default to
   capacity 1.
 

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<N, R: Clone + Eq + Hash> Scheduler<N, R> {
     /// 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<F, Fut>(
         sched: &Arc<Mutex<Self>>,
         run: F,
@@ -207,7 +212,18 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
         };
         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<N, R: Clone + Eq + Hash> Scheduler<N, R> {
     }
 }
 
+/// Drive `fut` to completion, returning `Err` with the panic message if any
+/// poll of it panics.
+async fn catch_panic<T>(fut: impl Future<Output = T>) -> Result<T, String> {
+    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::<String>().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