From 1789ab69d11da3163c54ea9f696af7d7f16c7715 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Thu, 7 May 2026 10:52:34 -0700 Subject: [PATCH] add plugin list and scheduler timing logs --- codex-rs/app-server/src/message_processor.rs | 5 +- .../src/request_processors/plugins.rs | 117 +++++++++++++++-- .../app-server/src/request_serialization.rs | 124 ++++++++++++++++-- 3 files changed, 226 insertions(+), 20 deletions(-) diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 2e7414a178..e9edd29e93 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -768,6 +768,8 @@ impl MessageProcessor { ); let serialization_scope = codex_request.serialization_scope(); + let serialization_method = codex_request.method(); + let serialization_request_id = connection_request_id.request_id.to_string(); let app_server_client_name = session.app_server_client_name().map(str::to_string); let client_version = session.client_version().map(str::to_string); let device_key_requests_allowed = session.allows_device_key_requests(); @@ -794,7 +796,8 @@ impl MessageProcessor { } } .instrument(span), - ); + ) + .with_log_metadata(serialization_method, serialization_request_id); if let Some(scope) = serialization_scope { let (key, access) = RequestSerializationQueueKey::from_scope(connection_id, scope); diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 3616d1ebbe..a36077d86b 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -348,6 +348,7 @@ impl PluginRequestProcessor { &self, params: PluginListParams, ) -> Result { + let request_started_at = Instant::now(); let plugins_manager = self.thread_manager.plugins_manager(); let PluginListParams { cwds, @@ -358,35 +359,80 @@ impl PluginRequestProcessor { let marketplace_kinds = marketplace_kinds.unwrap_or_else(|| vec![PluginListMarketplaceKind::Local]); let include_local = marketplace_kinds.contains(&PluginListMarketplaceKind::Local); + warn!( + roots_count = roots.len(), + explicit_marketplace_kinds, + marketplace_kinds_count = marketplace_kinds.len(), + include_local, + "plugin/list timing started" + ); + let config_started_at = Instant::now(); let config = self.load_latest_config(/*fallback_cwd*/ None).await?; + warn!( + elapsed_ms = config_started_at.elapsed().as_millis(), + plugins_enabled = config.features.enabled(Feature::Plugins), + remote_plugins_enabled = config.features.enabled(Feature::RemotePlugin), + "plugin/list timing loaded config" + ); let empty_response = || PluginListResponse { marketplaces: Vec::new(), marketplace_load_errors: Vec::new(), featured_plugin_ids: Vec::new(), }; if !config.features.enabled(Feature::Plugins) { + warn!( + elapsed_ms = request_started_at.elapsed().as_millis(), + "plugin/list timing completed with plugins disabled" + ); return Ok(empty_response()); } + let auth_started_at = Instant::now(); let auth = self.auth_manager.auth().await; - if !self + warn!( + elapsed_ms = auth_started_at.elapsed().as_millis(), + has_auth = auth.is_some(), + "plugin/list timing loaded auth" + ); + let workspace_setting_started_at = Instant::now(); + let workspace_plugins_enabled = self .workspace_codex_plugins_enabled(&config, auth.as_ref()) - .await - { + .await; + warn!( + elapsed_ms = workspace_setting_started_at.elapsed().as_millis(), + workspace_plugins_enabled, "plugin/list timing checked workspace setting" + ); + if !workspace_plugins_enabled { + warn!( + elapsed_ms = request_started_at.elapsed().as_millis(), + "plugin/list timing completed with workspace plugins disabled" + ); return Ok(empty_response()); } let plugins_input = config.plugins_config_input(); let (mut data, marketplace_load_errors) = if include_local { + let background_tasks_started_at = Instant::now(); plugins_manager.maybe_start_plugin_list_background_tasks_for_config( &plugins_input, auth.clone(), &roots, Some(self.effective_plugins_changed_callback()), ); + warn!( + elapsed_ms = background_tasks_started_at.elapsed().as_millis(), + "plugin/list timing started background tasks" + ); let config_for_marketplace_listing = plugins_input.clone(); let plugins_manager_for_marketplace_listing = plugins_manager.clone(); + let shared_plugin_ids_started_at = Instant::now(); let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(&config); + warn!( + elapsed_ms = shared_plugin_ids_started_at.elapsed().as_millis(), + shared_plugin_ids_count = shared_plugin_ids_by_local_path.len(), + "plugin/list timing loaded shared plugin ids" + ); + let local_listing_started_at = Instant::now(); match tokio::task::spawn_blocking(move || { let outcome = plugins_manager_for_marketplace_listing .list_marketplaces_for_config(&config_for_marketplace_listing, &roots)?; @@ -447,7 +493,21 @@ impl PluginRequestProcessor { }) .await { - Ok(Ok(outcome)) => outcome, + Ok(Ok(outcome)) => { + let plugin_count = outcome + .0 + .iter() + .map(|marketplace| marketplace.plugins.len()) + .sum::(); + warn!( + elapsed_ms = local_listing_started_at.elapsed().as_millis(), + marketplace_count = outcome.0.len(), + plugin_count, + load_error_count = outcome.1.len(), + "plugin/list timing listed local marketplaces" + ); + outcome + } Ok(Err(err)) => { return Err(Self::marketplace_error(err, "list marketplace plugins")); } @@ -472,6 +532,7 @@ impl PluginRequestProcessor { remote_sources.push(RemoteMarketplaceSource::SharedWithMe); } if !remote_sources.is_empty() { + let remote_marketplaces_started_at = Instant::now(); let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; @@ -483,6 +544,7 @@ impl PluginRequestProcessor { .await { Ok(remote_marketplaces) => { + let fetched_remote_marketplace_count = remote_marketplaces.len(); for remote_marketplace in remote_marketplaces .into_iter() .map(remote_marketplace_to_info) @@ -496,11 +558,25 @@ impl PluginRequestProcessor { data.push(remote_marketplace); } } + warn!( + elapsed_ms = remote_marketplaces_started_at.elapsed().as_millis(), + remote_source_count = remote_sources.len(), + fetched_remote_marketplace_count, + merged_marketplace_count = data.len(), + "plugin/list timing fetched remote marketplaces" + ); } Err( - RemotePluginCatalogError::AuthRequired - | RemotePluginCatalogError::UnsupportedAuthMode, - ) => {} + err @ (RemotePluginCatalogError::AuthRequired + | RemotePluginCatalogError::UnsupportedAuthMode), + ) => { + warn!( + elapsed_ms = remote_marketplaces_started_at.elapsed().as_millis(), + remote_source_count = remote_sources.len(), + error = %err, + "plugin/list timing skipped remote marketplaces" + ); + } Err(err) => { warn!( error = %err, @@ -508,12 +584,15 @@ impl PluginRequestProcessor { ); } } + } else { + warn!("plugin/list timing skipped remote marketplaces"); } - let featured_plugin_ids = if data + let curated_marketplace_present = data .iter() - .any(|marketplace| marketplace.name == OPENAI_CURATED_MARKETPLACE_NAME) - { + .any(|marketplace| marketplace.name == OPENAI_CURATED_MARKETPLACE_NAME); + let featured_plugin_ids_started_at = Instant::now(); + let featured_plugin_ids = if curated_marketplace_present { match plugins_manager .featured_plugin_ids_for_config(&plugins_input, auth.as_ref()) .await @@ -530,6 +609,24 @@ impl PluginRequestProcessor { } else { Vec::new() }; + warn!( + elapsed_ms = featured_plugin_ids_started_at.elapsed().as_millis(), + curated_marketplace_present, + featured_plugin_id_count = featured_plugin_ids.len(), + "plugin/list timing loaded featured ids" + ); + let plugin_count = data + .iter() + .map(|marketplace| marketplace.plugins.len()) + .sum::(); + warn!( + elapsed_ms = request_started_at.elapsed().as_millis(), + marketplace_count = data.len(), + plugin_count, + load_error_count = marketplace_load_errors.len(), + featured_plugin_id_count = featured_plugin_ids.len(), + "plugin/list timing completed" + ); Ok(PluginListResponse { marketplaces: data, diff --git a/codex-rs/app-server/src/request_serialization.rs b/codex-rs/app-server/src/request_serialization.rs index a84c43a062..0155435d72 100644 --- a/codex-rs/app-server/src/request_serialization.rs +++ b/codex-rs/app-server/src/request_serialization.rs @@ -4,6 +4,7 @@ use std::future::Future; use std::path::PathBuf; use std::pin::Pin; use std::sync::Arc; +use std::time::Instant; use codex_app_server_protocol::ClientRequestSerializationScope; use tokio::sync::Mutex; @@ -105,6 +106,8 @@ impl RequestSerializationQueueKey { pub(crate) struct QueuedInitializedRequest { gate: Arc, future: BoxFutureUnit, + method: String, + request_id: String, } impl QueuedInitializedRequest { @@ -115,11 +118,19 @@ impl QueuedInitializedRequest { Self { gate, future: Box::pin(future), + method: "".to_string(), + request_id: "".to_string(), } } + pub(crate) fn with_log_metadata(mut self, method: String, request_id: String) -> Self { + self.method = method; + self.request_id = request_id; + self + } + pub(crate) async fn run(self) { - let Self { gate, future } = self; + let Self { gate, future, .. } = self; gate.run(future).await; } } @@ -127,6 +138,46 @@ impl QueuedInitializedRequest { struct QueuedSerializedRequest { access: RequestSerializationAccess, request: QueuedInitializedRequest, + enqueued_at: Instant, +} + +impl QueuedSerializedRequest { + async fn run( + self, + key: RequestSerializationQueueKey, + batch_size: usize, + queue_depth_after_pop: usize, + ) { + let Self { + access, + request, + enqueued_at, + } = self; + let method = request.method.clone(); + let request_id = request.request_id.clone(); + tracing::warn!( + ?key, + ?access, + method, + request_id, + queue_wait_ms = enqueued_at.elapsed().as_millis(), + batch_size, + queue_depth_after_pop, + "serialized request started" + ); + + let started_at = Instant::now(); + request.run().await; + + tracing::warn!( + ?key, + ?access, + method, + request_id, + run_ms = started_at.elapsed().as_millis(), + "serialized request completed" + ); + } } #[derive(Default)] @@ -203,22 +254,74 @@ impl RequestSerializationQueues { access: RequestSerializationAccess, request: QueuedInitializedRequest, ) { - let request = QueuedSerializedRequest { access, request }; - let ready_requests = { + let method = request.method.clone(); + let request_id = request.request_id.clone(); + let request = QueuedSerializedRequest { + access, + request, + enqueued_at: Instant::now(), + }; + let ( + ready_requests, + queue_depth_before, + queued_exclusive_count, + head_access, + head_method, + head_request_id, + queue_depth_after_pop, + ) = { let mut queues = self.inner.lock().await; let queue = queues.entry(key.clone()).or_default(); + let queue_depth_before = queue.pending.len(); + let queued_exclusive_count = queue + .pending + .iter() + .filter(|request| request.access == RequestSerializationAccess::Exclusive) + .count(); + let head_request = queue.pending.front(); + let head_access = head_request.map(|request| request.access); + let head_method = head_request + .map(|request| request.request.method.clone()) + .unwrap_or_else(|| "".to_string()); + let head_request_id = head_request + .map(|request| request.request.request_id.clone()) + .unwrap_or_else(|| "".to_string()); queue.enqueue(request); - queue.take_ready_requests() + let ready_requests = queue.take_ready_requests(); + ( + ready_requests, + queue_depth_before, + queued_exclusive_count, + head_access, + head_method, + head_request_id, + queue.pending.len(), + ) }; + tracing::warn!( + ?key, + ?access, + method, + request_id, + queue_depth_before, + queue_depth_after = queue_depth_before + 1, + queued_exclusive_count, + ?head_access, + head_method, + head_request_id, + "serialized request queued" + ); - self.spawn_ready_requests(key, ready_requests); + self.spawn_ready_requests(key, ready_requests, queue_depth_after_pop); } fn spawn_ready_requests( &self, key: RequestSerializationQueueKey, requests: Vec, + queue_depth_after_pop: usize, ) { + let batch_size = requests.len(); for request in requests { let queues = self.clone(); let request_key = key.clone(); @@ -226,7 +329,9 @@ impl RequestSerializationQueues { tokio::spawn( async move { let access = request.access; - request.request.run().await; + request + .run(request_key.clone(), batch_size, queue_depth_after_pop) + .await; queues.complete(request_key, access).await; } .instrument(span), @@ -239,21 +344,22 @@ impl RequestSerializationQueues { key: RequestSerializationQueueKey, access: RequestSerializationAccess, ) { - let ready_requests = { + let (ready_requests, queue_depth_after_pop) = { let mut queues = self.inner.lock().await; let Some(queue) = queues.get_mut(&key) else { return; }; queue.complete(access); let ready_requests = queue.take_ready_requests(); + let queue_depth_after_pop = queue.pending.len(); let should_remove = queue.is_idle(); if should_remove { queues.remove(&key); } - ready_requests + (ready_requests, queue_depth_after_pop) }; - self.spawn_ready_requests(key, ready_requests); + self.spawn_ready_requests(key, ready_requests, queue_depth_after_pop); } }