diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 0fd577ea9b..6f191667a1 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/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>, tool_suggest_metadata_cache: ToolSuggestMetadataCache, remote_installed_plugins_cache: RwLock>>, + remote_installed_plugin_bundle_sync_gate: Arc, remote_installed_plugins_cache_refresh_state: RwLock, remote_catalog_cache_refresh_state: RwLock, restriction_product: Option, @@ -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( diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index b5088686ab..a6570c9602 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/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 { diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 0e4434ee9f..050c1e551e 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/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; diff --git a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs index a0a4504125..7560c81e61 100644 --- a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs +++ b/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>, +static REMOTE_INSTALLED_PLUGIN_BUNDLE_SYNC_GATES: OnceLock< + Mutex>>, > = OnceLock::new(); static REMOTE_PLUGIN_CACHE_MUTATIONS_IN_FLIGHT: OnceLock< Mutex>, @@ -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, - on_local_cache_changed: Option< - Arc, - >, -) { - let Some(auth) = auth else { - return; +pub(crate) fn remote_installed_plugin_bundle_sync_gate(codex_home: &Path) -> Arc { + 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;