diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 506b943b1e..997986c4d1 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -18,6 +18,7 @@ use codex_plugin::app_connector_ids_from_declarations; use codex_plugin::prompt_safe_plugin_description; use codex_utils_absolute_path::AbsolutePathBuf; use reqwest::RequestBuilder; +use reqwest::header::RETRY_AFTER; use serde::Deserialize; use serde::Serialize; use serde_json::Value as JsonValue; @@ -82,6 +83,7 @@ const OPENAI_CURATED_REMOTE_COLLECTION_KEY: &str = "vertical"; const OAI_PRODUCT_SKU_HEADER: &str = "OAI-Product-Sku"; const CODEX_PRODUCT_SKU: &str = "codex"; const REMOTE_PLUGIN_CATALOG_TIMEOUT: Duration = Duration::from_secs(30); +const REMOTE_PLUGIN_CATALOG_GET_RETRY_DELAY: Duration = Duration::from_millis(250); const RECOMMENDED_PLUGINS_TIMEOUT: Duration = Duration::from_secs(5); const REMOTE_PLUGIN_LIST_PAGE_LIMIT: u32 = 200; const MAX_RECOMMENDED_PLUGINS: usize = 50; @@ -828,7 +830,8 @@ pub async fn fetch_recommended_plugins( let request = authenticated_request(client.get(&url), auth)? .timeout(RECOMMENDED_PLUGINS_TIMEOUT) .query(&[("scope", "GLOBAL")]); - let response: RecommendedPluginsResponse = send_and_decode(request, &url).await?; + let response: RecommendedPluginsResponse = + send_and_decode_idempotent_get_with_retry(request, &url).await?; Ok(recommended_plugins_mode(response)) } @@ -1132,7 +1135,8 @@ pub async fn fetch_remote_plugin_skill_detail( let url = remote_plugin_skill_detail_url(config, plugin_id, skill_name)?; let client = build_reqwest_client(); let request = authenticated_request(client.get(&url), auth)?; - let response: RemotePluginSkillDetailResponse = send_and_decode(request, &url).await?; + let response: RemotePluginSkillDetailResponse = + send_and_decode_idempotent_get_with_retry(request, &url).await?; if response.plugin_id != plugin_id { return Err(RemotePluginCatalogError::UnexpectedPluginId { expected: plugin_id.to_string(), @@ -1798,7 +1802,7 @@ async fn get_remote_plugin_list_page( if let Some(page_token) = page_token { request = request.query(&[("pageToken", page_token)]); } - send_and_decode(request, &url).await + send_and_decode_idempotent_get_with_retry(request, &url).await } async fn get_remote_shared_workspace_plugins_page( @@ -1814,7 +1818,7 @@ async fn get_remote_shared_workspace_plugins_page( if let Some(page_token) = page_token { request = request.query(&[("pageToken", page_token)]); } - send_and_decode(request, &url).await + send_and_decode_idempotent_get_with_retry(request, &url).await } async fn get_remote_plugin_installed_page( @@ -1835,7 +1839,7 @@ async fn get_remote_plugin_installed_page( if let Some(page_token) = page_token { request = request.query(&[("pageToken", page_token)]); } - send_and_decode(request, &url).await + send_and_decode_idempotent_get_with_retry(request, &url).await } async fn fetch_plugin_detail( @@ -1851,7 +1855,7 @@ async fn fetch_plugin_detail( if include_download_urls { request = request.query(&[("includeDownloadUrls", true)]); } - send_and_decode(request, &url).await + send_and_decode_idempotent_get_with_retry(request, &url).await } fn remote_plugin_skill_detail_url( @@ -1899,25 +1903,135 @@ async fn send_and_decode Deserialize<'de>>( request: RequestBuilder, url: &str, ) -> Result { - let response = request - .send() + send_and_decode_attempt(request, url) .await - .map_err(|source| RemotePluginCatalogError::Request { + .map_err(|err| err.error) +} + +struct RemotePluginCatalogAttemptError { + error: RemotePluginCatalogError, + retry_after: RetryAfterDelay, +} + +impl RemotePluginCatalogAttemptError { + fn new(error: RemotePluginCatalogError) -> Self { + Self { + error, + retry_after: RetryAfterDelay::None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RetryAfterDelay { + None, + Delay(Duration), + Unparsable, +} + +async fn send_and_decode_attempt Deserialize<'de>>( + request: RequestBuilder, + url: &str, +) -> Result { + let response = request.send().await.map_err(|source| { + RemotePluginCatalogAttemptError::new(RemotePluginCatalogError::Request { url: url.to_string(), source, - })?; + }) + })?; let status = response.status(); - let body = response.text().await.unwrap_or_default(); - if !status.is_success() { - return Err(RemotePluginCatalogError::UnexpectedStatus { + let retry_after = parse_retry_after_delay(response.headers()); + let body = response.text().await.map_err(|source| { + RemotePluginCatalogAttemptError::new(RemotePluginCatalogError::Request { url: url.to_string(), - status, - body, + source, + }) + })?; + if !status.is_success() { + return Err(RemotePluginCatalogAttemptError { + error: RemotePluginCatalogError::UnexpectedStatus { + url: url.to_string(), + status, + body, + }, + retry_after, }); } - serde_json::from_str(&body).map_err(|source| RemotePluginCatalogError::Decode { - url: url.to_string(), - source, + serde_json::from_str(&body).map_err(|source| { + RemotePluginCatalogAttemptError::new(RemotePluginCatalogError::Decode { + url: url.to_string(), + source, + }) }) } + +async fn send_and_decode_idempotent_get_with_retry Deserialize<'de>>( + request: RequestBuilder, + url: &str, +) -> Result { + let retry_request = request.try_clone(); + let err = match send_and_decode_attempt(request, url).await { + Ok(response) => return Ok(response), + Err(err) => err, + }; + let retry_after = err.retry_after; + let err = err.error; + + let Some(retry_delay) = retry_delay_for_remote_plugin_catalog_get_error(&err, retry_after) + else { + return Err(err); + }; + + let Some(retry_request) = retry_request else { + return Err(err); + }; + + tracing::warn!( + error = %err, + "remote plugin catalog GET failed; retrying once" + ); + tokio::time::sleep(retry_delay).await; + send_and_decode(retry_request, url).await +} + +fn retry_delay_for_remote_plugin_catalog_get_error( + err: &RemotePluginCatalogError, + retry_after: RetryAfterDelay, +) -> Option { + match err { + RemotePluginCatalogError::Request { source, .. } => (source.is_timeout() + || source.is_connect()) + .then_some(REMOTE_PLUGIN_CATALOG_GET_RETRY_DELAY), + RemotePluginCatalogError::UnexpectedStatus { status, .. } + if *status == reqwest::StatusCode::TOO_MANY_REQUESTS => + { + match retry_after { + RetryAfterDelay::Delay(delay) if delay <= REMOTE_PLUGIN_CATALOG_GET_RETRY_DELAY => { + Some(delay) + } + RetryAfterDelay::Delay(_) | RetryAfterDelay::Unparsable => None, + RetryAfterDelay::None => Some(REMOTE_PLUGIN_CATALOG_GET_RETRY_DELAY), + } + } + RemotePluginCatalogError::UnexpectedStatus { status, .. } => { + (*status == reqwest::StatusCode::REQUEST_TIMEOUT || status.is_server_error()) + .then_some(REMOTE_PLUGIN_CATALOG_GET_RETRY_DELAY) + } + _ => None, + } +} + +fn parse_retry_after_delay(headers: &reqwest::header::HeaderMap) -> RetryAfterDelay { + let Some(retry_after) = headers.get(RETRY_AFTER) else { + return RetryAfterDelay::None; + }; + + retry_after + .to_str() + .ok() + .and_then(|value| value.trim().parse::().ok()) + .map(Duration::from_secs) + .map(RetryAfterDelay::Delay) + .unwrap_or(RetryAfterDelay::Unparsable) +} diff --git a/codex-rs/core-plugins/src/remote/share.rs b/codex-rs/core-plugins/src/remote/share.rs index c400d27c47..e9bdc065fa 100644 --- a/codex-rs/core-plugins/src/remote/share.rs +++ b/codex-rs/core-plugins/src/remote/share.rs @@ -384,7 +384,7 @@ async fn get_created_workspace_plugins_page( if let Some(page_token) = page_token { request = request.query(&[("pageToken", page_token)]); } - send_and_decode(request, &url).await + send_and_decode_idempotent_get_with_retry(request, &url).await } async fn create_workspace_plugin_upload( diff --git a/codex-rs/core-plugins/src/remote/share/tests.rs b/codex-rs/core-plugins/src/remote/share/tests.rs index ed48341237..79cdf06654 100644 --- a/codex-rs/core-plugins/src/remote/share/tests.rs +++ b/codex-rs/core-plugins/src/remote/share/tests.rs @@ -11,6 +11,9 @@ use std::fs; use std::io::Read; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use tempfile::TempDir; use wiremock::Mock; use wiremock::MockServer; @@ -88,6 +91,17 @@ fn archive_file_entries(archive_bytes: &[u8]) -> BTreeMap> { .collect() } +fn fail_once_then_succeed_json(body: serde_json::Value) -> impl wiremock::Respond { + let attempts = Arc::new(AtomicUsize::new(0)); + move |_request: &wiremock::Request| { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + ResponseTemplate::new(503).set_body_string("temporary failure") + } else { + ResponseTemplate::new(200).set_body_json(body.clone()) + } + } +} + fn remote_plugin_json(plugin_id: &str) -> serde_json::Value { json!({ "id": plugin_id, @@ -694,6 +708,37 @@ async fn list_remote_plugin_shares_fetches_created_workspace_plugins() { ); } +#[tokio::test] +async fn list_remote_plugin_shares_retries_transient_created_workspace_plugins_failure() { + let codex_home = TempDir::new().unwrap(); + let server = MockServer::start().await; + let config = test_config(&server); + let auth = test_auth(); + + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/workspace/created")) + .and(header("authorization", "Bearer Access Token")) + .and(header("chatgpt-account-id", "account_id")) + .and(query_param( + "limit", + REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string(), + )) + .and(query_param_is_missing("pageToken")) + .respond_with(fail_once_then_succeed_json(json!({ + "plugins": [], + "pagination": empty_pagination_json(), + }))) + .expect(2) + .mount(&server) + .await; + + let result = list_remote_plugin_shares(&config, Some(&auth), codex_home.path()) + .await + .unwrap(); + + assert_eq!(result, Vec::new()); +} + #[tokio::test] async fn delete_remote_plugin_share_deletes_workspace_plugin() { let codex_home = TempDir::new().unwrap(); diff --git a/codex-rs/core-plugins/src/remote_tests.rs b/codex-rs/core-plugins/src/remote_tests.rs index f1d4478d7f..914231b8ac 100644 --- a/codex-rs/core-plugins/src/remote_tests.rs +++ b/codex-rs/core-plugins/src/remote_tests.rs @@ -1,5 +1,16 @@ use super::*; use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::time::Duration; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; +use wiremock::matchers::query_param; #[test] fn build_remote_marketplace_preserves_directory_order_and_appends_installed_only_plugins() { @@ -33,6 +44,109 @@ fn build_remote_marketplace_preserves_directory_order_and_appends_installed_only ); } +#[tokio::test] +async fn fetch_remote_marketplaces_retries_transient_workspace_installed_failure() { + let server = MockServer::start().await; + let directory_body = remote_plugin_page_body(/*enabled*/ None); + let installed_body = remote_plugin_page_body(/*enabled*/ Some(true)); + mount_workspace_directory( + &server, + ResponseTemplate::new(200).set_body_string(directory_body), + ) + .await; + mount_workspace_installed(&server, fail_once_then_succeed(installed_body)).await; + + let marketplaces = fetch_remote_marketplaces( + &remote_plugin_service_config(&server), + Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()), + &[RemoteMarketplaceSource::WorkspaceDirectory], + /*global_catalog_cache_path*/ None, + ) + .await + .expect("workspace marketplace should load after retry"); + + assert_eq!(marketplaces.len(), 1); + assert_eq!(marketplaces[0].plugins.len(), 1); + assert_eq!(marketplaces[0].plugins[0].name, "workspace-linear"); + assert!(marketplaces[0].plugins[0].installed); + assert!(marketplaces[0].plugins[0].enabled); +} + +#[tokio::test] +async fn fetch_remote_marketplaces_retries_transient_workspace_directory_failure() { + let server = MockServer::start().await; + let directory_body = remote_plugin_page_body(/*enabled*/ None); + mount_workspace_directory(&server, fail_once_then_succeed(directory_body)).await; + mount_workspace_installed( + &server, + ResponseTemplate::new(200).set_body_string(empty_plugin_page_body()), + ) + .await; + + let marketplaces = fetch_remote_marketplaces( + &remote_plugin_service_config(&server), + Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()), + &[RemoteMarketplaceSource::WorkspaceDirectory], + /*global_catalog_cache_path*/ None, + ) + .await + .expect("workspace marketplace should load after retry"); + + assert_eq!(marketplaces.len(), 1); + assert_eq!(marketplaces[0].plugins.len(), 1); + assert_eq!(marketplaces[0].plugins[0].name, "workspace-linear"); + assert!(!marketplaces[0].plugins[0].installed); + assert!(!marketplaces[0].plugins[0].enabled); +} + +#[test] +fn remote_plugin_catalog_get_retry_delay_uses_short_retry_after() { + let err = RemotePluginCatalogError::UnexpectedStatus { + url: "https://chatgpt.example/backend-api/ps/plugins/list".to_string(), + status: reqwest::StatusCode::TOO_MANY_REQUESTS, + body: "throttled".to_string(), + }; + + assert_eq!( + retry_delay_for_remote_plugin_catalog_get_error( + &err, + RetryAfterDelay::Delay(Duration::from_millis(100)) + ), + Some(Duration::from_millis(100)) + ); +} + +#[test] +fn remote_plugin_catalog_get_retry_delay_skips_long_retry_after() { + let err = RemotePluginCatalogError::UnexpectedStatus { + url: "https://chatgpt.example/backend-api/ps/plugins/list".to_string(), + status: reqwest::StatusCode::TOO_MANY_REQUESTS, + body: "throttled".to_string(), + }; + + assert_eq!( + retry_delay_for_remote_plugin_catalog_get_error( + &err, + RetryAfterDelay::Delay(Duration::from_secs(1)) + ), + None + ); +} + +#[test] +fn remote_plugin_catalog_get_retry_delay_skips_unparsable_retry_after() { + let err = RemotePluginCatalogError::UnexpectedStatus { + url: "https://chatgpt.example/backend-api/ps/plugins/list".to_string(), + status: reqwest::StatusCode::TOO_MANY_REQUESTS, + body: "throttled".to_string(), + }; + + assert_eq!( + retry_delay_for_remote_plugin_catalog_get_error(&err, RetryAfterDelay::Unparsable), + None + ); +} + fn directory_plugin(id: &str, name: &str) -> RemotePluginDirectoryItem { RemotePluginDirectoryItem { id: id.to_string(), @@ -78,6 +192,97 @@ fn directory_plugin(id: &str, name: &str) -> RemotePluginDirectoryItem { } } +fn remote_plugin_service_config(server: &MockServer) -> RemotePluginServiceConfig { + RemotePluginServiceConfig { + chatgpt_base_url: format!("{}/backend-api/", server.uri()), + } +} + +async fn mount_workspace_directory( + server: &MockServer, + response: impl wiremock::Respond + 'static, +) { + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/list")) + .and(query_param("scope", "WORKSPACE")) + .and(query_param("limit", "200")) + .and(header("authorization", "Bearer Access Token")) + .and(header("chatgpt-account-id", "account_id")) + .respond_with(response) + .expect(1..=2) + .mount(server) + .await; +} + +async fn mount_workspace_installed( + server: &MockServer, + response: impl wiremock::Respond + 'static, +) { + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/installed")) + .and(query_param("scope", "WORKSPACE")) + .and(header("authorization", "Bearer Access Token")) + .and(header("chatgpt-account-id", "account_id")) + .respond_with(response) + .expect(1..=2) + .mount(server) + .await; +} + +fn fail_once_then_succeed(body: String) -> impl wiremock::Respond { + let attempts = Arc::new(AtomicUsize::new(0)); + move |_request: &wiremock::Request| { + if attempts.fetch_add(1, Ordering::SeqCst) == 0 { + ResponseTemplate::new(503).set_body_string("temporary failure") + } else { + ResponseTemplate::new(200).set_body_string(body.clone()) + } + } +} + +fn empty_plugin_page_body() -> String { + serde_json::json!({ + "plugins": [], + "pagination": { + "limit": 50, + "next_page_token": null, + }, + }) + .to_string() +} + +fn remote_plugin_page_body(enabled: Option) -> String { + let mut plugin = serde_json::json!({ + "id": "plugins~Plugin_11111111111111111111111111111111", + "name": "workspace-linear", + "scope": "WORKSPACE", + "discoverability": "LISTED", + "installation_policy": "AVAILABLE", + "authentication_policy": "ON_USE", + "status": "AVAILABLE", + "release": { + "display_name": "Workspace Linear", + "description": "Track workspace work", + "app_ids": [], + "interface": {}, + "skills": [], + }, + }); + if let Some(enabled) = enabled { + plugin["enabled"] = serde_json::json!(enabled); + plugin["disabled_skill_names"] = serde_json::json!([]); + } + + serde_json::json!({ + "plugins": [plugin], + "pagination": { + "limit": 50, + "next_page_token": null, + }, + }) + .to_string() +} + #[test] fn remote_plugin_interface_maps_dark_logo_url() { let mut plugin = directory_plugin("plugin-linear", "linear");