Add regression tests and standalone repro for BatchSpanProcessor Tokio deadlock

Author: timvw

Cargo.toml

@@ -5,7 +5,7 @@ members = [
     "opentelemetry-*/examples/*",
     "opentelemetry-otlp/tests/*",
     "examples/*",
-    "stress",
+    "stress", "batch-span-processor-hang-repro",
 ]
 resolver = "2"
 # Avoid applying patch to force use of workspace members for this

batch-span-processor-hang-repro/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "batch-span-processor-hang-repro"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+opentelemetry = { workspace = true, features = ["trace"] }
+opentelemetry_sdk = { workspace = true, features = ["trace", "experimental_trace_batch_span_processor_with_async_runtime", "rt-tokio"] }
+tokio = { workspace = true, features = ["rt", "macros"] }
+futures-util = { workspace = true, features = ["std"] }
+
+[lints]
+workspace = true

batch-span-processor-hang-repro/README.md

@@ -0,0 +1,25 @@
+# Batch Span Processor Tokio deadlock reproduction
+
+This crate demonstrates how the asynchronous BatchSpanProcessor from
+`opentelemetry-sdk` deadlocks when `force_flush` is invoked from a Tokio
+current-thread runtime.
+
+## Steps
+
+```bash
+# run the example; it will print the first line and then hang
+cargo run -p batch-span-processor-hang-repro
+```
+
+The program configures `BatchSpanProcessor::builder(..., runtime::Tokio)` and
+then calls `SdkTracerProvider::force_flush()` while running inside a
+`#[tokio::main(flavor = "current_thread")]` context. The processor calls
+`futures_executor::block_on` internally, so the same Tokio thread that is blocked
+is also responsible for driving the background worker, resulting in a deadlock.
+
+To see the hang, you can run with a timeout:
+
+```bash
+# exits with status 124 because the process never completes
+timeout 5 cargo run -p batch-span-processor-hang-repro
+```

batch-span-processor-hang-repro/src/main.rs

@@ -0,0 +1,39 @@
+use futures_util::future;
+use opentelemetry::trace::{Span, Tracer, TracerProvider};
+use opentelemetry_sdk::error::OTelSdkResult;
+use opentelemetry_sdk::runtime;
+use opentelemetry_sdk::trace::span_processor_with_async_runtime::BatchSpanProcessor;
+use opentelemetry_sdk::trace::{SdkTracerProvider, SpanData, SpanExporter};
+
+#[derive(Debug)]
+struct ReadyExporter;
+
+impl SpanExporter for ReadyExporter {
+    fn export(
+        &self,
+        _batch: Vec<SpanData>,
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
+        future::ready(Ok(()))
+    }
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    // Build a provider that uses the async-runtime batch span processor with the Tokio runtime.
+    let batch_processor = BatchSpanProcessor::builder(ReadyExporter, runtime::Tokio).build();
+
+    let provider = SdkTracerProvider::builder()
+        .with_span_processor(batch_processor)
+        .build();
+
+    // End a span so the processor has some work queued.
+    provider.tracer("repro").start("blocking-call").end();
+
+    println!("Calling force_flush... this blocks forever");
+
+    // This never returns because force_flush uses futures_executor::block_on,
+    // which cannot make progress while we're on Tokio's current-thread scheduler.
+    provider.force_flush().expect("force_flush should succeed");
+
+    println!("force_flush returned");
+}

opentelemetry-sdk/src/trace/span_processor_with_async_runtime.rs

@@ -703,4 +703,84 @@ mod tests {
             "exports overlapped even though max_concurrent_exports was 1"
         );
     }
+
+    #[test]
+    #[cfg(feature = "rt-tokio")]
+    fn batch_span_processor_force_flush_deadlocks_helper() {
+        if std::env::var_os("OTEL_RUN_BSP_DEADLOCK_HELPER").is_none() {
+            // Running under the normal test harness: skip so the suite does not hang.
+            return;
+        }
+
+        run_force_flush_deadlock();
+    }
+
+    #[test]
+    #[cfg(feature = "rt-tokio")]
+    fn batch_span_processor_force_flush_deadlocks_current_thread_runtime() {
+        use std::{process::Command, thread, time::Duration};
+
+        let mut child = Command::new(
+            std::env::current_exe().expect("failed to locate current test binary"),
+        )
+            .arg("--exact")
+            .arg(
+                "trace::span_processor_with_async_runtime::tests::batch_span_processor_force_flush_deadlocks_helper",
+            )
+            .env("OTEL_RUN_BSP_DEADLOCK_HELPER", "1")
+            .spawn()
+            .expect("failed to spawn helper test");
+
+        // Give the helper a short window to reach the blocking call.
+        thread::sleep(Duration::from_millis(200));
+
+        let still_running = child
+            .try_wait()
+            .expect("failed to query helper status")
+            .is_none();
+
+        // Always terminate the helper before making assertions to avoid leaking child processes.
+        let _ = child.kill();
+        let _ = child.wait();
+
+        assert!(
+            still_running,
+            "helper process exited unexpectedly; force_flush no longer blocks on the current-thread runtime"
+        );
+    }
+
+    #[cfg(feature = "rt-tokio")]
+    fn run_force_flush_deadlock() -> ! {
+        use crate::error::OTelSdkResult;
+        use crate::testing::trace::new_test_export_span_data;
+        use crate::trace::{SpanData, SpanExporter};
+
+        #[derive(Debug)]
+        struct ReadyExporter;
+
+        impl SpanExporter for ReadyExporter {
+            async fn export(&self, _batch: Vec<SpanData>) -> OTelSdkResult {
+                Ok(())
+            }
+        }
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("failed to build current-thread runtime");
+
+        runtime.block_on(async {
+            // Building the processor inside the runtime ensures the worker task is spawned via
+            // `tokio::spawn`, matching application usage.
+            let processor = BatchSpanProcessor::builder(ReadyExporter, runtime::Tokio).build();
+
+            processor.on_end(new_test_export_span_data());
+
+            // Calling force_flush from within the same current-thread runtime blocks forever.
+            let _ = processor.force_flush();
+            panic!("force_flush unexpectedly returned");
+        });
+
+        unreachable!("the helper should never return once force_flush blocks");
+    }
 }