fix(native): run shutdown in thread in uwsgi

ddtrace/internal/native/_native.pyi

@@ -180,6 +180,15 @@ class SharedRuntime:
     def shutdown(self, timeout_ms: Optional[int] = None) -> None:
         """Gracefully shut down the shared runtime.
 
+        Args:
+            timeout_ms: Maximum time in milliseconds to wait for shutdown.
+                If None, waits indefinitely.
+        """
+        ...
+    def shutdown_in_thread(self, timeout_ms: Optional[int] = None) -> None:
+        """Gracefully shut down the shared runtime.
+        The code is run in a separate thread to bypass a stale thread local storage.
+
         Args:
             timeout_ms: Maximum time in milliseconds to wait for shutdown.
                 If None, waits indefinitely.

ddtrace/internal/native_runtime.py

@@ -1,3 +1,4 @@
+import importlib.util
 import logging
 from typing import Optional
 
@@ -28,6 +29,7 @@ def __init__(self) -> None:
         forksafe.register_after_parent(self.after_fork_parent)
         forksafe.register(self.after_fork_child)
         atexit.register(self._atexit)
+        atexit.register_on_exit_signal(self._atexit)
 
     def _atexit(self) -> None:
         try:
@@ -43,7 +45,11 @@ def shutdown(self, timeout_ms: Optional[int] = None) -> None:
                 If None, waits indefinitely — only safe if all workers have
                 already been stopped (e.g. via TraceExporter.shutdown).
         """
-        super().shutdown(timeout_ms=timeout_ms)
+        using_uwsgi = importlib.util.find_spec("uwsgi") is not None
+        if not using_uwsgi:
+            super().shutdown(timeout_ms=timeout_ms)
+        else:
+            super().shutdown_in_thread(timeout_ms=timeout_ms)
         atexit.unregister(self._atexit)
         forksafe.unregister_before_fork(self.before_fork)
         forksafe.unregister_parent(self.after_fork_parent)

ddtrace/internal/writer/writer.py

@@ -851,7 +851,9 @@ def set_test_session_token(self, token: Optional[str]) -> None:
         :param token: The test session token to use for authentication.
         """
         self._test_session_token = token
+        old_exporter = self._exporter
         self._exporter = self._create_exporter()
+        old_exporter.shutdown(3_000_000_000)
 
     def recreate(
         self,
@@ -890,7 +892,9 @@ def _downgrade(self, status, client):
         if client.ENDPOINT == "v0.5/traces":
             self._clients = [AgentWriterClientV4(self._buffer_size, self._max_payload_size)]
             self._api_version = "v0.4"
+            old_exporter = self._exporter
             self._exporter = self._create_exporter()
+            old_exporter.shutdown(3_000_000_000)
 
             # Since we have to change the encoding in this case, the payload
             # would need to be converted to the downgraded encoding before

releasenotes/notes/fix-uwsgi-panic-in-worker-8e0cffa85b25592b.yaml

@@ -0,0 +1,3 @@
+fixes:
+  - |
+    Fix crashes in uwsgi worker when exiting from SIGTERM.

src/native/data_pipeline/mod.rs

@@ -269,14 +269,6 @@ impl TraceExporterPy {
     }
 }
 
-impl Drop for TraceExporterPy {
-    fn drop(&mut self) {
-        if let Some(exporter) = self.inner.take() {
-            let _ = exporter.shutdown(Some(Duration::from_secs(3)));
-        }
-    }
-}
-
 #[pymodule]
 pub fn register_data_pipeline(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<TraceExporterBuilderPy>()?;

src/native/shared_runtime/mod.rs

@@ -1,6 +1,7 @@
 use libdd_shared_runtime::SharedRuntime;
 use pyo3::prelude::*;
 use std::sync::Arc;
+use std::thread;
 use std::time::Duration;
 
 mod exceptions;
@@ -51,6 +52,19 @@ impl SharedRuntimePy {
             .map_err(shared_runtime_error_to_pyerr)
     }
 
+    /// Shutdown the runtime in a new thread.
+    /// This is can be used when thread local storage have been destroyed.
+    fn shutdown_in_thread(&self, timeout_ms: Option<u64>) -> PyResult<()> {
+        let timeout = timeout_ms.map(Duration::from_millis);
+        let inner = self.inner.clone();
+        let result = thread::spawn(move || inner.shutdown(timeout))
+            .join()
+            .map_err(|_| {
+                pyo3::exceptions::PyRuntimeError::new_err("Failed to join shutdown thread")
+            })?;
+        result.map_err(shared_runtime_error_to_pyerr)
+    }
+
     fn debug(&self) -> String {
         format!("{:?}", self.inner)
     }