diff --git a/app/src-tauri/src/commands/marketplace_commands.rs b/app/src-tauri/src/commands/marketplace_commands.rs index 583dda1..7e0bae0 100644 --- a/app/src-tauri/src/commands/marketplace_commands.rs +++ b/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 = 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() diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index e9097a3..5e90539 100644 --- a/app/src-tauri/src/lib.rs +++ b/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, ¤t, &m.id).await; if let Some(e) = snap.fetch_error { log::warn!("Marketplace \"{}\" could not be refreshed at startup: {}", m.name, e); } diff --git a/app/src-tauri/src/marketplace/mod.rs b/app/src-tauri/src/marketplace/mod.rs index 34f6bdb..9474bd3 100644 --- a/app/src-tauri/src/marketplace/mod.rs +++ b/app/src-tauri/src/marketplace/mod.rs @@ -37,9 +37,10 @@ pub struct MarketplaceManager { snapshots: Mutex>, reports: Mutex>, gh_login_cancel: tokio::sync::Mutex>>, - /// 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>>>, /// 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>>>, @@ -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> { + 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 { 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, ¤t, "m1"), + refresh_marketplace(&mgr, ¤t, "m1"), + refresh_marketplace(&mgr, ¤t, "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]);