consensus: map spawn_blocking cancellation during shutdown to error instead of panicking (#27513)

## Description

During the 2026-07-14 mainnet rolling restart, a validator panicked on
shutdown at the synchronizer's `.expect("Spawn blocking should not
fail")`: tokio cancels `spawn_blocking` tasks when the runtime shuts
down, so the JoinHandle resolves to `JoinError::Cancelled` and the panic
alert pages on an otherwise clean restart. Consensus has four such call
sites.

Adds a `task::spawn_blocking` wrapper that resolves to
`ConsensusError::Shutdown` on cancellation and resumes real panics on
the calling thread, and uses it at all four sites. `Shutdown` is already
handled quietly by callers as the established signal for components
going away mid-request.

## Test plan

New unit tests in `task.rs` cover the wrapper contract: a
`spawn_blocking` cancelled by runtime shutdown maps to
`ConsensusError::Shutdown`, and panics from the closure still propagate.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] Indexing Framework:

Author: alex-mysten

consensus/core/src/authority_service.rs

@@ -29,6 +29,7 @@ use crate::{
     network::{BlockStream, ExtendedSerializedBlock, PeerId, ValidatorNetworkService},
     round_tracker::RoundTracker,
     synchronizer::SynchronizerHandle,
+    task::spawn_blocking,
     transaction_vote_tracker::TransactionVoteTracker,
 };
 
@@ -176,20 +177,18 @@ impl<C: CoreThreadDispatcher> ValidatorNetworkService for AuthorityService<C> {
         // Reject blocks failing parsing and validations.
         let block_verifier = self.block_verifier.clone();
         let serialized = serialized_block.block.clone();
-        let (verified_block, reject_txn_votes) = tokio::task::spawn_blocking(move || {
-            block_verifier.verify_and_vote(signed_block, serialized)
-        })
-        .await
-        .expect("verify_and_vote blocking task panicked")
-        .tap_err(|e| {
-            self.context
-                .metrics
-                .node_metrics
-                .invalid_blocks
-                .with_label_values(&[peer_hostname.as_str(), "handle_send_block", e.name()])
-                .inc();
-            info!("Invalid block from {}: {}", peer, e);
-        })?;
+        let (verified_block, reject_txn_votes) =
+            spawn_blocking(move || block_verifier.verify_and_vote(signed_block, serialized))
+                .await?
+                .tap_err(|e| {
+                    self.context
+                        .metrics
+                        .node_metrics
+                        .invalid_blocks
+                        .with_label_values(&[peer_hostname.as_str(), "handle_send_block", e.name()])
+                        .inc();
+                    info!("Invalid block from {}: {}", peer, e);
+                })?;
         let excluded_ancestors = self
             .parse_excluded_ancestors(peer, &verified_block, serialized_block.excluded_ancestors)
             .tap_err(|e| {

consensus/core/src/commit_observer.rs

@@ -17,6 +17,7 @@ use crate::{
     error::ConsensusResult,
     linearizer::Linearizer,
     storage::Store,
+    task::spawn_blocking,
     transaction_vote_tracker::TransactionVoteTracker,
 };
 
@@ -73,16 +74,17 @@ impl CommitObserver {
         // Recover blocks needed for future commits (and block proposals).
         // Some blocks might have been recovered as committed blocks in recover_and_send_commits().
         // They will just be ignored.
-        tokio::runtime::Handle::current()
-            .spawn_blocking({
-                let transaction_vote_tracker = observer.transaction_vote_tracker.clone();
-                let gc_round = observer.dag_state.read().gc_round();
-                move || {
-                    transaction_vote_tracker.recover_blocks_after_round(gc_round);
-                }
-            })
-            .await
-            .expect("Spawn blocking should not fail");
+        if let Err(e) = spawn_blocking({
+            let transaction_vote_tracker = observer.transaction_vote_tracker.clone();
+            let gc_round = observer.dag_state.read().gc_round();
+            move || {
+                transaction_vote_tracker.recover_blocks_after_round(gc_round);
+            }
+        })
+        .await
+        {
+            info!("Skipping block recovery for transaction voting: {e}");
+        }
 
         observer
     }

consensus/core/src/commit_syncer.rs

@@ -39,7 +39,6 @@ use mysten_metrics::spawn_logged_monitored_task;
 use parking_lot::RwLock;
 use rand::{prelude::SliceRandom as _, rngs::ThreadRng};
 use tokio::{
-    runtime::Handle,
     sync::oneshot,
     task::{JoinHandle, JoinSet},
     time::{MissedTickBehavior, sleep},
@@ -63,7 +62,7 @@ use crate::{
     peers_pool::PeersPool,
     round_tracker::RoundTracker,
     stake_aggregator::{QuorumThreshold, StakeAggregator},
-    task::{join_and_propagate_panic, shutdown_join_set},
+    task::{join_and_propagate_panic, shutdown_join_set, spawn_blocking},
     transaction_vote_tracker::TransactionVoteTracker,
 };
 
@@ -570,24 +569,22 @@ where
         // 2. Verify the response contains blocks that can certify the last returned commit,
         // and the returned commits are chained by digests, so earlier commits are certified
         // as well.
-        let (commits, commit_certifying_blocks_and_votes) = Handle::current()
-            .spawn_blocking({
-                let context = inner.context.clone();
-                let block_verifier = inner.block_verifier.clone();
-                let peer = target_peer.clone();
-                move || {
-                    Inner::<VC, OC>::verify_commits(
-                        &context,
-                        block_verifier.as_ref(),
-                        peer,
-                        commit_range,
-                        serialized_commits,
-                        serialized_blocks,
-                    )
-                }
-            })
-            .await
-            .expect("Spawn blocking should not fail")?;
+        let (commits, commit_certifying_blocks_and_votes) = spawn_blocking({
+            let context = inner.context.clone();
+            let block_verifier = inner.block_verifier.clone();
+            let peer = target_peer.clone();
+            move || {
+                Inner::<VC, OC>::verify_commits(
+                    &context,
+                    block_verifier.as_ref(),
+                    peer,
+                    commit_range,
+                    serialized_commits,
+                    serialized_blocks,
+                )
+            }
+        })
+        .await??;
 
         // Only the vote tracker needs the reject votes, so move them into it without cloning.
         // Cheap clones of the blocks are enough for the rest of the fetch handling.

consensus/core/src/synchronizer.rs

@@ -22,7 +22,6 @@ use rand::{prelude::SliceRandom as _, rngs::ThreadRng};
 use sui_macros::fail_point_async;
 use tap::TapFallible;
 use tokio::{
-    runtime::Handle,
     sync::{mpsc::error::TrySendError, oneshot},
     task::JoinSet,
     time::{Instant, sleep, sleep_until, timeout},
@@ -41,7 +40,7 @@ use crate::{
     network::{ObserverNetworkClient, PeerId, SynchronizerClient, ValidatorNetworkClient},
     peers_pool::PeersPool,
     round_tracker::RoundTracker,
-    task::shutdown_join_set,
+    task::{shutdown_join_set, spawn_blocking},
 };
 use crate::{core_thread::CoreThreadDispatcher, transaction_vote_tracker::TransactionVoteTracker};
 
@@ -592,15 +591,13 @@ where
         serialized_blocks.truncate(context.parameters.max_blocks_per_sync);
 
         // Verify all the fetched blocks
-        let (blocks, voted_blocks) = Handle::current()
-            .spawn_blocking({
-                let block_verifier = block_verifier.clone();
-                let context = context.clone();
-                let peer = peer.clone();
-                move || Self::verify_blocks(serialized_blocks, block_verifier, &context, peer)
-            })
-            .await
-            .expect("Spawn blocking should not fail")?;
+        let (blocks, voted_blocks) = spawn_blocking({
+            let block_verifier = block_verifier.clone();
+            let context = context.clone();
+            let peer = peer.clone();
+            move || Self::verify_blocks(serialized_blocks, block_verifier, &context, peer)
+        })
+        .await??;
 
         if context.protocol_config.transaction_voting_enabled() {
             transaction_vote_tracker.add_voted_blocks(voted_blocks);

consensus/core/src/task.rs

@@ -4,6 +4,8 @@
 use futures::FutureExt as _;
 use tokio::task::{JoinHandle, JoinSet};
 
+use crate::error::{ConsensusError, ConsensusResult};
+
 /// Awaits the task and propagates its panic if it panicked. Task cancellation is ignored.
 pub(crate) async fn join_and_propagate_panic<T>(task: JoinHandle<T>) {
     if let Err(error) = task.await
@@ -47,3 +49,58 @@ pub(crate) fn reap_finished_task(task: &mut JoinHandle<()>) -> bool {
     }
     true
 }
+
+/// Runs the closure on the blocking pool, like tokio::task::spawn_blocking, but resolves
+/// to ConsensusError::Shutdown instead of a JoinError when the task is cancelled by
+/// runtime shutdown. Panics from the closure are resumed on the calling thread.
+pub(crate) async fn spawn_blocking<F, T>(f: F) -> ConsensusResult<T>
+where
+    F: FnOnce() -> T + Send + 'static,
+    T: Send + 'static,
+{
+    tokio::task::spawn_blocking(f)
+        .await
+        .map_err(spawn_blocking_join_error)
+}
+
+/// Converts a JoinError from a spawn_blocking task into a ConsensusError.
+/// Panics from the blocking task are resumed on the calling thread. Otherwise the task
+/// was cancelled, which only happens when the runtime is shutting down.
+fn spawn_blocking_join_error(e: tokio::task::JoinError) -> ConsensusError {
+    if e.is_panic() {
+        std::panic::resume_unwind(e.into_panic());
+    }
+    ConsensusError::Shutdown
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    /// Reproduces the production scenario: spawn_blocking racing with runtime shutdown
+    /// gets its task cancelled instead of run, so its JoinHandle resolves to
+    /// JoinError::Cancelled.
+    #[tokio::test]
+    async fn test_spawn_blocking_join_error_on_runtime_shutdown() {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+            .unwrap();
+        let handle = runtime.handle().clone();
+        runtime.shutdown_background();
+
+        let cancelled = handle.spawn_blocking(|| ());
+        let join_error = cancelled.await.unwrap_err();
+        assert!(join_error.is_cancelled());
+        assert!(matches!(
+            spawn_blocking_join_error(join_error),
+            ConsensusError::Shutdown
+        ));
+    }
+
+    #[tokio::test]
+    #[should_panic(expected = "boom")]
+    async fn test_spawn_blocking_resumes_panic() {
+        let _ = spawn_blocking(|| panic!("boom")).await;
+    }
+}