diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2d47976c66..f90ba32e77 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3846,6 +3846,7 @@ dependencies = [ "eventsource-stream", "gethostname", "http 1.4.0", + "libc", "opentelemetry", "opentelemetry-appender-tracing", "opentelemetry-otlp", @@ -3854,6 +3855,7 @@ dependencies = [ "os_info", "pretty_assertions", "reqwest 0.12.28", + "seccompiler", "serde", "serde_json", "strum_macros 0.28.0", diff --git a/codex-rs/core/src/otel_init.rs b/codex-rs/core/src/otel_init.rs index 1f36a6502f..9e99b54d19 100644 --- a/codex-rs/core/src/otel_init.rs +++ b/codex-rs/core/src/otel_init.rs @@ -80,7 +80,7 @@ pub fn build_provider( let service_name = service_name_override.unwrap_or(originator.value.as_str()); let runtime_metrics = config.features.enabled(Feature::RuntimeMetrics); - OtelProvider::from(&OtelSettings { + OtelProvider::try_new(&OtelSettings { service_name: service_name.to_string(), service_version: service_version.to_string(), codex_home: config.codex_home.to_path_buf(), diff --git a/codex-rs/otel/Cargo.toml b/codex-rs/otel/Cargo.toml index d96850616b..6efa93043d 100644 --- a/codex-rs/otel/Cargo.toml +++ b/codex-rs/otel/Cargo.toml @@ -63,3 +63,9 @@ opentelemetry_sdk = { workspace = true, features = [ "testing", ] } pretty_assertions = { workspace = true } + +[target.'cfg(unix)'.dev-dependencies] +libc = { workspace = true } + +[target.'cfg(target_os = "linux")'.dev-dependencies] +seccompiler = { workspace = true } diff --git a/codex-rs/otel/README.md b/codex-rs/otel/README.md index afae1b2f03..9deab4d511 100644 --- a/codex-rs/otel/README.md +++ b/codex-rs/otel/README.md @@ -43,7 +43,7 @@ let settings = OtelSettings { tracestate: std::collections::BTreeMap::new(), }; -if let Some(provider) = OtelProvider::from(&settings)? { +if let Some(provider) = OtelProvider::try_new(&settings)? { let registry = tracing_subscriber::registry() .with(provider.logger_layer()) .with(provider.tracing_layer()); diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index 513f3ed2b5..b7b36e30c1 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -44,6 +44,7 @@ use std::io; use std::mem::ManuallyDrop; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use std::sync::mpsc; use std::time::Duration; use tracing::debug; use tracing_subscriber::Layer; @@ -64,6 +65,7 @@ pub struct OtelProvider { pub tracer: Option, pub metrics: Option, shutdown_started: AtomicBool, + shutdown_worker: Option>, } struct ShutdownWorker { @@ -71,6 +73,11 @@ struct ShutdownWorker { completed_tx: tokio::sync::oneshot::Sender<()>, } +struct ShutdownWorkerStartup { + worker_rx: mpsc::Receiver, + ready_tx: mpsc::SyncSender<()>, +} + #[derive(Debug)] struct GlobalTracer { service_name: &'static str, @@ -102,37 +109,74 @@ impl OtelProvider { } } - /// Shuts down exporters on a detached thread within an external time budget. - pub async fn shutdown_with_timeout(self, timeout: Duration) -> io::Result<()> { - self.shutdown_with_timeout_and_spawner(timeout, |worker| { + /// Starts the detached shutdown worker before shutdown-time resource pressure. + fn prepare_shutdown_worker(&mut self) -> io::Result<()> { + self.prepare_shutdown_worker_with_spawner(|startup| { std::thread::Builder::new() .name("codex-otel-shutdown".to_string()) .spawn(move || { + if startup.ready_tx.send(()).is_err() { + return; + } + let Ok(worker) = startup.worker_rx.recv() else { + return; + }; let provider = ManuallyDrop::into_inner(worker.provider); provider.shutdown(); drop(provider); let _ = worker.completed_tx.send(()); }) }) - .await } - async fn shutdown_with_timeout_and_spawner( - self, - timeout: Duration, - spawn: F, - ) -> io::Result<()> + fn prepare_shutdown_worker_with_spawner(&mut self, spawn: F) -> io::Result<()> where - F: FnOnce(ShutdownWorker) -> io::Result>, + F: FnOnce(ShutdownWorkerStartup) -> io::Result>, { + if self.shutdown_worker.is_some() { + return Ok(()); + } + + let (worker_tx, worker_rx) = mpsc::sync_channel(/*bound*/ 1); + let (ready_tx, ready_rx) = mpsc::sync_channel(/*bound*/ 1); + let startup = ShutdownWorkerStartup { + worker_rx, + ready_tx, + }; + let _shutdown_worker = spawn(startup)?; + ready_rx.recv().map_err(|_| { + io::Error::new( + io::ErrorKind::BrokenPipe, + "telemetry shutdown worker stopped before initializing", + ) + })?; + self.shutdown_worker = Some(worker_tx); + Ok(()) + } + + /// Shuts down exporters on a prepared detached thread within a time budget. + pub async fn shutdown_with_timeout(mut self, timeout: Duration) -> io::Result<()> { + let Some(worker_tx) = self.shutdown_worker.take() else { + // Best-effort shutdown must not run a potentially blocking destructor + // when its worker could not be prepared. + let _provider = ManuallyDrop::new(self); + return Err(io::Error::new( + io::ErrorKind::NotConnected, + "telemetry shutdown worker was not initialized", + )); + }; + let (completed_tx, completed_rx) = tokio::sync::oneshot::channel(); - // A failed spawn drops its closure on the caller. Keep the provider - // from synchronously running its potentially blocking destructor. let worker = ShutdownWorker { provider: ManuallyDrop::new(self), completed_tx, }; - let _shutdown_worker = spawn(worker)?; + worker_tx.send(worker).map_err(|_| { + io::Error::new( + io::ErrorKind::BrokenPipe, + "telemetry shutdown worker stopped before receiving the provider", + ) + })?; match tokio::time::timeout(timeout, completed_rx).await { Ok(Ok(())) => Ok(()), @@ -147,7 +191,7 @@ impl OtelProvider { } } - pub fn from(settings: &OtelSettings) -> Result, Box> { + pub fn try_new(settings: &OtelSettings) -> Result, Box> { let log_enabled = !matches!(settings.exporter, OtelExporter::None); let trace_enabled = !matches!(settings.trace_exporter, OtelExporter::None); let metric_exporter = crate::config::resolve_exporter(&settings.metrics_exporter); @@ -170,7 +214,7 @@ impl OtelProvider { } crate::trace_context::validate_tracestate_entries(&settings.tracestate)?; - let mut metrics = if matches!(metric_exporter, OtelExporter::None) { + let metrics = if matches!(metric_exporter, OtelExporter::None) { None } else { let mut config = MetricsConfig::otlp( @@ -205,12 +249,22 @@ impl OtelProvider { .as_ref() .map(|provider| provider.tracer(settings.service_name.clone())); + let mut provider = Self { + logger, + tracer_provider, + tracer, + metrics, + shutdown_started: AtomicBool::default(), + shutdown_worker: None, + }; + provider.prepare_shutdown_worker()?; + crate::trace_context::set_tracestate_entries(settings.tracestate.clone())?; - if let Some(provider) = tracer_provider.clone() { - global::set_tracer_provider(provider); + if let Some(tracer_provider) = provider.tracer_provider.clone() { + global::set_tracer_provider(tracer_provider); global::set_text_map_propagator(TraceContextPropagator::new()); } - if let Some(metrics) = metrics.as_mut() { + if let Some(metrics) = provider.metrics.as_mut() { *metrics = crate::metrics::install_global(metrics.clone()); if matches!(settings.metrics_exporter, OtelExporter::Statsig) { crate::metrics::install_global_statsig_settings(StatsigMetricsSettings { @@ -218,13 +272,7 @@ impl OtelProvider { }); } } - Ok(Some(Self { - logger, - tracer_provider, - tracer, - metrics, - shutdown_started: AtomicBool::default(), - })) + Ok(Some(provider)) } pub fn logger_layer(&self) -> Option + Send + Sync> diff --git a/codex-rs/otel/src/provider_shutdown_tests.rs b/codex-rs/otel/src/provider_shutdown_tests.rs index e68c8ac879..768aa0f473 100644 --- a/codex-rs/otel/src/provider_shutdown_tests.rs +++ b/codex-rs/otel/src/provider_shutdown_tests.rs @@ -7,6 +7,8 @@ use super::SpanData; use super::SpanProcessor; use pretty_assertions::assert_eq; use std::io::ErrorKind; +#[cfg(any(target_os = "linux", target_os = "macos"))] +use std::process::Command; use std::sync::Arc; use std::sync::Condvar; use std::sync::Mutex; @@ -16,6 +18,59 @@ use std::sync::atomic::Ordering; use std::sync::mpsc; use std::time::Duration; +#[cfg(any(target_os = "linux", target_os = "macos"))] +const GUARD_PAGE_FAILURE_CHILD_TEST: &str = + "provider::shutdown_tests::bounded_shutdown_survives_worker_guard_page_failure_child"; + +#[cfg(target_os = "macos")] +static GUARD_PAGE_INJECTION_ENABLED: AtomicBool = AtomicBool::new(false); +#[cfg(target_os = "macos")] +static GUARD_PAGE_INJECTION_ARMED: AtomicBool = AtomicBool::new(false); +#[cfg(target_os = "macos")] +static GUARD_PAGE_INJECTION_OBSERVED: AtomicBool = AtomicBool::new(false); + +#[cfg(target_os = "macos")] +#[unsafe(export_name = "mprotect")] +unsafe extern "C" fn fault_injected_mprotect( + address: *mut libc::c_void, + length: usize, + protection: libc::c_int, +) -> libc::c_int { + let original_symbol = unsafe { libc::dlsym(libc::RTLD_NEXT, c"mprotect".as_ptr()) }; + let original_mprotect: unsafe extern "C" fn( + *mut libc::c_void, + usize, + libc::c_int, + ) -> libc::c_int = unsafe { std::mem::transmute(original_symbol) }; + + if GUARD_PAGE_INJECTION_ENABLED.load(Ordering::Relaxed) + && protection == libc::PROT_NONE + && length == unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as usize + { + let mut thread_name = [0; 64]; + let named_shutdown_worker = unsafe { + libc::pthread_getname_np( + libc::pthread_self(), + thread_name.as_mut_ptr(), + thread_name.len(), + ) + } == 0 + && unsafe { std::ffi::CStr::from_ptr(thread_name.as_ptr()) } + .to_bytes() + .starts_with(b"codex-otel-shut"); + + if named_shutdown_worker { + GUARD_PAGE_INJECTION_OBSERVED.store(/*val*/ true, Ordering::Relaxed); + if GUARD_PAGE_INJECTION_ARMED.load(Ordering::Relaxed) { + unsafe { *libc::__error() = libc::ENOMEM }; + return -1; + } + } + } + + unsafe { original_mprotect(address, length, protection) } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ShutdownBehavior { Complete, @@ -105,6 +160,7 @@ fn test_provider(behavior: ShutdownBehavior) -> TestProvider { tracer: None, metrics: None, shutdown_started: AtomicBool::default(), + shutdown_worker: None, }, state, started, @@ -115,39 +171,201 @@ fn test_provider(behavior: ShutdownBehavior) -> TestProvider { #[tokio::test(flavor = "current_thread")] async fn bounded_shutdown_does_not_flush_when_worker_creation_fails() { let TestProvider { - provider, + mut provider, state, started, completed, } = test_provider(ShutdownBehavior::Complete); - let result = provider - .shutdown_with_timeout_and_spawner(Duration::from_secs(/*secs*/ 1), |_worker| { - Err(std::io::Error::new( - ErrorKind::WouldBlock, - "shutdown worker could not be created", - )) - }) - .await; + let preparation = provider.prepare_shutdown_worker_with_spawner(|_startup| { + Err(std::io::Error::new( + ErrorKind::WouldBlock, + "shutdown worker could not be created", + )) + }); assert_eq!( - result.as_ref().map_err(std::io::Error::kind), + preparation.as_ref().map_err(std::io::Error::kind), Err(ErrorKind::WouldBlock) ); + + let result = provider + .shutdown_with_timeout(Duration::from_secs(/*secs*/ 1)) + .await; + assert_eq!( + result.as_ref().map_err(std::io::Error::kind), + Err(ErrorKind::NotConnected) + ); assert_eq!(state.shutdowns.load(Ordering::Relaxed), 0); assert_eq!(state.force_flushes.load(Ordering::Relaxed), 0); assert_eq!(started.try_recv(), Err(mpsc::TryRecvError::Empty)); assert_eq!(completed.try_recv(), Err(mpsc::TryRecvError::Empty)); } +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[test] +fn bounded_shutdown_survives_worker_guard_page_failure() { + let unique_suffix = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("current time follows Unix epoch") + .as_nanos(); + let temporary_directory = std::env::temp_dir().join(format!( + "codex-otel-guard-page-{}-{unique_suffix}", + std::process::id() + )); + std::fs::create_dir(&temporary_directory).expect("create fault injector directory"); + + let observed_path = temporary_directory.join("guard_page_fault.observed"); + let mut subprocess = Command::new(std::env::current_exe().expect("current test binary")); + subprocess + .arg("--exact") + .arg(GUARD_PAGE_FAILURE_CHILD_TEST) + .arg("--ignored") + .arg("--nocapture") + .arg("--test-threads=1") + .env("CODEX_OTEL_GUARD_PAGE_FAILURE_CHILD", "1") + .env("CODEX_OTEL_GUARD_PAGE_FAILURE_OBSERVED", &observed_path); + + let output = subprocess + .output() + .expect("run guard-page failure subprocess"); + let injection_was_observed = observed_path.is_file(); + + let _ = std::fs::remove_dir_all(&temporary_directory); + assert!( + output.status.success(), + "bounded telemetry shutdown crashed when its worker guard page could not be allocated\n\ + status: {}\nstdout:\n{}\nstderr:\n{}", + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + injection_was_observed, + "guard-page fault injection never became active on {}-{}", + std::env::consts::ARCH, + std::env::consts::OS + ); +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[test] +// The parent regression invokes this ignored test in a fresh subprocess with +// `--exact --ignored`, isolating fatal native-thread initialization failures. +#[ignore] +fn bounded_shutdown_survives_worker_guard_page_failure_child() { + if std::env::var_os("CODEX_OTEL_GUARD_PAGE_FAILURE_CHILD").is_none() { + return; + } + + let TestProvider { + mut provider, + state, + .. + } = test_provider(ShutdownBehavior::Complete); + + #[cfg(target_os = "macos")] + GUARD_PAGE_INJECTION_ENABLED.store(/*val*/ true, Ordering::Relaxed); + + provider + .prepare_shutdown_worker() + .expect("pre-initialize bounded shutdown worker"); + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .expect("create current-thread runtime"); + + #[cfg(target_os = "macos")] + { + assert!( + GUARD_PAGE_INJECTION_OBSERVED.load(Ordering::Relaxed), + "Rust mprotect interposer did not observe shutdown-worker guard-page setup" + ); + GUARD_PAGE_INJECTION_ARMED.store(/*val*/ true, Ordering::Relaxed); + } + + #[cfg(target_os = "linux")] + { + use seccompiler::BpfProgram; + use seccompiler::SeccompAction; + use seccompiler::SeccompCmpArgLen; + use seccompiler::SeccompCmpOp; + use seccompiler::SeccompCondition; + use seccompiler::SeccompFilter; + use seccompiler::SeccompRule; + + let page_size = + usize::try_from(unsafe { libc::sysconf(libc::_SC_PAGESIZE) }).expect("valid page size"); + let mapped_page = unsafe { + libc::mmap( + std::ptr::null_mut(), + page_size, + libc::PROT_READ | libc::PROT_WRITE, + libc::MAP_PRIVATE | libc::MAP_ANONYMOUS, + /*fd*/ -1, + /*offset*/ 0, + ) + }; + assert_ne!( + mapped_page, + libc::MAP_FAILED, + "map a page to verify guard-page fault injection: {}", + std::io::Error::last_os_error() + ); + + let protection_is_none = SeccompCondition::new( + /*arg_index*/ 2, + SeccompCmpArgLen::Dword, + SeccompCmpOp::Eq, + libc::PROT_NONE as u64, + ) + .expect("create guard-page seccomp condition"); + let rule = + SeccompRule::new(vec![protection_is_none]).expect("create guard-page seccomp rule"); + let filter = SeccompFilter::new( + std::collections::BTreeMap::from([(libc::SYS_mprotect, vec![rule])]), + SeccompAction::Allow, + SeccompAction::Errno(libc::ENOMEM as u32), + std::env::consts::ARCH + .try_into() + .expect("supported seccomp architecture"), + ) + .expect("create guard-page seccomp filter"); + let program: BpfProgram = filter + .try_into() + .expect("compile guard-page seccomp filter"); + seccompiler::apply_filter(&program).expect("install guard-page seccomp filter"); + + let protection_result = unsafe { libc::mprotect(mapped_page, page_size, libc::PROT_NONE) }; + let protection_error = std::io::Error::last_os_error(); + assert_eq!(protection_result, -1); + assert_eq!(protection_error.raw_os_error(), Some(libc::ENOMEM)); + assert_eq!(unsafe { libc::munmap(mapped_page, page_size) }, 0); + } + + let observed_path = std::env::var_os("CODEX_OTEL_GUARD_PAGE_FAILURE_OBSERVED") + .expect("guard-page fault observation path"); + std::fs::write(observed_path, "observed").expect("record guard-page fault injection"); + + runtime + .block_on(provider.shutdown_with_timeout(Duration::from_secs(/*secs*/ 1))) + .expect("bounded telemetry shutdown should not create a new native thread"); + + assert_eq!(state.shutdowns.load(Ordering::Relaxed), 1); +} + #[tokio::test(flavor = "current_thread")] async fn bounded_shutdown_times_out_without_blocking_the_runtime() { let TestProvider { - provider, + mut provider, state, started, completed, } = test_provider(ShutdownBehavior::WaitForRelease); + provider + .prepare_shutdown_worker() + .expect("pre-initialize bounded shutdown worker"); let result = provider .shutdown_with_timeout(Duration::from_millis(/*millis*/ 50)) @@ -172,11 +390,14 @@ async fn bounded_shutdown_times_out_without_blocking_the_runtime() { async fn assert_bounded_shutdown_completes() { let TestProvider { - provider, + mut provider, state, started, completed, } = test_provider(ShutdownBehavior::Complete); + provider + .prepare_shutdown_worker() + .expect("pre-initialize bounded shutdown worker"); provider .shutdown_with_timeout(Duration::from_secs(/*secs*/ 1)) diff --git a/codex-rs/otel/tests/suite/otlp_http_loopback.rs b/codex-rs/otel/tests/suite/otlp_http_loopback.rs index 3de87435c3..bcfe02b47f 100644 --- a/codex-rs/otel/tests/suite/otlp_http_loopback.rs +++ b/codex-rs/otel/tests/suite/otlp_http_loopback.rs @@ -343,7 +343,7 @@ fn otlp_http_exporter_sends_logs_to_collector() let _ = tx.send(captured); }); - let otel = OtelProvider::from(&OtelSettings { + let otel = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -402,7 +402,7 @@ fn otlp_http_exporter_sends_logs_to_collector() #[test] fn otel_provider_rejects_header_unsafe_configured_tracestate() { - let result = OtelProvider::from(&OtelSettings { + let result = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -467,7 +467,7 @@ fn otlp_http_exporter_sends_traces_to_collector() let _ = tx.send(captured); }); - let otel = OtelProvider::from(&OtelSettings { + let otel = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -612,7 +612,7 @@ async fn otlp_http_exporter_sends_traces_to_collector_with_bounded_shutdown_in_t let _ = tx.send(captured); }); - let otel = OtelProvider::from(&OtelSettings { + let otel = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -705,7 +705,7 @@ fn otlp_http_exporter_times_out_when_collector_stalls_during_bounded_shutdown() .build() .expect("build tokio runtime"); let (result, elapsed) = runtime.block_on(async move { - let otel = OtelProvider::from(&OtelSettings { + let otel = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -804,7 +804,7 @@ fn otlp_http_exporter_sends_traces_to_collector_in_current_thread_tokio_runtime( .expect("current-thread runtime"); let result = runtime.block_on(async move { - let otel = OtelProvider::from(&OtelSettings { + let otel = OtelProvider::try_new(&OtelSettings { environment: "test".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), diff --git a/codex-rs/windows-sandbox-rs/src/wfp_setup.rs b/codex-rs/windows-sandbox-rs/src/wfp_setup.rs index 351d2edd2f..688878583b 100644 --- a/codex-rs/windows-sandbox-rs/src/wfp_setup.rs +++ b/codex-rs/windows-sandbox-rs/src/wfp_setup.rs @@ -46,7 +46,7 @@ fn build_wfp_metrics_provider( // depends on this crate, so the parent process passes only the resolved // Statsig environment in the elevation payload. Other exporters are // intentionally omitted from this helper path. - OtelProvider::from(&OtelSettings { + OtelProvider::try_new(&OtelSettings { environment: otel.environment.clone(), service_name: WFP_SETUP_SERVICE_NAME.to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(),