fix(forge): stop a default-branch change from landing on a branch a push is deleting

crates/forge/src/cgi.rs

@@ -502,6 +502,11 @@ fn environment(l: &Launch, marker: &std::path::Path) -> Vec<(String, String)> {
     var("FORGE_GIT_BIN", git.bin().display().to_string());
     var("FORGE_USER_ID", l.plan.user.id.to_string());
     var("FORGE_ACCEPT_MARKER", marker.display().to_string());
+    // Test-only knob, read from the server's own environment: see `hook::hold`.
+    #[cfg(debug_assertions)]
+    if let Some(p) = std::env::var_os("SC_FORGE_HOOK_HOLD") {
+        var("FORGE_TEST_HOOK_HOLD", p.to_string_lossy().into_owned());
+    }
     env
 }
 

crates/forge/src/hook.rs

@@ -197,6 +197,20 @@ fn retain_deleted_heads(
     Ok(())
 }
 
+/// Test-only knob (`SC_FORGE_HOOK_HOLD` on the server): create `marker` once the row has been
+/// read, then wait (at most 30 s) until the test removes it, so a test can act between this
+/// push's read of the default branch and its ref update. Debug builds only.
+#[cfg(debug_assertions)]
+fn hold(marker: &Path) {
+    if std::fs::write(marker, b"").is_err() {
+        return;
+    }
+    let end = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    while marker.exists() && std::time::Instant::now() < end {
+        std::thread::sleep(std::time::Duration::from_millis(10));
+    }
+}
+
 /// Entry point: returns the process exit code.
 pub fn run() -> i32 {
     let (Some(data), Some(id)) = (
@@ -224,6 +238,10 @@ pub fn run() -> i32 {
             return 1;
         }
     };
+    #[cfg(debug_assertions)]
+    if let Some(p) = std::env::var_os("FORGE_TEST_HOOK_HOLD") {
+        hold(Path::new(&p));
+    }
     let git_bin = std::env::var_os("FORGE_GIT_BIN").map(PathBuf::from);
     let is_ff = |old: &str, new: &str| match &git_bin {
         Some(b) => is_ff_in(Path::new(&data), id, b, old, new),

crates/forge/src/lifecycle.rs

@@ -15,7 +15,8 @@
 //! shared lifecycle lock, each taken once. Nothing beneath that boundary may take the lifecycle lock
 //! again (no `with_read` helper): the lock is writer-preferring, so a queued delete between two
 //! shared acquisitions by one request deadlocks it. Never take the gate while holding the lifecycle
-//! lock.
+//! lock. The transport registry's mutex is innermost: never take the lifecycle lock or the gate
+//! under it (`Registry::if_no_writer_since`, `with_only`).
 
 use crate::gitproc::{Git, GitError, Identity, Spec};
 use crate::refname;
@@ -28,6 +29,9 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
+/// How long a default-branch change waits for the repository's pushes to finish.
+const PUSH_WAIT: Duration = Duration::from_secs(10);
+
 /// Exclusive hold on the data directory for the life of the process.
 pub struct DirLock(#[allow(dead_code)] File);
 
@@ -75,6 +79,9 @@ pub enum LifecycleError {
     /// A branch protection rule forbids the operation.
     #[error("{0}")]
     Protected(String),
+    /// A push or maintenance pass is writing the repository; the caller may retry.
+    #[error("busy")]
+    Busy,
     /// A concurrent change kept winning; the caller may retry.
     #[error("{0}")]
     Conflict(String),
@@ -118,6 +125,9 @@ pub struct Lifecycle {
     gates: Mutex<HashMap<i64, Arc<crate::limits::Gate>>>,
     #[cfg(test)]
     crash_at: Mutex<Option<&'static str>>,
+    /// Tests: a push registers (and leaves) right after `set_default_branch` wrote `HEAD`.
+    #[cfg(test)]
+    push_after_head: std::sync::atomic::AtomicBool,
 }
 
 impl Lifecycle {
@@ -140,6 +150,8 @@ impl Lifecycle {
             gates: Mutex::default(),
             #[cfg(test)]
             crash_at: Mutex::default(),
+            #[cfg(test)]
+            push_after_head: Default::default(),
         })
     }
 
@@ -486,13 +498,29 @@ impl Lifecycle {
     }
 
     /// Make `branch` the default: it must exist; bare `HEAD` is written first, then the row.
+    ///
+    /// The `pre-receive` hook reads the default branch when a push starts, so a push already in
+    /// flight could delete the new default after it is set. The row is therefore committed only if
+    /// no push (or maintenance pass) has registered since the branch was checked; otherwise `HEAD`
+    /// is put back and the answer is `Busy`. A push that registers later reads the new row.
     pub fn set_default_branch(
         &self,
         conn: &Connection,
         repo: &Repo,
         branch: &str,
         now: i64,
     ) -> Result<(), LifecycleError> {
+        // Naming the current default changes nothing, so it need not wait for pushes (clients that
+        // send the whole repository object on every edit do this). A concurrent PATCH may have
+        // moved it since; this one is then ordered before that one.
+        if repos::find_by_id(conn, repo.id)?.is_some_and(|r| r.default_branch == branch) {
+            return Ok(());
+        }
+        // Before the lock: waiting for a push must not hold up the repository's readers.
+        let since = self
+            .registry
+            .writers_idle(repo.id, PUSH_WAIT)
+            .ok_or(LifecycleError::Busy)?;
         let lock = self.lock_for(repo.id);
         let _held = lock.write().unwrap_or_else(|e| e.into_inner());
         let current = repos::find_by_id(conn, repo.id)?
@@ -508,14 +536,31 @@ impl Lifecycle {
         }
         self.run(&dir, &["symbolic-ref", "HEAD", &full], None, None)?;
         self.step("head-written")?;
+        #[cfg(test)]
+        if self
+            .push_after_head
+            .load(std::sync::atomic::Ordering::SeqCst)
+        {
+            drop(self.registry.register(current.id, true, || true));
+        }
         let patch = RepoPatch {
             default_branch: Some(branch.to_string()),
             ..Default::default()
         };
-        if !repos::update_live(conn, current.id, &patch, now)? {
-            return Err(LifecycleError::Gone);
-        }
-        Ok(())
+        let committed = self.registry.if_no_writer_since(current.id, since, || {
+            repos::update_live(conn, current.id, &patch, now)
+        });
+        let failed = match committed {
+            Some(Ok(true)) => return Ok(()),
+            Some(Ok(false)) => return Err(LifecycleError::Gone),
+            Some(Err(e)) => e.into(),
+            None => LifecycleError::Busy,
+        };
+        // The row still names the old default: put `HEAD` back. A failed restore is what the
+        // reconciler's HEAD repair fixes, as after a crash at `head-written`.
+        let old = refname::branch_ref(&current.default_branch);
+        self.run(&dir, &["symbolic-ref", "HEAD", &old], None, None)?;
+        Err(failed)
     }
 
     /// Delete a branch (the caller has refused the default branch). `NoSuchBranch` when absent;
@@ -1048,6 +1093,32 @@ mod tests {
         assert!(!dir.exists());
     }
 
+    #[test]
+    fn a_push_that_registers_during_the_change_puts_head_back_and_answers_busy() {
+        let e = env("default-race");
+        let r = make(&e, "alpha", true).unwrap();
+        let dir = e.lc.repo_dir(r.id);
+        let main = rev(&e, &r, "main").unwrap();
+        e.lc.run(&dir, &["update-ref", "refs/heads/dev", &main], None, None)
+            .unwrap();
+        e.lc.push_after_head
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        assert!(matches!(
+            e.lc.set_default_branch(&e.conn, &r, "dev", 20),
+            Err(LifecycleError::Busy)
+        ));
+        assert_eq!(head(&e, &r), "refs/heads/main");
+        let row = repos::find_by_id(&e.conn, r.id).unwrap().unwrap();
+        assert_eq!(row.default_branch, "main");
+        e.lc.push_after_head
+            .store(false, std::sync::atomic::Ordering::SeqCst);
+        e.lc.set_default_branch(&e.conn, &r, "dev", 21).unwrap();
+        assert_eq!(head(&e, &r), "refs/heads/dev");
+        // A live push makes a real change wait, but not one that names the current default.
+        let _push = e.lc.registry.register(r.id, true, || true).unwrap();
+        e.lc.set_default_branch(&e.conn, &r, "dev", 22).unwrap();
+    }
+
     #[test]
     fn set_default_branch_requires_an_existing_branch_and_writes_head_then_row() {
         let e = env("default");

crates/forge/src/registry.rs

@@ -64,6 +64,8 @@ struct Slot {
     closed: bool,
     /// Cancel handle and whether the run may write (a push, a maintenance pass).
     live: HashMap<u64, (Cancel, bool)>,
+    /// Writers registered so far: lets [`Registry::if_no_writer_since`] see one that came and went.
+    writers: u64,
 }
 
 /// Live transports by repository id.
@@ -107,11 +109,9 @@ impl Registry {
         }
         let key = self.next.fetch_add(1, Ordering::SeqCst);
         let cancel = Cancel::default();
-        slots
-            .entry(repo)
-            .or_default()
-            .live
-            .insert(key, (cancel.clone(), writes));
+        let slot = slots.entry(repo).or_default();
+        slot.writers += u64::from(writes);
+        slot.live.insert(key, (cancel.clone(), writes));
         Some(Registration {
             reg: self.clone(),
             repo,
@@ -133,9 +133,6 @@ impl Registry {
         let mut slots = self.lock();
         if let Some(s) = slots.get_mut(&repo) {
             s.closed = false;
-            if s.live.is_empty() {
-                slots.remove(&repo);
-            }
         }
     }
 
@@ -174,6 +171,33 @@ impl Registry {
         }
     }
 
+    /// Wait up to `limit` until no writer (a push, a maintenance pass) is registered on `repo`, and
+    /// return the writer count to hand to [`Registry::if_no_writer_since`]. `None` on timeout.
+    pub fn writers_idle(&self, repo: i64, limit: Duration) -> Option<u64> {
+        let end = Instant::now() + limit;
+        let mut slots = self.lock();
+        loop {
+            match slots.get(&repo) {
+                Some(s) if s.live.values().any(|(_, w)| *w) => {}
+                s => return Some(s.map_or(0, |s| s.writers)),
+            }
+            let left = end.checked_duration_since(Instant::now())?;
+            slots = self
+                .idle
+                .wait_timeout(slots, left)
+                .unwrap_or_else(|e| e.into_inner())
+                .0;
+        }
+    }
+
+    /// Run `f` only if no writer has registered on `repo` since [`Registry::writers_idle`] returned
+    /// `since`, holding the registry lock so none can register meanwhile: a push that starts after
+    /// `f` sees what `f` committed. `f` must be short and must not call back into the registry.
+    pub fn if_no_writer_since<T>(&self, repo: i64, since: u64, f: impl FnOnce() -> T) -> Option<T> {
+        let slots = self.lock();
+        (slots.get(&repo).map_or(0, |s| s.writers) == since).then(f)
+    }
+
     /// Drop the (closed) slot of a repository that no longer exists.
     pub fn forget(&self, repo: i64) {
         self.lock().remove(&repo);
@@ -222,10 +246,8 @@ impl Drop for Registration {
     fn drop(&mut self) {
         let mut slots = self.reg.lock();
         if let Some(s) = slots.get_mut(&self.repo) {
+            // The slot stays (it keeps the writer count); `forget` drops it with the repository.
             s.live.remove(&self.key);
-            if s.live.is_empty() && !s.closed {
-                slots.remove(&self.repo);
-            }
         }
         self.reg.idle.notify_all();
     }
@@ -257,6 +279,26 @@ mod tests {
         assert!(r.register(1, true, || true).is_some());
     }
 
+    #[test]
+    fn a_writer_that_came_and_went_still_counts() {
+        let r = Arc::new(Registry::default());
+        let fetch = r.register(3, false, || true).unwrap();
+        let since = r
+            .writers_idle(3, Duration::ZERO)
+            .expect("a fetch is not a writer");
+        assert_eq!(r.if_no_writer_since(3, since, || 1), Some(1));
+        drop(fetch); // the slot empties: its writer count must survive that
+        drop(r.register(3, true, || true).unwrap()); // a push starts and ends
+        assert_eq!(r.if_no_writer_since(3, since, || 1), None);
+        let push = r.register(3, true, || true).unwrap();
+        assert!(
+            r.writers_idle(3, Duration::from_millis(20)).is_none(),
+            "a live push"
+        );
+        drop(push);
+        assert!(r.writers_idle(3, Duration::ZERO).is_some());
+    }
+
     #[test]
     fn the_liveness_check_runs_under_the_lock_and_can_refuse() {
         let r = Arc::new(Registry::default());

crates/forge/src/repo_api.rs

@@ -33,6 +33,11 @@ fn life_err(app: &App, e: LifecycleError) -> Resp {
         LifecycleError::Protected(m) => error(&app.base, 403, &m),
         LifecycleError::Conflict(m) => error(&app.base, 409, &m),
         LifecycleError::Gone => not_found(&app.base, "The target couldn't be found.", None),
+        LifecycleError::Busy => error(
+            &app.base,
+            503,
+            "this repository is being written to, try again",
+        ),
         LifecycleError::Git(crate::gitproc::GitError::Busy) => {
             error(&app.base, 503, "too many git operations in flight")
         }