feat(utilities): add utilities for propagating errors across async tasks

Author: kaimast

Cargo.lock

@@ -124,7 +124,7 @@ package = [
   "dep:dotenvy"
 ]
 
-async = [ "snarkvm-ledger/async", "snarkvm-synthesizer/async" ]
+async = [ "snarkvm-ledger?/async", "snarkvm-synthesizer?/async", "snarkvm-utilities?/async" ]
 cuda = [ "snarkvm-algorithms/cuda" ]
 history = [ "snarkvm-synthesizer/history" ]
 parameters_no_std_out = [ "snarkvm-parameters/no_std_out" ]
@@ -515,6 +515,10 @@ default-features = false
 [workspace.dependencies.smallvec]
 version = "1.14"
 
+[workspace.dependencies.tokio]
+version = "1"
+default-features = false
+
 [workspace.dependencies.tempfile]
 version = "3.15"
 

Cargo.toml

@@ -39,7 +39,7 @@ features = [ "preserve_order" ]
 
 [dependencies.tokio]
 optional = true
-version = "1"
+workspace = true
 features = [ "rt" ]
 
 [dev-dependencies.snarkvm-ledger-block]

ledger/narwhal/data/Cargo.toml

@@ -40,6 +40,11 @@ workspace = true
 [dependencies.colored]
 workspace = true
 
+[dependencies.tokio]
+workspace = true
+optional = true
+features = ["rt"]
+
 [dependencies.num_cpus]
 version = "1"
 
@@ -83,6 +88,7 @@ workspace = true
 
 [features]
 default = [ "derive" ]
+async = [ "tokio" ]
 derive = [ "snarkvm-utilities-derives" ]
 dev-print = [ ]
 serial = [ "derive" ]

utilities/Cargo.toml

@@ -58,5 +58,11 @@ pub use serialize::*;
 pub mod errors;
 pub use errors::*;
 
+#[cfg(feature = "async")]
+/// Helpers to spawn async tasks.
+pub mod task;
+#[cfg(feature = "async")]
+pub use task::*;
+
 /// Use old name for backward-compatibility.
 pub use errors::io_error as error;

utilities/src/lib.rs

@@ -0,0 +1,71 @@
+// Copyright (c) 2019-2025 Provable Inc.
+// This file is part of the snarkVM library.
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+
+// http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// Wrapper around `tokio::JoinHandle` that propagates panics.
+pub struct JoinHandle<R: Send + 'static> {
+    inner: tokio::task::JoinHandle<R>,
+}
+
+/// Wrapper around `tokio::spawn_blocking` that propagates panics.
+pub fn spawn_blocking<F, R>(f: F) -> JoinHandle<R>
+where
+    F: FnOnce() -> R + Send + 'static,
+    R: Send + 'static,
+{
+    JoinHandle { inner: tokio::task::spawn_blocking(f) }
+}
+
+/// Wrapper around `tokio::spawn` that propagates panics.
+pub fn spawn<F>(f: F) -> JoinHandle<F::Output>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    JoinHandle { inner: tokio::task::spawn(f) }
+}
+
+impl<R: Send + 'static> JoinHandle<R> {
+    pub fn abort(&self) {
+        self.inner.abort();
+    }
+}
+
+impl<R: Send + 'static> Future for JoinHandle<R> {
+    type Output = R;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let Poll::Ready(result) = std::pin::pin!(&mut self.inner).poll(cx) else {
+            return Poll::Pending;
+        };
+
+        match result {
+            Ok(value) => Poll::Ready(value),
+            Err(err) => {
+                if err.is_panic() {
+                    // Resume the panic on the main task
+                    std::panic::resume_unwind(err.into_panic());
+                } else {
+                    panic!("Got unexpected tokio error: {err}");
+                }
+            }
+        }
+    }
+}