mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
Add debug timing for thread item pagination (#59497)
## What changed - Log `thread/items/list` request IDs, cursor kinds, page sizes, next-page availability, elapsed time, and success status. - Log remote-control item-page response preparation and WebSocket write times, along with request and stream IDs, sequence IDs, frame counts, wire bytes, and success, interruption, or write-error outcomes. Collect these transport metrics only when debug logging is enabled. GitOrigin-RevId: 0f26bef6d6b8cacbd5a5c76c49a5f882669f6598
This commit is contained in:
@@ -1,5 +1,7 @@
|
||||
use crate::outgoing_message::OutgoingMessage;
|
||||
use codex_app_server_protocol::ClientResponsePayload;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
use std::io;
|
||||
@@ -169,6 +171,20 @@ pub enum ServerEvent {
|
||||
}
|
||||
|
||||
impl ServerEvent {
|
||||
pub(super) fn item_list_response_id(&self) -> Option<&RequestId> {
|
||||
if let Self::ServerMessage { message } = self
|
||||
&& let OutgoingMessage::Response(response) = message.as_ref()
|
||||
&& matches!(
|
||||
response.result.as_ref(),
|
||||
ClientResponsePayload::ThreadItemsList(_)
|
||||
)
|
||||
{
|
||||
Some(&response.id)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn segment_id(&self) -> Option<usize> {
|
||||
match self {
|
||||
Self::ServerMessageChunk { segment_id, .. } => Some(*segment_id),
|
||||
|
||||
@@ -51,6 +51,7 @@ use std::collections::VecDeque;
|
||||
use std::io;
|
||||
use std::io::ErrorKind;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::oneshot;
|
||||
@@ -1050,7 +1051,18 @@ impl RemoteControlWebsocket {
|
||||
}
|
||||
}
|
||||
};
|
||||
let (payloads, write_complete_tx) = {
|
||||
let item_page = queued_server_envelope
|
||||
.event
|
||||
.item_list_response_id()
|
||||
.filter(|_| tracing::enabled!(tracing::Level::DEBUG))
|
||||
.map(|id| {
|
||||
(
|
||||
id.clone(),
|
||||
queued_server_envelope.stream_id.clone(),
|
||||
Instant::now(),
|
||||
)
|
||||
});
|
||||
let (payloads, seq_id) = {
|
||||
let mut state = state.lock().await;
|
||||
let seq_key = (
|
||||
queued_server_envelope.client_id.clone(),
|
||||
@@ -1090,20 +1102,49 @@ impl RemoteControlWebsocket {
|
||||
.next_seq_id_by_stream
|
||||
.insert(seq_key, seq_id.saturating_add(1));
|
||||
|
||||
(payloads, queued_server_envelope.write_complete_tx)
|
||||
(payloads, seq_id)
|
||||
};
|
||||
|
||||
let log_item_page_write = item_page.map(|(request_id, stream_id, prepare_started)| {
|
||||
let frame_count = payloads.len();
|
||||
let wire_bytes = payloads.iter().map(String::len).sum::<usize>();
|
||||
let write_started = Instant::now();
|
||||
move |outcome: &str| {
|
||||
tracing::debug!(
|
||||
%request_id,
|
||||
stream_id = %stream_id.0,
|
||||
seq_id,
|
||||
frame_count,
|
||||
wire_bytes,
|
||||
prepare_ms = (write_started - prepare_started).as_millis() as u64,
|
||||
write_ms = write_started.elapsed().as_millis() as u64,
|
||||
outcome,
|
||||
"remote control item page write finished"
|
||||
)
|
||||
}
|
||||
});
|
||||
for payload in payloads {
|
||||
tokio::select! {
|
||||
_ = shutdown_token.cancelled() => return Ok(()),
|
||||
_ = shutdown_token.cancelled() => {
|
||||
if let Some(log) = log_item_page_write {
|
||||
log("interrupted");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
send_result = websocket_writer.send(tungstenite::Message::Text(payload.into())) => {
|
||||
if let Err(err) = send_result {
|
||||
if let Some(log) = log_item_page_write {
|
||||
log("write_error");
|
||||
}
|
||||
return Err(io::Error::other(err));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(write_complete_tx) = write_complete_tx {
|
||||
if let Some(log) = log_item_page_write {
|
||||
log("success");
|
||||
}
|
||||
if let Some(write_complete_tx) = queued_server_envelope.write_complete_tx {
|
||||
let _ = write_complete_tx.send(());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1620,7 +1620,9 @@ impl MessageProcessor {
|
||||
self.thread_processor.thread_turns_list(params).await
|
||||
}
|
||||
ClientRequest::ThreadItemsList { params, .. } => {
|
||||
self.thread_processor.thread_items_list(params).await
|
||||
self.thread_processor
|
||||
.thread_items_list(&request_id, params)
|
||||
.await
|
||||
}
|
||||
ClientRequest::ThreadItemsRead { params, .. } => {
|
||||
self.thread_processor.thread_items_read(params).await
|
||||
|
||||
@@ -1001,11 +1001,29 @@ impl ThreadRequestProcessor {
|
||||
|
||||
pub(crate) async fn thread_items_list(
|
||||
&self,
|
||||
request_id: &ConnectionRequestId,
|
||||
params: ThreadItemsListParams,
|
||||
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
|
||||
self.thread_items_list_response_inner(params)
|
||||
.await
|
||||
.map(|response| Some(response.into()))
|
||||
let cursor_kind = match ¶ms.cursor {
|
||||
None => "none",
|
||||
Some(ThreadItemsListCursor::Opaque(_)) => "opaque",
|
||||
Some(ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { .. })) => {
|
||||
"item_anchor"
|
||||
}
|
||||
};
|
||||
let started = std::time::Instant::now();
|
||||
let result = self.thread_items_list_response_inner(params).await;
|
||||
let page = result.as_ref().ok();
|
||||
tracing::debug!(
|
||||
request_id = %request_id.request_id,
|
||||
cursor_kind,
|
||||
item_count = page.map(|page| page.data.len()),
|
||||
has_next_page = page.map(|page| page.next_cursor.is_some()),
|
||||
duration_ms = started.elapsed().as_millis() as u64,
|
||||
success = result.is_ok(),
|
||||
"thread/items/list finished"
|
||||
);
|
||||
result.map(|response| Some(response.into()))
|
||||
}
|
||||
|
||||
pub(crate) async fn thread_items_read(
|
||||
|
||||
Reference in New Issue
Block a user