Deduplicate remote plugin bundle syncs with shared semaphores (#39240)

## What changed

- Give plugin managers for the same cache root a shared semaphore that permits
  only one background installed-plugin bundle sync at a time.
- Hold the permit for the lifetime of the spawned sync task and refresh the
  installed-plugin cache after a successful local cache change.

## Testing

- Add a test that starts bundle syncs from two plugin managers sharing a Codex
  home and verifies that only one request is made.

GitOrigin-RevId: 86c39476c3a9015b9862ab75e8f11f110b0284fc

codex-rs/core-plugins/src/manager.rs

@@ -47,7 +47,6 @@ use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
 use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
 use crate::remote::RecommendedPluginsMode;
 use crate::remote::RemoteInstalledPlugin;
-use crate::remote::RemoteInstalledPluginBundleSyncOutcome;
 use crate::remote::RemotePluginCatalogError;
 use crate::remote::RemotePluginMaterialization;
 use crate::remote::RemotePluginScope;
@@ -444,6 +443,7 @@ pub struct PluginsManager {
     skill_root_loader: Arc<dyn SkillRootLoader<PluginSkillRoot>>,
     tool_suggest_metadata_cache: ToolSuggestMetadataCache,
     remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
+    remote_installed_plugin_bundle_sync_gate: Arc<Semaphore>,
     remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
     remote_catalog_cache_refresh_state: RwLock<RemoteCatalogCacheRefreshState>,
     restriction_product: Option<Product>,
@@ -524,6 +524,8 @@ impl PluginsManager {
         // already-admitted plugins may continue exposing MCP servers/tools from shared local state.
         //
         // This assumes a single CODEX_HOME is only used by one product.
+        let remote_installed_plugin_bundle_sync_gate =
+            crate::remote::remote_installed_plugin_bundle_sync_gate(&codex_home);
         Self {
             codex_home: codex_home.clone(),
             store: PluginStore::new(codex_home),
@@ -544,6 +546,7 @@ impl PluginsManager {
             skill_root_loader,
             tool_suggest_metadata_cache: ToolSuggestMetadataCache::new(),
             remote_installed_plugins_cache: RwLock::new(None),
+            remote_installed_plugin_bundle_sync_gate,
             remote_installed_plugins_cache_refresh_state: RwLock::new(
                 RemoteInstalledPluginsCacheRefreshState::default(),
             ),
@@ -1095,28 +1098,53 @@ impl PluginsManager {
             return;
         }
 
+        let Some(auth) = auth else {
+            return;
+        };
+        let Ok(permit) =
+            Arc::clone(&self.remote_installed_plugin_bundle_sync_gate).try_acquire_owned()
+        else {
+            return;
+        };
         let manager = Arc::clone(self);
         let config_for_refresh = config.clone();
-        let auth_for_refresh = auth.clone();
-        let on_local_cache_changed =
-            Arc::new(move |outcome: RemoteInstalledPluginBundleSyncOutcome| {
-                manager.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
-                    &config_for_refresh,
-                    auth_for_refresh.clone(),
-                    RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh,
-                    on_effective_plugins_changed.clone(),
-                    EffectivePluginsChange {
-                        materialized_remote_plugins: outcome.materialized_remote_plugins,
-                    },
-                );
-            });
-
-        crate::remote::maybe_start_remote_installed_plugin_bundle_sync(
-            self.codex_home.clone(),
-            remote_plugin_service_config(config),
-            auth,
-            Some(on_local_cache_changed),
-        );
+        let service_config = remote_plugin_service_config(config);
+        tokio::spawn(async move {
+            let _permit = permit;
+            let result = crate::remote::sync_remote_installed_plugin_bundles_once(
+                manager.codex_home.clone(),
+                &service_config,
+                Some(&auth),
+            )
+            .await;
+            match result {
+                Ok(outcome) => {
+                    tracing::info!(
+                        materialized_remote_plugins = ?outcome.materialized_remote_plugins,
+                        removed_cache_plugin_ids = ?outcome.removed_cache_plugin_ids,
+                        failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids,
+                        "completed remote installed plugin bundle sync"
+                    );
+                    if outcome.changed_local_cache() {
+                        manager.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
+                            &config_for_refresh,
+                            Some(auth),
+                            RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh,
+                            on_effective_plugins_changed,
+                            EffectivePluginsChange {
+                                materialized_remote_plugins: outcome.materialized_remote_plugins,
+                            },
+                        );
+                    }
+                }
+                Err(err) => {
+                    warn!(
+                        error = %err,
+                        "remote installed plugin bundle sync failed"
+                    );
+                }
+            }
+        });
     }
 
     fn maybe_start_remote_catalog_cache_refresh(

codex-rs/core-plugins/src/manager_tests.rs

@@ -6776,6 +6776,61 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() {
     );
 }
 
+#[tokio::test]
+async fn background_remote_installed_bundle_sync_stays_deduped() {
+    let codex_home = TempDir::new().unwrap();
+    write_file(
+        &codex_home.path().join(CONFIG_TOML_FILE),
+        r#"[features]
+plugins = true
+remote_plugin = true
+"#,
+    );
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/installed"))
+        .and(query_param("includeDownloadUrls", "true"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_delay(Duration::from_millis(200))
+                .set_body_json(serde_json::json!({
+                    "plugins": [],
+                    "pagination": {"next_page_token": null},
+                })),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+
+    let mut config = load_config(codex_home.path(), codex_home.path()).await;
+    config.chatgpt_base_url = format!("{}/backend-api", server.uri());
+    let first_manager = std::sync::Arc::new(test_plugins_manager_with_options(
+        codex_home.path().to_path_buf(),
+        Some(Product::Codex),
+        Some(AuthMode::Chatgpt),
+    ));
+    let second_manager = std::sync::Arc::new(test_plugins_manager_with_options(
+        codex_home.path().to_path_buf(),
+        Some(Product::Codex),
+        Some(AuthMode::Chatgpt),
+    ));
+    let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
+
+    first_manager.maybe_start_remote_installed_plugin_bundle_sync(
+        &config,
+        Some(auth.clone()),
+        /*on_effective_plugins_changed*/ None,
+    );
+    second_manager.maybe_start_remote_installed_plugin_bundle_sync(
+        &config,
+        Some(auth),
+        /*on_effective_plugins_changed*/ None,
+    );
+
+    tokio::time::sleep(Duration::from_millis(400)).await;
+    server.verify().await;
+}
+
 #[test]
 fn plugin_install_error_preserves_store_io_sub_error_type() {
     let error = PluginInstallError::Store(PluginStoreError::Io {

codex-rs/core-plugins/src/remote.rs

@@ -60,7 +60,7 @@ pub use remote_installed_plugin_sync::RemoteInstalledPluginBundleSyncOutcome;
 pub use remote_installed_plugin_sync::RemotePluginCacheMutationGuard;
 pub use remote_installed_plugin_sync::RemotePluginMaterialization;
 pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight;
-pub(crate) use remote_installed_plugin_sync::maybe_start_remote_installed_plugin_bundle_sync;
+pub(crate) use remote_installed_plugin_sync::remote_installed_plugin_bundle_sync_gate;
 pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once;
 pub use search::RemotePluginSearchPage;
 pub use search::RemotePluginSearchRequest;

codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs

@@ -20,18 +20,18 @@ use codex_plugin::PluginId;
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 use std::collections::HashMap;
-use std::collections::HashSet;
 use std::fs;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::OnceLock;
-use tracing::info;
+use std::sync::Weak;
+use tokio::sync::Semaphore;
 use tracing::warn;
 
-static REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_IN_FLIGHT: OnceLock<
-    Mutex<HashSet<RemoteInstalledPluginBundleSyncKey>>,
+static REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_GATES: OnceLock<
+    Mutex<HashMap<PathBuf, Weak<Semaphore>>>,
 > = OnceLock::new();
 static REMOTE_PLUGIN_CACHE_MUTATIONS_IN_FLIGHT: OnceLock<
     Mutex<HashMap<RemotePluginCacheMutationKey, usize>>,
@@ -68,11 +68,6 @@ pub enum RemoteInstalledPluginBundleSyncError {
     Store(#[from] PluginStoreError),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-struct RemoteInstalledPluginBundleSyncKey {
-    plugin_cache_root: PathBuf,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct RemotePluginCacheMutationKey {
     plugin_cache_root: PathBuf,
@@ -84,50 +79,20 @@ pub struct RemotePluginCacheMutationGuard {
     key: RemotePluginCacheMutationKey,
 }
 
-pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync(
-    codex_home: PathBuf,
-    config: RemotePluginServiceConfig,
-    auth: Option<CodexAuth>,
-    on_local_cache_changed: Option<
-        Arc<dyn Fn(RemoteInstalledPluginBundleSyncOutcome) + Send + Sync + 'static>,
-    >,
-) {
-    let Some(auth) = auth else {
-        return;
+pub(crate) fn remote_installed_plugin_bundle_sync_gate(codex_home: &Path) -> Arc<Semaphore> {
+    let plugin_cache_root = remote_plugin_cache_root(codex_home);
+    let gates =
+        REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_GATES.get_or_init(|| Mutex::new(HashMap::new()));
+    let mut gates = match gates.lock() {
+        Ok(gates) => gates,
+        Err(err) => err.into_inner(),
     };
-    let key = RemoteInstalledPluginBundleSyncKey {
-        plugin_cache_root: remote_plugin_cache_root(&codex_home),
-    };
-    if !mark_remote_installed_plugin_bundle_sync_in_flight(key.clone()) {
-        return;
+    if let Some(gate) = gates.get(&plugin_cache_root).and_then(Weak::upgrade) {
+        return gate;
     }
-
-    tokio::spawn(async move {
-        let result =
-            sync_remote_installed_plugin_bundles_once(codex_home, &config, Some(&auth)).await;
-        match result {
-            Ok(outcome) => {
-                info!(
-                    materialized_remote_plugins = ?outcome.materialized_remote_plugins,
-                    removed_cache_plugin_ids = ?outcome.removed_cache_plugin_ids,
-                    failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids,
-                    "completed remote installed plugin bundle sync"
-                );
-                if outcome.changed_local_cache()
-                    && let Some(on_local_cache_changed) = on_local_cache_changed
-                {
-                    on_local_cache_changed(outcome);
-                }
-            }
-            Err(err) => {
-                warn!(
-                    error = %err,
-                    "remote installed plugin bundle sync failed"
-                );
-            }
-        }
-        clear_remote_installed_plugin_bundle_sync_in_flight(&key);
-    });
+    let gate = Arc::new(Semaphore::new(/*permits*/ 1));
+    gates.insert(plugin_cache_root, Arc::downgrade(&gate));
+    gate
 }
 
 pub async fn sync_remote_installed_plugin_bundles_once(
@@ -431,29 +396,6 @@ fn is_remote_plugin_cache_mutation_in_flight(
     })
 }
 
-fn mark_remote_installed_plugin_bundle_sync_in_flight(
-    key: RemoteInstalledPluginBundleSyncKey,
-) -> bool {
-    let syncs =
-        REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_IN_FLIGHT.get_or_init(|| Mutex::new(HashSet::new()));
-    let mut syncs = match syncs.lock() {
-        Ok(syncs) => syncs,
-        Err(err) => err.into_inner(),
-    };
-    syncs.insert(key)
-}
-
-fn clear_remote_installed_plugin_bundle_sync_in_flight(key: &RemoteInstalledPluginBundleSyncKey) {
-    let Some(syncs) = REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_IN_FLIGHT.get() else {
-        return;
-    };
-    let mut syncs = match syncs.lock() {
-        Ok(syncs) => syncs,
-        Err(err) => err.into_inner(),
-    };
-    syncs.remove(key);
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -467,27 +409,6 @@ mod tests {
     use wiremock::matchers::query_param;
     use wiremock::matchers::query_param_is_missing;
 
-    #[test]
-    fn remote_installed_plugin_sync_in_flight_dedupes_by_cache_root() {
-        let codex_home = tempfile::tempdir().expect("create codex home");
-        let key = RemoteInstalledPluginBundleSyncKey {
-            plugin_cache_root: remote_plugin_cache_root(codex_home.path()),
-        };
-
-        assert!(mark_remote_installed_plugin_bundle_sync_in_flight(
-            key.clone()
-        ));
-        assert!(!mark_remote_installed_plugin_bundle_sync_in_flight(
-            key.clone()
-        ));
-
-        clear_remote_installed_plugin_bundle_sync_in_flight(&key);
-        assert!(mark_remote_installed_plugin_bundle_sync_in_flight(
-            key.clone()
-        ));
-        clear_remote_installed_plugin_bundle_sync_in_flight(&key);
-    }
-
     #[tokio::test]
     async fn sync_same_version_backfills_metadata_without_materialization() {
         let server = MockServer::start().await;