From da21366e9c0e1bb9da9ac1b2ce33066788ce7f06 Mon Sep 17 00:00:00 2001
From: Alex Tharaldsen <alexander@tharaldsen.org>
Date: Sun, 31 May 2026 18:29:19 +0200
Subject: [PATCH] fix(consensusmanager): don't panic on spawn_blocking
 cancellation during shutdown (#949)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`ConsensusSessionOwned::spawn_blocking` did `spawn_blocking(...).await.unwrap()`.
When the Tokio runtime shuts down it cancels not-yet-started blocking tasks, so
the join returns `Err(JoinError::Cancelled)` and `.unwrap()` panics a
runtime-worker thread mid-teardown (observed during graceful shutdown under slow
I/O, e.g. flushing a large block backlog to an HDD).

Handle the join result explicitly:
- Cancellation only happens during runtime shutdown and has no result to return,
  so park until the task is dropped as part of teardown instead of panicking.
- A genuine panic inside the closure is still propagated via `resume_unwind`,
  preserving the previous behavior (we don't mask real bugs).

Adds unit tests covering all three paths (result returned, cancellation parks,
closure panic propagates) using a local multi-thread runtime — no extra deps.
---
 components/consensusmanager/src/session.rs | 61 +++++++++++++++++++++-
 1 file changed, 60 insertions(+), 1 deletion(-)

diff --git a/components/consensusmanager/src/session.rs b/components/consensusmanager/src/session.rs
index e1b183d298..1904ca0f44 100644
--- a/components/consensusmanager/src/session.rs
+++ b/components/consensusmanager/src/session.rs
@@ -165,7 +165,19 @@ impl ConsensusSessionOwned {
         F: FnOnce(&dyn ConsensusApi) -> R + Send + 'static,
         R: Send + 'static,
     {
-        spawn_blocking(move || f(self.consensus.as_ref())).await.unwrap()
+        join_blocking_or_park(spawn_blocking(move || f(self.consensus.as_ref()))).await
+    }
+}
+
+/// Awaits a `spawn_blocking` join handle, handling the two [`tokio::task::JoinError`] cases:
+/// - Cancellation only occurs while the Tokio runtime is shutting down, where there is no result to
+///   return; in that case park until this task is dropped during teardown rather than panicking (see #949).
+/// - A genuine panic inside the closure is propagated to the caller, preserving the previous behavior.
+async fn join_blocking_or_park<R>(handle: tokio::task::JoinHandle<R>) -> R {
+    match handle.await {
+        Ok(result) => result,
+        Err(err) if err.is_cancelled() => std::future::pending().await,
+        Err(err) => std::panic::resume_unwind(err.into_panic()),
     }
 }
 
@@ -504,3 +516,50 @@ impl ConsensusSessionOwned {
 }
 
 pub type ConsensusProxy = ConsensusSessionOwned;
+
+#[cfg(test)]
+mod tests {
+    use super::join_blocking_or_park;
+
+    fn runtime() -> tokio::runtime::Runtime {
+        tokio::runtime::Builder::new_multi_thread().worker_threads(2).build().unwrap()
+    }
+
+    #[test]
+    fn returns_closure_result() {
+        let rt = runtime();
+        let value = rt.block_on(async { join_blocking_or_park(tokio::task::spawn_blocking(|| 42u32)).await });
+        assert_eq!(value, 42);
+    }
+
+    #[test]
+    fn parks_on_cancellation() {
+        // A cancelled join handle (the runtime-shutdown case) must park rather than panic or resolve,
+        // since there is no result to return. We emulate the cancellation by aborting a task.
+        let rt = runtime();
+        rt.block_on(async {
+            let cancelled = tokio::spawn(std::future::pending::<u32>());
+            cancelled.abort();
+
+            let parked = tokio::spawn(join_blocking_or_park(cancelled));
+            for _ in 0..1000 {
+                tokio::task::yield_now().await;
+            }
+            assert!(!parked.is_finished(), "helper should park on cancellation rather than resolve");
+            parked.abort();
+        });
+    }
+
+    #[test]
+    fn propagates_closure_panic() {
+        // A genuine panic inside the closure must propagate to the caller (not be swallowed).
+        let rt = runtime();
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            rt.block_on(async { join_blocking_or_park(tokio::task::spawn_blocking(|| -> u32 { panic!("boom-{}", 42) })).await })
+        }));
+        let payload = result.expect_err("closure panic should propagate through join_blocking_or_park");
+        let message =
+            payload.downcast_ref::<String>().map(String::as_str).or_else(|| payload.downcast_ref::<&str>().copied()).unwrap_or("");
+        assert!(message.contains("boom-42"), "panic payload should be preserved, got: {message:?}");
+    }
+}