fix(core): keep Tokio runtime alive for OTel batch exporter

.github/workflows/integration.yaml

@@ -53,3 +53,40 @@ jobs:
 
       - name: ${{ matrix.suite.name }} tests
         run: make ${{ matrix.suite.target }}
+
+  # ----------------------------------------------------------------------------
+  # Non-default feature verification (integration tests)
+  # ----------------------------------------------------------------------------
+
+  experimental-features:
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: read
+    strategy:
+      fail-fast: false
+      matrix:
+        crate:
+          - praxis-tests-integration
+    name: experimental-features (${{ matrix.crate }})
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Set up Rust
+        uses: praxis-proxy/conventions/.github/actions/setup-rust@dc77dbd3a55ce96e8e31a3c8d36cea4952617dd3 # v0.1.0
+
+      - name: Cache cargo-hack
+        id: cache-hack
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
+        with:
+          path: ~/.cargo/bin/cargo-hack
+          key: cargo-hack-0.6.45-${{ runner.os }}
+
+      - name: Install cargo-hack
+        if: steps.cache-hack.outputs.cache-hit != 'true'
+        run: cargo install cargo-hack@0.6.45 --locked
+
+      - name: Test non-default features
+        run: >
+          cargo hack test --each-feature
+          --exclude-no-default-features --exclude-features default,no-mac-cert-rotation-tests
+          -p ${{ matrix.crate }}

core/src/logging.rs

@@ -33,6 +33,9 @@ pub struct TracingGuard {
     /// Tracer provider to shut down when the guard is dropped.
     #[cfg(feature = "otel")]
     provider: Option<opentelemetry_sdk::trace::SdkTracerProvider>,
+    /// Tokio runtime kept alive for the tonic gRPC exporter.
+    #[cfg(feature = "otel")]
+    _otel_runtime: Option<::tokio::runtime::Runtime>,
 }
 
 #[cfg(feature = "otel")]
@@ -135,7 +138,7 @@ fn init_with_otel(
 ) -> Result<TracingGuard, ProxyError> {
     use opentelemetry::trace::TracerProvider as _;
 
-    let provider = build_otel_provider(telemetry)?;
+    let (provider, otel_runtime) = build_otel_provider(telemetry)?;
 
     // `OpenTelemetryLayer<S>` requires `S` to match the composed subscriber type.
     // JSON and text fmt produce different types, preventing a shared binding.
@@ -164,7 +167,10 @@ fn init_with_otel(
             .init();
     }
 
-    Ok(TracingGuard { provider })
+    Ok(TracingGuard {
+        provider,
+        _otel_runtime: otel_runtime,
+    })
 }
 
 /// Initialize the layered subscriber with fmt only (no `otel` feature).
@@ -203,9 +209,15 @@ fn init_fmt_only(env_filter: tracing_subscriber::EnvFilter, json: bool) {
 #[cfg(feature = "otel")]
 fn build_otel_provider(
     config: &crate::config::TelemetryConfig,
-) -> Result<Option<opentelemetry_sdk::trace::SdkTracerProvider>, ProxyError> {
+) -> Result<
+    (
+        Option<opentelemetry_sdk::trace::SdkTracerProvider>,
+        Option<::tokio::runtime::Runtime>,
+    ),
+    ProxyError,
+> {
     let Some(endpoint) = config.otlp_endpoint.as_deref() else {
-        return Ok(None);
+        return Ok((None, None));
     };
 
     if let Ok(protocol) = std::env::var(crate::config::OTLP_PROTOCOL_ENV_VAR)
@@ -217,7 +229,13 @@ fn build_otel_provider(
         )));
     }
 
-    let exporter = build_span_exporter(endpoint, &config.otlp_headers)?;
+    let runtime = ::tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_all()
+        .build()
+        .map_err(|e| ProxyError::Config(format!("failed to create OTel runtime: {e}")))?;
+
+    let exporter = runtime.block_on(async { build_span_exporter(endpoint, &config.otlp_headers) })?;
     let batch_processor = build_batch_processor(exporter, config);
     let resource = build_otel_resource(config);
 
@@ -238,7 +256,7 @@ fn build_otel_provider(
 
     opentelemetry::global::set_tracer_provider(provider.clone());
 
-    Ok(Some(provider))
+    Ok((Some(provider), Some(runtime)))
 }
 
 // -----------------------------------------------------------------------------
@@ -751,7 +769,7 @@ telemetry:
         let config = crate::config::TelemetryConfig::default();
         let provider = build_otel_provider(&config).expect("should succeed with no endpoint");
         assert!(
-            provider.is_none(),
+            provider.0.is_none(),
             "provider should be None when no endpoint configured"
         );
     }

tests/integration/Cargo.toml

@@ -41,5 +41,6 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-
 tokio-rustls = { workspace = true }
 tokio-stream = { workspace = true }
 tokio-tungstenite = { workspace = true }
+opentelemetry-proto = { version = "0.32.0", default-features = false, features = ["gen-tonic", "trace"] }
 tonic = { workspace = true }
 tonic-prost = { workspace = true }

tests/integration/tests/suite/examples/tracing_otlp.rs

@@ -1,26 +1,12 @@
 // SPDX-License-Identifier: MIT
 // Copyright (c) 2026 Praxis Contributors
 
-//! Functional test for the OTLP tracing example config.
-//!
-//! Verifies the proxy starts and serves traffic with the `otel` feature
-//! enabled and an OTLP endpoint configured via YAML. The batch exporter
-//! buffers spans internally when no collector is reachable, so no
-//! external service is needed for this test.
-//!
-//! Note: env var fallback (`OTEL_EXPORTER_OTLP_ENDPOINT`) is tested via
-//! pure unit tests in `TelemetryConfig::resolved()`. A full integration
-//! test for env var → OTLP activation would require spawning the proxy
-//! binary as a subprocess, which is out of scope for this PR.
+//! Functional tests for the OTLP tracing example config.
 
 use std::collections::HashMap;
 
 use praxis_test_utils::{free_port, http_get, start_proxy};
 
-// -----------------------------------------------------------------------------
-// Tests
-// -----------------------------------------------------------------------------
-
 #[test]
 fn tracing_otlp() {
     let proxy_port = free_port();
@@ -30,3 +16,105 @@ fn tracing_otlp() {
     let (status, _body) = http_get(proxy.addr(), "/", None);
     assert_eq!(status, 200, "proxy with OTLP tracing config should serve requests");
 }
+
+#[cfg(feature = "otel")]
+mod collector_test {
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+
+    use opentelemetry_proto::tonic::collector::trace::v1::{
+        ExportTraceServiceRequest, ExportTraceServiceResponse,
+        trace_service_server::{TraceService, TraceServiceServer},
+    };
+    use praxis_core::config::Config;
+    use praxis_test_utils::{free_port, http_get, start_proxy};
+
+    struct FakeCollector {
+        span_count: Arc<AtomicUsize>,
+    }
+
+    #[tonic::async_trait]
+    impl TraceService for FakeCollector {
+        async fn export(
+            &self,
+            request: tonic::Request<ExportTraceServiceRequest>,
+        ) -> Result<tonic::Response<ExportTraceServiceResponse>, tonic::Status> {
+            let msg = request.into_inner();
+            let count: usize = msg
+                .resource_spans
+                .iter()
+                .flat_map(|rs| &rs.scope_spans)
+                .map(|ss| ss.spans.len())
+                .sum();
+            self.span_count.fetch_add(count, Ordering::Relaxed);
+            Ok(tonic::Response::new(ExportTraceServiceResponse {
+                partial_success: None,
+            }))
+        }
+    }
+
+    #[test]
+    fn otlp_exporter_delivers_spans() {
+        let collector_port = free_port();
+        let span_count = Arc::new(AtomicUsize::new(0));
+
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .expect("tokio runtime");
+
+        let collector = FakeCollector {
+            span_count: Arc::clone(&span_count),
+        };
+        let addr: std::net::SocketAddr = ([127, 0, 0, 1], collector_port).into();
+        rt.spawn(async move {
+            tonic::transport::Server::builder()
+                .add_service(TraceServiceServer::new(collector))
+                .serve(addr)
+                .await
+                .expect("collector server");
+        });
+
+        std::thread::sleep(std::time::Duration::from_millis(200));
+
+        let proxy_port = free_port();
+        let yaml = format!(
+            "\
+listeners:
+  - name: default
+    address: \"127.0.0.1:{proxy_port}\"
+    filter_chains: [main]
+filter_chains:
+  - name: main
+    filters:
+      - filter: request_id
+      - filter: static_response
+        status: 200
+telemetry:
+  otlp_endpoint: \"http://127.0.0.1:{collector_port}\"
+  sampling_rate: 1.0
+  batch_size: 1
+  batch_interval_secs: 1
+"
+        );
+        let config = Config::from_yaml(&yaml).expect("parse inline OTLP config");
+
+        let _tracing_guard = praxis_core::logging::init_tracing(&config).expect("init tracing with OTLP");
+
+        let proxy = start_proxy(&config);
+
+        let (status, _) = http_get(proxy.addr(), "/", None);
+        assert_eq!(status, 200);
+
+        // Wait for the batch exporter to flush (interval=1s, add margin).
+        std::thread::sleep(std::time::Duration::from_secs(3));
+
+        let received = span_count.load(Ordering::Relaxed);
+        assert!(
+            received > 0,
+            "fake collector should have received spans, got {received}"
+        );
+    }
+}