Retry transient remote plugin catalog GETs

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<T: for<'de> Deserialize<'de>>(
     request: RequestBuilder,
     url: &str,
 ) -> Result<T, RemotePluginCatalogError> {
-    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<T: for<'de> Deserialize<'de>>(
+    request: RequestBuilder,
+    url: &str,
+) -> Result<T, RemotePluginCatalogAttemptError> {
+    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<T: for<'de> Deserialize<'de>>(
+    request: RequestBuilder,
+    url: &str,
+) -> Result<T, RemotePluginCatalogError> {
+    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<Duration> {
+    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::<u64>().ok())
+        .map(Duration::from_secs)
+        .map(RetryAfterDelay::Delay)
+        .unwrap_or(RetryAfterDelay::Unparsable)
+}

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(

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<String, Vec<u8>> {
         .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();

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<bool>) -> 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");