volume: one long-lived tokio runtime for blocking tiered S3 reads (#11448)

* volume: one long-lived tokio runtime for blocking tiered S3 reads

block_on_tier_future, behind read_range_blocking and delete_file_blocking,
spawned an OS thread and built a fresh current-thread tokio runtime on every
call, then tore the runtime down. On an S3-tiered volume that is once per
needle read, per streamed 64 KiB chunk, per DatScanPlan record and per
destroy. The SDK client's pooled HTTPS connections are driven by tasks on the
runtime a request ran on, so each teardown dropped the pool and every call
re-dialed and re-handshook TLS. A panic inside the SDK was also flattened to
the fixed string "tier runtime thread panicked".

Now one process-wide runtime (OnceLock, multi_thread, 2 workers named
tier-io) drives all tier I/O; block_on_tier_future spawns onto it and parks
the caller on an mpsc channel for the JoinHandle result. Blocking the caller
is unavoidable (the storage layer is synchronous) and is what the old code
did through thread::spawn().join(). Handle::block_on is not used because the
wrappers are also reached from inside another runtime's worker, where it
panics with "Cannot start a runtime from within a runtime". JoinError panics
are downcast to &str/String and the payload is kept in the error.

Tests cover runtime reuse (Handle::id equal across calls, thread name
tier-io), calls from a std thread, from spawn_blocking, and directly from
current-thread and multi-thread runtime contexts, and the panic payload.
Against the old body 7 of 9 fail.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* volume: return a tier runtime build failure instead of panicking

Review follow-up. tier_runtime() expect'ed the runtime build, so an OS
refusing threads panicked inside Volume::destroy (after the volume left
the in-memory map, before its files were removed) and inside needle
reads, bypassing their error paths. Keep the runtime in a
Mutex<Option<Runtime>> behind tier_handle() -> Result<Handle, String>:
a failed build is returned to the caller through block_on_tier_future's
existing Result and is not cached, so a later call retries once the
pressure is gone. The lock is held only while building.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* volume: trim comments on the shared tier I/O runtime

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/remote_storage/s3_tier.rs

@@ -604,18 +604,192 @@ pub fn global_s3_tier_registry() -> &'static RwLock<S3TierRegistry> {
     GLOBAL_S3_TIER_REGISTRY.get_or_init(|| RwLock::new(S3TierRegistry::new()))
 }
 
+/// The one process-wide runtime for tiered-S3 I/O issued from synchronous
+/// storage code. A per-call runtime tore down the SDK's pooled connections
+/// after every 64 KiB chunk, re-dialing TLS per read; a long-lived runtime
+/// keeps the pool warm.
+///
+/// Built on first use. A build failure is returned, not cached or panicked:
+/// callers sit inside `Volume::destroy` and needle reads, whose own error
+/// paths must run, and a later call may succeed.
+static TIER_RUNTIME: std::sync::Mutex<Option<tokio::runtime::Runtime>> =
+    std::sync::Mutex::new(None);
+
+fn tier_handle() -> Result<tokio::runtime::Handle, String> {
+    let mut slot = TIER_RUNTIME
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if slot.is_none() {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .thread_name("tier-io")
+            .enable_all()
+            .build()
+            .map_err(|e| format!("failed to build the tier I/O tokio runtime: {}", e))?;
+        *slot = Some(runtime);
+    }
+    Ok(slot.as_ref().expect("just initialised").handle().clone())
+}
+
+/// Run `future` on the tier runtime and block the calling thread until it
+/// finishes. The caller may be a worker of *another* tokio runtime, so this
+/// waits on a channel rather than `Handle::block_on`, which panics when
+/// called from inside any runtime context.
 fn block_on_tier_future<F, T>(future: F) -> Result<T, String>
 where
     F: Future<Output = Result<T, String>> + Send + 'static,
     T: Send + 'static,
 {
-    std::thread::spawn(move || {
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .map_err(|e| format!("failed to build tokio runtime: {}", e))?;
-        runtime.block_on(future)
-    })
-    .join()
-    .map_err(|_| "tier runtime thread panicked".to_string())?
+    let handle = tier_handle()?;
+    let task = handle.spawn(future);
+    let (tx, rx) = std::sync::mpsc::sync_channel(1);
+    handle.spawn(async move {
+        // The receiver only goes away if the caller was unwound; nothing to
+        // report then.
+        let _ = tx.send(task.await);
+    });
+    match rx.recv() {
+        Ok(Ok(result)) => result,
+        Ok(Err(join_error)) => Err(describe_join_error(join_error)),
+        Err(_) => Err("tier I/O runtime dropped the task before it finished".to_string()),
+    }
+}
+
+/// Turn a `JoinError` into a message that keeps the panic payload, so an
+/// SDK panic surfaces as "boom" rather than a fixed "thread panicked".
+fn describe_join_error(join_error: tokio::task::JoinError) -> String {
+    if join_error.is_panic() {
+        let payload = join_error.into_panic();
+        let message = if let Some(s) = payload.downcast_ref::<&str>() {
+            (*s).to_string()
+        } else if let Some(s) = payload.downcast_ref::<String>() {
+            s.clone()
+        } else {
+            "non-string panic payload".to_string()
+        };
+        format!("tier I/O task panicked: {}", message)
+    } else {
+        format!("tier I/O task failed: {}", join_error)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashSet;
+    use tokio::runtime::Handle;
+
+    fn probe() -> Result<(tokio::runtime::Id, Option<String>), String> {
+        block_on_tier_future(async {
+            Ok((
+                Handle::current().id(),
+                std::thread::current().name().map(str::to_string),
+            ))
+        })
+    }
+
+    #[test]
+    fn block_on_tier_future_reuses_one_runtime() {
+        let (first_runtime, first_thread) = probe().expect("first call");
+        let (second_runtime, second_thread) = probe().expect("second call");
+        assert_eq!(
+            first_runtime, second_runtime,
+            "each call must run on the same long-lived tier runtime"
+        );
+        assert_eq!(first_thread.as_deref(), Some("tier-io"));
+        assert_eq!(second_thread.as_deref(), Some("tier-io"));
+
+        let mut runtimes = HashSet::new();
+        for _ in 0..20 {
+            let (id, _) = probe().expect("probe");
+            runtimes.insert(id);
+        }
+        assert_eq!(runtimes.len(), 1);
+    }
+
+    #[test]
+    fn block_on_tier_future_returns_the_value_and_the_error() {
+        assert_eq!(block_on_tier_future(async { Ok(7u32) }), Ok(7));
+        assert_eq!(
+            block_on_tier_future::<_, u32>(async { Err("nope".to_string()) }),
+            Err("nope".to_string())
+        );
+    }
+
+    #[test]
+    fn block_on_tier_future_works_from_a_std_thread() {
+        let (id, _) = std::thread::spawn(probe)
+            .join()
+            .expect("probe thread")
+            .expect("probe");
+        assert_eq!(id, tier_handle().expect("tier runtime").id());
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn block_on_tier_future_works_from_spawn_blocking() {
+        let (id, _) = tokio::task::spawn_blocking(probe)
+            .await
+            .expect("spawn_blocking")
+            .expect("probe");
+        assert_eq!(id, tier_handle().expect("tier runtime").id());
+        assert_ne!(id, Handle::current().id());
+    }
+
+    // Called straight from another runtime's async context: the case that
+    // would panic with `Handle::block_on` ("Cannot start a runtime from
+    // within a runtime").
+    #[tokio::test]
+    async fn block_on_tier_future_works_from_a_current_thread_runtime() {
+        let (id, _) = probe().expect("probe");
+        assert_eq!(id, tier_handle().expect("tier runtime").id());
+        assert_ne!(id, Handle::current().id());
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn block_on_tier_future_works_from_a_multi_thread_runtime_worker() {
+        let (id, _) = probe().expect("probe");
+        assert_eq!(id, tier_handle().expect("tier runtime").id());
+        assert_ne!(id, Handle::current().id());
+    }
+
+    #[test]
+    fn block_on_tier_future_reports_the_panic_payload() {
+        let err = block_on_tier_future::<_, ()>(async {
+            if std::hint::black_box(true) {
+                panic!("boom {}", 42);
+            }
+            Ok(())
+        })
+        .expect_err("a panicking future must be an error");
+        assert!(err.contains("boom 42"), "got: {err}");
+        assert!(err.contains("panicked"), "got: {err}");
+    }
+
+    #[test]
+    fn block_on_tier_future_reports_a_str_panic_payload() {
+        let err = block_on_tier_future::<_, ()>(async {
+            if std::hint::black_box(true) {
+                panic!("static boom");
+            }
+            Ok(())
+        })
+        .expect_err("a panicking future must be an error");
+        assert!(err.contains("static boom"), "got: {err}");
+    }
+
+    #[test]
+    fn backend_name_to_type_id_splits_on_dot() {
+        assert_eq!(
+            backend_name_to_type_id("s3"),
+            ("s3".to_string(), "default".to_string())
+        );
+        assert_eq!(
+            backend_name_to_type_id("s3.eu"),
+            ("s3".to_string(), "eu".to_string())
+        );
+        assert_eq!(
+            backend_name_to_type_id("s3.a.b"),
+            (String::new(), String::new())
+        );
+    }
 }