From e5b9a9050b05b99ce09af1b7fd44519ef7323b67 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Fri, 26 Jun 2026 12:08:24 -0400 Subject: [PATCH 1/7] fix(shared-runtime): guard shutdown() against Tokio TLS destruction during CPython finalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit During CPython interpreter finalization, thread-local storage is destroyed before atexit handlers fire. SharedRuntime::shutdown() calls runtime.block_on() which internally calls context::enter() to set up Tokio's CONTEXT thread-local. If that TLS slot is already destroyed, context::enter() panics with "The Tokio context thread-local variable has been destroyed", which PyO3 converts to a pyo3_runtime.PanicException. This causes a crash on every uWSGI worker shutdown when using ddtrace >=4.9.x. Fix: check Handle::try_current().is_thread_local_destroyed() before calling block_on(). If TLS is gone, return Ok(()) early — the OS will clean up remaining Tokio threads on process exit. This eliminates both the panic and the subsequent 60s hang/SIGKILL. Reproducer: uWSGI app with lazy-apps=true, ddtrace imported via uwsgi import=, 4 workers. SIGTERM triggers the panic on every worker. Co-Authored-By: Claude Sonnet 4.6 --- .../src/shared_runtime/mod.rs | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/libdd-shared-runtime/src/shared_runtime/mod.rs b/libdd-shared-runtime/src/shared_runtime/mod.rs index 058bf730a5..fdebc6de39 100644 --- a/libdd-shared-runtime/src/shared_runtime/mod.rs +++ b/libdd-shared-runtime/src/shared_runtime/mod.rs @@ -234,6 +234,17 @@ mod native { timeout: Option, ) -> Result<(), SharedRuntimeError> { debug!(?timeout, "Shutting down SharedRuntime"); + // block_on calls context::enter() which accesses Tokio's CONTEXT thread-local. + // During CPython interpreter finalization, TLS is destroyed before atexit handlers + // fire, causing a panic. Detect this via try_current() and bail out early — + // the OS will clean up remaining threads on process exit. + if matches!( + tokio::runtime::Handle::try_current(), + Err(ref e) if e.is_thread_local_destroyed() + ) { + debug!("Tokio TLS destroyed during interpreter finalization, skipping shutdown"); + return Ok(()); + } match self.runtime.lock_or_panic().take() { Some(runtime) => { if let Some(timeout) = timeout { @@ -669,6 +680,23 @@ mod tests { assert_eq!(last, -1); } + #[test] + fn test_shutdown_is_idempotent() { + // Calling shutdown() twice must not panic or error. The second call hits the + // None-guard (runtime already taken). This covers the same early-return path as + // the TLS-destroyed guard added for CPython atexit finalization ordering. + let shared_runtime = SharedRuntime::new().unwrap(); + let (worker, receiver) = make_test_worker(); + + let _ = shared_runtime.spawn_worker(worker, true).unwrap(); + receiver + .recv_timeout(Duration::from_secs(1)) + .expect("worker did not run"); + + assert!(shared_runtime.shutdown(None).is_ok()); + assert!(shared_runtime.shutdown(None).is_ok(), "second shutdown must not panic"); + } + #[test] fn test_after_fork_child_drops_worker_not_restart_on_fork() { let shared_runtime = SharedRuntime::new().unwrap(); From 1742952f25b63887c133063205471ba88ce2c939 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Mon, 29 Jun 2026 11:44:20 -0400 Subject: [PATCH 2/7] style: rustfmt Co-Authored-By: Claude Sonnet 4.6 --- libdd-shared-runtime/src/shared_runtime/fork_safe.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs index 5f0b531b49..1d7f41dbab 100644 --- a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs +++ b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs @@ -447,6 +447,9 @@ mod tests { .expect("worker did not run"); assert!(shared_runtime.shutdown(None).is_ok()); - assert!(shared_runtime.shutdown(None).is_ok(), "second shutdown must not panic"); + assert!( + shared_runtime.shutdown(None).is_ok(), + "second shutdown must not panic" + ); } } From e85cc8665fb204b756acd162db9b1febe7640337 Mon Sep 17 00:00:00 2001 From: quinna-h Date: Thu, 9 Jul 2026 16:12:34 -0400 Subject: [PATCH 3/7] fix(shared-runtime): make ForkSafeRuntime teardown structurally TLS/poison-safe Replaces the symptomatic shutdown()-only TLS guard with two structural invariants so teardown does not depend on the destruction order of any other global/thread-local state during (embedded) interpreter finalization: A. Terminal teardown never blocks or enters the Tokio context. Add `Drop for ForkSafeRuntime` that detaches the owned runtime via `shutdown_background()` instead of a normal blocking `Runtime` drop (which joins worker threads and touches the CONTEXT thread-local, panicking with "The Tokio context thread-local variable has been destroyed" when that TLS is already gone). Done unconditionally, so there is no finalization-detection branch a future teardown scenario can slip past. The graceful `shutdown()` path (which flushes) still runs when a live context exists (`can_block_on`); if it already ran, Drop is a no-op. B. Shared-state access is poison-immune. Add `MutexExt::lock_or_recover` (recovers the guard via `into_inner()` instead of unwrapping) and use it on all runtime/workers lifecycle locks, so one thread panicking cannot cascade into a `PoisonError` on every other thread that later takes the lock. Adds tests: drop-without-shutdown detaches cleanly, and shutdown recovers from a poisoned lock. Co-Authored-By: Claude Opus 4.8 (1M context) --- libdd-common/src/lib.rs | 16 +++ .../src/shared_runtime/fork_safe.rs | 128 +++++++++++++++--- 2 files changed, 123 insertions(+), 21 deletions(-) diff --git a/libdd-common/src/lib.rs b/libdd-common/src/lib.rs index b10df80d75..5edefe3ede 100644 --- a/libdd-common/src/lib.rs +++ b/libdd-common/src/lib.rs @@ -79,6 +79,17 @@ pub mod unix_utils; /// ``` pub trait MutexExt { fn lock_or_panic(&self) -> MutexGuard<'_, T>; + + /// Acquires the lock, recovering the guard if the mutex is poisoned instead of + /// panicking. + /// + /// Poisoning only means a previous holder panicked while holding the lock; the + /// data itself is still structurally valid to access. Use this on teardown, fork, + /// and shutdown paths where one thread panicking (e.g. during interpreter + /// finalization) must not cascade into every other thread that later takes the + /// same lock. Prefer [`MutexExt::lock_or_panic`] on hot paths where a poisoned + /// lock genuinely indicates a bug that should surface. + fn lock_or_recover(&self) -> MutexGuard<'_, T>; } impl MutexExt for Mutex { @@ -88,6 +99,11 @@ impl MutexExt for Mutex { #[allow(clippy::unwrap_used)] self.lock().unwrap() } + + #[inline(always)] + fn lock_or_recover(&self) -> MutexGuard<'_, T> { + self.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } } /// Extension trait for `RwLock` to provide methods that acquire read/write locks, panicking if diff --git a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs index 1d7f41dbab..bfde12d685 100644 --- a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs +++ b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs @@ -51,11 +51,11 @@ impl ForkSafeRuntime { /// Pauses all workers before `fork()`. Worker pause errors are logged, not propagated. pub fn before_fork(&self) { debug!("before_fork: pausing all workers"); - let mut runtime_lock = self.runtime.lock_or_panic(); + let mut runtime_lock = self.runtime.lock_or_recover(); let Some(runtime) = runtime_lock.take() else { return; }; - let mut workers_lock = self.workers.lock_or_panic(); + let mut workers_lock = self.workers.lock_or_recover(); runtime.block_on(async { let futures: FuturesUnordered<_> = workers_lock .iter_mut() @@ -71,7 +71,7 @@ impl ForkSafeRuntime { } fn restart_runtime(&self) -> Result<(), SharedRuntimeError> { - let mut runtime_lock = self.runtime.lock_or_panic(); + let mut runtime_lock = self.runtime.lock_or_recover(); if runtime_lock.is_none() { *runtime_lock = Some(Arc::new(build_runtime(self.worker_threads)?)); } @@ -83,7 +83,7 @@ impl ForkSafeRuntime { debug!("after_fork_parent: restarting runtime and workers"); self.restart_runtime()?; - let runtime_lock = self.runtime.lock_or_panic(); + let runtime_lock = self.runtime.lock_or_recover(); let handle = runtime_lock .as_ref() .ok_or(SharedRuntimeError::RuntimeUnavailable)? @@ -91,7 +91,7 @@ impl ForkSafeRuntime { .clone(); drop(runtime_lock); - let mut workers_lock = self.workers.lock_or_panic(); + let mut workers_lock = self.workers.lock_or_recover(); for worker_entry in workers_lock.iter_mut() { worker_entry.worker.start(tokio_spawn_fn(&handle))?; @@ -107,7 +107,7 @@ impl ForkSafeRuntime { debug!("after_fork_child: reinitializing runtime and workers"); self.restart_runtime()?; - let runtime_lock = self.runtime.lock_or_panic(); + let runtime_lock = self.runtime.lock_or_recover(); let handle = runtime_lock .as_ref() .ok_or(SharedRuntimeError::RuntimeUnavailable)? @@ -115,7 +115,7 @@ impl ForkSafeRuntime { .clone(); drop(runtime_lock); - let mut workers_lock = self.workers.lock_or_panic(); + let mut workers_lock = self.workers.lock_or_recover(); workers_lock.retain(|entry| entry.restart_on_fork); @@ -129,20 +129,23 @@ impl ForkSafeRuntime { /// Shuts down all workers synchronously. Returns `ShutdownTimedOut` if `timeout` is /// exceeded. + /// + /// This is the *graceful* path: it drives each worker's shutdown to completion so + /// in-flight work (e.g. a final trace flush) is not lost. It requires a live Tokio + /// context, so it is only attempted when one is available — see [`Self::can_block_on`]. + /// When it is not (interpreter finalization), teardown is left to [`Drop`], which + /// detaches the runtime without blocking. This split means neither path depends on + /// the destruction order of any other global/TLS state. pub fn shutdown(&self, timeout: Option) -> Result<(), SharedRuntimeError> { debug!(?timeout, "Shutting down ForkSafeRuntime"); - // block_on calls context::enter() which accesses Tokio's CONTEXT thread-local. - // During CPython interpreter finalization, TLS is destroyed before atexit handlers - // fire, causing a panic. Detect this via try_current() and bail out early — - // the OS will clean up remaining threads on process exit. - if matches!( - tokio::runtime::Handle::try_current(), - Err(ref e) if e.is_thread_local_destroyed() - ) { - debug!("Tokio TLS destroyed during interpreter finalization, skipping shutdown"); + if !Self::can_block_on() { + // block_on -> context::enter() would access Tokio's CONTEXT thread-local, + // which is already destroyed during (embedded) interpreter finalization. + // Skip the graceful path; Drop detaches the runtime safely. + debug!("No live Tokio context (finalization); deferring teardown to Drop"); return Ok(()); } - match self.runtime.lock_or_panic().take() { + match self.runtime.lock_or_recover().take() { Some(runtime) => { if let Some(timeout) = timeout { match runtime.block_on(async { @@ -161,6 +164,19 @@ impl ForkSafeRuntime { } } + /// Whether the current thread can safely enter a Tokio context (i.e. call `block_on`). + /// + /// Returns `false` only when the Tokio CONTEXT thread-local has been *destroyed*, + /// which happens during interpreter/thread finalization. A merely-missing context + /// (the normal case for a thread that never entered a runtime) still returns `true`, + /// because `block_on` establishes its own context in that case. + fn can_block_on() -> bool { + !matches!( + tokio::runtime::Handle::try_current(), + Err(ref e) if e.is_thread_local_destroyed() + ) + } + fn push_worker( &self, workers_guard: &mut std::sync::MutexGuard>, @@ -180,6 +196,34 @@ impl ForkSafeRuntime { } } +impl Drop for ForkSafeRuntime { + /// Terminal teardown for the owned Tokio runtime. + /// + /// A normal `Runtime` drop blocks the current thread to join its worker threads, + /// and that join path touches the Tokio CONTEXT thread-local. During (embedded) + /// interpreter finalization — e.g. a uWSGI worker exiting — that thread-local may + /// already be destroyed, so a normal drop panics with "The Tokio context + /// thread-local variable has been destroyed" (and, via a poisoned mutex, cascades). + /// + /// `shutdown_background` instead signals shutdown and returns immediately, detaching + /// the worker threads; the OS reclaims them on process exit. This is done + /// *unconditionally* — we do not try to detect whether we are in finalization — + /// so there is no ordering-dependent branch that a future teardown scenario can + /// slip past. If `shutdown` already took the runtime (the graceful path ran), this + /// is a no-op. + fn drop(&mut self) { + if let Some(runtime) = self.runtime.lock_or_recover().take() { + match Arc::try_unwrap(runtime) { + Ok(runtime) => runtime.shutdown_background(), + // Another Arc clone is still alive (e.g. an in-flight block_on holds + // one). We cannot take ownership to detach; drop our reference and let + // the last owner's drop run. This is not the finalization path. + Err(_runtime) => {} + } + } + } +} + impl SharedRuntime for ForkSafeRuntime { fn new() -> Result { Self::with_worker_threads(1) @@ -197,8 +241,8 @@ impl SharedRuntime for ForkSafeRuntime { // Hold both locks together (runtime → workers, per struct lock order) so // before_fork cannot interleave between start and push. If runtime is already // None (fork window), skip start; after_fork_* will pick it up. - let runtime_guard = self.runtime.lock_or_panic(); - let mut workers_guard = self.workers.lock_or_panic(); + let runtime_guard = self.runtime.lock_or_recover(); + let mut workers_guard = self.workers.lock_or_recover(); if let Some(rt) = runtime_guard.as_ref() { pausable_worker.start(tokio_spawn_fn(rt.handle()))?; @@ -210,7 +254,7 @@ impl SharedRuntime for ForkSafeRuntime { async fn shutdown_async(&self) { debug!("Shutting down all workers asynchronously"); let workers = { - let mut workers_lock = self.workers.lock_or_panic(); + let mut workers_lock = self.workers.lock_or_recover(); std::mem::take(&mut *workers_lock) }; @@ -232,7 +276,7 @@ impl SharedRuntime for ForkSafeRuntime { impl BlockingRuntime for ForkSafeRuntime { /// Falls back to a temporary current-thread runtime in the fork window. fn block_on(&self, f: F) -> Result { - let runtime = match self.runtime.lock_or_panic().as_ref() { + let runtime = match self.runtime.lock_or_recover().as_ref() { None => Arc::new(Builder::new_current_thread().enable_all().build()?), Some(runtime) => runtime.clone(), }; @@ -452,4 +496,46 @@ mod tests { "second shutdown must not panic" ); } + + #[test] + fn test_drop_without_shutdown_detaches_cleanly() { + // Dropping a ForkSafeRuntime that still owns a live runtime + worker (i.e. + // shutdown() was never called) must not panic or hang. Drop routes through + // shutdown_background(), which detaches worker threads without blocking or + // entering the Tokio context — the structural terminal-teardown path that + // makes finalization (destroyed-TLS) drops safe. + let shared_runtime = ForkSafeRuntime::new().unwrap(); + let (worker, receiver) = make_test_worker(); + + let _ = shared_runtime.spawn_worker(worker, true).unwrap(); + receiver + .recv_timeout(Duration::from_secs(1)) + .expect("worker did not run"); + + // No explicit shutdown() — just drop. Must return promptly without panicking. + drop(shared_runtime); + } + + #[test] + fn test_lock_recovers_from_poison() { + // A panic while holding one of the runtime's locks must not cascade into + // subsequent lifecycle calls (the original PoisonError second-panic). After a + // poisoning panic, shutdown()/Drop must still succeed. + use std::panic::{catch_unwind, AssertUnwindSafe}; + + let shared_runtime = ForkSafeRuntime::new().unwrap(); + let (worker, _receiver) = make_test_worker(); + let _ = shared_runtime.spawn_worker(worker, true).unwrap(); + + let _ = catch_unwind(AssertUnwindSafe(|| { + let _guard = shared_runtime.workers.lock_or_recover(); + panic!("poison the workers mutex while holding it"); + })); + + // The mutex is now poisoned; lock_or_recover must still work rather than panic. + assert!( + shared_runtime.shutdown(None).is_ok(), + "shutdown must recover from a poisoned lock" + ); + } } From 302c31d550d9821da4e16dd8621e1ce807801b51 Mon Sep 17 00:00:00 2001 From: quinna-h Date: Fri, 10 Jul 2026 11:40:46 -0400 Subject: [PATCH 4/7] fix(shared-runtime): flush workers from a helper thread when TLS is destroyed shutdown() previously skipped the graceful shutdown entirely once the calling thread's Tokio CONTEXT thread-local was destroyed (embedded interpreter finalization), deferring to Drop's shutdown_background(), which detaches without flushing in-flight work. Since only the calling thread's TLS is affected, not the runtime's own worker threads, shutdown() now hands the same graceful shutdown off to a freshly spawned thread whose TLS was never touched, and waits for it via a plain channel. This lets a final flush actually happen during finalization instead of being silently dropped. --- .../src/shared_runtime/fork_safe.rs | 177 +++++++++++++----- 1 file changed, 132 insertions(+), 45 deletions(-) diff --git a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs index bfde12d685..6ef9c8bb4a 100644 --- a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs +++ b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs @@ -7,6 +7,7 @@ use libdd_common::MutexExt; use std::io; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; +use std::time::Duration; use tokio::runtime::{Builder, Runtime}; use tracing::{debug, error}; @@ -130,37 +131,75 @@ impl ForkSafeRuntime { /// Shuts down all workers synchronously. Returns `ShutdownTimedOut` if `timeout` is /// exceeded. /// - /// This is the *graceful* path: it drives each worker's shutdown to completion so - /// in-flight work (e.g. a final trace flush) is not lost. It requires a live Tokio - /// context, so it is only attempted when one is available — see [`Self::can_block_on`]. - /// When it is not (interpreter finalization), teardown is left to [`Drop`], which - /// detaches the runtime without blocking. This split means neither path depends on - /// the destruction order of any other global/TLS state. - pub fn shutdown(&self, timeout: Option) -> Result<(), SharedRuntimeError> { + /// This always drives each worker's shutdown to completion so in-flight work (e.g. a + /// final trace flush) is not lost. Normally that happens by calling `block_on` on the + /// current thread. But `block_on` needs a live Tokio context, which requires the + /// calling thread's CONTEXT thread-local — and during (embedded) interpreter + /// finalization that thread-local may already be destroyed (see + /// [`Self::can_block_on`]). The runtime's own worker threads are unaffected by that; + /// only the *calling* thread's TLS is gone. So in that case we hand the same graceful + /// shutdown off to a freshly spawned thread, whose TLS was never touched, and just wait + /// for it to finish (bounded by `timeout`, if given) using plain OS-level + /// synchronization that doesn't touch Tokio's TLS. + pub fn shutdown(&self, timeout: Option) -> Result<(), SharedRuntimeError> { debug!(?timeout, "Shutting down ForkSafeRuntime"); - if !Self::can_block_on() { - // block_on -> context::enter() would access Tokio's CONTEXT thread-local, - // which is already destroyed during (embedded) interpreter finalization. - // Skip the graceful path; Drop detaches the runtime safely. - debug!("No live Tokio context (finalization); deferring teardown to Drop"); + let Some(runtime) = self.runtime.lock_or_recover().take() else { return Ok(()); + }; + + if Self::can_block_on() { + return Self::block_on_shutdown(&runtime, &self.workers, timeout); } - match self.runtime.lock_or_recover().take() { - Some(runtime) => { - if let Some(timeout) = timeout { - match runtime.block_on(async { - tokio::time::timeout(timeout, ::shutdown_async(self)) - .await - }) { - Ok(()) => Ok(()), - Err(_) => Err(SharedRuntimeError::ShutdownTimedOut(timeout)), - } - } else { - runtime.block_on(::shutdown_async(self)); - Ok(()) + + debug!("No live Tokio context (finalization); flushing from a helper thread"); + Self::shutdown_from_helper_thread(runtime, self.workers.clone(), timeout) + } + + /// Drives `shutdown_workers` to completion via `runtime.block_on` on the calling + /// thread. Only safe to call when [`Self::can_block_on`] is true. + fn block_on_shutdown( + runtime: &Runtime, + workers: &Arc>>, + timeout: Option, + ) -> Result<(), SharedRuntimeError> { + match timeout { + Some(timeout) => { + match runtime.block_on(async { + tokio::time::timeout(timeout, shutdown_workers(workers)).await + }) { + Ok(()) => Ok(()), + Err(_) => Err(SharedRuntimeError::ShutdownTimedOut(timeout)), } } - None => Ok(()), + None => { + runtime.block_on(shutdown_workers(workers)); + Ok(()) + } + } + } + + /// Runs [`Self::block_on_shutdown`] on a fresh, detached thread (whose Tokio TLS was + /// never touched) and blocks the calling thread on its completion via a channel — + /// plain OS-level synchronization, safe even with the calling thread's Tokio TLS gone. + /// + /// A missing `timeout` waits indefinitely for completion, matching the behavior of + /// [`Self::block_on_shutdown`] itself when called with `None`. + fn shutdown_from_helper_thread( + runtime: Arc, + workers: Arc>>, + timeout: Option, + ) -> Result<(), SharedRuntimeError> { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let result = Self::block_on_shutdown(&runtime, &workers, timeout); + let _ = done_tx.send(result); + }); + + match timeout { + Some(timeout) => done_rx + .recv_timeout(timeout) + .unwrap_or(Err(SharedRuntimeError::ShutdownTimedOut(timeout))), + None => done_rx.recv().unwrap_or(Ok(())), } } @@ -209,8 +248,13 @@ impl Drop for ForkSafeRuntime { /// the worker threads; the OS reclaims them on process exit. This is done /// *unconditionally* — we do not try to detect whether we are in finalization — /// so there is no ordering-dependent branch that a future teardown scenario can - /// slip past. If `shutdown` already took the runtime (the graceful path ran), this - /// is a no-op. + /// slip past. + /// + /// [`Self::shutdown`] always takes the runtime out (even during finalization, via a + /// helper thread — see its docs), so this only fires when `shutdown` was never called + /// at all before drop; that caller-error case is the one scenario this crate can't + /// flush on behalf of the caller, so it just detaches cleanly instead of risking a + /// panic. fn drop(&mut self) { if let Some(runtime) = self.runtime.lock_or_recover().take() { match Arc::try_unwrap(runtime) { @@ -252,25 +296,33 @@ impl SharedRuntime for ForkSafeRuntime { } async fn shutdown_async(&self) { - debug!("Shutting down all workers asynchronously"); - let workers = { - let mut workers_lock = self.workers.lock_or_recover(); - std::mem::take(&mut *workers_lock) - }; + shutdown_workers(&self.workers).await; + } +} - let futures: FuturesUnordered<_> = workers - .into_iter() - .map(|mut worker_entry| async move { - if let Err(e) = worker_entry.worker.pause().await { - error!("Worker failed to shutdown: {:?}", e); - return; - } - worker_entry.worker.shutdown().await; - }) - .collect(); +/// Drains and gracefully shuts down every registered worker. Standalone (rather than a +/// `&self` method) so it can run identically whether driven from the calling thread +/// ([`ForkSafeRuntime::block_on_shutdown`]) or from the helper thread spawned by +/// [`ForkSafeRuntime::shutdown_from_helper_thread`]. +async fn shutdown_workers(workers: &Mutex>) { + debug!("Shutting down all workers asynchronously"); + let workers = { + let mut workers_lock = workers.lock_or_recover(); + std::mem::take(&mut *workers_lock) + }; + + let futures: FuturesUnordered<_> = workers + .into_iter() + .map(|mut worker_entry| async move { + if let Err(e) = worker_entry.worker.pause().await { + error!("Worker failed to shutdown: {:?}", e); + return; + } + worker_entry.worker.shutdown().await; + }) + .collect(); - futures.collect::<()>().await; - } + futures.collect::<()>().await; } impl BlockingRuntime for ForkSafeRuntime { @@ -497,6 +549,41 @@ mod tests { ); } + #[test] + fn test_shutdown_from_helper_thread_flushes_workers() { + // Exercises the destroyed-TLS fallback mechanism directly (this test's thread has + // a perfectly fine Tokio context, but shutdown_from_helper_thread is the exact + // code path shutdown() delegates to when the calling thread's context is gone). + // Confirms the worker's shutdown() -- the "final flush" -- actually runs to + // completion on the helper thread, not just that teardown avoids a panic. + let shared_runtime = ForkSafeRuntime::new().unwrap(); + let (worker, receiver) = make_test_worker(); + + let _ = shared_runtime.spawn_worker(worker, true).unwrap(); + receiver + .recv_timeout(Duration::from_secs(1)) + .expect("worker did not run"); + + let runtime = shared_runtime.runtime.lock_or_recover().take().unwrap(); + let result = ForkSafeRuntime::shutdown_from_helper_thread( + runtime, + shared_runtime.workers.clone(), + Some(Duration::from_secs(1)), + ); + assert!(result.is_ok()); + + let mut last = receiver + .recv_timeout(Duration::from_secs(1)) + .expect("helper thread did not flush worker shutdown"); + while let Ok(v) = receiver.try_recv() { + last = v; + } + assert_eq!( + last, -1, + "worker shutdown (the flush) must actually run on the helper thread" + ); + } + #[test] fn test_drop_without_shutdown_detaches_cleanly() { // Dropping a ForkSafeRuntime that still owns a live runtime + worker (i.e. From 0ab449081a8850143cfcb89b401e71792b9a2097 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Mon, 13 Jul 2026 16:32:05 -0400 Subject: [PATCH 5/7] fix(shared-runtime): scope lock_or_recover to terminal teardown only Address review feedback on #2220: lock_or_recover was applied to every lock site (fork handling, spawn, restart), silently continuing past a poisoned lock even where poisoning indicates a real bug. Restrict it to shutdown()/Drop/shutdown_workers, the terminal paths where recovering from poison is the point; lifecycle operations keep lock_or_panic. --- libdd-common/src/lib.rs | 12 +++-------- .../src/shared_runtime/fork_safe.rs | 20 +++++++++---------- 2 files changed, 13 insertions(+), 19 deletions(-) diff --git a/libdd-common/src/lib.rs b/libdd-common/src/lib.rs index 5edefe3ede..523e395b53 100644 --- a/libdd-common/src/lib.rs +++ b/libdd-common/src/lib.rs @@ -80,15 +80,9 @@ pub mod unix_utils; pub trait MutexExt { fn lock_or_panic(&self) -> MutexGuard<'_, T>; - /// Acquires the lock, recovering the guard if the mutex is poisoned instead of - /// panicking. - /// - /// Poisoning only means a previous holder panicked while holding the lock; the - /// data itself is still structurally valid to access. Use this on teardown, fork, - /// and shutdown paths where one thread panicking (e.g. during interpreter - /// finalization) must not cascade into every other thread that later takes the - /// same lock. Prefer [`MutexExt::lock_or_panic`] on hot paths where a poisoned - /// lock genuinely indicates a bug that should surface. + /// Acquires the lock, recovering the guard instead of panicking if poisoned. + /// Reserve for terminal teardown (`shutdown`/`Drop`); elsewhere prefer + /// [`MutexExt::lock_or_panic`] since poisoning there indicates a real bug. fn lock_or_recover(&self) -> MutexGuard<'_, T>; } diff --git a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs index 6ef9c8bb4a..4d2f666192 100644 --- a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs +++ b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs @@ -52,11 +52,11 @@ impl ForkSafeRuntime { /// Pauses all workers before `fork()`. Worker pause errors are logged, not propagated. pub fn before_fork(&self) { debug!("before_fork: pausing all workers"); - let mut runtime_lock = self.runtime.lock_or_recover(); + let mut runtime_lock = self.runtime.lock_or_panic(); let Some(runtime) = runtime_lock.take() else { return; }; - let mut workers_lock = self.workers.lock_or_recover(); + let mut workers_lock = self.workers.lock_or_panic(); runtime.block_on(async { let futures: FuturesUnordered<_> = workers_lock .iter_mut() @@ -72,7 +72,7 @@ impl ForkSafeRuntime { } fn restart_runtime(&self) -> Result<(), SharedRuntimeError> { - let mut runtime_lock = self.runtime.lock_or_recover(); + let mut runtime_lock = self.runtime.lock_or_panic(); if runtime_lock.is_none() { *runtime_lock = Some(Arc::new(build_runtime(self.worker_threads)?)); } @@ -84,7 +84,7 @@ impl ForkSafeRuntime { debug!("after_fork_parent: restarting runtime and workers"); self.restart_runtime()?; - let runtime_lock = self.runtime.lock_or_recover(); + let runtime_lock = self.runtime.lock_or_panic(); let handle = runtime_lock .as_ref() .ok_or(SharedRuntimeError::RuntimeUnavailable)? @@ -92,7 +92,7 @@ impl ForkSafeRuntime { .clone(); drop(runtime_lock); - let mut workers_lock = self.workers.lock_or_recover(); + let mut workers_lock = self.workers.lock_or_panic(); for worker_entry in workers_lock.iter_mut() { worker_entry.worker.start(tokio_spawn_fn(&handle))?; @@ -108,7 +108,7 @@ impl ForkSafeRuntime { debug!("after_fork_child: reinitializing runtime and workers"); self.restart_runtime()?; - let runtime_lock = self.runtime.lock_or_recover(); + let runtime_lock = self.runtime.lock_or_panic(); let handle = runtime_lock .as_ref() .ok_or(SharedRuntimeError::RuntimeUnavailable)? @@ -116,7 +116,7 @@ impl ForkSafeRuntime { .clone(); drop(runtime_lock); - let mut workers_lock = self.workers.lock_or_recover(); + let mut workers_lock = self.workers.lock_or_panic(); workers_lock.retain(|entry| entry.restart_on_fork); @@ -285,8 +285,8 @@ impl SharedRuntime for ForkSafeRuntime { // Hold both locks together (runtime → workers, per struct lock order) so // before_fork cannot interleave between start and push. If runtime is already // None (fork window), skip start; after_fork_* will pick it up. - let runtime_guard = self.runtime.lock_or_recover(); - let mut workers_guard = self.workers.lock_or_recover(); + let runtime_guard = self.runtime.lock_or_panic(); + let mut workers_guard = self.workers.lock_or_panic(); if let Some(rt) = runtime_guard.as_ref() { pausable_worker.start(tokio_spawn_fn(rt.handle()))?; @@ -328,7 +328,7 @@ async fn shutdown_workers(workers: &Mutex>) { impl BlockingRuntime for ForkSafeRuntime { /// Falls back to a temporary current-thread runtime in the fork window. fn block_on(&self, f: F) -> Result { - let runtime = match self.runtime.lock_or_recover().as_ref() { + let runtime = match self.runtime.lock_or_panic().as_ref() { None => Arc::new(Builder::new_current_thread().enable_all().build()?), Some(runtime) => runtime.clone(), }; From 0af3e0b13427b923c61b3e589e91da11b51364cf Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Mon, 13 Jul 2026 17:11:08 -0400 Subject: [PATCH 6/7] fix(shared-runtime): give MutexExt::lock_or_recover a default impl CI's semver-checks flagged lock_or_recover as a breaking change: adding a method to a non-sealed public trait with no default breaks existing implementors. Give it a default that delegates to lock_or_panic so the trait addition is non-breaking; Mutex's impl still overrides it with the actual poison-recovery behavior. --- libdd-common/src/lib.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/libdd-common/src/lib.rs b/libdd-common/src/lib.rs index 523e395b53..295889ee17 100644 --- a/libdd-common/src/lib.rs +++ b/libdd-common/src/lib.rs @@ -83,7 +83,13 @@ pub trait MutexExt { /// Acquires the lock, recovering the guard instead of panicking if poisoned. /// Reserve for terminal teardown (`shutdown`/`Drop`); elsewhere prefer /// [`MutexExt::lock_or_panic`] since poisoning there indicates a real bug. - fn lock_or_recover(&self) -> MutexGuard<'_, T>; + /// + /// Default implementation just delegates to `lock_or_panic` (a safe, behavior- + /// preserving default for existing implementors); `Mutex`'s impl below is the + /// one that actually recovers. + fn lock_or_recover(&self) -> MutexGuard<'_, T> { + self.lock_or_panic() + } } impl MutexExt for Mutex { From 8dd25edb404f92a3fd51aaf458335733198ab970 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Wed, 15 Jul 2026 11:17:38 -0400 Subject: [PATCH 7/7] fix(shared-runtime): guard block_on against destroyed TLS via block_on_send ForkSafeRuntime::shutdown() already guards against the calling thread's Tokio CONTEXT thread-local being destroyed during (embedded) interpreter finalization, but BlockingRuntime::block_on() does not. TraceExporter:: shutdown() goes through block_on(), not shutdown(), so it still panics with "The Tokio context thread-local variable has been destroyed" on uWSGI worker teardown. Add a new BlockingRuntime::block_on_send method (Send + 'static bounded, defaulting to block_on) so fork-safe implementations can fall back to a helper thread when TLS is gone, mirroring shutdown_from_helper_thread. Switch only TraceExporter::shutdown() to use it; every other block_on call site is untouched. --- libdd-data-pipeline/src/trace_exporter/mod.rs | 5 +- .../src/shared_runtime/fork_safe.rs | 77 +++++++++++++++++++ .../src/shared_runtime/mod.rs | 14 ++++ 3 files changed, 94 insertions(+), 2 deletions(-) diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 5cdefa3db9..ce62f07241 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -258,10 +258,11 @@ impl< #[cfg(not(target_arch = "wasm32"))] pub fn shutdown(self, timeout: Option) -> Result<(), TraceExporterError> where - R: BlockingRuntime, + R: BlockingRuntime + Send + Sync + 'static, + Self: Send, { let runtime = self.shared_runtime.clone(); - runtime.block_on(self.shutdown_async(timeout))? + runtime.block_on_send(self.shutdown_async(timeout))? } /// Async version of [`Self::shutdown`]. diff --git a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs index 4d2f666192..e48b1f4bfc 100644 --- a/libdd-shared-runtime/src/shared_runtime/fork_safe.rs +++ b/libdd-shared-runtime/src/shared_runtime/fork_safe.rs @@ -334,6 +334,53 @@ impl BlockingRuntime for ForkSafeRuntime { }; Ok(runtime.block_on(f)) } + + /// Entering a Tokio context, or building a new runtime, touches thread-locals that + /// panic if the calling thread's TLS is already destroyed (same hazard as + /// [`Self::shutdown`]). When that's the case, run `f` on a helper thread instead. + fn block_on_send( + &self, + f: F, + ) -> Result + where + F::Output: Send + 'static, + { + if Self::can_block_on() { + return self.block_on(f); + } + + debug!("No live Tokio context (finalization); running block_on from a helper thread"); + let runtime = self.runtime.lock_or_recover().clone(); + Self::block_on_from_helper_thread(runtime, f) + } +} + +impl ForkSafeRuntime { + /// Runs `f` on a fresh thread (whose Tokio TLS was never touched) and blocks the + /// calling thread on its result via a plain OS channel, safe even with the calling + /// thread's TLS gone. Mirrors [`Self::shutdown_from_helper_thread`]. + fn block_on_from_helper_thread( + runtime: Option>, + f: F, + ) -> Result + where + F::Output: Send + 'static, + { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + std::thread::Builder::new().spawn(move || { + let result = (|| -> Result { + let runtime = match runtime { + Some(runtime) => runtime, + None => Arc::new(Builder::new_current_thread().enable_all().build()?), + }; + Ok(runtime.block_on(f)) + })(); + let _ = done_tx.send(result); + })?; + done_rx + .recv() + .map_err(|_| io::Error::other("block_on helper thread died without a result"))? + } } #[cfg(test)] @@ -625,4 +672,34 @@ mod tests { "shutdown must recover from a poisoned lock" ); } + + #[test] + fn test_block_on_send_runs_future_on_calling_thread() { + // Sanity check for the live-TLS path: block_on_send must still just drive `f` to + // completion on the calling thread and return its output, matching block_on. + let shared_runtime = ForkSafeRuntime::new().unwrap(); + let result = shared_runtime.block_on_send(async { 21 * 2 }).unwrap(); + assert_eq!(result, 42); + } + + #[test] + fn test_block_on_from_helper_thread_runs_future_and_returns_value() { + // Exercises the fallback directly; real TLS destruction can't be triggered + // deterministically from a unit test. Mirrors + // test_shutdown_from_helper_thread_flushes_workers. + let shared_runtime = ForkSafeRuntime::new().unwrap(); + let runtime = shared_runtime.runtime.lock_or_recover().clone(); + + let result = ForkSafeRuntime::block_on_from_helper_thread(runtime, async { 1 + 1 }); + assert_eq!(result.unwrap(), 2); + } + + #[test] + fn test_block_on_from_helper_thread_builds_runtime_when_none() { + // Same as above but for the (runtime already taken) case block_on's own None + // branch handles unsafely — the helper thread must build its own fresh runtime + // instead, since building on the *calling* thread is what panics once TLS is gone. + let result = ForkSafeRuntime::block_on_from_helper_thread(None, async { 40 + 2 }); + assert_eq!(result.unwrap(), 42); + } } diff --git a/libdd-shared-runtime/src/shared_runtime/mod.rs b/libdd-shared-runtime/src/shared_runtime/mod.rs index 852283179f..f99d2cddef 100644 --- a/libdd-shared-runtime/src/shared_runtime/mod.rs +++ b/libdd-shared-runtime/src/shared_runtime/mod.rs @@ -82,6 +82,20 @@ pub trait BlockingRuntime: SharedRuntime { /// /// Returns an [`io::Error`] if the executor cannot be accessed or constructed. fn block_on(&self, f: F) -> Result; + + /// Like [`Self::block_on`], but for callers that can guarantee `F`/its output are + /// `Send + 'static`, letting fork-safe impls hand `f` off to a helper thread when the + /// calling thread can't safely enter a Tokio context (see + /// [`ForkSafeRuntime::block_on_send`]). Defaults to forwarding to [`Self::block_on`]. + fn block_on_send( + &self, + f: F, + ) -> Result + where + F::Output: Send + 'static, + { + self.block_on(f) + } } /// Handle to a worker registered on a [`SharedRuntime`].