From 622a79ed5667571e12e46b199fbc225fbd4ea00f Mon Sep 17 00:00:00 2001 From: stevenlee-oai Date: Wed, 15 Jul 2026 20:33:56 +0000 Subject: [PATCH] Fetch workspace connectors concurrently (#33421) ## Why Workspace accounts fetched the paginated connector directory before starting the independent workspace connector request, adding the latency of both request chains. ## What changed Start the workspace connector request alongside the public directory lookup. Continue to ignore workspace lookup failures and filter hidden workspace apps before merging the results. ## Testing Add a regression test that blocks the directory response until the workspace request starts, verifying that both lookups overlap and their connectors are returned. GitOrigin-RevId: fa29023a4154b555a31756054109e706e61c32c0 --- codex-rs/connectors/Cargo.toml | 2 +- codex-rs/connectors/src/lib.rs | 99 ++++++++++++++++++++++++++-------- 2 files changed, 79 insertions(+), 22 deletions(-) diff --git a/codex-rs/connectors/Cargo.toml b/codex-rs/connectors/Cargo.toml index eb2306495b..ac3ce9241c 100644 --- a/codex-rs/connectors/Cargo.toml +++ b/codex-rs/connectors/Cargo.toml @@ -20,12 +20,12 @@ serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha1 = { workspace = true } tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tracing = { workspace = true } urlencoding = { workspace = true } [dev-dependencies] pretty_assertions = { workspace = true } -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } [lib] doctest = false diff --git a/codex-rs/connectors/src/lib.rs b/codex-rs/connectors/src/lib.rs index e98c894f50..b8e0110432 100644 --- a/codex-rs/connectors/src/lib.rs +++ b/codex-rs/connectors/src/lib.rs @@ -172,10 +172,28 @@ where return Ok(cached_connectors); } - let mut apps = list_directory_connectors(&mut fetch_page).await?; - if is_workspace_account { - apps.extend(list_workspace_connectors(&mut fetch_page).await?); - } + let apps = if is_workspace_account { + // The workspace directory is independent from the paginated public directory. + // Start both before awaiting either so workspace accounts do not pay for the + // two request chains back-to-back. + let workspace_connectors = + fetch_page("/connectors/directory/list_workspace?external_logos=true".to_string()); + let directory_connectors = list_directory_connectors(&mut fetch_page); + let (directory_connectors, workspace_connectors) = + tokio::join!(directory_connectors, workspace_connectors); + let mut apps = directory_connectors?; + if let Ok(response) = workspace_connectors { + apps.extend( + response + .apps + .into_iter() + .filter(|app| !is_hidden_directory_app(app)), + ); + } + apps + } else { + list_directory_connectors(&mut fetch_page).await? + }; let mut connectors = merge_directory_apps(apps) .into_iter() @@ -260,23 +278,6 @@ where Ok(apps) } -async fn list_workspace_connectors(fetch_page: &mut F) -> anyhow::Result> -where - F: FnMut(String) -> Fut, - Fut: Future>, -{ - let response = - fetch_page("/connectors/directory/list_workspace?external_logos=true".to_string()).await; - match response { - Ok(response) => Ok(response - .apps - .into_iter() - .filter(|app| !is_hidden_directory_app(app)) - .collect()), - Err(_) => Ok(Vec::new()), - } -} - fn merge_directory_apps(apps: Vec) -> Vec { let mut merged: HashMap = HashMap::new(); for app in apps { @@ -518,7 +519,9 @@ mod tests { use std::sync::Mutex; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; + use std::time::Duration; use tempfile::TempDir; + use tokio::sync::Notify; static CONNECTOR_DIRECTORY_CACHE_TEST_LOCK: LazyLock> = LazyLock::new(|| tokio::sync::Mutex::new(())); @@ -733,6 +736,60 @@ mod tests { Ok(()) } + #[tokio::test] + #[expect( + clippy::await_holding_invalid_type, + reason = "test serializes access to the shared connector cache for its full duration" + )] + async fn list_all_connectors_overlaps_workspace_and_directory_requests() -> anyhow::Result<()> { + let _cache_guard = CONNECTOR_DIRECTORY_CACHE_TEST_LOCK.lock().await; + + let codex_home = TempDir::new()?; + let cache_context = cache_context(&codex_home, "overlap"); + let workspace_started = Arc::new(Notify::new()); + + // The public directory response waits until the workspace request is polled. + // Without overlap this future cannot complete; the timeout only bounds a + // regression instead of supplying the ordering. + let connectors = tokio::time::timeout( + Duration::from_secs(1), + list_all_connectors_with_options( + cache_context, + /*is_workspace_account*/ true, + /*force_refetch*/ true, + move |path| { + let workspace_started = Arc::clone(&workspace_started); + async move { + if path.starts_with("/connectors/directory/list_workspace") { + workspace_started.notify_one(); + Ok(DirectoryListResponse { + apps: vec![app("workspace", "Workspace")], + next_token: None, + }) + } else { + workspace_started.notified().await; + Ok(DirectoryListResponse { + apps: vec![app("directory", "Directory")], + next_token: None, + }) + } + } + }, + ), + ) + .await + .expect("workspace request should start while directory request is pending")?; + + assert_eq!( + connectors + .into_iter() + .map(|connector| connector.id) + .collect::>(), + vec!["directory".to_string(), "workspace".to_string()] + ); + Ok(()) + } + #[tokio::test] #[expect( clippy::await_holding_invalid_type,