Marketplace: lock each cache on its own; a removed marketplace stays removed (PR review #5, #6)

The single global repo lock becomes one lock per marketplace, and
refresh_pins takes each marketplace's lock only while setting its pins, so
a slow fetch no longer queues installs and refreshes of other marketplaces.

refresh_marketplace now takes the lock first and reads the settings store
under it (a closure, not a copy captured earlier); a marketplace removed
meanwhile gets no cache and no snapshot. Removing a marketplace deletes its
snapshot and cache under the same lock (remove_marketplace_cache).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/src-tauri/src/commands/marketplace_commands.rs

@@ -322,7 +322,8 @@ pub(crate) mod ops {
                 branch: None,
                 account_id: None,
             });
-            let snap = crate::marketplace::refresh_marketplace(&mgr, &settings, "m1").await;
+            let snap =
+                crate::marketplace::refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
 
             let repo = crate::marketplace::git::cache_path(data.path(), "m1");
             let tree = GitTree::open(&repo, &head).unwrap();
@@ -568,21 +569,23 @@ async fn snapshot_blocking(
 }
 
 /// Make each cache's pin refs exactly the commits installs reference, so a
-/// pinned version can never be garbage-collected away. Under the repo lock
-/// (pre-flight F11): a concurrent fetch writes refs in the same repos.
+/// pinned version can never be garbage-collected away. Each marketplace's
+/// pins are set under that marketplace's repo lock only (pre-flight F11, PR
+/// review #5): a concurrent fetch writes refs in the same repo.
 pub(crate) async fn refresh_pins(state: &AppState) {
     let settings = state.settings_store.get();
     let pins = mk::pins_by_marketplace(&settings, &state.projects_store.list());
     let root = state.marketplace.data_root().to_path_buf();
-    let ids: Vec<String> = settings.marketplaces.iter().map(|m| m.id.clone()).collect();
-    let _repo_guard = state.marketplace.repo_lock().lock().await;
-    let _ = tokio::task::spawn_blocking(move || {
-        for id in ids {
-            let repo = git::cache_path(&root, &id);
+    for m in &settings.marketplaces {
+        let lock = state.marketplace.repo_lock(&m.id);
+        let _repo_guard = lock.lock().await;
+        let repo = git::cache_path(&root, &m.id);
+        let commits = pins.get(&m.id).cloned().unwrap_or_default();
+        let id = m.id.clone();
+        let _ = tokio::task::spawn_blocking(move || {
             if !repo.exists() {
-                continue;
+                return;
             }
-            let commits = pins.get(&id).cloned().unwrap_or_default();
             if let Err(e) = git::set_pins(&repo, &commits) {
                 log::warn!(
                     "Could not update the pinned commits of marketplace {}: {}",
@@ -590,28 +593,14 @@ pub(crate) async fn refresh_pins(state: &AppState) {
                     e
                 );
             }
-        }
-    })
-    .await;
+        })
+        .await;
+    }
 }
 
-/// Forget a marketplace's snapshot and delete its cache, under the repo lock.
+/// Forget a marketplace's snapshot and delete its cache, under its repo lock.
 pub(crate) async fn remove_cache(state: &AppState, marketplace_id: &str) {
-    state.marketplace.remove_snapshot(marketplace_id);
-    let path = git::cache_path(state.marketplace.data_root(), marketplace_id);
-    let _repo_guard = state.marketplace.repo_lock().lock().await;
-    let _ = tokio::task::spawn_blocking(move || {
-        if path.exists() {
-            if let Err(e) = std::fs::remove_dir_all(&path) {
-                log::warn!(
-                    "Could not delete the marketplace cache {}: {}",
-                    path.display(),
-                    e
-                );
-            }
-        }
-    })
-    .await;
+    mk::remove_marketplace_cache(&state.marketplace, marketplace_id).await;
 }
 
 fn save_new_account(
@@ -665,7 +654,7 @@ pub async fn refresh_marketplaces(
         .iter()
         .filter(|m| marketplace_id.as_deref().is_none_or(|id| id == m.id))
     {
-        mk::refresh_marketplace(&state.marketplace, &settings, &m.id).await;
+        mk::refresh_marketplace(&state.marketplace, &|| state.settings_store.get(), &m.id).await;
     }
     refresh_pins(&state).await;
     list_marketplace_snapshots(state).await
@@ -696,7 +685,8 @@ pub async fn add_marketplace(
 
     let mut trial = settings.clone();
     trial.marketplaces.push(m.clone());
-    let snap = mk::refresh_marketplace(&state.marketplace, &trial, &m.id).await;
+    // Not yet in the store: the trial settings stand in for it.
+    let snap = mk::refresh_marketplace(&state.marketplace, &|| trial.clone(), &m.id).await;
     let failure = snap.fetch_error.clone().or_else(|| {
         snap.head_commit
             .is_none()

app/src-tauri/src/lib.rs

@@ -311,10 +311,14 @@ pub fn run() {
             // Failures are logged, not toasted — the Marketplace tab shows them.
             {
                 let settings = settings_store_setup.get();
+                let settings_store = settings_store_setup.clone();
                 let marketplace = marketplace_setup.clone();
                 tauri::async_runtime::spawn(async move {
                     for m in &settings.marketplaces {
-                        let snap = crate::marketplace::refresh_marketplace(&marketplace, &settings, &m.id).await;
+                        // Reads the store again under the lock: one removed
+                        // since startup is skipped (PR review #6).
+                        let current = || settings_store.get();
+                        let snap = crate::marketplace::refresh_marketplace(&marketplace, &current, &m.id).await;
                         if let Some(e) = snap.fetch_error {
                             log::warn!("Marketplace \"{}\" could not be refreshed at startup: {}", m.name, e);
                         }

app/src-tauri/src/marketplace/mod.rs

@@ -37,9 +37,10 @@ pub struct MarketplaceManager {
     snapshots: Mutex<HashMap<String, MarketplaceSnapshot>>,
     reports: Mutex<HashMap<String, SyncReport>>,
     gh_login_cancel: tokio::sync::Mutex<Option<oneshot::Sender<()>>>,
-    /// Serialises writers of the bare caches (fetch, pins, cache removal) so
-    /// concurrent refreshes never race on gix ref locks (pre-flight F11a).
-    repo_lock: tokio::sync::Mutex<()>,
+    /// One lock per marketplace cache, serialising its writers (fetch, pins,
+    /// cache removal) so they never race on gix ref locks (pre-flight F11a),
+    /// without one marketplace's fetch holding up another (PR review #5).
+    repo_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
     /// One lock per project, held for a whole `sync_project`, so a start sync
     /// and Apply now never run `sync.sh` in one container at once (F11b).
     sync_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
@@ -62,7 +63,7 @@ impl MarketplaceManager {
             snapshots: Mutex::new(HashMap::new()),
             reports: Mutex::new(HashMap::new()),
             gh_login_cancel: tokio::sync::Mutex::new(None),
-            repo_lock: tokio::sync::Mutex::new(()),
+            repo_locks: Mutex::new(HashMap::new()),
             sync_locks: Mutex::new(HashMap::new()),
         }
     }
@@ -71,10 +72,16 @@ impl MarketplaceManager {
         &self.data_root
     }
 
-    /// Hold while writing to any marketplace cache (fetch, `git::set_pins`,
-    /// removing a cache).
-    pub fn repo_lock(&self) -> &tokio::sync::Mutex<()> {
-        &self.repo_lock
+    /// The marketplace's cache lock. Hold it while writing to that cache
+    /// (fetch, `git::set_pins`, removing it) and never drop it mid-fetch: a
+    /// blocking fetch keeps running after its future is cancelled.
+    pub fn repo_lock(&self, marketplace_id: &str) -> Arc<tokio::sync::Mutex<()>> {
+        self.repo_locks
+            .lock()
+            .unwrap()
+            .entry(marketplace_id.to_string())
+            .or_default()
+            .clone()
     }
 
     /// The project's sync lock; see `sync_project`.
@@ -240,14 +247,23 @@ fn failed_snapshot(
     snap
 }
 
-/// Refresh one marketplace: resolve the credential, fetch (blocking task, under
-/// the repo lock), parse the catalog at head and store the snapshot. On failure
-/// the previous items and head are kept and `fetch_error` is set.
+/// Refresh one marketplace: resolve the credential, fetch (blocking task),
+/// parse the catalog at head and store the snapshot. On failure the previous
+/// items and head are kept and `fetch_error` is set.
+///
+/// Everything runs under the marketplace's repo lock, and `current_settings`
+/// (the settings store as it is *now*, not a copy taken before the lock) is
+/// read only once the lock is held: a marketplace removed meanwhile gets no
+/// cache and no snapshot (PR review #6), since its removal deletes both
+/// under the same lock.
 pub async fn refresh_marketplace(
     mgr: &MarketplaceManager,
-    settings: &AppSettings,
+    current_settings: &(dyn Fn() -> AppSettings + Sync),
     marketplace_id: &str,
 ) -> MarketplaceSnapshot {
+    let lock = mgr.repo_lock(marketplace_id);
+    let _repo_guard = lock.lock().await;
+    let settings = current_settings();
     let Some(m) = settings
         .marketplaces
         .iter()
@@ -275,15 +291,12 @@ pub async fn refresh_marketplace(
 
     let repo = git::cache_path(mgr.data_root(), &m.id);
     let (url, branch) = (m.url.clone(), m.branch.clone());
-    let joined = {
-        let _repo_guard = mgr.repo_lock.lock().await;
-        tokio::task::spawn_blocking(move || {
-            let head = git::fetch(&repo, &url, branch.as_deref(), cred)?;
-            let items = parse_at(&repo, &head).map_err(git::FetchError::Other)?;
-            Ok::<_, git::FetchError>((head, items))
-        })
-        .await
-    };
+    let joined = tokio::task::spawn_blocking(move || {
+        let head = git::fetch(&repo, &url, branch.as_deref(), cred)?;
+        let items = parse_at(&repo, &head).map_err(git::FetchError::Other)?;
+        Ok::<_, git::FetchError>((head, items))
+    })
+    .await;
 
     match joined {
         Ok(Ok((head, items))) => {
@@ -306,6 +319,28 @@ pub async fn refresh_marketplace(
     }
 }
 
+/// Forget a marketplace's snapshot and delete its cache, under its repo lock,
+/// so a refresh already under way either finishes first (and is then
+/// deleted) or sees the marketplace gone and stores nothing.
+pub async fn remove_marketplace_cache(mgr: &MarketplaceManager, marketplace_id: &str) {
+    let lock = mgr.repo_lock(marketplace_id);
+    let _repo_guard = lock.lock().await;
+    mgr.remove_snapshot(marketplace_id);
+    let path = git::cache_path(mgr.data_root(), marketplace_id);
+    let _ = tokio::task::spawn_blocking(move || {
+        if path.exists() {
+            if let Err(e) = std::fs::remove_dir_all(&path) {
+                log::warn!(
+                    "Could not delete the marketplace cache {}: {}",
+                    path.display(),
+                    e
+                );
+            }
+        }
+    })
+    .await;
+}
+
 fn item_changed(repo: &Path, inst: &MarketplaceInstall, head: &str) -> Result<bool, String> {
     let old = GitTree::open(repo, &inst.commit)?;
     let new = GitTree::open(repo, head)?;
@@ -498,7 +533,7 @@ mod tests {
         let data = tempfile::tempdir().unwrap();
         let mgr = MarketplaceManager::new(data.path().to_path_buf());
 
-        let snap = refresh_marketplace(&mgr, &settings_with(&fx.url()), "m1").await;
+        let snap = refresh_marketplace(&mgr, &|| settings_with(&fx.url()), "m1").await;
 
         assert_eq!(snap.fetch_error, None);
         assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
@@ -530,11 +565,11 @@ mod tests {
         let data = tempfile::tempdir().unwrap();
         let mgr = MarketplaceManager::new(data.path().to_path_buf());
         let settings = settings_with(&url);
-        let first = refresh_marketplace(&mgr, &settings, "m1").await;
+        let first = refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
         assert_eq!(first.fetch_error, None);
 
         drop(fx); // the source repository disappears (offline, deleted, …)
-        let second = refresh_marketplace(&mgr, &settings, "m1").await;
+        let second = refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
 
         assert!(second.fetch_error.is_some(), "expected a fetch error");
         assert_eq!(second.head_commit.as_deref(), Some(c1.as_str()));
@@ -550,10 +585,11 @@ mod tests {
         let mgr = MarketplaceManager::new(data.path().to_path_buf());
         let settings = settings_with(&fx.url());
 
-        let guard = mgr.repo_lock().lock().await;
+        let lock = mgr.repo_lock("m1");
+        let guard = lock.lock().await;
         let blocked = tokio::time::timeout(
             std::time::Duration::from_millis(300),
-            refresh_marketplace(&mgr, &settings, "m1"),
+            refresh_marketplace(&mgr, &|| settings.clone(), "m1"),
         )
         .await;
         assert!(blocked.is_err(), "refresh must not fetch while the repo lock is held");
@@ -563,10 +599,102 @@ mod tests {
         );
         drop(guard);
 
-        let snap = refresh_marketplace(&mgr, &settings, "m1").await;
+        let snap = refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
         assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
     }
 
+    #[test]
+    fn repo_locks_are_per_marketplace() {
+        let mgr = MarketplaceManager::new(std::env::temp_dir());
+        let a1 = mgr.repo_lock("a");
+        let a2 = mgr.repo_lock("a");
+        let b = mgr.repo_lock("b");
+        assert!(Arc::ptr_eq(&a1, &a2), "one lock per marketplace");
+        assert!(!Arc::ptr_eq(&a1, &b), "marketplaces do not block each other");
+    }
+
+    /// PR review #5: a long fetch of one marketplace must not hold up work
+    /// (another refresh, pins, installs) on a different one.
+    #[tokio::test]
+    async fn a_busy_marketplace_does_not_block_another() {
+        let Some(fx) = GitFixture::new() else { return };
+        let c1 = fx.with_all_kinds();
+        let data = tempfile::tempdir().unwrap();
+        let mgr = MarketplaceManager::new(data.path().to_path_buf());
+        let settings = settings_with(&fx.url());
+
+        let other = mgr.repo_lock("some-other-marketplace");
+        let _busy = other.lock().await;
+        let snap = tokio::time::timeout(
+            std::time::Duration::from_secs(20),
+            refresh_marketplace(&mgr, &|| settings.clone(), "m1"),
+        )
+        .await
+        .expect("m1 must not wait for another marketplace's lock");
+        assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
+    }
+
+    /// PR review #6: a refresh that was already under way when the
+    /// marketplace was removed must not recreate its cache or snapshot.
+    #[tokio::test]
+    async fn a_refresh_of_a_removed_marketplace_leaves_nothing_behind() {
+        let Some(fx) = GitFixture::new() else { return };
+        fx.with_all_kinds();
+        let data = tempfile::tempdir().unwrap();
+        let mgr = MarketplaceManager::new(data.path().to_path_buf());
+        let current = Mutex::new(settings_with(&fx.url()));
+        let read_current = || current.lock().unwrap().clone();
+
+        let lock = mgr.repo_lock("m1");
+        let guard = lock.lock().await;
+        let refresh = refresh_marketplace(&mgr, &read_current, "m1");
+        tokio::pin!(refresh);
+        // The refresh starts, then waits for the lock…
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(100), &mut refresh)
+                .await
+                .is_err()
+        );
+        // …while the marketplace is removed from settings.
+        current.lock().unwrap().marketplaces.clear();
+        drop(guard);
+        let snap = refresh.await;
+
+        assert!(
+            snap.fetch_error.as_deref().unwrap_or("").contains("no longer configured"),
+            "{snap:?}"
+        );
+        assert!(!git::cache_path(data.path(), "m1").exists(), "cache recreated");
+        assert_eq!(mgr.snapshot("m1"), None, "snapshot stored");
+    }
+
+    #[tokio::test]
+    async fn removing_a_cache_waits_for_the_marketplaces_lock() {
+        let Some(fx) = GitFixture::new() else { return };
+        fx.with_all_kinds();
+        let data = tempfile::tempdir().unwrap();
+        let mgr = MarketplaceManager::new(data.path().to_path_buf());
+        let settings = settings_with(&fx.url());
+        refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
+        let cache = git::cache_path(data.path(), "m1");
+        assert!(cache.exists());
+
+        let lock = mgr.repo_lock("m1");
+        let guard = lock.lock().await;
+        let blocked = tokio::time::timeout(
+            std::time::Duration::from_millis(200),
+            remove_marketplace_cache(&mgr, "m1"),
+        )
+        .await;
+        assert!(blocked.is_err(), "removal must wait for an in-flight fetch");
+        assert!(cache.exists());
+        drop(guard);
+
+        remove_marketplace_cache(&mgr, "m1").await;
+        assert!(!cache.exists());
+        assert_eq!(mgr.snapshot("m1"), None);
+    }
+
     #[tokio::test]
     async fn concurrent_refreshes_all_succeed() {
         let Some(fx) = GitFixture::new() else { return };
@@ -575,10 +703,11 @@ mod tests {
         let mgr = MarketplaceManager::new(data.path().to_path_buf());
         let settings = settings_with(&fx.url());
 
+        let current = || settings.clone();
         let (a, b, c) = tokio::join!(
-            refresh_marketplace(&mgr, &settings, "m1"),
-            refresh_marketplace(&mgr, &settings, "m1"),
-            refresh_marketplace(&mgr, &settings, "m1"),
+            refresh_marketplace(&mgr, &current, "m1"),
+            refresh_marketplace(&mgr, &current, "m1"),
+            refresh_marketplace(&mgr, &current, "m1"),
         );
         for snap in [a, b, c] {
             assert_eq!(snap.fetch_error, None);
@@ -594,7 +723,7 @@ mod tests {
         let settings = settings_with(&fx.url());
         {
             let mgr = MarketplaceManager::new(data.path().to_path_buf());
-            refresh_marketplace(&mgr, &settings, "m1").await;
+            refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
         }
         drop(fx);
         let mgr = MarketplaceManager::new(data.path().to_path_buf());
@@ -622,7 +751,7 @@ mod tests {
         ];
         let mut project = crate::models::Project::new("p".into(), vec![]);
         project.marketplace_installs = vec![install(ItemKind::Skill, "example-skill", &c1)];
-        refresh_marketplace(&mgr, &settings, "m1").await;
+        refresh_marketplace(&mgr, &|| settings.clone(), "m1").await;
 
         let updates = compute_updates(&mgr, &settings, &[project]);