diff --git a/codex-rs/exec-server/src/noise_relay/harness.rs b/codex-rs/exec-server/src/noise_relay/harness.rs index 66db172867..d1d6670d02 100644 --- a/codex-rs/exec-server/src/noise_relay/harness.rs +++ b/codex-rs/exec-server/src/noise_relay/harness.rs @@ -50,6 +50,11 @@ use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT; use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT_REASON; use crate::websocket_pong_watchdog::WebSocketPongWatchdog; +#[path = "harness_diagnostics.rs"] +mod diagnostics; +use diagnostics::TransportExitReason; +use diagnostics::WebsocketWriteError; + /// Values that bind one harness websocket to the intended executor registration. /// /// These fields all come from the same registry response. Keeping them together @@ -111,7 +116,7 @@ where let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY); let (disconnected_tx, disconnected_rx) = watch::channel(false); let (handshake_ready_tx, handshake_ready_rx) = oneshot::channel(); - let stream_span = tracing::debug_span!("noise_relay.stream", noise_side = "harness",); + let stream_span = tracing::debug_span!("noise_relay.stream", noise_side = "harness", stream_id); debug!( environment_id, executor_registration_id, stream_id, "Noise harness relay details" @@ -261,6 +266,7 @@ where match initiator_handshake.finish(&response) { Ok(transport) => { info!( + target: "codex_otel.trace_safe", noise_event = "handshake", noise_outcome = "ok", noise_side = "harness", @@ -333,12 +339,26 @@ where let mut pending_outbound = None; let mut force_incoming = false; let mut frames_drained_after_pong_deadline = 0usize; - 'relay: loop { + let connected_at = Instant::now(); + let mut last_read_at = connected_at; + let mut last_write_at = connected_at; + let mut write_started_at = None; + let mut write_kind = "none"; + let mut records_written = 0_u64; + let mut frames_read = 0_u64; + let mut read_reason = "none"; + let mut messages_dequeued = 0_u64; + let mut messages_written = 0_u64; + let mut pending_request: Option<(tracing::Span, Instant)> = None; + let mut heartbeat = diagnostics::HeartbeatDiagnostics::default(); + let exit_reason = 'relay: loop { // Consume a due tick before the always-ready record arm below can win // another select iteration and postpone the keepalive. if pong_watchdog.deadline().is_none() && keepalive.tick().now_or_never().is_some() { + write_started_at = Some(Instant::now()); + write_kind = "keepalive"; if let Err(error) = send_keepalive_ping( &mut websocket, &mut pong_watchdog, @@ -347,8 +367,11 @@ where .await { warn!("failed to write Noise relay keepalive ping: {error}"); - break; + break error.exit_reason(); } + last_write_at = Instant::now(); + write_started_at = None; + heartbeat.record(diagnostics::HeartbeatProgress::PingWritten, &stream_id); frames_drained_after_pong_deadline = 0; continue; } @@ -368,16 +391,10 @@ where { force_incoming = true; } else { - warn!( - noise_reason = WEBSOCKET_PONG_TIMEOUT_REASON, - "Noise harness rendezvous websocket disconnected" - ); - send_disconnected( - &incoming_tx, - &disconnected_tx, - WEBSOCKET_PONG_TIMEOUT_REASON.to_string(), - ); - return; + let _ = incoming_tx.try_send(JsonRpcConnectionEvent::Disconnected { + reason: Some(WEBSOCKET_PONG_TIMEOUT_REASON.to_string()), + }); + break TransportExitReason::PongTimeout; } } @@ -397,13 +414,36 @@ where tokio::select! { maybe_message = outgoing_rx.recv(), if pending_outbound.is_none() && !force_incoming && !pong_deadline_expired => { let Some(message) = maybe_message else { - break; + break TransportExitReason::OutgoingClosed; }; + messages_dequeued += 1; + if let codex_exec_server_protocol::JSONRPCMessage::Request(request) = &message { + // Link transport progress to the existing RPC trace without copying + // its method, ID, payload or unvalidated trace carrier into fields. + let span = tracing::info_span!("codex.exec_server.noise_request_write", stream_id); + if let Some(trace) = &request.trace { + codex_otel::set_parent_from_w3c_trace_context(&span, trace); + } + // Live JSON correlation reads OtelData before this span closes. + // Setting a remote parent alone leaves its trace ID lazy. + let _ = codex_otel::span_w3c_trace_context(&span); + tracing::event!( + name: "codex.exec_server.request_dequeued", + target: "codex_otel.trace_safe", + parent: &span, + tracing::Level::INFO, + event.name = "codex.exec_server.request_dequeued", + stream_id, + message_sequence = messages_dequeued, + outgoing_queue_depth = outgoing_rx.len(), + ); + pending_request = Some((span, Instant::now())); + } let framed = match frame_jsonrpc_message(&message) { Ok(framed) => framed, Err(error) => { warn!("failed to frame JSON-RPC payload for Noise relay: {error}"); - break; + break TransportExitReason::Framing; } }; let request_trace = match message { @@ -419,7 +459,7 @@ where Ok(seq) => seq, Err(error) => { warn!("Noise relay sequence exhausted: {error}"); - break 'relay; + break 'relay TransportExitReason::SequenceExhausted; } }; let (ciphertext, next_offset, message_complete, request_trace) = { @@ -431,7 +471,7 @@ where Ok(ciphertext) => ciphertext, Err(error) => { warn!("failed to encrypt JSON-RPC payload for Noise relay: {error}"); - break 'relay; + break 'relay TransportExitReason::Encryption; } }; ( @@ -447,6 +487,8 @@ where // combined sink and stream. A single bounded record can therefore hit the // deadline and disconnect with that Pong queued. Treat that as write // backpressure; this loop yields only between records. + write_started_at = Some(Instant::now()); + write_kind = "data"; if let Err(error) = send_websocket_message( &mut websocket, Message::Binary(encode_relay_message_frame(&frame).into()), @@ -455,9 +497,25 @@ where .await { warn!("failed to write Noise relay websocket: {error}"); - break 'relay; + break 'relay error.exit_reason(); } + last_write_at = Instant::now(); + write_started_at = None; + records_written += 1; if message_complete { + messages_written += 1; + if let Some((span, started_at)) = pending_request.take() { + tracing::event!( + name: "codex.exec_server.request_written", + target: "codex_otel.trace_safe", + parent: &span, + tracing::Level::INFO, + event.name = "codex.exec_server.request_written", + stream_id, + message_sequence = messages_dequeued, + write_elapsed_ms = started_at.elapsed().as_millis() as u64, + ); + } pending_outbound = None; } else if let Some((_framed, offset, _request_trace)) = pending_outbound.as_mut() { *offset = next_offset; @@ -467,6 +525,8 @@ where continue; } _ = keepalive.tick(), if pong_watchdog.deadline().is_none() => { + write_started_at = Some(Instant::now()); + write_kind = "keepalive"; if let Err(error) = send_keepalive_ping( &mut websocket, &mut pong_watchdog, @@ -475,15 +535,22 @@ where .await { warn!("failed to write Noise relay keepalive ping: {error}"); - break; + break error.exit_reason(); } + last_write_at = Instant::now(); + write_started_at = None; + heartbeat.record(diagnostics::HeartbeatProgress::PingWritten, &stream_id); frames_drained_after_pong_deadline = 0; } incoming_message = websocket.next() => { force_incoming = false; let Some(incoming_message) = incoming_message else { - break; + break TransportExitReason::WebsocketEnded; }; + if incoming_message.is_ok() { + last_read_at = Instant::now(); + frames_read += 1; + } // Count each completed read after expiry. If only the deadline arm // advanced this counter, reads won by a simultaneously ready incoming // arm would not count toward the 32-frame cap. @@ -499,7 +566,7 @@ where Ok(frame) => frame, Err(error) => { send_malformed(&incoming_tx, error.to_string()); - break; + break TransportExitReason::InvalidFrame; } }; if frame.stream_id != stream_id { @@ -511,7 +578,7 @@ where Ok(data) => data, Err(error) => { send_malformed(&incoming_tx, error.to_string()); - break; + break TransportExitReason::InvalidFrame; } }; if let Err(error) = receive_data( @@ -525,10 +592,10 @@ where .await { if matches!(error, ExecServerError::Closed) { - break; + break TransportExitReason::ApplicationDelivery; } send_malformed(&incoming_tx, error.to_string()); - break; + break TransportExitReason::InvalidData; } } Ok(RelayFrameBodyKind::Reset) => { @@ -539,7 +606,7 @@ where ), }, ); - break; + break TransportExitReason::RelayReset; } Ok( RelayFrameBodyKind::Ack @@ -551,12 +618,13 @@ where &incoming_tx, "Noise relay received invalid post-handshake frame".to_string(), ); - break; + break TransportExitReason::InvalidFrame; } } } - Ok(Message::Close(_)) => break, + Ok(Message::Close(_)) => break TransportExitReason::WebsocketClose, Ok(Message::Pong(_)) => { + heartbeat.record(diagnostics::HeartbeatProgress::PongReceived, &stream_id); pong_watchdog.received_pong(); frames_drained_after_pong_deadline = 0; } @@ -566,16 +634,46 @@ where &incoming_tx, "Noise relay transport expects binary protobuf frames".to_string(), ); - break; + break TransportExitReason::UnexpectedText; } Err(error) => { debug!("Noise relay websocket read failed: {error}"); - break; + read_reason = websocket_read_error_reason(&error); + break TransportExitReason::WebsocketRead; } } } } - } + }; + // Fixed reasons and local counters only. Raw errors, frames and peer-provided + // close/reset text must remain outside the trace-safe target. + tracing::event!( + name: "codex.exec_server.noise_transport_closed", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.exec_server.noise_transport_closed", + noise_side = "harness", + noise_reason = exit_reason.as_str(), + read_reason, + stream_id, + messages_dequeued, + messages_written, + pings_written = heartbeat.pings_written, + pongs_received = heartbeat.pongs_received, + connection_age_ms = connected_at.elapsed().as_millis() as u64, + last_read_age_ms = last_read_at.elapsed().as_millis() as u64, + last_write_age_ms = last_write_at.elapsed().as_millis() as u64, + write_pending = write_started_at.is_some(), + write_kind = if write_started_at.is_some() { write_kind } else { "none" }, + write_wait_ms = write_started_at.map(|at| at.elapsed().as_millis() as u64).unwrap_or(0), + pong_pending = pong_watchdog.deadline().is_some(), + pong_deadline_overdue_ms = pong_watchdog.deadline() + .map(|at| Instant::now().saturating_duration_since(at).as_millis() as u64).unwrap_or(0), + frames_read, + records_written, + outgoing_queue_depth = outgoing_rx.len(), + incoming_queue_depth = incoming_tx.max_capacity() - incoming_tx.capacity(), + ); let _ = disconnected_tx.send(true); } .instrument(stream_span)); @@ -618,15 +716,15 @@ async fn send_websocket_message( websocket: &mut T, message: Message, deadline: tokio::time::Instant, -) -> Result<(), String> +) -> Result<(), WebsocketWriteError> where T: Sink + Unpin, E: std::fmt::Display, { match tokio::time::timeout_at(deadline, websocket.send(message)).await { Ok(Ok(())) => Ok(()), - Ok(Err(error)) => Err(error.to_string()), - Err(_) => Err("websocket write timed out".to_string()), + Ok(Err(error)) => Err(WebsocketWriteError::Transport(error.to_string())), + Err(_) => Err(WebsocketWriteError::Timeout), } } @@ -634,7 +732,7 @@ async fn send_keepalive_ping( websocket: &mut T, pong_watchdog: &mut WebSocketPongWatchdog, pong_deadline: std::pin::Pin<&mut tokio::time::Sleep>, -) -> Result<(), String> +) -> Result<(), WebsocketWriteError> where T: Sink + Unpin, E: std::fmt::Display, @@ -692,13 +790,21 @@ async fn send_incoming_event( event: JsonRpcConnectionEvent, deadline: tokio::time::Instant, ) -> Result<(), ExecServerError> { + let delivery_started = Instant::now(); match tokio::time::timeout_at(deadline, incoming_tx.send(event)).await { Ok(Ok(())) => Ok(()), Ok(Err(_)) => Err(ExecServerError::Closed), Err(_) => { - warn!( + tracing::event!( + name: "codex.exec_server.noise_application_backpressure", + target: "codex_otel.trace_safe", + tracing::Level::WARN, + event.name = "codex.exec_server.noise_application_backpressure", + noise_side = "harness", noise_reason = "application_backpressure", - "Noise harness application event delivery timed out" + delivery_wait_ms = delivery_started.elapsed().as_millis() as u64, + incoming_queue_depth = incoming_tx.max_capacity() - incoming_tx.capacity(), + incoming_queue_capacity = incoming_tx.max_capacity(), ); Err(ExecServerError::Closed) } diff --git a/codex-rs/exec-server/src/noise_relay/harness_diagnostics.rs b/codex-rs/exec-server/src/noise_relay/harness_diagnostics.rs new file mode 100644 index 0000000000..a053392c76 --- /dev/null +++ b/codex-rs/exec-server/src/noise_relay/harness_diagnostics.rs @@ -0,0 +1,109 @@ +//! Fixed transport exit categories and bounded heartbeat progress telemetry. +//! +//! Keep wire values and error text out of these categories: CCA retains this +//! telemetry even when ordinary exec-server diagnostics are filtered out. + +pub(super) enum TransportExitReason { + PongTimeout, + OutgoingClosed, + Framing, + SequenceExhausted, + Encryption, + WebsocketWriteTimeout, + WebsocketWrite, + WebsocketEnded, + InvalidFrame, + ApplicationDelivery, + InvalidData, + RelayReset, + WebsocketClose, + UnexpectedText, + WebsocketRead, +} + +impl TransportExitReason { + pub(super) fn as_str(&self) -> &'static str { + match self { + Self::PongTimeout => "pong_timeout", + Self::OutgoingClosed => "outgoing_closed", + Self::Framing => "framing", + Self::SequenceExhausted => "sequence_exhausted", + Self::Encryption => "encryption", + Self::WebsocketWriteTimeout => "websocket_write_timeout", + Self::WebsocketWrite => "websocket_write", + Self::WebsocketEnded => "websocket_ended", + Self::InvalidFrame => "invalid_frame", + Self::ApplicationDelivery => "application_delivery", + Self::InvalidData => "invalid_data", + Self::RelayReset => "relay_reset", + Self::WebsocketClose => "websocket_close", + Self::UnexpectedText => "unexpected_text", + Self::WebsocketRead => "websocket_read", + } + } +} + +#[derive(Debug, thiserror::Error)] +pub(super) enum WebsocketWriteError { + #[error("websocket write timed out")] + Timeout, + #[error("{0}")] + Transport(String), +} + +impl WebsocketWriteError { + pub(super) fn exit_reason(&self) -> TransportExitReason { + match self { + Self::Timeout => TransportExitReason::WebsocketWriteTimeout, + Self::Transport(_) => TransportExitReason::WebsocketWrite, + } + } +} + +/// WebSocket heartbeat progress only proves reachability of the relay peer, not +/// delivery to the executor behind it. Emit at most once per local Ping or Pong. +pub(super) enum HeartbeatProgress { + PingWritten, + PongReceived, +} + +#[derive(Default)] +pub(super) struct HeartbeatDiagnostics { + pub(super) pings_written: u64, + pub(super) pongs_received: u64, + last_ping_at: Option, +} + +impl HeartbeatDiagnostics { + pub(super) fn record(&mut self, progress: HeartbeatProgress, stream_id: &str) { + let phase = match progress { + HeartbeatProgress::PingWritten => { + self.pings_written += 1; + self.last_ping_at = Some(tokio::time::Instant::now()); + "ping_written" + } + HeartbeatProgress::PongReceived => { + // Unsolicited Pongs must not create unbounded telemetry volume. + if self.last_ping_at.is_none() { + return; + } + self.pongs_received += 1; + "pong_received" + } + }; + tracing::event!( + name: "codex.exec_server.noise_heartbeat", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.exec_server.noise_heartbeat", + stream_id, + phase, + pings_written = self.pings_written, + pongs_received = self.pongs_received, + ping_elapsed_ms = self.last_ping_at.map(|at| at.elapsed().as_millis() as u64).unwrap_or(0), + ); + if matches!(progress, HeartbeatProgress::PongReceived) { + self.last_ping_at = None; + } + } +} diff --git a/codex-rs/exec-server/src/noise_relay/harness_tests.rs b/codex-rs/exec-server/src/noise_relay/harness_tests.rs index e8b9e98831..4c2d77d2d4 100644 --- a/codex-rs/exec-server/src/noise_relay/harness_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/harness_tests.rs @@ -66,6 +66,28 @@ fn websocket_read_diagnostics_use_typed_errors_only() { #[tokio::test(start_paused = true)] async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Result<()> { + use opentelemetry::trace::TracerProvider as _; + use opentelemetry_sdk::trace::InMemorySpanExporter; + use opentelemetry_sdk::trace::SdkTracerProvider; + use tracing_subscriber::prelude::*; + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let events = SafeEvents::default(); + let _subscriber = tracing::subscriber::set_default( + tracing_subscriber::registry() + .with( + events + .clone() + .with_filter(tracing_subscriber::filter::LevelFilter::INFO), + ) + .with( + tracing_opentelemetry::layer() + .with_tracer(provider.tracer("harness-test")) + .with_filter(tracing_subscriber::filter::LevelFilter::INFO), + ), + ); let (connection, mut control, mut outbound_rx) = connected_controlled_harness().await?; let traceparent = "00-00000000000000000000000000000001-0000000000000002-01"; let tracestate = "dd=s:1"; @@ -86,12 +108,34 @@ async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Resul .await?; control.wait_for_blocked_write(/*expected*/ 1).await?; + { + let captured = events.0.lock().unwrap(); + let dequeue = captured + .iter() + .find(|event| { + event + .get("event.name") + .is_some_and(|name| name.contains("request_dequeued")) + }) + .context("live dequeue")?; + assert_eq!( + dequeue.get("trace_id").map(String::as_str), + Some("00000000000000000000000000000001") + ); + assert!( + exporter + .get_finished_spans()? + .iter() + .all(|span| span.name != "codex.exec_server.noise_request_write") + ); + } control.grant_writes(/*count*/ 1); let first = read_outbound_frame(&mut outbound_rx).await?; assert_eq!( (first.traceparent.as_deref(), first.tracestate.as_deref()), (Some(traceparent), Some(tracestate)) ); + let stream_id = first.stream_id.clone(); let first_payload = first.into_data()?.payload; assert!( !first_payload @@ -103,6 +147,43 @@ async fn first_encrypted_request_frame_exposes_only_its_trace_context() -> Resul control.grant_writes(/*count*/ 1); let second = read_outbound_frame(&mut outbound_rx).await?; assert_eq!((second.traceparent, second.tracestate), (None, None)); + assert!(!format!("{:?}", events.0.lock().unwrap()).contains("request_written")); + control.wait_for_blocked_write(/*expected*/ 3).await?; + control.grant_writes(/*count*/ 1); + read_outbound_frame(&mut outbound_rx).await?; + let spans = exporter.get_finished_spans()?; + let write_span = spans + .iter() + .find(|span| span.name == "codex.exec_server.noise_request_write") + .context("expected finished request write span")?; + assert_eq!( + write_span.span_context.trace_id().to_string(), + "00000000000000000000000000000001" + ); + assert_eq!(write_span.parent_span_id.to_string(), "0000000000000002"); + let captured = events.0.lock().unwrap(); + let progress: Vec<_> = captured + .iter() + .filter(|e| { + e.get("event.name").is_some_and(|name| { + name.contains("request_dequeued") || name.contains("request_written") + }) + }) + .collect(); + assert_eq!(progress.len(), 2); + for event in &progress { + assert_eq!( + event.get("trace_id").map(String::as_str), + Some("00000000000000000000000000000001") + ); + } + assert_eq!(progress[0]["stream_id"], format!("{stream_id:?}")); + assert_eq!(progress[0]["stream_id"], progress[1]["stream_id"]); + assert_eq!( + progress[0]["message_sequence"], + progress[1]["message_sequence"] + ); + drop(captured); for task in &connection.task_handles { task.abort(); @@ -294,8 +375,13 @@ async fn pong_keeps_harness_alive_until_peer_stops_responding() -> Result<()> { Ok(()) } -#[tokio::test] +/// A full consumer queue times out with occupancy and wait duration, without logging its data. +#[tokio::test(start_paused = true)] async fn application_event_delivery_is_bounded() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); let (incoming_tx, _incoming_rx) = mpsc::channel(1); incoming_tx .send(JsonRpcConnectionEvent::MalformedMessage { @@ -313,6 +399,14 @@ async fn application_event_delivery_is_bounded() -> Result<()> { .await; assert!(matches!(result, Err(ExecServerError::Closed))); + let events = events.0.lock().unwrap(); + assert_eq!(events.len(), 1); + let event = &events[0]; + assert_eq!(event["noise_reason"], "\"application_backpressure\""); + assert_eq!(event["incoming_queue_depth"], "1"); + assert_eq!(event["incoming_queue_capacity"], "1"); + assert!(event["delivery_wait_ms"].parse::()? >= 10); + assert!(!format!("{events:?}").contains("blocked event")); Ok(()) } @@ -343,6 +437,18 @@ async fn connected_controlled_harness() -> Result<( JsonRpcConnection, ControlledWebSocketHandle, futures_mpsc::UnboundedReceiver, +)> { + let (connection, control, outbound_rx, _transport, _stream_id) = + connected_controlled_harness_with_transport().await?; + Ok((connection, control, outbound_rx)) +} + +async fn connected_controlled_harness_with_transport() -> Result<( + JsonRpcConnection, + ControlledWebSocketHandle, + futures_mpsc::UnboundedReceiver, + NoiseTransport, + String, )> { let (websocket, control, mut outbound_rx) = ControlledWebSocket::new(/*write_permits*/ 2); let executor_identity = NoiseChannelIdentity::generate()?; @@ -381,11 +487,12 @@ async fn connected_controlled_harness() -> Result<( &prologue, &handshake.into_handshake_payload()?, )?; - let (_transport, response) = pending.complete()?; + let (transport, response) = pending.complete()?; control.send_inbound(Message::Binary( - encode_relay_message_frame(&RelayMessageFrame::handshake(stream_id, response)).into(), + encode_relay_message_frame(&RelayMessageFrame::handshake(stream_id.clone(), response)) + .into(), ))?; - Ok((connection, control, outbound_rx)) + Ok((connection, control, outbound_rx, transport, stream_id)) } struct ControlledWebSocket { @@ -528,3 +635,324 @@ impl futures::Stream for ControlledWebSocket { result } } + +#[derive(Clone, Default)] +struct SafeEvents(Arc>>>); + +impl tracing_subscriber::Layer for SafeEvents +where + S: tracing::Subscriber + for<'lookup> tracing_subscriber::registry::LookupSpan<'lookup>, +{ + fn on_event( + &self, + event: &tracing::Event<'_>, + context: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().target() != "codex_otel.trace_safe" { + return; + } + #[derive(Default)] + struct Fields(std::collections::BTreeMap); + impl tracing::field::Visit for Fields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().to_owned(), format!("{value:?}")); + } + } + let mut fields = Fields::default(); + event.record(&mut fields); + // Match CCA's live JSON lookup: inspecting OtelData must not materialize + // the context and accidentally hide a missing producer initialization. + if let Some(trace_id) = context.event_scope(event).and_then(|mut scope| { + scope.find_map(|span| { + span.extensions() + .get::()? + .trace_id() + }) + }) { + fields.0.insert("trace_id".to_owned(), trace_id.to_string()); + } + self.0.lock().unwrap().push(fields.0); + } +} + +/// A blocked write can time out with a Pong queued; its diagnostic must distinguish this +/// local write stall from the read loop's missing-Pong watchdog. +#[tokio::test(start_paused = true)] +async fn blocked_write_reports_timeout_with_pong_still_queued() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?; + control.wait_for_blocked_write(/*expected*/ 1).await?; + control.grant_writes(/*count*/ 1); + let Message::Ping(payload) = outbound_rx.next().await.context("expected Ping")? else { + anyhow::bail!("expected Ping before blocked write"); + }; + connection + .outgoing_tx + .send(JSONRPCMessage::Request(JSONRPCRequest { + id: RequestId::Integer(1), + method: "SYNTHETIC_PRIVATE_METHOD".to_owned(), + params: Some(serde_json::json!({"secret": "SYNTHETIC_PRIVATE_PAYLOAD"})), + trace: None, + })) + .await?; + control.wait_for_blocked_write(/*expected*/ 2).await?; + connection + .outgoing_tx + .send(JSONRPCMessage::Request(JSONRPCRequest { + id: RequestId::Integer(2), + method: "SYNTHETIC_PRIVATE_QUEUED".to_owned(), + params: None, + trace: None, + })) + .await?; + let reads = control.inbound_reads(); + control.send_inbound(Message::Pong(payload))?; + tokio::time::advance(WEBSOCKET_PONG_TIMEOUT + Duration::from_millis(1)).await; + for task in connection.task_handles.drain(..) { + task.await?; + } + assert_eq!(control.inbound_reads(), reads); + assert!(*connection.disconnected_rx.borrow()); + let events = events.0.lock().unwrap(); + let terminal: Vec<_> = events + .iter() + .filter(|e| { + e.get("event.name") + .is_some_and(|name| name == "\"codex.exec_server.noise_transport_closed\"") + }) + .collect(); + assert_eq!(terminal.len(), 1); + let event = terminal[0]; + assert_eq!(event["noise_reason"], "\"websocket_write_timeout\""); + assert_eq!(event["write_pending"], "true"); + assert_eq!(event["write_kind"], "\"data\""); + assert_eq!(event["pong_pending"], "true"); + assert_eq!(event["records_written"], "0"); + assert_eq!(event["messages_dequeued"], "1"); + assert_eq!(event["messages_written"], "0"); + assert_eq!(event["outgoing_queue_depth"], "1"); + assert_eq!(event["pings_written"], "1"); + assert_eq!(event["pongs_received"], "0"); + assert!(!format!("{events:?}").contains("request_written")); + assert!(event["write_wait_ms"].parse::()? >= WEBSOCKET_PONG_TIMEOUT.as_millis() as u64); + assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE")); + Ok(()) +} + +/// Peer-controlled read errors must produce a fixed terminal category without retaining +/// the raw error, even though ordinary debug logs still carry it. +#[tokio::test(start_paused = true)] +async fn read_error_reports_safe_terminal_category() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, control, _outbound_rx) = connected_controlled_harness().await?; + control + .inbound_tx + .unbounded_send(Err(WebSocketError::Io(std::io::Error::other( + "SYNTHETIC_PRIVATE_ERROR", + ))))?; + for task in connection.task_handles.drain(..) { + task.await?; + } + let events = events.0.lock().unwrap(); + let event = events.last().context("expected terminal event")?; + assert_eq!( + event["event.name"], + "\"codex.exec_server.noise_transport_closed\"" + ); + assert_eq!(event["noise_reason"], "\"websocket_read\""); + assert_eq!(event["read_reason"], "\"websocket_read_io_other\""); + assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE_ERROR")); + Ok(()) +} + +/// A keepalive write stall has no outstanding Pong yet; retain that distinction +/// when a disconnect aligns with the periodic keepalive schedule. +#[tokio::test(start_paused = true)] +async fn blocked_keepalive_reports_write_phase_without_pong_deadline() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, mut control, _outbound_rx) = connected_controlled_harness().await?; + control.wait_for_blocked_write(/*expected*/ 1).await?; + tokio::time::advance(WEBSOCKET_PONG_TIMEOUT + Duration::from_millis(1)).await; + for task in connection.task_handles.drain(..) { + task.await?; + } + let events = events.0.lock().unwrap(); + let event = events.last().context("expected terminal event")?; + assert_eq!(event["noise_reason"], "\"websocket_write_timeout\""); + assert_eq!(event["write_kind"], "\"keepalive\""); + assert_eq!(event["pong_pending"], "false"); + assert_eq!(event["write_pending"], "true"); + Ok(()) +} + +/// Successful local writes with no inbound traffic reproduce a blackholed path, +/// which must report a Pong timeout rather than a blocked write. +#[tokio::test(start_paused = true)] +async fn blackholed_peer_reports_pong_timeout_after_successful_writes() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?; + control.wait_for_blocked_write(/*expected*/ 1).await?; + control.grant_writes(/*count*/ 1); + assert!(matches!(outbound_rx.next().await, Some(Message::Ping(_)))); + connection + .outgoing_tx + .send(JSONRPCMessage::Request(JSONRPCRequest { + id: RequestId::Integer(1), + method: "test".to_owned(), + params: None, + trace: None, + })) + .await?; + control.wait_for_blocked_write(/*expected*/ 2).await?; + control.grant_writes(/*count*/ 1); + read_outbound_data(&mut outbound_rx).await?; + tokio::time::advance(WEBSOCKET_PONG_TIMEOUT / 2).await; + assert!(!*connection.disconnected_rx.borrow()); + tokio::time::advance(WEBSOCKET_PONG_TIMEOUT / 2 + Duration::from_millis(1)).await; + for task in connection.task_handles.drain(..) { + task.await?; + } + let events = events.0.lock().unwrap(); + let event = events.last().context("expected terminal event")?; + assert_eq!(event["noise_reason"], "\"pong_timeout\""); + assert_eq!(event["write_pending"], "false"); + assert_eq!(event["records_written"], "1"); + assert_eq!(event["messages_dequeued"], "1"); + assert_eq!(event["messages_written"], "1"); + assert_eq!(event["pings_written"], "1"); + assert_eq!(event["pongs_received"], "0"); + assert!(format!("{events:?}").contains("request_written")); + assert_eq!(event["pong_pending"], "true"); + Ok(()) +} + +/// A stalled application consumer can prevent the shared loop from reading an +/// already queued Pong; retain queue saturation separately from network timeout. +#[tokio::test(start_paused = true)] +async fn full_application_queue_reports_backpressure_with_pong_queued() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, mut control, mut outbound_rx, mut transport, stream_id) = + connected_controlled_harness_with_transport().await?; + control.wait_for_blocked_write(/*expected*/ 1).await?; + control.grant_writes(/*count*/ 1); + let Some(Message::Ping(payload)) = outbound_rx.next().await else { + anyhow::bail!("expected Ping"); + }; + let framed = frame_jsonrpc_message(&JSONRPCMessage::Response( + codex_exec_server_protocol::JSONRPCResponse { + id: RequestId::Integer(1), + result: serde_json::json!({"private": "SYNTHETIC_PRIVATE_RESPONSE"}), + }, + ))?; + let reads = control.inbound_reads(); + for seq in 0..=CHANNEL_CAPACITY { + let frame = RelayMessageFrame::data( + stream_id.clone(), + seq as u32, + transport.encrypt(&framed)?, + /*trace*/ None, + ); + control.send_inbound(Message::Binary(encode_relay_message_frame(&frame).into()))?; + } + control.send_inbound(Message::Pong(payload))?; + // Awaiting the task lets the paused clock advance only once frame processing + // has filled the application queue and is waiting on its delivery deadline. + for task in connection.task_handles.drain(..) { + task.await?; + } + assert_eq!(control.inbound_reads() - reads, CHANNEL_CAPACITY + 1); + assert_eq!(connection.incoming_rx.len(), CHANNEL_CAPACITY); + let events = events.0.lock().unwrap(); + let backpressure = events + .iter() + .find(|event| { + event + .get("noise_reason") + .is_some_and(|value| value == "\"application_backpressure\"") + }) + .context("expected application backpressure")?; + assert_eq!( + backpressure["incoming_queue_depth"], + CHANNEL_CAPACITY.to_string() + ); + let terminal = events.last().context("expected terminal event")?; + assert_eq!(terminal["noise_reason"], "\"application_delivery\""); + assert_eq!(terminal["pong_pending"], "true"); + assert!(!format!("{events:?}").contains("SYNTHETIC_PRIVATE_RESPONSE")); + Ok(()) +} + +/// A responsive relay can coexist with a missing executor response. Ping/Pong +/// progress must remain distinct from successful application delivery. +#[tokio::test(start_paused = true)] +async fn responsive_relay_without_application_response_reports_progress() -> Result<()> { + use tracing_subscriber::prelude::*; + let events = SafeEvents::default(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?; + connection + .outgoing_tx + .send(JSONRPCMessage::Request(JSONRPCRequest { + id: RequestId::Integer(1), + method: "SYNTHETIC_PRIVATE_METHOD".to_owned(), + params: None, + trace: None, + })) + .await?; + control.wait_for_blocked_write(/*expected*/ 1).await?; + control.grant_writes(/*count*/ 1); + read_outbound_data(&mut outbound_rx).await?; + for expected in 2..=3 { + control.wait_for_blocked_write(expected).await?; + control.grant_writes(/*count*/ 1); + let Some(Message::Ping(payload)) = outbound_rx.next().await else { + anyhow::bail!("expected Ping") + }; + for _ in 0..3 { + control.send_inbound(Message::Pong(payload.clone()))?; + } + } + // A subsequent Close is ordered behind the last Pong and completes the summary. + control.send_inbound(Message::Close(None))?; + for task in connection.task_handles.drain(..) { + task.await?; + } + assert!(connection.incoming_rx.is_empty()); + let captured = events.0.lock().unwrap(); + let terminal = captured.last().context("expected terminal")?; + assert_eq!(terminal["noise_reason"], "\"websocket_close\""); + assert_eq!(terminal["messages_written"], "1"); + assert_eq!(terminal["pings_written"], "2"); + assert_eq!(terminal["pongs_received"], "2"); + let heartbeats: Vec<_> = captured + .iter() + .filter_map(|event| event.get("phase")) + .collect(); + assert_eq!( + heartbeats, + vec![ + "\"ping_written\"", + "\"pong_received\"", + "\"ping_written\"", + "\"pong_received\"" + ] + ); + assert!(!format!("{captured:?}").contains("SYNTHETIC_PRIVATE")); + Ok(()) +} diff --git a/codex-rs/exec-server/src/noise_relay/stream_handler.rs b/codex-rs/exec-server/src/noise_relay/stream_handler.rs index b0c610933c..e4727e7f1e 100644 --- a/codex-rs/exec-server/src/noise_relay/stream_handler.rs +++ b/codex-rs/exec-server/src/noise_relay/stream_handler.rs @@ -49,9 +49,18 @@ impl NoiseStreamHandler for ConnectionProcessor { type Outgoing = JSONRPCMessage; fn decode(payload: Bytes) -> Result { - Ok(JsonRpcConnectionEvent::message(serde_json::from_slice( - &payload, - )?)) + let event = JsonRpcConnectionEvent::message(serde_json::from_slice(&payload)?); + if let JsonRpcConnectionEvent::QueuedRequest { request_span, .. } = &event { + // The request span can remain open while queued or executing. Finish a + // payload-free receipt marker now so a stalled executor is distinguishable + // from a request that never arrived, without waiting for that span to close. + drop(tracing::info_span!( + target: "codex_otel.trace_safe", + parent: request_span, + "codex.exec_server.noise_request_received", + )); + } + Ok(event) } fn encode(message: Self::Outgoing) -> Result { diff --git a/codex-rs/exec-server/src/noise_relay/stream_handler_tests.rs b/codex-rs/exec-server/src/noise_relay/stream_handler_tests.rs index 0015176684..335066b24b 100644 --- a/codex-rs/exec-server/src/noise_relay/stream_handler_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/stream_handler_tests.rs @@ -12,14 +12,28 @@ use crate::server::ConnectionProcessor; #[test] fn local_decode_starts_request_span_before_queueing() -> Result<()> { - let _subscriber = tracing::subscriber::set_default(tracing_subscriber::registry()); + use opentelemetry::trace::TracerProvider as _; + use opentelemetry_sdk::trace::InMemorySpanExporter; + use opentelemetry_sdk::trace::SdkTracerProvider; + use tracing_subscriber::prelude::*; + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let _subscriber = + tracing::subscriber::set_default(tracing_subscriber::registry().with( + tracing_opentelemetry::layer().with_tracer(provider.tracer("noise-receipt-test")), + )); tracing::callsite::rebuild_interest_cache(); let request = JSONRPCRequest { id: RequestId::Integer(1), method: "test/queued".to_string(), - params: None, - trace: None, + params: Some(serde_json::json!({"secret": "SYNTHETIC_PRIVATE_PAYLOAD"})), + trace: Some(codex_protocol::protocol::W3cTraceContext { + traceparent: Some("00-00000000000000000000000000000001-0000000000000002-01".to_owned()), + tracestate: None, + }), }; let before_decode = Instant::now(); let event = ::decode( @@ -37,5 +51,23 @@ fn local_decode_starts_request_span_before_queueing() -> Result<()> { assert_eq!(actual_request, request); assert!(!request_span.is_disabled()); assert!((before_decode..=after_decode).contains(&queued_at)); + let spans = exporter.get_finished_spans()?; + assert_eq!( + spans.len(), + 1, + "receipt must export while the request is still queued" + ); + assert_eq!(spans[0].name, "codex.exec_server.noise_request_received"); + assert_eq!( + spans[0].span_context.trace_id().to_string(), + "00000000000000000000000000000001" + ); + assert!(!format!("{spans:?}").contains("SYNTHETIC_PRIVATE_PAYLOAD")); + assert!(!format!("{spans:?}").contains("test/queued")); + let receipt_parent = spans[0].parent_span_id; + drop(request_span); + let spans = exporter.get_finished_spans()?; + assert_eq!(spans.len(), 2); + assert_eq!(receipt_parent, spans[1].span_context.span_id()); Ok(()) } diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index cbe30bb37f..e60530fd55 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -28,6 +28,7 @@ use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::sync::watch; use tokio::task::JoinHandle; +use tokio::time::Instant; use tokio::time::timeout; use crate::client_telemetry::record_client_request; @@ -740,6 +741,7 @@ impl RpcClient { response_rx: oneshot::Receiver, call_timeout: RpcCallTimeout, ) -> Result { + let enqueue_started = Instant::now(); if self .write_tx .send(JSONRPCMessage::Request(JSONRPCRequest { @@ -759,6 +761,9 @@ impl RpcClient { tracing::Level::INFO, event.name = "codex.exec_server.request_enqueued", rpc.method = method, + enqueue_wait_ms = enqueue_started.elapsed().as_millis() as u64, + outgoing_queue_depth = self.write_tx.max_capacity() - self.write_tx.capacity(), + outgoing_queue_capacity = self.write_tx.max_capacity(), ); // Do not race in-flight requests directly against the transport-close // watch value. The connection reader receives JSON-RPC messages and