fix: serialize concurrent clones & recover from leftover dirs

CHANGELOG.md

@@ -26,11 +26,14 @@
 - Restructure the core crate into focused modules for announcements, pull requests, git events, inbox, and upstream state
 - Publish nostr events directly through the client with relay acceptance required, removing the grasp push helper and its per-server retry loop so failures surface immediately
 - Poll the repository sync drift check every 180 seconds instead of 60
+- Track sync status per view instead of globally
 - Merge the `PixelAvatar` into the `Avatar` component. Seed user pixel avatars from the public key instead of the display name, so avatars stay stable when a user renames
 
 ### Fixed
 
 - Fix the repository view refreshing its mirror from the network while a push is in progress, so the refresh no longer races the push
+- Fix opening a repository failing with a `destination already exists` error when two loads raced to clone its mirror, clones are now serialized and later loads reuse the finished clone
+- Fix a repository never opening again after an interrupted clone left an unusable mirror directory behind, which is now removed and re-cloned
 - Fix signing in with a `bunker://` URI silently generating a new identity instead of erroring when its master key cannot be parsed
 
 ### Removed

crates/signed_git/src/cache.rs

@@ -1,6 +1,8 @@
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, PoisonError};
 
-use anyhow::{Context, Result};
+use anyhow::Result;
 use signed_core::{Announcement, RepoAddr};
 
 use crate::repo::Repo;
@@ -8,12 +10,16 @@ use crate::repo::Repo;
 #[derive(Debug, Clone)]
 pub struct GitCache {
     root: PathBuf,
+    clone_locks: Arc<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>>,
 }
 
 impl GitCache {
     /// Creates a cache rooted at `root`.
     pub fn new(root: PathBuf) -> Self {
-        Self { root }
+        Self {
+            root,
+            clone_locks: Arc::default(),
+        }
     }
 
     /// Returns the cache root directory.
@@ -38,20 +44,40 @@ impl GitCache {
         }
     }
 
+    /// Returns the clone lock for `path`, creating it on first use.
+    fn clone_lock(&self, path: &Path) -> Arc<Mutex<()>> {
+        let mut locks = self
+            .clone_locks
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner);
+        Arc::clone(locks.entry(path.to_path_buf()).or_default())
+    }
+
     /// Returns the cached repository, cloning from one of `clone_urls` on first use.
     pub fn ensure_clone<U: AsRef<str>>(&self, addr: &RepoAddr, clone_urls: &[U]) -> Result<Repo> {
         let path = self.repo_path(addr);
 
-        if let Some(repo) = self.open(addr)? {
+        // Lock the clone lock for this path to prevent concurrent clones.
+        let clone_lock = self.clone_lock(&path);
+        let _clone_guard = clone_lock.lock().unwrap_or_else(PoisonError::into_inner);
+
+        // Refreshes the cached repository if it is already cloned, or clones it from one of `clone_urls` otherwise.
+        if let Ok(Some(repo)) = self.open(addr) {
             if let Err(error) = repo.fetch() {
                 log::warn!("failed to refresh the cached repository: {error:#}");
             }
             return Ok(repo);
         }
 
+        // Create the parent directory if it does not exist.
         if let Some(parent) = path.parent() {
-            std::fs::create_dir_all(parent)
+            std::fs::create_dir_all(parent)?;
-                .with_context(|| format!("failed to create {}", parent.display()))?;
+        }
+
+        // Remove the cached clone if it exists but is unusable.
+        if path.exists() {
+            log::warn!("removing unusable cached clone at {}", path.display());
+            std::fs::remove_dir_all(&path)?;
         }
 
         Repo::clone(clone_urls, &path)

crates/signed_git/src/tests.rs

@@ -1,6 +1,9 @@
 use std::path::{Path, PathBuf};
 use std::process::Command;
 
+use nostr::prelude::Keys;
+use signed_core::RepoAddr;
+
 use super::*;
 
 #[test]
@@ -11,6 +14,70 @@ fn blocks_parent_components() {
     assert_eq!(GitCache::sanitize_path_component("a/../b"), "a_.._b");
 }
 
+#[test]
+fn concurrent_ensure_clone_succeeds_for_every_caller() {
+    let server = tempfile::tempdir().expect("tempdir");
+    let server_repo = bare_server(server.path(), "npub1test", "repo");
+
+    let (work_dir, work_repo) = fixture(&[("a.txt", b"one")]);
+    commit_all(&work_repo, "initial");
+    let work_dir = work_dir.path();
+
+    let base_url = format!("file://{}", server.path().display());
+    Repo::open(work_dir)
+        .expect("open")
+        .push_all(&base_url, "npub1test", "repo")
+        .expect("push");
+
+    let cache_dir = tempfile::tempdir().expect("tempdir");
+    let cache = GitCache::new(cache_dir.path().to_path_buf());
+    let addr = RepoAddr::new(Keys::generate().public_key(), "repo");
+    let clone_url = format!("file://{}", server_repo.display());
+
+    std::thread::scope(|scope| {
+        for _ in 0..4 {
+            scope.spawn(|| {
+                cache
+                    .ensure_clone(&addr, &[&clone_url])
+                    .expect("ensure_clone");
+            });
+        }
+    });
+
+    assert!(cache.open(&addr).expect("open").is_some());
+}
+
+#[test]
+fn ensure_clone_recovers_from_a_leftover_directory() {
+    let server = tempfile::tempdir().expect("tempdir");
+    let server_repo = bare_server(server.path(), "npub1test", "repo");
+
+    let (work_dir, work_repo) = fixture(&[("a.txt", b"one")]);
+    commit_all(&work_repo, "initial");
+    let work_dir = work_dir.path();
+
+    let base_url = format!("file://{}", server.path().display());
+    Repo::open(work_dir)
+        .expect("open")
+        .push_all(&base_url, "npub1test", "repo")
+        .expect("push");
+
+    let cache_dir = tempfile::tempdir().expect("tempdir");
+    let cache = GitCache::new(cache_dir.path().to_path_buf());
+    let addr = RepoAddr::new(Keys::generate().public_key(), "repo");
+    let clone_url = format!("file://{}", server_repo.display());
+
+    // Simulate the remains of an interrupted clone.
+    let leftover = cache.repo_path(&addr);
+    std::fs::create_dir_all(&leftover).expect("mkdir");
+
+    cache
+        .ensure_clone(&addr, &[&clone_url])
+        .expect("ensure_clone");
+
+    assert!(cache.open(&addr).expect("open").is_some());
+}
+
 #[test]
 fn root_commit_reports_the_first_ancestor() {
     let (dir, repo) = fixture(&[("a.txt", b"one")]);