diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index b1cb0074cb..5d651a5663 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -2628,27 +2628,6 @@ impl CodexMessageProcessor { }; } - if let Some(otel_reloader) = listener_task_context.otel_reloader.as_ref() { - let otel_reload_error = otel_reloader - .reload_from_config(&config) - .err() - .map(|err| err.to_string()); - if let Some(err) = otel_reload_error { - listener_task_context - .outgoing - .send_error( - request_id, - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to reload otel config: {err}"), - data: None, - }, - ) - .await; - return; - } - } - let instruction_sources = Self::instruction_sources_from_config(&config).await; let dynamic_tools = dynamic_tools.unwrap_or_default(); let core_dynamic_tools = if dynamic_tools.is_empty() { @@ -2677,6 +2656,31 @@ impl CodexMessageProcessor { .collect() }; let core_dynamic_tool_count = core_dynamic_tools.len(); + let otel_reload = if let Some(otel_reloader) = listener_task_context.otel_reloader.as_ref() + { + let otel_reload_result = otel_reloader + .reload_from_config(&config) + .map_err(|err| err.to_string()); + match otel_reload_result { + Ok(otel_reload) => Some(otel_reload), + Err(err) => { + listener_task_context + .outgoing + .send_error( + request_id, + JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to reload otel config: {err}"), + data: None, + }, + ) + .await; + return; + } + } + } else { + None + }; match listener_task_context .thread_manager @@ -2702,6 +2706,9 @@ impl CodexMessageProcessor { .await { Ok(new_conv) => { + if let Some(otel_reload) = otel_reload { + otel_reload.commit(); + } let NewThread { thread_id, thread, diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 6bf9d1fb3b..b2ef8b8df3 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -517,13 +517,12 @@ pub async fn run_main_with_transport( let log_db_layer = log_db .clone() .map(|layer| layer.with_filter(Targets::new().with_default(Level::TRACE))); - let otel_tracing_layer = otel.as_ref().and_then(|provider| provider.tracing_layer()); - let (otel_layer, otel_reloader) = + let (otel_layer, otel_trace_layer, otel_reloader) = otel_reload::OtelReloader::new(&config, otel, default_analytics_enabled); let _ = tracing_subscriber::registry() .with(stderr_fmt) .with(otel_layer) - .with(otel_tracing_layer) + .with(otel_trace_layer) .with(feedback_layer) .with(feedback_metadata_layer) .with(log_db_layer) diff --git a/codex-rs/app-server/src/otel_reload.rs b/codex-rs/app-server/src/otel_reload.rs index 5412fd24e8..dd09b7e143 100644 --- a/codex-rs/app-server/src/otel_reload.rs +++ b/codex-rs/app-server/src/otel_reload.rs @@ -1,17 +1,21 @@ //! App-server OpenTelemetry reloading for configuration resolved after startup. //! -//! The app server installs its tracing subscriber once, but `thread/start` can -//! later load project-scoped config from the requested cwd. This module keeps -//! the installed log layer stable while swapping the underlying OTel provider -//! when that effective thread config changes. +//! The app server installs its tracing subscriber once, before `thread/start` +//! can load project-scoped config from the requested cwd. This module keeps the +//! installed OTel layers stable while initializing their provider from the first +//! effective thread config. use codex_config::types::OtelConfig; use codex_core::config::Config; use codex_otel::OtelLoggerLayer; use codex_otel::OtelProvider; +use codex_otel::OtelTraceLayer; +use codex_otel::OtelTraceLayerHandle; use std::error::Error; use std::sync::Arc; use std::sync::Mutex; +use tracing::Subscriber; +use tracing_subscriber::registry::LookupSpan; #[derive(Clone, Debug, PartialEq)] struct OtelProviderKey { @@ -23,45 +27,68 @@ struct OtelProviderKey { struct OtelReloadState { key: Option, provider: Option, + retired_providers: Vec, + initialized_from_thread_config: bool, } #[derive(Clone)] pub(crate) struct OtelReloader { logger_layer: OtelLoggerLayer, + trace_layer: OtelTraceLayerHandle, state: Arc>, default_analytics_enabled: bool, } impl OtelReloader { - pub(crate) fn new( + pub(crate) fn new( initial_config: &Config, provider: Option, default_analytics_enabled: bool, - ) -> (OtelLoggerLayer, OtelReloader) { + ) -> (OtelLoggerLayer, OtelTraceLayer, OtelReloader) + where + S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static, + { let logger_layer = OtelLoggerLayer::from_provider(provider.as_ref()); + let (trace_layer, trace_layer_handle) = OtelTraceLayer::from_provider(provider.as_ref()); ( logger_layer.clone(), + trace_layer, OtelReloader { logger_layer, + trace_layer: trace_layer_handle, state: Arc::new(Mutex::new(OtelReloadState { key: Some(provider_key(initial_config, default_analytics_enabled)), provider, + retired_providers: Vec::new(), + initialized_from_thread_config: false, })), default_analytics_enabled, }, ) } - pub(crate) fn reload_from_config(&self, config: &Config) -> Result<(), Box> { + pub(crate) fn reload_from_config( + &self, + config: &Config, + ) -> Result> { let next_key = provider_key(config, self.default_analytics_enabled); - { - let state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if state.key.as_ref() == Some(&next_key) { - return Ok(()); - } + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.key.as_ref() == Some(&next_key) { + return Ok(OtelReloadCommit::new( + Arc::clone(&self.state), + next_key, + state.key.clone(), + /*old_provider*/ None, + self.logger_layer.clone(), + self.trace_layer.clone(), + /*applied_reload*/ false, + )); + } + if state.initialized_from_thread_config { + return Err("app-server OTel config is already initialized from a different effective thread config; restart the app server to use a different project OTel config".into()); } let next_provider = codex_core::otel_init::build_provider( @@ -71,23 +98,104 @@ impl OtelReloader { self.default_analytics_enabled, )?; self.logger_layer.replace_provider(next_provider.as_ref()); - - let mut state = self - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - state.key = Some(next_key); - state.provider = next_provider; - Ok(()) + self.trace_layer.replace_provider(next_provider.as_ref()); + let previous_key = state.key.clone(); + state.key = Some(next_key.clone()); + let old_provider = std::mem::replace(&mut state.provider, next_provider); + Ok(OtelReloadCommit::new( + Arc::clone(&self.state), + next_key, + previous_key, + old_provider, + self.logger_layer.clone(), + self.trace_layer.clone(), + /*applied_reload*/ true, + )) } pub(crate) fn shutdown(&self) { self.logger_layer.replace_provider(/*provider*/ None); + self.trace_layer.shutdown(); let mut state = self .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); state.provider.take(); + state.retired_providers.clear(); + } +} + +pub(crate) struct OtelReloadCommit { + state: Arc>, + key: OtelProviderKey, + previous_key: Option, + old_provider: Option, + logger_layer: OtelLoggerLayer, + trace_layer: OtelTraceLayerHandle, + applied_reload: bool, + committed: bool, +} + +impl OtelReloadCommit { + fn new( + state: Arc>, + key: OtelProviderKey, + previous_key: Option, + old_provider: Option, + logger_layer: OtelLoggerLayer, + trace_layer: OtelTraceLayerHandle, + applied_reload: bool, + ) -> Self { + Self { + state, + key, + previous_key, + old_provider, + logger_layer, + trace_layer, + applied_reload, + committed: false, + } + } + + pub(crate) fn commit(mut self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.key.as_ref() == Some(&self.key) { + if let Some(old_provider) = self.old_provider.take() { + state.retired_providers.push(old_provider); + } + state.initialized_from_thread_config = true; + self.committed = true; + } + } +} + +impl Drop for OtelReloadCommit { + fn drop(&mut self) { + if self.committed { + return; + } + + if !self.applied_reload { + return; + } + + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !state.initialized_from_thread_config && state.key.as_ref() == Some(&self.key) { + let _abandoned_provider = + std::mem::replace(&mut state.provider, self.old_provider.take()); + state.key = self.previous_key.clone(); + self.logger_layer.replace_provider(state.provider.as_ref()); + self.trace_layer.replace_provider(state.provider.as_ref()); + } else if let Some(old_provider) = self.old_provider.take() { + state.retired_providers.push(old_provider); + } } } @@ -130,12 +238,14 @@ mod tests { .codex_home(codex_home.path().to_path_buf()) .build() .await?; - let (layer, reloader) = OtelReloader::new( + let (logger_layer, trace_layer, reloader) = OtelReloader::new( &initial_config, /*provider*/ None, /*default_analytics_enabled*/ false, ); - let subscriber = tracing_subscriber::registry().with(layer); + let subscriber = tracing_subscriber::registry() + .with(logger_layer) + .with(trace_layer); let _guard = tracing::subscriber::set_default(subscriber); let mut project_config = initial_config.clone(); @@ -146,7 +256,7 @@ mod tests { tls: None, }; - reloader.reload_from_config(&project_config)?; + reloader.reload_from_config(&project_config)?.commit(); tracing::event!( target: "codex_otel.log_only", tracing::Level::INFO, @@ -164,6 +274,113 @@ mod tests { Ok(()) } + #[tokio::test(flavor = "multi_thread")] + async fn reload_from_config_rejects_different_second_config() -> Result<(), Box> { + let server = MockServer::start().await; + let codex_home = TempDir::new()?; + let initial_config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .build() + .await?; + let (_logger_layer, _trace_layer, reloader): ( + _, + OtelTraceLayer, + _, + ) = OtelReloader::new( + &initial_config, + /*provider*/ None, + /*default_analytics_enabled*/ false, + ); + + let mut first_config = initial_config.clone(); + first_config.otel.exporter = OtelExporterKind::OtlpHttp { + endpoint: format!("{}/v1/logs", server.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + }; + reloader.reload_from_config(&first_config)?.commit(); + + let mut second_config = initial_config; + second_config.otel.exporter = OtelExporterKind::OtlpHttp { + endpoint: format!("{}/other/logs", server.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + }; + let err = match reloader.reload_from_config(&second_config) { + Ok(_) => panic!("different effective OTel config should be rejected"), + Err(err) => err, + }; + assert!( + err.to_string() + .contains("already initialized from a different effective thread config"), + "unexpected error: {err}" + ); + + reloader.shutdown(); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread")] + async fn reload_from_config_does_not_pin_until_committed() -> Result<(), Box> { + let server = MockServer::start().await; + let codex_home = TempDir::new()?; + let initial_config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .build() + .await?; + let (_logger_layer, _trace_layer, reloader): ( + _, + OtelTraceLayer, + _, + ) = OtelReloader::new( + &initial_config, + /*provider*/ None, + /*default_analytics_enabled*/ false, + ); + + let mut first_config = initial_config.clone(); + first_config.otel.exporter = OtelExporterKind::OtlpHttp { + endpoint: format!("{}/v1/logs", server.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + }; + let first_reload = reloader.reload_from_config(&first_config)?; + drop(first_reload); + { + let state = reloader + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert!(state.provider.is_none()); + assert!(state.retired_providers.is_empty()); + } + + let mut second_config = initial_config; + second_config.otel.exporter = OtelExporterKind::OtlpHttp { + endpoint: format!("{}/other/logs", server.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + }; + reloader.reload_from_config(&second_config)?.commit(); + + let err = match reloader.reload_from_config(&first_config) { + Ok(_) => panic!("different effective OTel config should be rejected after commit"), + Err(err) => err, + }; + assert!( + err.to_string() + .contains("already initialized from a different effective thread config"), + "unexpected error: {err}" + ); + + reloader.shutdown(); + Ok(()) + } + async fn wait_for_otel_logs_payload(server: &MockServer) -> Result, Box> { let body = timeout(Duration::from_secs(10), async { loop { diff --git a/codex-rs/otel/src/lib.rs b/codex-rs/otel/src/lib.rs index 854768cd39..e3033feb50 100644 --- a/codex-rs/otel/src/lib.rs +++ b/codex-rs/otel/src/lib.rs @@ -4,6 +4,7 @@ mod logger_layer; pub(crate) mod metrics; pub(crate) mod provider; pub(crate) mod trace_context; +mod trace_layer; mod otlp; mod targets; @@ -32,6 +33,8 @@ pub use crate::trace_context::set_parent_from_context; pub use crate::trace_context::set_parent_from_w3c_trace_context; pub use crate::trace_context::span_w3c_trace_context; pub use crate::trace_context::traceparent_context_from_env; +pub use crate::trace_layer::OtelTraceLayer; +pub use crate::trace_layer::OtelTraceLayerHandle; pub use codex_utils_string::sanitize_metric_tag_value; #[derive(Debug, Clone, Serialize, Display)] diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index 7850fabdf9..9933e84682 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -84,9 +84,8 @@ impl OtelProvider { Some(MetricsClient::new(config)?) }; - crate::metrics::replace_global(metrics.clone()); - if !log_enabled && !trace_enabled && metrics.is_none() { + crate::metrics::replace_global(/*metrics*/ None); debug!("No OTEL exporter enabled in settings."); return Ok(None); } @@ -109,6 +108,7 @@ impl OtelProvider { global::set_tracer_provider(provider); global::set_text_map_propagator(TraceContextPropagator::new()); } + crate::metrics::replace_global(metrics.clone()); Ok(Some(Self { logger, tracer_provider, diff --git a/codex-rs/otel/src/trace_layer.rs b/codex-rs/otel/src/trace_layer.rs new file mode 100644 index 0000000000..270a905849 --- /dev/null +++ b/codex-rs/otel/src/trace_layer.rs @@ -0,0 +1,331 @@ +//! Reloadable OpenTelemetry trace layer. +//! +//! This layer stays registered with `tracing-subscriber` while allowing callers +//! to replace the OTel tracer behind it. It is used by long-lived processes that +//! discover their effective telemetry config after subscriber initialization. + +use crate::OtelProvider; +use opentelemetry_sdk::trace::Tracer; +use std::any::TypeId; +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::RwLock; +use tracing::Event; +use tracing::Subscriber; +use tracing::span::Attributes; +use tracing::span::Id; +use tracing::span::Record; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::registry::LookupSpan; + +type TraceLayer = tracing_opentelemetry::OpenTelemetryLayer; +type ReplaceProviderFn = dyn Fn(Option<&OtelProvider>) + Send + Sync; + +#[derive(Clone, Default)] +pub struct OtelTraceLayer { + state: Arc>>, +} + +struct OtelTraceLayerState { + current: Option>>, + retired: Vec>>, + active_spans: HashMap>>, +} + +impl Default for OtelTraceLayerState { + fn default() -> Self { + Self { + current: None, + retired: Vec::new(), + active_spans: HashMap::new(), + } + } +} + +#[derive(Clone)] +pub struct OtelTraceLayerHandle { + replace_provider: Arc, + shutdown: Arc, +} + +impl OtelTraceLayerHandle { + pub fn replace_provider(&self, provider: Option<&OtelProvider>) { + (self.replace_provider)(provider); + } + + pub fn shutdown(&self) { + (self.shutdown)(); + } +} + +impl OtelTraceLayer +where + S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static, +{ + pub fn from_provider(provider: Option<&OtelProvider>) -> (Self, OtelTraceLayerHandle) { + let state = Arc::new(RwLock::new(OtelTraceLayerState { + current: trace_layer(provider), + retired: Vec::new(), + active_spans: HashMap::new(), + })); + let layer = Self { + state: Arc::clone(&state), + }; + let replace_state = Arc::clone(&state); + let shutdown_state = Arc::clone(&state); + let handle = OtelTraceLayerHandle { + replace_provider: Arc::new(move |provider| { + replace_trace_layer(&replace_state, trace_layer(provider)); + }), + shutdown: Arc::new(move || { + let mut state = shutdown_state + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.current = None; + state.retired.clear(); + state.active_spans.clear(); + }), + }; + (layer, handle) + } + + fn current_layer(&self) -> Option>> { + self.state + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .current + .clone() + } + + fn record_span_layer(&self, id: &Id, layer: Arc>) { + self.state + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .active_spans + .insert(id.clone(), layer); + } + + fn layer_for_span(&self, id: &Id) -> Option>> { + let state = self + .state + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state + .active_spans + .get(id) + .cloned() + .or_else(|| state.current.clone()) + } + + fn remove_span_layer(&self, id: &Id) -> Option>> { + let mut state = self + .state + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state + .active_spans + .remove(id) + .or_else(|| state.current.clone()) + } +} + +impl Layer for OtelTraceLayer +where + S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + if !OtelProvider::trace_export_filter(attrs.metadata()) { + return; + } + + if let Some(layer) = self.current_layer() { + layer.on_new_span(attrs, id, ctx); + self.record_span_layer(id, layer); + } + } + + fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) { + if let Some(layer) = self.layer_for_span(id) { + layer.on_record(id, values, ctx); + } + } + + fn on_follows_from(&self, id: &Id, follows: &Id, ctx: Context<'_, S>) { + if let Some(layer) = self.layer_for_span(id) { + layer.on_follows_from(id, follows, ctx); + } + } + + fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) { + if !OtelProvider::trace_export_filter(event.metadata()) { + return; + } + + if let Some(layer) = self.current_layer() { + layer.on_event(event, ctx); + } + } + + fn on_enter(&self, id: &Id, ctx: Context<'_, S>) { + if let Some(layer) = self.layer_for_span(id) { + layer.on_enter(id, ctx); + } + } + + fn on_exit(&self, id: &Id, ctx: Context<'_, S>) { + if let Some(layer) = self.layer_for_span(id) { + layer.on_exit(id, ctx); + } + } + + fn on_close(&self, id: Id, ctx: Context<'_, S>) { + if let Some(layer) = self.remove_span_layer(&id) { + layer.on_close(id, ctx); + } + } + + fn on_id_change(&self, old: &Id, new: &Id, ctx: Context<'_, S>) { + if let Some(layer) = self.layer_for_span(old) { + layer.on_id_change(old, new, ctx); + let mut state = self + .state + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(layer) = state.active_spans.remove(old) { + state.active_spans.insert(new.clone(), layer); + } + } + } + + unsafe fn downcast_raw(&self, id: TypeId) -> Option<*const ()> { + let state = self + .state + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let layer = state.current.as_ref()?; + unsafe { Layer::::downcast_raw(layer.as_ref(), id) } + } +} + +fn replace_trace_layer( + state: &RwLock>, + next_layer: Option>>, +) { + let mut state = state + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(current) = state.current.take() { + state.retired.push(current); + } + state.current = next_layer; +} + +fn trace_layer(provider: Option<&OtelProvider>) -> Option>> +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + provider + .and_then(|provider| provider.tracer.clone()) + .map(|tracer| Arc::new(tracing_opentelemetry::layer().with_tracer(tracer))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::OtelProvider; + use crate::current_span_trace_id; + use opentelemetry::trace::TracerProvider as _; + use opentelemetry_sdk::error::OTelSdkResult; + use opentelemetry_sdk::trace::SdkTracerProvider; + use opentelemetry_sdk::trace::SpanData; + use opentelemetry_sdk::trace::SpanExporter; + use std::sync::Mutex; + use tracing::trace_span; + use tracing_subscriber::prelude::*; + + #[derive(Clone, Debug, Default)] + struct RecordingSpanExporter { + spans: Arc>>, + } + + impl SpanExporter for RecordingSpanExporter { + async fn export(&self, batch: Vec) -> OTelSdkResult { + self.spans + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .extend(batch); + Ok(()) + } + } + + fn provider_with_tracer(scope: &'static str) -> (OtelProvider, Arc>>) { + let exporter = RecordingSpanExporter::default(); + let spans = Arc::clone(&exporter.spans); + let tracer_provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter) + .build(); + let tracer = tracer_provider.tracer(scope); + let provider = OtelProvider { + logger: None, + tracer_provider: Some(tracer_provider), + tracer: Some(tracer), + metrics: None, + }; + (provider, spans) + } + + #[test] + fn trace_layer_preserves_span_context_downcasting() { + let tracer_provider = SdkTracerProvider::builder().build(); + let tracer = tracer_provider.tracer("codex-otel-tests"); + let provider = OtelProvider { + logger: None, + tracer_provider: Some(tracer_provider), + tracer: Some(tracer), + metrics: None, + }; + let (layer, _handle) = OtelTraceLayer::from_provider(Some(&provider)); + let subscriber = tracing_subscriber::registry().with(layer); + let _guard = tracing::subscriber::set_default(subscriber); + + let span = trace_span!("test_span"); + let _entered = span.enter(); + let trace_id = current_span_trace_id().expect("trace id"); + + assert_eq!(trace_id.len(), 32); + assert!(trace_id.chars().all(|ch| ch.is_ascii_hexdigit())); + assert_ne!(trace_id, "00000000000000000000000000000000"); + } + + #[test] + fn span_closes_with_original_layer_after_provider_replacement() { + let (first_provider, first_spans) = provider_with_tracer("first"); + let (second_provider, second_spans) = provider_with_tracer("second"); + let (layer, handle) = OtelTraceLayer::from_provider(Some(&first_provider)); + let subscriber = tracing_subscriber::registry().with(layer); + let _guard = tracing::subscriber::set_default(subscriber); + + let span = trace_span!("old_span"); + { + let _entered = span.enter(); + } + handle.replace_provider(Some(&second_provider)); + drop(span); + + assert_eq!( + first_spans + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len(), + 1 + ); + assert_eq!( + second_spans + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len(), + 0 + ); + } +}