add push

crates/signed_state/src/backend.rs

@@ -1,8 +1,9 @@
-use std::collections::HashMap;
 use std::collections::hash_map::DefaultHasher;
+use std::collections::{HashMap, HashSet};
 use std::hash::{Hash, Hasher};
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use anyhow::{Context as AnyhowContext, Error, anyhow, bail};
@@ -96,6 +97,13 @@ pub struct Backend {
     /// so duplicate requests within [`FETCH_DEDUP_WINDOW`] collapse into
     /// one. Entries are pruned lazily on the next request.
     recent_fetches: HashMap<u64, Instant>,
+    /// Repositories a push (mirror or checkout) is currently in flight
+    /// for. Concurrent pushes of the same refs — two panels of the same
+    /// repository, or the banner push racing the header's Republish — make
+    /// the losing push fail server-side with a compare-and-swap rejection
+    /// ("cannot lock ref … is at … but expected …"), so pushes are
+    /// single-flight per repository.
+    pushing_repos: Arc<Mutex<HashSet<RepoAddr>>>,
     tasks: Vec<Task<Result<(), Error>>>,
 }
 
@@ -103,6 +111,22 @@ struct GlobalBackend(Entity<Backend>);
 
 impl Global for GlobalBackend {}
 
+/// Removes its repository from the in-flight push set when dropped, so a
+/// push task that is cancelled (e.g. its panel closed mid-push) can never
+/// leave the repository locked for the rest of the session.
+struct PushGuard {
+    repos: Arc<Mutex<HashSet<RepoAddr>>>,
+    addr: RepoAddr,
+}
+
+impl Drop for PushGuard {
+    fn drop(&mut self) {
+        if let Ok(mut repos) = self.repos.lock() {
+            repos.remove(&self.addr);
+        }
+    }
+}
+
 impl EventEmitter<BackendEvent> for Backend {}
 
 impl Backend {
@@ -147,6 +171,7 @@ impl Backend {
             sync_progress: None,
             passphrase_required: false,
             recent_fetches: HashMap::new(),
+            pushing_repos: Arc::new(Mutex::new(HashSet::new())),
             tasks: vec![pump],
         };
 
@@ -758,9 +783,56 @@ impl Backend {
         announcement: Announcement,
         cx: &mut Context<Self>,
     ) -> Task<Result<(), Error>> {
-        let addr = announcement.addr();
         let cache = GitStore::global(cx).cache().clone();
-        let path = cache.repo_path(&addr);
+        let path = cache.repo_path(&announcement.addr());
+        self.push_repo_from(announcement, path, None, cx)
+    }
+
+    /// Push the refs of a local checkout (the working copy of the user's
+    /// own repository) to the grasp servers announced in the `relays` tag:
+    /// publishes a fresh state event, then pushes every branch and tag of
+    /// the checkout, like the init flow. `announced_head` keeps the state
+    /// event's `HEAD` on the repository's announced default branch when the
+    /// checkout is on a different branch.
+    pub fn push_checkout(
+        &mut self,
+        announcement: Announcement,
+        checkout: PathBuf,
+        announced_head: Option<String>,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<(), Error>> {
+        self.push_repo_from(announcement, checkout, announced_head, cx)
+    }
+
+    /// Shared body of the mirror-based and checkout-based pushes: publish
+    /// the repository state (the push authorization), then push every
+    /// branch and tag of `path` to each announced grasp server. Pushes are
+    /// single-flight per repository: two concurrent pushes of the same refs
+    /// (e.g. two panels of the same repository) make the losing push fail
+    /// server-side with a compare-and-swap rejection.
+    fn push_repo_from(
+        &mut self,
+        announcement: Announcement,
+        path: PathBuf,
+        announced_head: Option<String>,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<(), Error>> {
+        let addr = announcement.addr();
+        let guard = {
+            let mut pushing = self
+                .pushing_repos
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if !pushing.insert(addr.clone()) {
+                return Task::ready(Err(anyhow!(
+                    "A push to this repository is already in progress"
+                )));
+            }
+            PushGuard {
+                repos: self.pushing_repos.clone(),
+                addr: addr.clone(),
+            }
+        };
         let owner = announcement
             .owner
             .to_bech32()
@@ -769,11 +841,32 @@ impl Backend {
         let relays = announcement.relays.clone();
 
         cx.spawn(async move |this, cx| {
-            let work = cx.background_spawn({
-                let path = path.clone();
-                async move { signed_git::worktree_ref_state(&path) }
-            });
-            let state = work.await?;
+            // Held for the whole task; dropped (and the lock released) on
+            // completion, on error and on cancellation alike.
+            let _guard = guard;
+
+            let mut state = {
+                let work = cx.background_spawn({
+                    let path = path.clone();
+                    async move { signed_git::worktree_ref_state(&path) }
+                });
+                work.await?
+            };
+
+            // The state event announces the pushed refs. When the source is
+            // a checkout on a side branch, keep the repository's announced
+            // default branch (its `HEAD`) when that branch is among the
+            // pushed refs; otherwise the checkout's current branch.
+            let heads: Vec<&str> = state
+                .refs
+                .iter()
+                .filter_map(|(name, _)| name.strip_prefix("refs/heads/"))
+                .collect();
+            if let Some(head) = announced_head
+                && heads.iter().any(|branch| *branch == head)
+            {
+                state.head = Some(head);
+            }
 
             // Grasp servers authorize a push by the state they have seen.
             let refs = state.refs.clone();