fix(node): satisfy clippy on stale-local download test path join

Under-lock GET failures now always shed as RepoUnavailable, even when a local
copy exists, so a transient download error cannot publish over a newer stored
generation. PublishDurabilitySlot only synthesizes UploadUnknowable after
release starts, and receive-pack marks that boundary before spawning the tail.

Adds regression tests for both paths; inv22 F4 gate accepts the release wrapper.

.env.example

@@ -24,11 +24,26 @@ DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb
 # ── Database pool & startup resilience ────────────────────────────────────
 # Maximum connections in the PostgreSQL pool. A cap, not a floor —
 # connections open lazily. Size against the DB server's max_connections,
-# remembering admin tooling opens its own pool. Each concurrent write pins one
-# connection for its whole duration (the connection-affine advisory lock), so the
-# node REJECTS at boot any value below GITLAWB_MAX_CONCURRENT_GIT_PUSHES + 8
-# headroom — keep this comfortably above that (default 48 for pushes 32).
+# remembering admin tooling opens its own pool.
 GITLAWB_DB_MAX_CONNECTIONS=48
+# Maximum connections in the DEDICATED advisory-lock pool, separate from the
+# pool above. Every in-flight repo write pins one connection here for its whole
+# duration, so this is a hard ceiling on simultaneous writes node-wide: size it
+# to expected peak concurrent writers, not small. Keeping it separate is what
+# stops a push burst from starving ordinary request handlers. Budget
+# (GITLAWB_DB_MAX_CONNECTIONS + this) per node against the server's
+# max_connections, times node count, plus admin tooling.
+GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS=40
+# Upper bound, in seconds, on any object-storage transfer that runs while a
+# per-repo write lock is HELD (the archive download inside acquire_write and the
+# upload inside release). These were free before the lock's connection was
+# pinned to the guard; now an unbounded stall holds a lock-pool slot, and enough
+# stalls deny every write on the node. The bound applies PER SPAN and there are
+# two (the acquire-side refresh, which covers the existence check and download
+# together, and the release-side upload), so worst-case slot occupancy is about
+# twice this value plus the git work between them. Read it together with the pool
+# size above and with GITLAWB_GIT_SERVICE_TIMEOUT_SECS.
+GITLAWB_LOCK_HELD_TRANSFER_TIMEOUT_SECS=300
 # Seconds a request waits for a pool connection before failing with 503.
 GITLAWB_DB_ACQUIRE_TIMEOUT_SECS=5
 # Upper bound on each startup connect+migrate attempt, in seconds. Keep it

crates/git-remote-gitlawb/src/main.rs

@@ -287,10 +287,8 @@ fn handle_connect<R: Read>(
     //    this into one POST deadlocks a multi-round fetch (#117).
     //
     //  git-receive-pack (push):
-    //    Git sends the ref-update command list (pkt-lines ending in a flush),
-    //    then the complete PACK blob only when a command has a non-zero new-oid.
-    //    A delete-only push sends no pack, and git holds its write pipe open for
-    //    the report-status response, so EOF cannot delimit the request (#369).
+    //    Git sends ref-update commands + the complete PACK blob, then closes its
+    //    write pipe. read_to_end is safe and correct here, a single POST.
 
     let post_url = format!("{}/{}", repo_base, service);
 
@@ -298,7 +296,10 @@ fn handle_connect<R: Read>(
         return negotiate_upload_pack(&client, &post_url, service, signing_key, stdin, &mut stdout);
     }
 
-    let request_body = read_receive_pack_request(stdin)?;
+    let mut request_body = Vec::new();
+    stdin
+        .read_to_end(&mut request_body)
+        .context("reading receive-pack request")?;
     tracing::debug!("pack request: {} bytes from git", request_body.len());
 
     if request_body.is_empty() {
@@ -534,108 +535,6 @@ fn read_upload_pack_round<R: Read>(stdin: &mut R) -> Result<(Vec<u8>, RoundEnd)>
     }
 }
 
-/// Read one git-receive-pack request: optional `shallow` lines, then EITHER a
-/// command list (pkt-lines terminated by a flush) or a push-cert block
-/// (pkt-lines terminated by a "push-cert-end" line), then the negotiated
-/// push-options block, then the pack ONLY when a command carries a non-zero
-/// new-oid.
-///
-/// A delete-only push sends no pack, and git holds its write pipe open waiting
-/// for the report-status response, so a plain read_to_end deadlocks the two
-/// (#369). A signed push is the other trap: when the client sends a push-cert
-/// it sends NO command list at all (the commands live inside the cert), so the
-/// request starts with a "push-cert" line and no flush ever arrives. The
-/// negotiated capability list is on the push-cert line, or on the first
-/// command line after its NUL when there is no cert. EOF at any point also
-/// ends the read, covering git versions that close the pipe early.
-fn read_receive_pack_request<R: Read>(stdin: &mut R) -> Result<Vec<u8>> {
-    /// Read one pkt-line into `body` (length prefix + data), returning its data
-    /// or `None` on flush or EOF. Malformed lengths bail rather than collapse
-    /// into a synthesized flush (#192 F5).
-    fn read_pkt<R: Read>(stdin: &mut R, body: &mut Vec<u8>) -> Result<Option<Vec<u8>>> {
-        let mut len_bytes = [0u8; 4];
-        match stdin.read_exact(&mut len_bytes) {
-            Ok(()) => {}
-            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
-            Err(e) => return Err(e.into()),
-        }
-        let Ok(len_hex) = std::str::from_utf8(&len_bytes) else {
-            bail!("malformed pkt-line length prefix: non-UTF-8 bytes {len_bytes:02x?}");
-        };
-        let Ok(pkt_len) = usize::from_str_radix(len_hex, 16) else {
-            bail!("malformed pkt-line length prefix: non-hex {len_hex:?}");
-        };
-        body.extend_from_slice(&len_bytes);
-        if pkt_len == 0 {
-            return Ok(None);
-        }
-        if pkt_len < 4 {
-            bail!("invalid pkt-line length: {pkt_len}");
-        }
-        let mut data = vec![0u8; pkt_len - 4];
-        stdin
-            .read_exact(&mut data)
-            .context("reading pkt-line data")?;
-        body.extend_from_slice(&data);
-        Ok(Some(data))
-    }
-
-    /// "<40-hex old> <40-hex new> <refname>" — the command shape, whether it
-    /// appears in a command list or inside a push-cert.
-    fn is_command(data: &[u8]) -> bool {
-        data.len() >= 82
-            && data[..40].iter().all(|b| b.is_ascii_hexdigit())
-            && data[40] == b' '
-            && data[41..81].iter().all(|b| b.is_ascii_hexdigit())
-            && data[81] == b' '
-    }
-
-    let mut body = Vec::new();
-    let mut caps = String::new();
-    let mut pack_expected = false;
-    let mut in_cert = false;
-    loop {
-        let Some(data) = read_pkt(stdin, &mut body)? else {
-            // Flush ends a command list; EOF ends anything. A cert has no
-            // legal flush inside it, so the same break covers malformed
-            // input without hanging.
-            break;
-        };
-        if !in_cert && data.starts_with(b"push-cert") {
-            in_cert = true;
-            if let Some(nul) = data.iter().position(|&b| b == 0) {
-                caps = String::from_utf8_lossy(&data[nul + 1..]).into_owned();
-            }
-            continue;
-        }
-        if in_cert && data == b"push-cert-end\n" {
-            break;
-        }
-        if is_command(&data) {
-            if caps.is_empty() {
-                if let Some(nul) = data.iter().position(|&b| b == 0) {
-                    caps = String::from_utf8_lossy(&data[nul + 1..]).into_owned();
-                }
-            }
-            if data[41..81].iter().any(|&b| b != b'0') {
-                pack_expected = true;
-            }
-        }
-    }
-    // When both are negotiated the push-options block follows the cert; when
-    // only options are negotiated it follows the command flush. Either way it
-    // is pkt-lines to a flush.
-    if caps.split(' ').any(|c| c.starts_with("push-options")) {
-        while read_pkt(stdin, &mut body)?.is_some() {}
-    }
-    if pack_expected {
-        stdin
-            .read_to_end(&mut body)
-            .context("reading receive-pack pack")?;
-    }
-    Ok(body)
-}
-
 /// POST one self-contained stateless-RPC request body and stream the response to
 /// `stdout`. Signs the request when `signing_key` is present, so every round of a
 /// private-repo fetch carries the caller's signature (the node visibility-gates
@@ -2118,184 +2017,6 @@ mod tests {
         assert!(help.ends_with('\n'));
     }
 
-    // ── #369 delete-only push deadlock ───────────────────────────────────────
-
-    /// A delete-only receive-pack request ends at the command flush; git then
-    /// holds the write pipe open for the report and sends no pack. A reader that
-    /// blocks after the command list reproduces the hang that a Cursor (EOF)
-    /// hides. The read must return at the flush, inside the timeout.
-    #[test]
-    fn delete_only_receive_pack_request_returns_at_command_flush() {
-        let old = "a".repeat(40);
-        let zero = "0".repeat(40);
-        let mut seed = Vec::new();
-        seed.extend_from_slice(&pkt(&format!("{old} {zero} refs/heads/gone\n")));
-        seed.extend_from_slice(b"0000"); // command-list flush; git then blocks
-
-        let (tx, rx) = std::sync::mpsc::channel::<()>();
-        let mut reader = BlockAfterSeed {
-            seed: io::Cursor::new(seed),
-            gate: rx,
-        };
-
-        let (done_tx, done_rx) = std::sync::mpsc::channel::<Vec<u8>>();
-        let handle = std::thread::spawn(move || {
-            let _ = done_tx.send(read_receive_pack_request(&mut reader).unwrap());
-        });
-
-        let got = done_rx.recv_timeout(std::time::Duration::from_secs(2));
-        let _ = tx.send(()); // unblock the reader so the thread can exit
-        let _ = handle.join();
-
-        let body = got.expect(
-            "read_receive_pack_request blocked past the command flush waiting for \
-             a pack a delete-only push never sends (#369)",
-        );
-        assert!(
-            body.ends_with(b"0000"),
-            "the request body keeps the command list and its flush terminator"
-        );
-    }
-
-    /// A push carrying objects still reads the pack through to EOF, and the
-    /// command flush sits between the commands and the pack in the body.
-    #[test]
-    fn receive_pack_request_with_pack_reads_through_eof() {
-        let old = "a".repeat(40);
-        let new = "b".repeat(40);
-        let mut stream = Vec::new();
-        stream.extend_from_slice(&pkt(&format!(
-            "{old} {new} refs/heads/main\0 report-status-v2\n"
-        )));
-        stream.extend_from_slice(b"0000");
-        stream.extend_from_slice(b"PACK\x00\x00\x00\x02fakepack");
-
-        let mut stdin = io::Cursor::new(stream);
-        let body = read_receive_pack_request(&mut stdin).unwrap();
-        let pos = body.windows(4).position(|w| w == b"0000").unwrap();
-        assert_eq!(
-            &body[pos + 4..],
-            b"PACK\x00\x00\x00\x02fakepack",
-            "a non-zero new-oid means the pack follows the flush"
-        );
-    }
-
-    /// A signed delete-only push sends NO command list: the request starts
-    /// with the "push-cert" line, the commands live inside the cert, and the
-    /// cert ends with a "push-cert-end" pkt-line (not a flush). The reader
-    /// must return after push-cert-end without waiting on the pipe.
-    #[test]
-    fn signed_delete_only_push_returns_after_push_cert_end() {
-        let old = "a".repeat(40);
-        let zero = "0".repeat(40);
-        let mut seed = Vec::new();
-        seed.extend_from_slice(&pkt("push-cert\0 report-status-v2 push-cert\n"));
-        seed.extend_from_slice(&pkt("certificate version 0.1\n"));
-        seed.extend_from_slice(&pkt("pusher did:key:z6Mk\n"));
-        seed.extend_from_slice(&pkt("\n")); // empty line ends the headers
-        seed.extend_from_slice(&pkt(&format!("{old} {zero} refs/heads/gone\n")));
-        seed.extend_from_slice(&pkt("sig-line\n"));
-        seed.extend_from_slice(&pkt("push-cert-end\n"));
-        // git now blocks for the report; the pipe stays open.
-
-        let (tx, rx) = std::sync::mpsc::channel::<()>();
-        let mut reader = BlockAfterSeed {
-            seed: io::Cursor::new(seed),
-            gate: rx,
-        };
-
-        let (done_tx, done_rx) = std::sync::mpsc::channel::<Vec<u8>>();
-        let handle = std::thread::spawn(move || {
-            let _ = done_tx.send(read_receive_pack_request(&mut reader).unwrap());
-        });
-        let got = done_rx.recv_timeout(std::time::Duration::from_secs(2));
-        let _ = tx.send(());
-        let _ = handle.join();
-
-        let body = got.expect(
-            "read_receive_pack_request blocked past push-cert-end on a signed \
-             delete-only push (#369)",
-        );
-        assert!(
-            body.windows(6).any(|w| w == b"pusher"),
-            "the cert block is part of the request body"
-        );
-    }
-
-    /// A signed pack-bearing push: cert with the commands inside, then the
-    /// pack. Everything must land in the body.
-    #[test]
-    fn signed_pack_push_keeps_cert_and_pack() {
-        let old = "a".repeat(40);
-        let new = "b".repeat(40);
-        let mut stream = Vec::new();
-        stream.extend_from_slice(&pkt("push-cert\0 report-status-v2 push-cert\n"));
-        stream.extend_from_slice(&pkt("certificate version 0.1\n"));
-        stream.extend_from_slice(&pkt("\n"));
-        stream.extend_from_slice(&pkt(&format!("{old} {new} refs/heads/main\n")));
-        stream.extend_from_slice(&pkt("push-cert-end\n"));
-        stream.extend_from_slice(b"PACK\x00\x00\x00\x02fakepack");
-
-        let mut stdin = io::Cursor::new(stream);
-        let body = read_receive_pack_request(&mut stdin).unwrap();
-        assert!(
-            body.windows(5).any(|w| w == b"PACK\x00"),
-            "the pack follows the cert into the body"
-        );
-    }
-
-    /// A negotiated push-options block is pkt-lines ending in its own flush.
-    /// With no cert it follows the command flush; with a cert it follows
-    /// push-cert-end.
-    #[test]
-    fn receive_pack_request_consumes_a_push_options_block() {
-        let old = "a".repeat(40);
-        let new = "b".repeat(40);
-        let mut stream = Vec::new();
-        stream.extend_from_slice(&pkt(&format!(
-            "{old} {new} refs/heads/main\0 report-status-v2 push-options\n"
-        )));
-        stream.extend_from_slice(b"0000"); // command flush
-        stream.extend_from_slice(&pkt("ci.skip\n"));
-        stream.extend_from_slice(b"0000"); // options flush
-        stream.extend_from_slice(b"PACK\x00\x00\x00\x02fakepack");
-
-        let mut stdin = io::Cursor::new(stream);
-        let body = read_receive_pack_request(&mut stdin).unwrap();
-        assert!(body.windows(7).any(|w| w == b"ci.skip"), "options survived");
-        assert!(
-            body.ends_with(b"PACK\x00\x00\x00\x02fakepack"),
-            "the pack lands after the options block"
-        );
-    }
-
-    /// Cert and options together: the cert comes first and the options block
-    /// sits between push-cert-end and the pack.
-    #[test]
-    fn signed_push_with_options_reads_cert_then_options_then_pack() {
-        let old = "a".repeat(40);
-        let new = "b".repeat(40);
-        let mut stream = Vec::new();
-        stream.extend_from_slice(&pkt(
-            "push-cert\0 report-status-v2 push-cert push-options\n",
-        ));
-        stream.extend_from_slice(&pkt("certificate version 0.1\n"));
-        stream.extend_from_slice(&pkt("\n"));
-        stream.extend_from_slice(&pkt(&format!("{old} {new} refs/heads/main\n")));
-        stream.extend_from_slice(&pkt("push-cert-end\n"));
-        stream.extend_from_slice(&pkt("ci.skip\n"));
-        stream.extend_from_slice(b"0000"); // options flush
-        stream.extend_from_slice(b"PACK\x00\x00\x00\x02fakepack");
-
-        let mut stdin = io::Cursor::new(stream);
-        let body = read_receive_pack_request(&mut stdin).unwrap();
-        assert!(body.windows(7).any(|w| w == b"ci.skip"), "options survived");
-        assert!(
-            body.ends_with(b"PACK\x00\x00\x00\x02fakepack"),
-            "the pack lands after the options block"
-        );
-    }
-
     // ── #117 multi-round fetch negotiation ───────────────────────────────────
 
     /// Encode a git pkt-line: 4-byte hex length (incl. the 4 bytes) + data.

crates/git-remote-gitlawb/tests/real_git_fetch.rs

@@ -95,36 +95,6 @@ fn upload_pack(repo: &Path, advertise: bool, input: &[u8]) -> Vec<u8> {
     out.stdout
 }
 
-/// Run `git receive-pack --stateless-rpc [--advertise-refs] <repo>`, optionally
-/// feeding `input` on stdin. Returns raw stdout (the wire bytes). The receive
-/// half of `upload_pack`, used so the shim can serve a real push.
-fn receive_pack(repo: &Path, advertise: bool, input: &[u8]) -> Vec<u8> {
-    let mut cmd = Command::new("git");
-    cmd.arg("receive-pack").arg("--stateless-rpc");
-    if advertise {
-        cmd.arg("--advertise-refs");
-    }
-    cmd.arg(repo)
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped());
-    let mut child = cmd.spawn().expect("spawn git receive-pack");
-    let mut stdin = child.stdin.take().unwrap();
-    let input = input.to_vec();
-    let writer = std::thread::spawn(move || {
-        let _ = stdin.write_all(&input);
-        // drop closes stdin
-    });
-    let out = child.wait_with_output().expect("wait receive-pack");
-    writer.join().ok();
-    assert!(
-        out.status.success(),
-        "git receive-pack failed: {}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-    out.stdout
-}
-
 #[derive(Clone, Copy)]
 enum ShimMode {
     /// Faithful v0 serving: pipe each POST to `git upload-pack --stateless-rpc`.
@@ -248,21 +218,7 @@ fn handle_conn(stream: TcpStream, repo: &Path, mode: ShimMode, posts: &AtomicUsi
         reader.read_exact(&mut body).ok();
     }
 
-    let (content_type, payload) = if method == "GET" && target.contains("service=git-receive-pack")
-    {
-        // Receive-pack advertisement, same wrapper shape as upload-pack.
-        let adv = receive_pack(repo, true, b"");
-        let mut wrapped = pkt(b"# service=git-receive-pack\n");
-        wrapped.extend_from_slice(b"0000");
-        wrapped.extend_from_slice(&adv);
-        ("application/x-git-receive-pack-advertisement", wrapped)
-    } else if method == "POST" && target.ends_with("/git-receive-pack") {
-        posts.fetch_add(1, Ordering::SeqCst);
-        (
-            "application/x-git-receive-pack-result",
-            receive_pack(repo, false, &body),
-        )
-    } else if method == "GET" && target.contains("/info/refs") {
+    let (content_type, payload) = if method == "GET" && target.contains("/info/refs") {
         // v0 advertisement, wrapped exactly as the node's info_refs does.
         let adv = upload_pack(repo, true, b"");
         let mut wrapped = pkt(b"# service=git-upload-pack\n");
@@ -337,34 +293,6 @@ fn fetch_with_helper(clone: &Path, node_url: &str) -> (bool, std::process::Outpu
     run_bounded(cmd, Duration::from_secs(30))
 }
 
-/// Run `git push` in `local` through the helper, same wiring and hard timeout as
-/// [`fetch_with_helper`]. `args` carries the push spec, e.g.
-/// `["--delete", "doomed"]`.
-fn push_with_helper(local: &Path, node_url: &str, args: &[&str]) -> (bool, std::process::Output) {
-    let helper_bin = PathBuf::from(env!("CARGO_BIN_EXE_git-remote-gitlawb"));
-    let helper_dir = helper_bin.parent().unwrap().to_path_buf();
-    let path_env = match std::env::var_os("PATH") {
-        Some(p) => {
-            let mut dirs = vec![helper_dir.clone()];
-            dirs.extend(std::env::split_paths(&p));
-            std::env::join_paths(dirs).unwrap()
-        }
-        None => helper_dir.clone().into_os_string(),
-    };
-
-    let mut cmd = Command::new("git");
-    cmd.arg("-C")
-        .arg(local)
-        .arg("push")
-        .arg("origin")
-        .args(args)
-        .env("PATH", path_env)
-        .env("LC_ALL", "C")
-        .env("GITLAWB_NODE", node_url)
-        .env("GITLAWB_KEY", "/nonexistent-key-for-anon-push");
-    run_bounded(cmd, Duration::from_secs(30))
-}
-
 /// A synthetic non-success `ExitStatus` used only on the unix timeout path, where
 /// `reap_group` has already consumed the child and the later `wait` returns ECHILD.
 /// The timeout branch already reports `completed == false`, so the exact status is
@@ -1314,77 +1242,3 @@ fn run_bounded_bounds_join_when_leader_exits_leaving_a_pipe_holder() {
          held pipes do not stall the joins)"
     );
 }
-
-/// #369 end-to-end: a real `git push --delete` through the helper completes.
-/// The receive-pack request for a delete-only push is commands + flush with NO
-/// pack, and git holds its write pipe open for the report — so a helper that
-/// reads stdin to EOF deadlocks against it. The shim pipes the POST to a real
-/// `git receive-pack --stateless-rpc`, so the branch actually has to be deleted
-/// server-side for the assertions to pass.
-#[test]
-fn real_git_push_delete_completes() {
-    let dir = tempfile::tempdir().unwrap();
-    let server = dir.path().join("server.git");
-    let local = dir.path().join("local");
-
-    // A bare repo holding the doomed branch, plus a live one.
-    git(
-        dir.path(),
-        &[
-            "init",
-            "--bare",
-            server.to_str().unwrap(),
-            "--initial-branch=main",
-        ],
-    );
-    let work = dir.path().join("work");
-    git(
-        dir.path(),
-        &["init", work.to_str().unwrap(), "--initial-branch=main"],
-    );
-    git(&work, &["config", "user.email", "t@t"]);
-    git(&work, &["config", "user.name", "t"]);
-    std::fs::write(work.join("f"), "x").unwrap();
-    git(&work, &["add", "f"]);
-    git(&work, &["commit", "-m", "c"]);
-    git(&work, &["branch", "doomed"]);
-    git(&work, &["push", server.to_str().unwrap(), "main", "doomed"]);
-
-    // The pusher: a local repo whose origin is the gitlawb remote pointing at
-    // the shim. Nothing needs to be fetched first; git takes the old-oid from
-    // the receive-pack advertisement.
-    git(
-        dir.path(),
-        &["init", local.to_str().unwrap(), "--initial-branch=main"],
-    );
-    git(
-        &local,
-        &["remote", "add", "origin", "gitlawb://did:key:z6MkTest/repo"],
-    );
-
-    let shim = start_shim(server.clone(), ShimMode::Normal);
-    let (completed, out) = push_with_helper(&local, &shim.base_url, &["--delete", "doomed"]);
-    let posts = shim.posts.load(Ordering::SeqCst);
-
-    assert!(
-        completed,
-        "git push --delete did not complete within the timeout (deadlock signature). stderr:\n{}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-    assert!(
-        out.status.success(),
-        "git push --delete failed. stderr:\n{}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-    assert_eq!(posts, 1, "a delete-only push is exactly one POST");
-    let gone = Command::new("git")
-        .arg("-C")
-        .arg(&server)
-        .args(["rev-parse", "--verify", "--quiet", "refs/heads/doomed"])
-        .output()
-        .unwrap();
-    assert!(
-        !gone.status.success(),
-        "refs/heads/doomed must be deleted in the server repo"
-    );
-}

crates/gitlawb-node/src/api/ipfs.rs

@@ -3758,9 +3758,13 @@ mod tests {
         // deterministically.
         let endpoint = crate::test_support::silent_http_endpoint().await;
         let tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
+        );
         state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
         let mut cfg = (*state.config).clone();
         cfg.git_acquire_timeout_secs = 1;
@@ -3831,9 +3835,13 @@ mod tests {
         // (endpoint-pinned test client, no AWS_* env reads).
         let endpoint = crate::test_support::silent_http_endpoint().await;
         let tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
+        );
         let mut cfg = (*state.config).clone();
         cfg.git_acquire_timeout_secs = 1;
         state.config = Arc::new(cfg);
@@ -7956,9 +7964,18 @@ mod tests {
         // (endpoint-pinned test client, no AWS_* env reads).
         let endpoint = crate::test_support::silent_http_endpoint().await;
         let tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
+        );
+        state
+            .db
+            .upsert_mirror_repo("z6f3budget", "ghost", "/unused-ghost", None, false)
+            .await
+            .unwrap();
         let mut cfg = (*state.config).clone();
         cfg.ipfs_request_budget_secs = 1;
         cfg.git_acquire_timeout_secs = 2;

crates/gitlawb-node/src/api/issues.rs

@@ -61,20 +61,30 @@ pub async fn create_issue(
     let json_str = serde_json::to_string(&issue)
         .map_err(|e| AppError::BadRequest(format!("serialization error: {e}")))?;
 
-    // Shed 503 + Retry-After on an exhausted write-lock POOL instead of a generic
-    // git 500 (#173 F1). This path holds no admission permit, so it reaches the pool
-    // unthrottled; reuse the push handler's mapping so the two cannot drift.
     let guard = state
         .repo_store
         .acquire_write(&record.owner_did, &record.name)
         .await
         .map_err(|e| crate::api::repos::acquire_write_app_error(&e, &repo))?;
-    let disk_path = guard.path().to_path_buf();
 
-    let create_result = git_issues::create_issue(&disk_path, &issue_id, &json_str);
+    let create_result = git_issues::create_issue(guard.path(), &issue_id, &json_str);
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(create_result.is_ok()).await;
+    let git_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs);
+    let deadline = std::time::Instant::now() + git_timeout;
+    let git_bin = state.git_bin.clone();
+    let issue_id_for_comp = issue_id.clone();
+
+    let release_result = if create_result.is_ok() {
+        guard
+            .release_compensating(true, move |path| {
+                git_issues::delete_issue_ref(&git_bin, path, &issue_id_for_comp, deadline)
+            })
+            .await
+            .into_result()
+    } else {
+        guard.release(false).await.into_result()
+    };
+    release_result?;
 
     create_result.map_err(|e| AppError::Git(e.to_string()))?;
 
@@ -225,6 +235,8 @@ pub async fn close_issue(
     State(state): State<AppState>,
     Extension(auth): Extension<AuthenticatedDid>,
     Path((owner, repo, issue_id)): Path<(String, String, String)>,
+    headers: axum::http::HeaderMap,
+    crate::rate_limit::PeerAddr(peer): crate::rate_limit::PeerAddr,
 ) -> Result<Json<serde_json::Value>> {
     let record = state
         .db
@@ -232,8 +244,161 @@ pub async fn close_issue(
         .await?
         .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
 
-    // Same capacity shed as create_issue above (#173 F1): an exhausted write-lock
-    // pool is a 503 + Retry-After, not a 500 git error.
+    // READ-GATE before any snapshot work or rate charging. The author fallback below
+    // needs the issue blob, which needs the repo tree, so authorship cannot be established
+    // without a download; but a caller who cannot even READ the repo must be
+    // stopped here, cheaply, before any Tigris transfer or extraction happens.
+    // Without this, any signed non-owner could issue parallel close requests for
+    // arbitrary issue ids and drive unbounded downloads and blocking extraction
+    // (a disposable-identity DoS), because the route has no other pre-authorization.
+    //
+    // Rate limiting runs AFTER this gate so a 429 cannot distinguish a hidden repo
+    // from a missing one (INV-12 read-denial status contract).
+    //
+    // mirror-rows-handled: a repo row synced from a peer is stored public and
+    // carries none of the owner's visibility rules, so for such a row this check
+    // can only return allow. That is deliberate rather than overlooked, and it is
+    // the same verdict every other read gate in this API reaches for one. Refusing
+    // instead would deny the repo's real owner and the issue's real author on any
+    // node whose only copy of the repo is a synced one, which is an ordinary state
+    // here, and it would not buy the protection it appears to, because a synced
+    // row's recorded owner comes from the peer that sent it. The expensive work
+    // this check guards is bounded ahead of it by the per-IP limiter and the read
+    // pool slot taken below, and the authoritative owner-or-author decision still
+    // runs below and again under the write lock.
+    {
+        // mirror-rows-handled: synced mirror rows carry no owner rules; read gate only.
+        let rules = state.db.list_visibility_rules(&record.id).await?;
+        let caller = auth.0.as_str();
+        if crate::visibility::visibility_check(
+            &rules,
+            record.is_public,
+            &record.owner_did,
+            Some(caller),
+            "/",
+        ) == crate::visibility::Decision::Deny
+        {
+            return Err(AppError::RepoNotFound(format!("{owner}/{repo}")));
+        }
+    }
+
+    // Per-IP flood brake, layered on the same shared limiter and trusted-proxy
+    // policy as the push advertisement. The pre-lock snapshot downloads the
+    // whole archive and runs a blocking extraction, so an unlimited route would
+    // let disposable identities drive unbounded transfer/CPU/disk with parallel
+    // close requests for arbitrary issue ids. Applied after the read gate so a
+    // denied reader still sees 404, not 429.
+    if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
+        if !state.close_issue_rate_limiter.check(&key).await {
+            tracing::warn!(repo = %repo, key = %key, "close_issue rate limited");
+            return Err(AppError::TooManyRequests(
+                "rate limit exceeded — try again later".into(),
+            ));
+        }
+    }
+
+    // AUTHORIZE BEFORE ACQUIRING. The per-repo advisory lock genuinely excludes
+    // now, so taking it first would hand any caller with read access a way to hold
+    // that lock on demand and be refused afterwards, while a legitimate writer
+    // burned its retry budget against it. On a public repo that is every
+    // permissionless identity. The lock must not be reachable by a caller who is
+    // about to be refused the write.
+    let is_owner = crate::api::require_repo_owner(&record, &auth.0).is_ok();
+    if !is_owner {
+        // Cap concurrent snapshot work on the shared read pool. The hourly rate
+        // bucket above is not a concurrent-work brake; without this, parallel close
+        // attempts from read-capable callers could each drive a full archive
+        // download and blocking extraction before the author denial below.
+        let _read_permit = state
+            .git_read_semaphore
+            .clone()
+            .try_acquire_owned()
+            .map_err(|_| {
+                tracing::warn!(
+                    repo = %repo,
+                    "close_issue snapshot refused — git read pool at capacity"
+                );
+                AppError::Overloaded("git service at capacity, retry shortly".into())
+            })?;
+
+        // Not the owner, so the author fallback decides it, and the author lives in
+        // the issue's git-JSON blob rather than a DB column.
+        //
+        // Read it WITHOUT the write lock, from a NON-MUTATING SNAPSHOT. The
+        // justification is NOT that authorship is immutable — it is not:
+        // `refs/gitlawb/**` is pushable, so a forged author blob can be pushed
+        // (tracked separately; it is what makes this fallback only as trustworthy
+        // as push authorization). The justification is that this read is only a
+        // PRE-CHECK, deciding whether to take the lock at all. It is NOT the
+        // authorization decision: `acquire_write` re-downloads the archive after
+        // locking, so the tree that gets mutated is routinely not this one, and the
+        // authoritative owner-or-author check runs again under the guard below.
+        // Refusing here early just keeps a caller who is already visibly
+        // unauthorized from reaching the lock.
+        //
+        // `read_snapshot`, not `acquire_fresh`: acquire's fast path returns as soon
+        // as the directory exists and never contacts object storage, so on a node
+        // with a stale copy the author's own issue would be invisible and the
+        // cannot-establish-authorship arm below would 403 a legitimate author.
+        // read_snapshot refreshes the same way, but unpacks into a throwaway temp
+        // dir instead of publishing into the live repo path — an unlocked
+        // pre-check must not delete or swap the directory under a concurrent
+        // guarded write on the same path.
+        // Pre-lock authorization only: bound with the read acquire budget, not the
+        // under-lock transfer timeout a guarded write may hold for minutes.
+        let snapshot_bound_secs = state.config.git_acquire_timeout_secs;
+        let snapshot = tokio::time::timeout(
+            std::time::Duration::from_secs(snapshot_bound_secs),
+            state
+                .repo_store
+                .read_snapshot(&record.owner_did, &record.name),
+        )
+        .await
+        .map_err(|_elapsed| {
+            tracing::warn!(
+                repo = %repo,
+                bound_secs = snapshot_bound_secs,
+                "close_issue snapshot exceeded the read acquire bound — shedding as a retryable refusal"
+            );
+            AppError::RepoUnavailable
+        })??;
+        let snapshot_path = snapshot.path().to_path_buf();
+
+        let author_did: Option<String> = match git_issues::get_issue(&snapshot_path, &issue_id) {
+            Ok(Some(raw)) => serde_json::from_str::<IssueRecord>(&raw)
+                .ok()
+                .and_then(|i| i.author),
+            // Cannot establish authorship, so fail closed. Deliberately 403 rather
+            // than 404 for a non-owner: a caller who is not authorized to write
+            // should not learn from this route whether the issue exists. Both arms
+            // below return None; they are split only so a read failure is visible
+            // to operators, since a genuinely absent issue and an unreadable one
+            // are the same answer to the client but not the same event.
+            Ok(None) => None,
+            Err(e) => {
+                tracing::warn!(
+                    repo = %repo,
+                    issue = %issue_id,
+                    err = %e,
+                    "get_issue failed during close_issue authorship pre-check"
+                );
+                None
+            }
+        };
+        let is_author = author_did
+            .as_deref()
+            .is_some_and(|a| crate::api::did_matches(&auth.0, a));
+        if !is_author {
+            return Err(AppError::Forbidden(
+                "only the repo owner or the issue author can close this issue".into(),
+            ));
+        }
+    }
+
+    // Authorized. Only now is the lock taken.
+    // Propagate rather than stringify: AppError's From<anyhow::Error> downcasts to
+    // sqlx::Error so a pool timeout or a database outage surfaces as a retryable
+    // 503. Calling .to_string() first destroys that and reports both as a 500.
     let guard = state
         .repo_store
         .acquire_write(&record.owner_did, &record.name)
@@ -241,37 +406,56 @@ pub async fn close_issue(
         .map_err(|e| crate::api::repos::acquire_write_app_error(&e, &repo))?;
     let disk_path = guard.path().to_path_buf();
 
-    // Owner OR issue author may close. The author lives in the issue's git-JSON
-    // blob (not a DB column); a None author (legacy issues) falls back to
-    // owner-only. Read it under the write guard, before mutating.
-    let author_did: Option<String> = match git_issues::get_issue(&disk_path, &issue_id) {
-        Ok(Some(raw)) => serde_json::from_str::<IssueRecord>(&raw)
-            .ok()
-            .and_then(|i| i.author),
+    // Re-read under the guard and RE-AUTHORIZE against what we read, rather than
+    // only confirming the issue still exists. The pre-lock read decided whether to
+    // take the lock; it cannot be the authorization decision, because acquire_write
+    // re-downloads the archive after locking, so this is frequently a different tree
+    // than the one the author was read from. Checking existence alone would leave the
+    // whole decision resting on the earlier read of a tree we are no longer looking
+    // at. The blob is already in hand here, so this costs a deserialize.
+    match git_issues::get_issue(&disk_path, &issue_id) {
+        Ok(Some(raw)) => {
+            let author_now: Option<String> = serde_json::from_str::<IssueRecord>(&raw)
+                .ok()
+                .and_then(|i| i.author);
+            let is_author_now = author_now
+                .as_deref()
+                .is_some_and(|a| crate::api::did_matches(&auth.0, a));
+            if !is_owner && !is_author_now {
+                // Consumed, NOT propagated, and that is deliberate at all three
+                // `release(false)` sites below. These release without
+                // publishing, so there is nothing for the store to refuse, and
+                // mapping the outcome here would let a 503 shadow the
+                // authorization answer this route exists to give.
+                let _ = guard.release(false).await;
+                return Err(AppError::Forbidden(
+                    "only the repo owner or the issue author can close this issue".into(),
+                ));
+            }
+        }
         Ok(None) => {
-            guard.release(false).await;
-            return Err(AppError::NotFound(format!("issue {issue_id} not found")));
+            let _ = guard.release(false).await;
+            // The owner keeps the informative 404; a non-owner must not learn from
+            // this route whether the issue exists, matching the pre-check above.
+            return Err(if is_owner {
+                AppError::NotFound(format!("issue {issue_id} not found"))
+            } else {
+                AppError::Forbidden(
+                    "only the repo owner or the issue author can close this issue".into(),
+                )
+            });
         }
         Err(e) => {
-            guard.release(false).await;
+            let _ = guard.release(false).await;
             return Err(AppError::Git(e.to_string()));
         }
-    };
-    let is_owner = crate::api::require_repo_owner(&record, &auth.0).is_ok();
-    let is_author = author_did
-        .as_deref()
-        .is_some_and(|a| crate::api::did_matches(&auth.0, a));
-    if !is_owner && !is_author {
-        guard.release(false).await;
-        return Err(AppError::Forbidden(
-            "only the repo owner or the issue author can close this issue".into(),
-        ));
     }
 
     let close_result = git_issues::close_issue(&disk_path, &issue_id);
 
     // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(close_result.is_ok()).await;
+    // Same short-circuit as create_issue, and before the 200 body below.
+    guard.release(close_result.is_ok()).await.into_result()?;
 
     let updated = close_result
         .map_err(|e| AppError::Git(e.to_string()))?
@@ -285,12 +469,344 @@ pub async fn close_issue(
     Ok(Json(issue))
 }
 
-/// #173 F1 follow-up: the two issue write paths reach `acquire_write` holding NO
-/// admission permit (unlike the push handler, which is capped by the git-push
-/// semaphore), so they are the callers most likely to meet an exhausted write-lock
-/// POOL under load. An exhausted pool is a capacity signal, so both must shed
-/// 503 + Retry-After (`AppError::Overloaded`) the way the push handler does, not
-/// report the generic 500 git error that says nothing about retrying.
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use sqlx::PgPool;
+
+    /// U7: once the advisory lock actually excludes, taking it BEFORE authorizing
+    /// turns close_issue into a wedge primitive. Any caller with repo read access
+    /// (on a public repo, any permissionless identity) could take the per-repo
+    /// write lock on demand and be refused the write afterwards, while the owner's
+    /// push burned its retry budget against a lock held by someone with no write
+    /// authorization.
+    ///
+    /// The observable: hold the lock from an independent session, then call the
+    /// handler as a stranger. If it authorizes first it refuses immediately; if it
+    /// acquires first it sits in the 60-attempt retry loop and the deadline fires.
+    #[sqlx::test]
+    async fn stranger_is_refused_without_waiting_on_the_write_lock(pool: PgPool) {
+        use sqlx::Connection;
+        let opts = (*pool.connect_options()).clone();
+        let state = crate::test_support::test_state(pool.clone()).await;
+
+        let owner = "did:key:z6MkU7Owner";
+        state
+            .db
+            .upsert_mirror_repo("z6MkU7Owner", "u7repo", "/tmp/u7repo", None, true)
+            .await
+            .expect("seed repo");
+        let record = state
+            .db
+            .get_repo("z6MkU7Owner", "u7repo")
+            .await
+            .expect("get_repo")
+            .expect("repo exists");
+
+        // An independent session holds the repo's write lock for the whole call.
+        let key = crate::git::repo_store::advisory_lock_key_for_test(
+            &record.owner_did.replace([':', '/'], "_"),
+            &record.name,
+        );
+        let mut holder = sqlx::PgConnection::connect_with(&opts).await.unwrap();
+        let held: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
+            .bind(key)
+            .fetch_one(&mut holder)
+            .await
+            .unwrap();
+        assert!(
+            held.0,
+            "the test must hold the lock for this to mean anything"
+        );
+        let _ = owner;
+
+        let stranger = crate::auth::AuthenticatedDid("did:key:z6MkU7Stranger".to_string());
+        let outcome = tokio::time::timeout(
+            std::time::Duration::from_secs(3),
+            close_issue(
+                axum::extract::State(state.clone()),
+                axum::Extension(stranger),
+                axum::extract::Path((
+                    "z6MkU7Owner".to_string(),
+                    "u7repo".to_string(),
+                    "1".to_string(),
+                )),
+                axum::http::HeaderMap::new(),
+                crate::rate_limit::PeerAddr(Some("203.0.113.64:5000".parse().unwrap())),
+            ),
+        )
+        .await;
+
+        let refused = outcome.expect(
+            "a caller with no write authorization must be refused WITHOUT waiting on the \
+             write lock; hitting this deadline means the handler tried to acquire first, \
+             which is the wedge primitive",
+        );
+        assert!(
+            matches!(refused, Err(AppError::Forbidden(_))),
+            "expected 403 Forbidden for a stranger, got {:?}",
+            refused.err().map(|e| format!("{e:?}"))
+        );
+    }
+
+    /// The read-gate's blind spot, pinned rather than left to be rediscovered.
+    ///
+    /// A repo row synced from a peer is stored public and carries none of the
+    /// owner's visibility rules, so the gate's own inputs can only produce allow
+    /// for it. The gate above therefore does not carry this class of row, and the
+    /// test that does cover it (`non_reader_is_refused_before_the_snapshot`) seeds
+    /// a locally created repo, which cannot observe this: a passing test there is
+    /// not coverage here.
+    ///
+    /// Two things are asserted, and the second is why the first is acceptable.
+    /// The gate's verdict for such a row is allow for an arbitrary caller, and the
+    /// handler still refuses that caller afterwards, because the decision that
+    /// matters is the owner-or-author check rather than this one. If a later change
+    /// makes the gate the load-bearing decision for this route, the first assertion
+    /// breaks and this comment is where to start.
+    #[sqlx::test]
+    async fn a_synced_row_is_not_gated_by_its_own_visibility(pool: PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+
+        // Only a synced row exists for this repo, with no locally created twin.
+        state
+            .db
+            .upsert_mirror_repo("z6MkSyncOwner", "syncrepo", "/tmp/syncrepo", None, false)
+            .await
+            .expect("seed synced repo"); // false = not quarantined, the ordinary case
+        let record = state
+            .db
+            .get_repo("z6MkSyncOwner", "syncrepo")
+            .await
+            .expect("get_repo")
+            .expect("repo exists")
+            .clone();
+        assert!(
+            record.id.contains('/'),
+            "this test is only meaningful against a synced row; got id {}",
+            record.id
+        );
+
+        // The gate's two inputs, and what they force.
+        let rules = state
+            .db
+            .list_visibility_rules(&record.id)
+            .await
+            .expect("list rules");
+        assert!(rules.is_empty(), "a synced row carries no rules of its own");
+        assert!(record.is_public, "a synced row is stored public");
+        assert_eq!(
+            crate::visibility::visibility_check(
+                &rules,
+                record.is_public,
+                &record.owner_did,
+                Some("did:key:z6MkSyncStranger"),
+                "/",
+            ),
+            crate::visibility::Decision::Allow,
+            "the gate can only allow for a synced row, which is the property the \
+             handler's mirror-rows-handled note records",
+        );
+
+        // So the refusal has to come from the decision that is actually load-bearing.
+        let stranger = crate::auth::AuthenticatedDid("did:key:z6MkSyncStranger".to_string());
+        let outcome = close_issue(
+            axum::extract::State(state.clone()),
+            axum::Extension(stranger),
+            axum::extract::Path((
+                "z6MkSyncOwner".to_string(),
+                "syncrepo".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some("203.0.113.99:5000".parse().unwrap())),
+        )
+        .await;
+        assert!(
+            outcome.is_err(),
+            "a stranger must still be refused on a synced row, gate or no gate",
+        );
+        let body = format!("{:?}", outcome.err().unwrap());
+        assert!(
+            !body.contains("syncrepo/") && !body.to_lowercase().contains("issue body"),
+            "the refusal must not leak repo contents: {body}",
+        );
+    }
+
+    /// The read-gate added for the pre-lock snapshot: a caller who cannot READ
+    /// the repo (private repo, no rule granting them access) must be refused with
+    /// a not-found BEFORE any snapshot download or extraction happens. The
+    /// observable is the refusal itself; the cheaper part (no Tigris work) is
+    /// structural (the gate precedes the snapshot call in the handler).
+    #[sqlx::test]
+    async fn non_reader_is_refused_before_the_snapshot(pool: PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+        let now = chrono::Utc::now();
+        state
+            .db
+            .create_repo(&crate::db::RepoRecord {
+                id: uuid::Uuid::new_v4().to_string(),
+                name: "priv-close".to_string(),
+                owner_did: "z6MkT3Owner".to_string(),
+                description: None,
+                is_public: false,
+                default_branch: "main".to_string(),
+                created_at: now,
+                updated_at: now,
+                disk_path: "/tmp/priv-close".to_string(),
+                forked_from: None,
+                machine_id: None,
+            })
+            .await
+            .expect("seed private repo");
+
+        let stranger = crate::auth::AuthenticatedDid("did:key:z6MkT3Stranger".to_string());
+        let res = close_issue(
+            axum::extract::State(state.clone()),
+            axum::Extension(stranger),
+            axum::extract::Path((
+                "z6MkT3Owner".to_string(),
+                "priv-close".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some("203.0.113.67:5000".parse().unwrap())),
+        )
+        .await;
+        assert!(
+            matches!(res, Err(AppError::RepoNotFound(_))),
+            "a non-reader must be refused as not-found, got {:?}",
+            res.err().map(|e| format!("{e:?}"))
+        );
+    }
+
+    async fn seed_repo_with_issue(
+        state: &crate::state::AppState,
+        owner_slug: &str,
+        owner_did: &str,
+        repo: &str,
+        issue_id: &str,
+        author_did: &str,
+    ) -> std::path::PathBuf {
+        state
+            .db
+            .upsert_mirror_repo(owner_slug, repo, "/unused", None, true)
+            .await
+            .expect("seed repo row");
+        // Seed at the path the HANDLER will resolve. upsert_mirror_repo stores the
+        // bare slug in owner_did, and close_issue resolves from record.owner_did, so
+        // seeding from the full did:key would create the repo in a different
+        // directory and the handler would find nothing.
+        let record = state
+            .db
+            .get_repo(owner_slug, repo)
+            .await
+            .expect("get_repo")
+            .expect("seeded repo exists");
+        let _ = owner_did;
+        let path = state
+            .repo_store
+            .acquire(&record.owner_did, &record.name)
+            .await
+            .expect("resolve disk path");
+        let _ = std::fs::remove_dir_all(&path);
+        crate::git::store::init_bare(&path).expect("init bare repo");
+        // Must deserialize as a real IssueRecord: `created_at` and `status` are
+        // required, and a parse failure would silently drop the author (the
+        // `.ok()` on from_str), which reads as a 403 rather than as a broken fixture.
+        let json = serde_json::to_string(&IssueRecord {
+            id: issue_id.to_string(),
+            title: "seeded".to_string(),
+            body: Some(String::new()),
+            author: Some(author_did.to_string()),
+            created_at: chrono::Utc::now().to_rfc3339(),
+            status: "open".to_string(),
+            signed_payload: None,
+        })
+        .expect("serialize seeded issue");
+        crate::git::issues::create_issue(&path, issue_id, &json).expect("seed issue blob");
+        path
+    }
+
+    /// INV-21(c) positive twin 1: the OWNER can still close. The reorder moved the
+    /// owner check above the lock, so this is the arm most likely to have broken,
+    /// and the deny test alone could not see it.
+    ///
+    /// The issue is seeded with a THIRD party as its author, deliberately. Seeding
+    /// the owner as their own author made this test unable to fail: with the owner
+    /// check disabled, the author fallback granted the close anyway and the test
+    /// stayed green. Only the owner arm can grant here now.
+    #[sqlx::test]
+    async fn owner_can_still_close_after_the_reorder(pool: PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+        let owner_did = "did:key:z6MkT1Owner";
+        seed_repo_with_issue(
+            &state,
+            "z6MkT1Owner",
+            owner_did,
+            "t1repo",
+            "1",
+            "did:key:z6MkT1Stranger",
+        )
+        .await;
+
+        let res = close_issue(
+            axum::extract::State(state.clone()),
+            axum::Extension(crate::auth::AuthenticatedDid(owner_did.to_string())),
+            axum::extract::Path((
+                "z6MkT1Owner".to_string(),
+                "t1repo".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some("203.0.113.65:5000".parse().unwrap())),
+        )
+        .await;
+        assert!(
+            res.is_ok(),
+            "the owner must still be able to close: {:?}",
+            res.err().map(|e| format!("{e:?}"))
+        );
+    }
+
+    /// INV-21(c) positive twin 2: the non-owner AUTHOR can still close, through both
+    /// the pre-lock check and the re-assertion under the guard.
+    ///
+    /// It does NOT cover the acquire-vs-acquire_fresh distinction, despite that being
+    /// the reason the call changed. `RepoStore::for_testing` hardcodes `tigris: None`,
+    /// which makes `acquire` and `acquire_fresh` identical in every test here, so
+    /// reverting that line leaves this green. Separating them needs an object-storage
+    /// seam, which is out of scope for this change and tracked separately. Claiming
+    /// the coverage here would be worse than admitting the gap.
+    #[sqlx::test]
+    async fn issue_author_who_is_not_the_owner_can_still_close(pool: PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+        let owner_did = "did:key:z6MkT2Owner";
+        let author_did = "did:key:z6MkT2Author";
+        seed_repo_with_issue(&state, "z6MkT2Owner", owner_did, "t2repo", "1", author_did).await;
+
+        let res = close_issue(
+            axum::extract::State(state.clone()),
+            axum::Extension(crate::auth::AuthenticatedDid(author_did.to_string())),
+            axum::extract::Path((
+                "z6MkT2Owner".to_string(),
+                "t2repo".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some("203.0.113.66:5000".parse().unwrap())),
+        )
+        .await;
+        assert!(
+            res.is_ok(),
+            "the issue author, who is NOT the repo owner, must still be able to close: {:?}",
+            res.err().map(|e| format!("{e:?}"))
+        );
+    }
+}
+
+/// #173 F1 follow-up: issue write paths reach `acquire_write` with no admission
+/// permit, so an exhausted write-lock pool must shed 503 + Retry-After.
 #[cfg(test)]
 mod lock_pool_shed_tests {
     use super::*;
@@ -314,22 +830,17 @@ mod lock_pool_shed_tests {
         }
     }
 
-    /// State whose repo store draws write locks from a ONE-connection pool with a
-    /// short checkout timeout, so a single held guard exhausts it promptly rather
-    /// than at the pool default.
     async fn one_connection_lock_pool_state(pool: &PgPool) -> AppState {
         let mut state = crate::test_support::test_state(pool.clone()).await;
         state.repo_store = crate::git::repo_store::RepoStore::new(
             std::path::PathBuf::from("/tmp/gitlawb-issues-lockpool"),
             None,
             crate::git::repo_store::build_lock_pool(pool, 1, std::time::Duration::from_secs(1)),
+            std::time::Duration::from_secs(300),
         );
         state
     }
 
-    /// The shed must be a real 503 carrying Retry-After, not just an internal enum
-    /// variant: assert on the rendered response so a remapping of `Overloaded` is
-    /// caught here too.
     fn assert_sheds_503_with_retry_after(err: AppError, what: &str) {
         let resp = err.into_response();
         assert_eq!(
@@ -346,10 +857,6 @@ mod lock_pool_shed_tests {
         );
     }
 
-    /// RED-before/GREEN-after for `create_issue`. Both directions: the shed while the
-    /// only lock-pool connection is held by a guard on a DIFFERENT repo (so this is
-    /// pool capacity, not advisory-lock contention on this repo), and the must-not
-    /// case once that connection is back.
     #[sqlx::test]
     async fn create_issue_lock_pool_exhaustion_sheds_503_not_500(pool: PgPool) {
         let owner = "did:key:zISSUECREATELOCKPOOLAAAAAAAAAAAAAAAAAAAA";
@@ -380,9 +887,7 @@ mod lock_pool_shed_tests {
         let err = shed.expect_err("an exhausted lock pool must fail the call");
         assert_sheds_503_with_retry_after(err, "create_issue");
 
-        // MUST-NOT: with the pool free again the call is not shed as capacity (it
-        // fails later on the nonexistent on-disk repo, which is a git 500).
-        held.release(false).await;
+        let _ = held.release(false).await;
         let admitted = create_issue(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),
@@ -401,7 +906,38 @@ mod lock_pool_shed_tests {
         );
     }
 
-    /// RED-before/GREEN-after for `close_issue`, same two directions.
+    #[sqlx::test]
+    async fn close_issue_read_pool_exhaustion_sheds_before_snapshot(pool: PgPool) {
+        use std::sync::Arc;
+        let owner = "did:key:zISSUECLOSEREADPOOLBBBBBBBBBBBBBBBBBBBBB";
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0));
+        state
+            .db
+            .create_repo(&seed_repo(owner, "read-cap"))
+            .await
+            .expect("seed repo");
+
+        let shed = close_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(
+                "did:key:zISSUECLOSEREADSTRANGER".to_string(),
+            )),
+            Path((
+                owner.to_string(),
+                "read-cap".to_string(),
+                "deadbeef".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(None),
+        )
+        .await;
+        assert!(
+            matches!(shed, Err(AppError::Overloaded(_))),
+            "an exhausted read pool must shed before snapshot work; got {shed:?}"
+        );
+    }
+
     #[sqlx::test]
     async fn close_issue_lock_pool_exhaustion_sheds_503_not_500(pool: PgPool) {
         let owner = "did:key:zISSUECLOSELOCKPOOLBBBBBBBBBBBBBBBBBBBBB";
@@ -426,12 +962,14 @@ mod lock_pool_shed_tests {
                 "lp-close".to_string(),
                 "deadbeef".to_string(),
             )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(None),
         )
         .await;
         let err = shed.expect_err("an exhausted lock pool must fail the call");
         assert_sheds_503_with_retry_after(err, "close_issue");
 
-        held.release(false).await;
+        let _ = held.release(false).await;
         let admitted = close_issue(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),
@@ -440,6 +978,8 @@ mod lock_pool_shed_tests {
                 "lp-close".to_string(),
                 "deadbeef".to_string(),
             )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(None),
         )
         .await;
         assert!(
@@ -448,4 +988,244 @@ mod lock_pool_shed_tests {
             admitted.err()
         );
     }
+
+    async fn assert_retryable_repo_acquire(err: AppError, what: &str, code: &str) {
+        let resp = err.into_response();
+        assert_eq!(
+            resp.status(),
+            StatusCode::SERVICE_UNAVAILABLE,
+            "{what}: a transient acquire refusal must shed 503, not a 500 git error"
+        );
+        let body = axum::body::to_bytes(resp.into_body(), 64 * 1024)
+            .await
+            .expect("body");
+        let body = String::from_utf8_lossy(&body);
+        assert!(
+            body.contains(code),
+            "{what}: the 503 must carry the {code} code, got {body}"
+        );
+    }
+
+    /// Advisory-lock contention on a non-push mutation must map through the shared
+    /// `acquire_write_app_error` classifier to `repo_busy`, not a 500 git_error.
+    #[sqlx::test]
+    async fn create_issue_contention_sheds_repo_busy_not_500(pool: PgPool) {
+        let owner = "did:key:zISSUECREATEBUSYAAAAAAAAAAAAAAAAAAAAA";
+        let repo_name = "busy-create";
+        let state = crate::test_support::test_state(pool.clone()).await;
+        state
+            .db
+            .create_repo(&seed_repo(owner, repo_name))
+            .await
+            .expect("seed repo");
+
+        let held = state
+            .repo_store
+            .acquire_write(owner, repo_name)
+            .await
+            .expect("the first writer takes the advisory lock");
+
+        let shed = create_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), repo_name.to_string())),
+            Json(CreateIssueRequest {
+                title: "t".to_string(),
+                body: None,
+                signed_payload: None,
+            }),
+        )
+        .await;
+        let err = shed.expect_err("contention must refuse the second writer");
+        assert_retryable_repo_acquire(err, "create_issue", "repo_busy").await;
+
+        let _ = held.release(false).await;
+    }
+
+    /// A refused under-lock refresh on a non-push mutation must map to
+    /// `repo_unavailable`, not expose the storage error as a 500 git_error.
+    #[sqlx::test]
+    async fn create_issue_unavailable_sheds_repo_unavailable_not_500(pool: PgPool) {
+        use axum::response::IntoResponse;
+
+        let app = axum::Router::new().route(
+            "/{*key}",
+            axum::routing::any(|method: axum::http::Method| async move {
+                if method == axum::http::Method::HEAD {
+                    let mut resp = axum::http::StatusCode::OK.into_response();
+                    resp.headers_mut()
+                        .insert("etag", axum::http::HeaderValue::from_static("\"gen-1\""));
+                    resp
+                } else {
+                    axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let owner = "did:key:zISSUEUNAVAILAAAAAAAAAAAAAAAAAAAAAAA";
+        let repo_name = "unavail-create";
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
+            std::path::PathBuf::from("/tmp/gitlawb-issue-unavail"),
+            crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
+        );
+        state
+            .db
+            .create_repo(&seed_repo(owner, repo_name))
+            .await
+            .expect("seed repo");
+
+        let shed = create_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), repo_name.to_string())),
+            Json(CreateIssueRequest {
+                title: "t".to_string(),
+                body: None,
+                signed_payload: None,
+            }),
+        )
+        .await;
+        let err = shed.expect_err("a failed archive download must refuse the write");
+        assert_retryable_repo_acquire(err, "create_issue", "repo_unavailable").await;
+
+        server.abort();
+    }
+
+    /// A denied reader on a private repo must see 404 even when the caller's rate
+    /// bucket is exhausted. Rate limiting runs after the read gate.
+    #[sqlx::test]
+    async fn close_issue_rate_limit_runs_after_the_read_gate(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let owner = "did:key:zCLOSERATEOWNERAAAAAAAAAAAAAAAAAAAAAAA";
+        let stranger = "did:key:zCLOSERATESTRANGERBBBBBBBBBBBBBBBBBB";
+        let mut state = crate::test_support::test_state(pool).await;
+        state.close_issue_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+
+        let mut repo = seed_repo(owner, "priv-close");
+        repo.is_public = false;
+        state.db.create_repo(&repo).await.expect("seed repo");
+
+        let peer: SocketAddr = "203.0.113.88:7000".parse().unwrap();
+        assert!(
+            state
+                .close_issue_rate_limiter
+                .check(&peer.ip().to_string())
+                .await,
+            "exhaust the close_issue bucket before close_issue"
+        );
+
+        let res = close_issue(
+            State(state),
+            Extension(AuthenticatedDid(stranger.to_string())),
+            Path((owner.to_string(), "priv-close".to_string(), "1".to_string())),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some(peer)),
+        )
+        .await;
+
+        assert!(
+            matches!(res, Err(AppError::RepoNotFound(_))),
+            "a non-reader must see 404, not 429, even when rate limited: {:?}",
+            res
+        );
+    }
+
+    /// close_issue and receive-pack must not share one per-IP bucket.
+    #[sqlx::test]
+    async fn close_issue_rate_limit_does_not_drain_push_bucket(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let mut state = crate::test_support::test_state(pool).await;
+        state.close_issue_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.push_rate_limiter = crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+
+        let peer: SocketAddr = "203.0.113.99:7000".parse().unwrap();
+        let key = peer.ip().to_string();
+        assert!(
+            state.close_issue_rate_limiter.check(&key).await,
+            "exhaust only the close_issue bucket"
+        );
+        assert!(
+            state.push_rate_limiter.check(&key).await,
+            "push traffic must keep its own bucket after close_issue is exhausted"
+        );
+    }
+
+    /// A refused publish must roll back the local issue ref so a retry does not
+    /// mint a second id for the same logical filing attempt.
+    #[sqlx::test]
+    async fn create_issue_rolls_back_local_ref_when_publish_refuses(pool: PgPool) {
+        use axum::response::IntoResponse;
+
+        let app = axum::Router::new().route(
+            "/{*key}",
+            axum::routing::any(|method: axum::http::Method| async move {
+                if method == axum::http::Method::HEAD {
+                    let mut resp = axum::http::StatusCode::OK.into_response();
+                    resp.headers_mut()
+                        .insert("etag", axum::http::HeaderValue::from_static("\"gen-1\""));
+                    resp
+                } else {
+                    axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let owner = "did:key:zISSUEROLLBACKAAAAAAAAAAAAAAAAAAAAAAA";
+        let repo_name = "rollback-create";
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        let disk = tempfile::TempDir::new().unwrap();
+        crate::git::store::init_bare(&disk.path().join("repo.git")).expect("bare repo");
+        let mut seeded = seed_repo(owner, repo_name);
+        seeded.disk_path = disk.path().join("repo.git").to_string_lossy().to_string();
+        state.db.create_repo(&seeded).await.expect("seed repo");
+        state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
+            disk.path().to_path_buf(),
+            crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
+        );
+
+        let shed = create_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), repo_name.to_string())),
+            Json(CreateIssueRequest {
+                title: "t".to_string(),
+                body: None,
+                signed_payload: None,
+            }),
+        )
+        .await;
+        assert!(
+            shed.is_err(),
+            "a failed archive upload must refuse the write"
+        );
+
+        let issues = git_issues::list_issues(&disk.path().join("repo.git")).expect("list issues");
+        assert!(
+            issues.is_empty(),
+            "the local issue ref must be rolled back after a refused publish, got {issues:?}"
+        );
+
+        server.abort();
+    }
 }

crates/gitlawb-node/src/api/pulls.rs

@@ -209,9 +209,6 @@ pub async fn merge_pr(
         return Err(AppError::BadRequest(format!("PR is already {}", pr.status)));
     }
 
-    // Shed 503 + Retry-After on an exhausted write-lock POOL instead of a generic
-    // git 500 (#173 F1). Merging holds no admission permit, so it reaches the pool
-    // unthrottled; reuse the push handler's mapping so the two cannot drift.
     let guard = state
         .repo_store
         .acquire_write(&record.owner_did, &record.name)
@@ -228,7 +225,10 @@ pub async fn merge_pr(
     );
 
     // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(merge_result.is_ok()).await;
+    // Short-circuit on a refused publish before the PR is marked merged and
+    // before the webhook fires. Both are irreversible announcements of a merge
+    // commit that only exists on this node's disk.
+    guard.release(merge_result.is_ok()).await.into_result()?;
 
     let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
 
@@ -428,12 +428,8 @@ pub async fn list_comments(
     Ok(Json(serde_json::json!({ "comments": comments })))
 }
 
-/// #173 F1 follow-up: `merge_pr` reaches `acquire_write` holding NO admission permit
-/// (unlike the push handler, which is capped by the git-push semaphore), so it is one
-/// of the callers most likely to meet an exhausted write-lock POOL under load. An
-/// exhausted pool is a capacity signal, so the merge must shed 503 + Retry-After
-/// (`AppError::Overloaded`) the way the push handler does, not report the generic
-/// 500 git error that says nothing about retrying.
+/// #173 F1 follow-up: merge holds no admission permit, so an exhausted write-lock
+/// pool must shed 503 + Retry-After.
 #[cfg(test)]
 mod lock_pool_shed_tests {
     use super::*;
@@ -457,20 +453,15 @@ mod lock_pool_shed_tests {
         }
     }
 
-    /// RED-before/GREEN-after for `merge_pr`. Both directions: the shed while the only
-    /// lock-pool connection is held by a guard on a DIFFERENT repo (so this is pool
-    /// capacity, not advisory-lock contention on this repo), and the must-not case
-    /// once that connection is back.
     #[sqlx::test]
     async fn merge_pr_lock_pool_exhaustion_sheds_503_not_500(pool: PgPool) {
         let owner = "did:key:zMERGELOCKPOOLOWNERAAAAAAAAAAAAAAAAAAAAA";
         let mut state = crate::test_support::test_state(pool.clone()).await;
-        // One lock-pool connection with a short checkout timeout, so a single held
-        // guard exhausts it promptly rather than at the pool default.
         state.repo_store = crate::git::repo_store::RepoStore::new(
             std::path::PathBuf::from("/tmp/gitlawb-pulls-lockpool"),
             None,
             crate::git::repo_store::build_lock_pool(&pool, 1, std::time::Duration::from_secs(1)),
+            std::time::Duration::from_secs(300),
         );
 
         let repo = seed_repo(owner, "lp-merge");
@@ -524,9 +515,7 @@ mod lock_pool_shed_tests {
             "merge_pr: a capacity shed must tell the client when to retry"
         );
 
-        // MUST-NOT: with the pool free again the merge is not shed as capacity (it
-        // fails later on the nonexistent on-disk repo, which is a git 500).
-        held.release(false).await;
+        let _ = held.release(false).await;
         let admitted = merge_pr(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),

crates/gitlawb-node/src/api/repos.rs

@@ -3,6 +3,7 @@ use axum::http::StatusCode;
 use axum::response::Response;
 use axum::Json;
 use bytes::Bytes;
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use crate::auth::{caller_authorized_to_push, AuthenticatedDid};
@@ -692,8 +693,17 @@ pub async fn git_info_refs(
         git_permit(&state.git_read_semaphore)?
     };
 
-    // For receive-pack (push), download the latest from Tigris so the client
-    // sees the same refs that acquire_write() will operate on.
+    // For receive-pack (push), read the latest from Tigris so the client sees
+    // the same refs that acquire_write() will operate on. A NON-MUTATING
+    // snapshot, not `acquire_fresh`: the advertisement runs WITHOUT the advisory
+    // lock, and acquire_fresh downloads and publishes into the live repo path
+    // (removing the existing directory and renaming the extract into place), so
+    // an unlocked advertisement could delete or swap the directory under a
+    // concurrent guarded write. In the worst ordering the guarded write has
+    // finished but `release` has not compressed the tree, and the guarded
+    // release uploads the replaced old tree with its still-valid ETag and
+    // reports success, losing the accepted write. The snapshot unpacks into a
+    // throwaway temp dir that is served from and then removed.
     //
     // Bound the acquire under `git_acquire_timeout_secs`: the concurrency permit is
     // already held above, and `git_service_timeout_secs` only starts once git spawns,
@@ -703,28 +713,46 @@ pub async fn git_info_refs(
     // so the shed frees the slot; return a bounded 503.
     let acquire_deadline = std::time::Duration::from_secs(state.config.git_acquire_timeout_secs);
     let acquire_fut = async {
-        if service == "git-receive-pack" {
+        let res = if service == "git-receive-pack" {
             state
                 .repo_store
-                .acquire_fresh(&record.owner_did, &record.name)
+                .read_snapshot(&record.owner_did, &record.name)
                 .await
+                .map(|s| (s.path().to_path_buf(), Some(s)))
         } else {
             state
                 .repo_store
                 .acquire(&record.owner_did, &record.name)
                 .await
-        }
+                .map(|p| (p, None))
+        };
+        res.map_err(|e| {
+            if is_expected_transient_acquire_failure(&e) {
+                tracing::warn!(repo = %name, service = %service, err = %e, "repo acquire failed");
+            } else {
+                tracing::error!(repo = %name, service = %service, err = %e, "repo acquire failed");
+            }
+            // This closure bypasses the `From<anyhow::Error>` chain, so a typed
+            // refusal would otherwise be stringified into a 500 `git_error`. Route
+            // just that one case through `From` and leave every other failure on
+            // exactly today's behavior: this call site also serves the read path via
+            // `acquire`, whose error vocabulary is out of scope here.
+            if e.is::<crate::git::repo_store::RepoUnavailable>() {
+                AppError::from(e)
+            } else {
+                AppError::Git(e.to_string())
+            }
+        })
     };
-    let disk_path = tokio::time::timeout(acquire_deadline, acquire_fut)
+    // The snapshot (if any) is kept alive for the whole handler scope below: its
+    // Drop removes the temp dir it was unpacked into, so dropping it here would
+    // delete the directory `info_refs` is about to serve from.
+    let (disk_path, _snapshot_keepalive) = tokio::time::timeout(acquire_deadline, acquire_fut)
         .await
         .map_err(|_elapsed| {
             tracing::warn!(repo = %name, service = %service, "repo acquire timed out; shedding with 503");
             AppError::Overloaded("git service acquisition timed out, retry shortly".into())
-        })?
-        .map_err(|e| {
-            tracing::error!(repo = %name, service = %service, err = %e, "repo acquire failed");
-            AppError::Git(e.to_string())
-        })?;
+        })??;
 
     // Move the admission permits into the guard so they release only after the spawned
     // git process group is confirmed reaped, on complete/timeout/disconnect — not the
@@ -978,6 +1006,7 @@ pub(crate) mod drain_faults {
         pub(crate) rules_read_failures_left: usize,
         pub(crate) repo_read_attempts: usize,
         pub(crate) rules_read_attempts: usize,
+        pub(crate) reread_exhausted: bool,
     }
 
     fn table() -> &'static Mutex<HashMap<String, Counters>> {
@@ -1008,6 +1037,25 @@ pub(crate) mod drain_faults {
             .unwrap_or_default()
     }
 
+    /// Whether the drain re-read loop exhausted its retry budget for `repo_id`.
+    pub(crate) fn reread_exhausted(repo_id: &str) -> bool {
+        table()
+            .lock()
+            .unwrap()
+            .get(repo_id)
+            .map(|c| c.reread_exhausted)
+            .unwrap_or(false)
+    }
+
+    pub(crate) fn mark_reread_exhausted(repo_id: &str) {
+        table()
+            .lock()
+            .unwrap()
+            .entry(repo_id.to_string())
+            .or_default()
+            .reread_exhausted = true;
+    }
+
     /// Production-path hook: count one repo re-read attempt, return whether it must fail.
     pub(crate) fn take_repo_read(repo_id: &str) -> bool {
         let mut map = table().lock().unwrap();
@@ -1129,6 +1177,8 @@ async fn drain_refresh_state(ctx: &EncryptTaskCtx) -> DrainRefresh {
         "coalesced drain: re-read failed on every attempt; the coalesced push's \
          pin/encrypt pass is dropped (no reconciliation sweep re-derives it)"
     );
+    #[cfg(test)]
+    drain_faults::mark_reread_exhausted(&ctx.repo_id);
     DrainRefresh::Failed
 }
 
@@ -1516,11 +1566,12 @@ async fn pin_and_encrypt_objects(
 /// Map an `acquire_write` failure to the right `AppError`. An exhausted repo write-lock
 /// POOL is a capacity signal, not a broken repo, so it sheds 503 + Retry-After the same
 /// way the admission caps around it do; it used to fall into the generic git 500, which
-/// tells the client nothing about retrying (#173 F1). Anything else stays a git error.
+/// tells the client nothing about retrying (#173 F1). Lock contention and a refused
+/// under-lock refresh are transient and map to fixed-body 503s via [`RepoBusy`] and
+/// [`RepoUnavailable`]; only genuine untyped git failures stay on the 500 path.
 ///
-/// Shared with the non-push `acquire_write` callers (`api/issues.rs`, `api/pulls.rs`)
-/// rather than copied: those hold no admission permit, so they meet an exhausted pool
-/// first, and a second copy of this mapping would be free to drift from the push path.
+/// Shared with every `acquire_write` caller (`receive-pack`, `api/issues.rs`,
+/// `api/pulls.rs`) so the write-acquisition contract cannot drift between routes.
 pub(crate) fn acquire_write_app_error(err: &anyhow::Error, repo: &str) -> AppError {
     if err
         .downcast_ref::<crate::git::repo_store::LockPoolBusy>()
@@ -1528,6 +1579,16 @@ pub(crate) fn acquire_write_app_error(err: &anyhow::Error, repo: &str) -> AppErr
     {
         tracing::warn!(repo = %repo, err = %err, "write-lock pool exhausted; shedding with 503");
         AppError::Overloaded("git write locks at capacity, retry shortly".into())
+    } else if is_expected_transient_acquire_failure(err) {
+        tracing::warn!(repo = %repo, err = %err, "acquire_write failed");
+        if err
+            .downcast_ref::<crate::git::repo_store::RepoBusy>()
+            .is_some()
+        {
+            AppError::RepoBusy
+        } else {
+            AppError::RepoUnavailable
+        }
     } else {
         tracing::error!(repo = %repo, err = %err, "acquire_write failed");
         AppError::Git(err.to_string())
@@ -1538,6 +1599,23 @@ pub(crate) fn acquire_write_app_error(err: &anyhow::Error, repo: &str) -> AppErr
 /// [`smart_http::GitServiceTimeout`] to 504, a malformed client request to 400,
 /// anything else to a 500 git error. Pure (no logging) so it is unit-testable;
 /// callers add their own tracing.
+/// Acquire failures that are ordinary and transient: lock contention
+/// ([`RepoBusy`]) and an under-lock refresh that could not reach object storage
+/// ([`RepoUnavailable`]). Both already log at their raise site and both map to a
+/// retryable 503, so the handler layer logs them at warn rather than paging.
+/// Best-effort, like the database startup classifier: anything this cannot
+/// recognize counts as NOT transient and keeps its error-level log.
+///
+/// [`RepoBusy`]: crate::git::repo_store::RepoBusy
+/// [`RepoUnavailable`]: crate::git::repo_store::RepoUnavailable
+fn is_expected_transient_acquire_failure(err: &anyhow::Error) -> bool {
+    err.downcast_ref::<crate::git::repo_store::RepoBusy>()
+        .is_some()
+        || err
+            .downcast_ref::<crate::git::repo_store::RepoUnavailable>()
+            .is_some()
+}
+
 fn git_service_app_error(err: &anyhow::Error) -> AppError {
     if err
         .downcast_ref::<smart_http::GitServiceTimeout>()
@@ -2297,19 +2375,7 @@ pub async fn git_receive_pack(
     // would return 200 to the pusher before the durable copy lands, which is a larger
     // change to the client contract than the window it closes.
     let push_succeeded = receive_result.is_ok();
-    if push_succeeded {
-        tokio::spawn(post_receive_replication_tail(
-            state.clone(),
-            record.clone(),
-            ref_updates.clone(),
-            disk_path.clone(),
-            auth.0.to_string(),
-        ));
-    }
-
-    // Always release the advisory lock — even on error — to prevent stale locks
-    // from blocking subsequent pushes. Only upload to Tigris when the push
-    // succeeded; uploading a half-applied repo would propagate corruption.
+    let publish_durability = PublishDurabilitySlot::new();
     // Reclaim the write lock from the shared cell (#173 F2). This is only reachable
     // once `receive_pack` has returned, so the admission guard's copy can only ever
     // DELAY release, never perform it early; on the disconnect path this line is not
@@ -2319,7 +2385,27 @@ pub async fn git_receive_pack(
         .expect("repo write-lock mutex poisoned")
         .take()
         .expect("the write lock is only taken here, and only once");
-    reclaimed.release(push_succeeded).await;
+    if push_succeeded {
+        publish_durability.mark_release_started();
+        tokio::spawn(post_receive_replication_tail(
+            state.clone(),
+            record.clone(),
+            ref_updates.clone(),
+            disk_path.clone(),
+            auth.0.to_string(),
+            Some(publish_durability.arc()),
+        ));
+    }
+    // Short-circuit on a refused publish BEFORE anything downstream observes
+    // the push. The pack is on local disk but not in object storage, so
+    // touching the repo, recording the push, bumping trust, issuing ref
+    // certificates or answering 200 would all be reporting a write no other
+    // node can read.
+    let outcome = reclaimed.release(push_succeeded).await;
+    if push_succeeded {
+        publish_durability.record(outcome).await;
+    }
+    outcome.into_result()?;
     // Clean path: clone (a) already dropped inside run_git_service when the receive-pack
     // group was reaped; clone (b) held here spanned the success-only Tigris upload that
     // ran inside release() above. Drop it now so a second same-repo push proceeds the
@@ -2436,13 +2522,122 @@ pub async fn git_receive_pack(
 /// the per-repo-coalesced pin/encrypt task, and this push's own Pinata + announce
 /// task. Split out of `git_receive_pack` so the ordering the coalescing gate depends
 /// on is directly testable; the handler spawns it and returns.
+/// Records the release-side publish outcome for the detached post-receive tail.
+/// If the handler future is dropped after [`Self::mark_release_started`] but
+/// before [`Self::record`], the drop sets [`ReleaseOutcome::UploadUnknowable`]
+/// so the tail can proceed on local disk without waiting out the full transfer
+/// bound. A drop before release starts leaves the slot empty so the tail fails
+/// closed rather than treating an unattempted publish as unknowably durable.
+struct PublishDurabilitySlot {
+    inner: Arc<std::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>>,
+    recorded: std::sync::atomic::AtomicBool,
+    release_started: std::sync::atomic::AtomicBool,
+}
+
+impl PublishDurabilitySlot {
+    fn new() -> Self {
+        Self {
+            inner: Arc::new(std::sync::Mutex::new(None)),
+            recorded: std::sync::atomic::AtomicBool::new(false),
+            release_started: std::sync::atomic::AtomicBool::new(false),
+        }
+    }
+
+    fn mark_release_started(&self) {
+        self.release_started
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+    }
+
+    fn arc(&self) -> Arc<std::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>> {
+        Arc::clone(&self.inner)
+    }
+
+    async fn record(&self, outcome: crate::git::repo_store::ReleaseOutcome) {
+        *self
+            .inner
+            .lock()
+            .expect("publish durability mutex poisoned") = Some(outcome);
+        self.recorded
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+    }
+}
+
+impl Drop for PublishDurabilitySlot {
+    fn drop(&mut self) {
+        if self.recorded.load(std::sync::atomic::Ordering::SeqCst) {
+            return;
+        }
+        if !self
+            .release_started
+            .load(std::sync::atomic::Ordering::SeqCst)
+        {
+            return;
+        }
+        let mut slot = self
+            .inner
+            .lock()
+            .expect("publish durability mutex poisoned");
+        if slot.is_none() {
+            *slot = Some(crate::git::repo_store::ReleaseOutcome::UploadUnknowable);
+        }
+    }
+}
+
+async fn publish_durability_confirmed(
+    slot: &Option<Arc<std::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>>>,
+    wait: std::time::Duration,
+) -> bool {
+    let Some(slot) = slot else {
+        return true;
+    };
+    let start = std::time::Instant::now();
+    loop {
+        if let Some(outcome) = *slot.lock().expect("publish durability mutex poisoned") {
+            return matches!(
+                outcome,
+                crate::git::repo_store::ReleaseOutcome::Released
+                    | crate::git::repo_store::ReleaseOutcome::UploadUnknowable
+            );
+        }
+        if start.elapsed() >= wait {
+            // No outcome within the release-side transfer bound plus slack. Fail
+            // closed when the slot is still empty; an abandoned handler installs
+            // UploadUnknowable via PublishDurabilitySlot::drop so the tail can
+            // proceed on local disk without waiting out the full bound.
+            return matches!(
+                *slot.lock().expect("publish durability mutex poisoned"),
+                Some(crate::git::repo_store::ReleaseOutcome::Released)
+                    | Some(crate::git::repo_store::ReleaseOutcome::UploadUnknowable)
+            );
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+}
+
 async fn post_receive_replication_tail(
     state: AppState,
     record: RepoRecord,
     ref_updates: Vec<RefUpdate>,
     disk_path: std::path::PathBuf,
     did: String,
+    publish_durability: Option<
+        Arc<std::sync::Mutex<Option<crate::git::repo_store::ReleaseOutcome>>>,
+    >,
 ) {
+    let durability_wait = std::time::Duration::from_secs(
+        state
+            .config
+            .lock_held_transfer_timeout_secs
+            .saturating_add(5),
+    );
+    if !publish_durability_confirmed(&publish_durability, durability_wait).await {
+        tracing::warn!(
+            repo = %record.id,
+            "skipping post-receive tail: publish durability was not confirmed"
+        );
+        return;
+    }
+
     // Replication enforcement (Phase 2): decide once per push whether the public
     // may read this repo at all and, if so, which blob OIDs must not leave the
     // node. `withheld == None` means this push pins nothing (private / mode A /
@@ -2994,6 +3189,129 @@ pub async fn list_federated_repos(
 
 // ── Fork ──────────────────────────────────────────────────────────────────
 
+/// Removes a fork's local mirror clone on every exit until disarmed after the
+/// database row commits.
+struct ForkCloneGuard(Option<PathBuf>);
+
+const FORK_CREATE_CONFIRM_ATTEMPTS: u32 = 5;
+
+/// After `create_repo` fails, re-read the row with bounded retries so a
+/// transport-level error does not skip archive compensation when the insert never
+/// landed, and so a successful insert is not mistaken for a failure.
+async fn confirm_fork_repo_row(
+    db: &crate::db::Db,
+    owner_short: &str,
+    fork_name: &str,
+) -> anyhow::Result<Option<crate::db::RepoRecord>> {
+    let mut delay = std::time::Duration::from_millis(25);
+    for attempt in 0..FORK_CREATE_CONFIRM_ATTEMPTS {
+        match db.get_repo(owner_short, fork_name).await {
+            Ok(row) => return Ok(row),
+            Err(e) if attempt + 1 < FORK_CREATE_CONFIRM_ATTEMPTS => {
+                tracing::warn!(
+                    fork = %fork_name,
+                    attempt,
+                    err = %e,
+                    "fork create_repo confirmation lookup failed — retrying"
+                );
+                tokio::time::sleep(delay).await;
+                delay = delay
+                    .saturating_mul(2)
+                    .min(std::time::Duration::from_secs(1));
+            }
+            Err(e) => return Err(e),
+        }
+    }
+    unreachable!("loop returns on every attempt")
+}
+
+/// When the confirmation lookup stays unavailable after bounded retries, keep
+/// retrying in the background and compensate only after establishing that no row
+/// exists for this fork name.
+async fn schedule_fork_create_recovery(
+    db: std::sync::Arc<crate::db::Db>,
+    repo_store: crate::git::repo_store::RepoStore,
+    owner_short: String,
+    fork_name: String,
+    owner_did: String,
+    disk_path: std::path::PathBuf,
+) {
+    let mut delay = std::time::Duration::from_millis(250);
+    for attempt in 0..12 {
+        match db.get_repo(&owner_short, &fork_name).await {
+            Ok(Some(_)) => {
+                tracing::info!(
+                    fork = %fork_name,
+                    attempt,
+                    "fork create_repo recovery found a committed row — no archive compensation"
+                );
+                return;
+            }
+            Ok(None) => {
+                tracing::warn!(
+                    fork = %fork_name,
+                    attempt,
+                    "fork create_repo recovery confirmed no row — compensating orphan archive"
+                );
+                repo_store
+                    .compensate_fork_archive(&owner_did, &fork_name, disk_path.as_path())
+                    .await;
+                return;
+            }
+            Err(e) if attempt + 1 < 12 => {
+                tracing::warn!(
+                    fork = %fork_name,
+                    attempt,
+                    err = %e,
+                    "fork create_repo recovery lookup failed — retrying"
+                );
+                tokio::time::sleep(delay).await;
+                delay = delay
+                    .saturating_mul(2)
+                    .min(std::time::Duration::from_secs(30));
+            }
+            Err(e) => {
+                tracing::error!(
+                    fork = %fork_name,
+                    err = %e,
+                    "fork create_repo recovery gave up — orphan archive may remain until operator cleanup"
+                );
+                return;
+            }
+        }
+    }
+}
+
+impl ForkCloneGuard {
+    fn new(path: crate::git::repo_store::ValidatedRepoDiskPath) -> Self {
+        Self(Some(path.into_path_buf()))
+    }
+
+    fn path(&self) -> &std::path::Path {
+        self.0
+            .as_deref()
+            .expect("fork clone guard disarmed while still in use")
+    }
+
+    fn disarm(&mut self) {
+        self.0 = None;
+    }
+}
+
+impl Drop for ForkCloneGuard {
+    fn drop(&mut self) {
+        if let Some(path) = self.0.take() {
+            if let Err(e) = std::fs::remove_dir_all(&path) {
+                tracing::warn!(
+                    path = %path.display(),
+                    err = %e,
+                    "failed to remove fork clone during attempt cleanup"
+                );
+            }
+        }
+    }
+}
+
 #[derive(Debug, Deserialize)]
 pub struct ForkRepoRequest {
     pub name: Option<String>, // defaults to source repo name
@@ -3064,7 +3382,12 @@ pub async fn fork_repo(
         .await
         .map_err(|e| AppError::Git(e.to_string()))?;
 
-    let disk_path = store::repo_disk_path(&state.config.repos_dir, &forker_did, &fork_name);
+    let disk_path = crate::git::repo_store::validated_repo_disk_path(
+        &state.config.repos_dir,
+        &forker_did,
+        &fork_name,
+    )
+    .map_err(|e| AppError::BadRequest(e.to_string()))?;
 
     // Clone the source repo as a mirror
     let output = std::process::Command::new("git")
@@ -3084,11 +3407,30 @@ pub async fn fork_repo(
         )));
     }
 
-    // Upload fork to Tigris
+    let mut clone_guard = ForkCloneGuard::new(disk_path.clone());
+
+    // Upload fork to Tigris. Create-only: a refused precondition means an orphan
+    // archive already sits under this key (a failed create_repo or another
+    // writer), and proceeding would create a DB record whose archive is shadowed
+    // by bytes that are not this fork. Refuse rather than accept a fork other
+    // nodes would fetch as unrelated content. The clone guard removes the local
+    // mirror on every upload failure path.
     state
         .repo_store
         .release_after_write(&forker_did, &fork_name)
-        .await;
+        .await
+        .map_err(|e| match e {
+            crate::git::tigris::UploadError::PreconditionLost { status } => {
+                tracing::warn!(
+                    forker = %forker_did,
+                    fork = %fork_name,
+                    status,
+                    "fork refused: an archive already exists under the fork's key"
+                );
+                AppError::RepoExists(fork_name.clone())
+            }
+            other => AppError::Git(format!("fork upload failed: {other}")),
+        })?;
 
     let now = Utc::now();
     let record = crate::db::RepoRecord {
@@ -3100,12 +3442,64 @@ pub async fn fork_repo(
         default_branch: source.default_branch.clone(),
         created_at: now,
         updated_at: now,
-        disk_path: disk_path.to_string_lossy().to_string(),
+        disk_path: clone_guard.path().to_string_lossy().to_string(),
         forked_from: Some(source.id.clone()),
         machine_id: state.machine_id.clone(),
     };
 
-    state.db.create_repo(&record).await?;
+    if let Err(e) = state.db.create_repo(&record).await {
+        match confirm_fork_repo_row(&state.db, forker_short, &fork_name).await {
+            Ok(Some(committed)) => {
+                clone_guard.disarm();
+                tracing::warn!(
+                    fork = %fork_name,
+                    forker = %forker_did,
+                    "fork create_repo returned an error but the row is present — treating as success"
+                );
+                return Ok((
+                    StatusCode::CREATED,
+                    Json(to_response(&committed, &state, 0)),
+                ));
+            }
+            Ok(None) => {
+                clone_guard.disarm();
+                state
+                    .repo_store
+                    .compensate_fork_archive(&forker_did, &fork_name, disk_path.as_path())
+                    .await;
+                return Err(e.into());
+            }
+            Err(lookup_err) => {
+                clone_guard.disarm();
+                let db = std::sync::Arc::clone(&state.db);
+                let repo_store = state.repo_store.clone();
+                let owner_short = forker_short.to_string();
+                let fork_name_cl = fork_name.clone();
+                let owner_did = forker_did.clone();
+                let disk_path_cl = disk_path.as_path().to_path_buf();
+                tokio::spawn(async move {
+                    schedule_fork_create_recovery(
+                        db,
+                        repo_store,
+                        owner_short,
+                        fork_name_cl,
+                        owner_did,
+                        disk_path_cl,
+                    )
+                    .await;
+                });
+                tracing::warn!(
+                    fork = %fork_name,
+                    create_err = %e,
+                    lookup_err = %lookup_err,
+                    "fork create_repo failed and confirmation lookup stayed unavailable — scheduled recovery"
+                );
+                return Err(AppError::RepoUnavailable);
+            }
+        }
+    }
+
+    clone_guard.disarm();
 
     // Persist the proof so the fork carries it when it propagates to peers.
     if let Some(p) = verified_proof {
@@ -3343,6 +3737,129 @@ mod tests {
     const OWNER_SHORT: &str = "z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH";
     const STRANGER_DID: &str = "did:key:z6Mkffonly5tranger0000000000000000000000000000000";
 
+    #[tokio::test]
+    async fn publish_durability_slot_drop_installs_unknowable_when_release_started_but_unrecorded()
+    {
+        let slot = PublishDurabilitySlot::new();
+        slot.mark_release_started();
+        let arc = slot.arc();
+        drop(slot);
+        let outcome = *arc.lock().expect("publish durability mutex poisoned");
+        assert_eq!(
+            outcome,
+            Some(crate::git::repo_store::ReleaseOutcome::UploadUnknowable),
+            "dropping an unrecorded slot after release starts must publish UploadUnknowable for the tail"
+        );
+    }
+
+    #[test]
+    fn publish_durability_slot_drop_leaves_empty_before_release_starts() {
+        let slot = PublishDurabilitySlot::new();
+        let arc = slot.arc();
+        drop(slot);
+        assert_eq!(
+            *arc.lock().expect("publish durability mutex poisoned"),
+            None,
+            "dropping before release starts must not synthesize durability for the tail"
+        );
+    }
+
+    #[test]
+    fn publish_durability_slot_drop_waits_for_contended_mutex() {
+        let slot = PublishDurabilitySlot::new();
+        slot.mark_release_started();
+        let arc = slot.arc();
+        let holder = {
+            let arc = Arc::clone(&arc);
+            std::thread::spawn(move || {
+                let _guard = arc.lock().expect("publish durability mutex poisoned");
+                std::thread::sleep(std::time::Duration::from_millis(100));
+            })
+        };
+        std::thread::sleep(std::time::Duration::from_millis(10));
+        drop(slot);
+        holder.join().expect("mutex holder thread");
+        let outcome = *arc.lock().expect("publish durability mutex poisoned");
+        assert_eq!(
+            outcome,
+            Some(crate::git::repo_store::ReleaseOutcome::UploadUnknowable),
+            "drop must block until it can install UploadUnknowable, not give up on try_lock"
+        );
+    }
+
+    #[tokio::test]
+    async fn publish_durability_confirmed_proceeds_quickly_after_unrecorded_slot_drop() {
+        let start = std::time::Instant::now();
+        let slot = PublishDurabilitySlot::new();
+        slot.mark_release_started();
+        let arc = slot.arc();
+        drop(slot);
+        let confirmed =
+            publish_durability_confirmed(&Some(arc), std::time::Duration::from_millis(50)).await;
+        assert!(confirmed);
+        assert!(
+            start.elapsed() < std::time::Duration::from_millis(200),
+            "the tail must not wait out the full transfer bound when the slot already carries UploadUnknowable"
+        );
+    }
+
+    #[tokio::test]
+    async fn publish_durability_confirmed_fails_closed_when_release_never_records() {
+        let slot = Arc::new(std::sync::Mutex::new(None));
+        let confirmed =
+            publish_durability_confirmed(&Some(slot), std::time::Duration::from_millis(30)).await;
+        assert!(
+            !confirmed,
+            "an empty slot after the bounded wait must not admit the tail; only an installed \
+             Released or UploadUnknowable outcome may"
+        );
+    }
+
+    #[tokio::test]
+    async fn publish_durability_confirmed_accepts_only_released() {
+        let slot = Arc::new(std::sync::Mutex::new(Some(
+            crate::git::repo_store::ReleaseOutcome::Released,
+        )));
+        assert!(
+            publish_durability_confirmed(&Some(slot), std::time::Duration::from_millis(5)).await
+        );
+
+        let slot = Arc::new(std::sync::Mutex::new(Some(
+            crate::git::repo_store::ReleaseOutcome::UploadUnknowable,
+        )));
+        assert!(
+            publish_durability_confirmed(&Some(slot), std::time::Duration::from_millis(5)).await,
+            "an unknowable upload still landed on local disk, so the tail may proceed"
+        );
+
+        let slot = Arc::new(std::sync::Mutex::new(Some(
+            crate::git::repo_store::ReleaseOutcome::UploadFailed,
+        )));
+        assert!(
+            !publish_durability_confirmed(&Some(slot), std::time::Duration::from_millis(5)).await
+        );
+    }
+
+    #[test]
+    fn fork_clone_guard_removes_mirror_on_drop() {
+        let root = tempfile::TempDir::new().unwrap();
+        let validated = crate::git::repo_store::validated_repo_disk_path(
+            root.path(),
+            "did:key:testfork",
+            "fork",
+        )
+        .expect("test fork path must validate");
+        std::fs::create_dir_all(validated.as_path()).unwrap();
+        {
+            let _guard = ForkCloneGuard::new(validated.clone());
+            assert!(validated.exists());
+        }
+        assert!(
+            !validated.exists(),
+            "dropping the fork clone guard must remove the mirror directory"
+        );
+    }
+
     #[test]
     fn upload_pack_request_finalizes_only_with_done_pktline() {
         let want = "0032want 1111111111111111111111111111111111111111\n";
@@ -3525,6 +4042,55 @@ mod tests {
         assert!(git_permit(&sem).is_ok());
     }
 
+    #[test]
+    fn is_expected_transient_matches_both_typed_refusals() {
+        // The real raise shape wraps the marker in a `.context()` layer naming the
+        // owner slug and repo, so the downcast has to survive that wrapping.
+        let busy = anyhow::Error::new(crate::git::repo_store::RepoBusy)
+            .context("another write is in progress for alice/demo");
+        assert!(is_expected_transient_acquire_failure(&busy));
+
+        let unavailable = anyhow::Error::new(crate::git::repo_store::RepoUnavailable)
+            .context("could not read the archive HEAD for alice/demo");
+        assert!(is_expected_transient_acquire_failure(&unavailable));
+    }
+
+    #[test]
+    fn is_expected_transient_rejects_unrelated_failures() {
+        // Anything the classifier cannot recognize keeps paging at error level.
+        let other = anyhow::anyhow!("disk on fire");
+        assert!(!is_expected_transient_acquire_failure(&other));
+    }
+
+    #[test]
+    fn acquire_write_app_error_maps_transient_markers_to_retryable_503() {
+        let busy = anyhow::Error::new(crate::git::repo_store::RepoBusy)
+            .context("another write is in progress for alice/demo");
+        assert!(matches!(
+            acquire_write_app_error(&busy, "demo"),
+            AppError::RepoBusy
+        ));
+
+        let unavailable = anyhow::Error::new(crate::git::repo_store::RepoUnavailable)
+            .context("could not read the archive HEAD for alice/demo");
+        assert!(matches!(
+            acquire_write_app_error(&unavailable, "demo"),
+            AppError::RepoUnavailable
+        ));
+
+        let pool = anyhow::Error::new(crate::git::repo_store::LockPoolBusy);
+        assert!(matches!(
+            acquire_write_app_error(&pool, "demo"),
+            AppError::Overloaded(_)
+        ));
+
+        let other = anyhow::anyhow!("disk on fire");
+        assert!(matches!(
+            acquire_write_app_error(&other, "demo"),
+            AppError::Git(_)
+        ));
+    }
+
     fn repo_owned_by(owner_did: &str) -> crate::db::RepoRecord {
         let now = chrono::Utc::now();
         crate::db::RepoRecord {
@@ -6255,14 +6821,12 @@ mod tests {
             repos_dir.path().to_path_buf(),
             None,
             crate::git::repo_store::build_lock_pool(&pool, 4, std::time::Duration::from_secs(5)),
+            std::time::Duration::from_secs(300),
         );
         let mut cfg = (*state.config).clone();
         // Long enough that the git-service timeout is never what ends this push; the
         // disconnect is.
         cfg.git_service_timeout_secs = 600;
-        // Lock-path test, not owner-push authorization: pin off so the default-on gate
-        // does not reject the proof DID before receive-pack runs.
-        cfg.enforce_owner_push = false;
         state.config = std::sync::Arc::new(cfg);
         state
             .db
@@ -6305,9 +6869,7 @@ mod tests {
                 let mut fut = Box::pin(git_receive_pack(
                     State(state.clone()),
                     Path((owner.to_string(), name.to_string())),
-                    Extension(crate::auth::AuthenticatedDid(
-                        "did:key:z6MkDisconnectWriteLockProofDidAAAAAAAA".to_string(),
-                    )),
+                    Extension(crate::auth::AuthenticatedDid("did:key:z6disc".to_string())),
                     crate::rate_limit::PeerAddr(Some(
                         "203.0.113.81:5000".parse::<SocketAddr>().unwrap(),
                     )),
@@ -6432,10 +6994,8 @@ mod tests {
             repos_dir.path().to_path_buf(),
             None,
             crate::git::repo_store::build_lock_pool(&pool, 4, std::time::Duration::from_secs(5)),
+            std::time::Duration::from_secs(300),
         );
-        let mut cfg = (*state.config).clone();
-        cfg.enforce_owner_push = false;
-        state.config = std::sync::Arc::new(cfg);
         state
             .db
             .upsert_mirror_repo(owner, name, "/tmp/z6succ-sc1", None, false)
@@ -6459,9 +7019,7 @@ mod tests {
                 git_receive_pack(
                     State(state.clone()),
                     Path((owner.to_string(), name.to_string())),
-                    Extension(crate::auth::AuthenticatedDid(
-                        "did:key:z6MkPushSuccessReleaseProofDidAAAAAAAA".to_string(),
-                    )),
+                    Extension(crate::auth::AuthenticatedDid("did:key:z6succ".to_string())),
                     crate::rate_limit::PeerAddr(Some(
                         "203.0.113.83:5000".parse::<SocketAddr>().unwrap(),
                     )),
@@ -6522,15 +7080,13 @@ mod tests {
         let owner = "z6lockpool";
         let name = "lp1";
         let mut state = crate::test_support::test_state(pool.clone()).await;
-        let mut cfg = (*state.config).clone();
-        cfg.enforce_owner_push = false;
-        state.config = std::sync::Arc::new(cfg);
         // One lock-pool connection, short checkout timeout so the exhaustion surfaces
         // promptly rather than at the handler's own acquire deadline.
         state.repo_store = crate::git::repo_store::RepoStore::new(
             std::path::PathBuf::from("/tmp/gitlawb-lockpool-shed"),
             None,
             crate::git::repo_store::build_lock_pool(&pool, 1, std::time::Duration::from_secs(1)),
+            std::time::Duration::from_secs(300),
         );
         state
             .db
@@ -6538,7 +7094,7 @@ mod tests {
             .await
             .unwrap();
 
-        let did = "did:key:z6MkLockPoolShedProofDidAAAAAAAAAAAAAAAAAA";
+        let did = "did:key:z6lockpool";
         let peer: SocketAddr = "203.0.113.71:5000".parse().unwrap();
 
         // Occupy the only lock-pool connection with a write on an UNRELATED repo.
@@ -6565,7 +7121,7 @@ mod tests {
 
         // MUST-NOT: with the pool free again, the push is not shed as capacity (it fails
         // later on the nonexistent on-disk repo, which is a git error, not Overloaded).
-        held.release(false).await;
+        let _ = held.release(false).await;
         let admitted = git_receive_pack(
             State(state.clone()),
             Path((owner.to_string(), name.to_string())),
@@ -9640,6 +10196,43 @@ mod tests {
 
     const F2A_PUSHER: &str = "did:key:z6MkF2aPusherAAAAAAAAAAAAAAAAAAAAAAAAAA";
 
+    /// When publish durability is fenced, the tail must not run walks or take
+    /// coalescing keys before returning.
+    #[cfg(unix)]
+    #[sqlx::test]
+    async fn post_receive_tail_skips_all_work_when_publish_fenced(pool: sqlx::PgPool) {
+        let repo = tempfile::TempDir::new().unwrap();
+        let bin = tempfile::TempDir::new().unwrap();
+        let log = bin.path().join("git.log");
+        let git_bin = f2a_logging_git(bin.path(), &log);
+        u5_init_repo(repo.path());
+        let c1 = u5_commit_file(repo.path(), "a.txt", "one\n");
+        let (state, rec) = f2a_state(pool, &git_bin, "z6f2afence", "fence-repo", false).await;
+        let slot = Arc::new(std::sync::Mutex::new(Some(
+            crate::git::repo_store::ReleaseOutcome::Fenced,
+        )));
+        post_receive_replication_tail(
+            state.clone(),
+            rec,
+            f2a_update("refs/heads/main", &c1),
+            repo.path().to_path_buf(),
+            F2A_PUSHER.to_string(),
+            Some(slot),
+        )
+        .await;
+        assert_eq!(
+            f2a_walks(&log),
+            0,
+            "a fenced publish must not run the replication walk; log:\n{}",
+            f2a_log(&log)
+        );
+        assert_eq!(
+            state.encrypt_inflight.len(),
+            0,
+            "a fenced publish must not take the per-repo coalescing key"
+        );
+    }
+
     /// Scenario 1 (the finding). A second rapid push to the same repo coalesces
     /// WITHOUT running the withheld walk. Asserted on the walk's git children, not
     /// on the `Coalesced` outcome: with `try_begin` below the walk (the pre-fix
@@ -9669,6 +10262,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
         let after_first = f2a_walks(&log);
@@ -9689,6 +10283,7 @@ mod tests {
             f2a_update("refs/heads/second", &c1),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -9762,6 +10357,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -9843,6 +10439,7 @@ mod tests {
             f2a_update("refs/heads/main", &c2),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         ));
         f2a_wait_for(|| started.exists(), "the admitted push's walk to start").await;
 
@@ -9956,6 +10553,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10234,6 +10832,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10289,6 +10888,7 @@ mod tests {
             f2a_update("refs/heads/main", &c1),
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
         tokio::time::sleep(std::time::Duration::from_millis(300)).await;
@@ -10330,6 +10930,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 
@@ -10427,6 +11028,7 @@ mod tests {
             }],
             repo.path().to_path_buf(),
             F2A_PUSHER.to_string(),
+            None,
         )
         .await;
 

crates/gitlawb-node/src/auth/mod.rs

@@ -517,6 +517,7 @@ mod tests {
             rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
             create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
             push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
+            close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
             ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
             ipfs_work_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
             ipfs_max_history_walks: crate::api::ipfs::MAX_HISTORY_WALKS_PER_REQUEST,

crates/gitlawb-node/src/config.rs

@@ -252,11 +252,9 @@ pub struct Config {
 
     /// Maximum connections in the PostgreSQL pool. This is a cap, not a floor
     /// (connections open lazily). Size against the database server's
-    /// max_connections, remembering admin tooling opens its own pool. Each
-    /// concurrent write pins one pooled connection for its whole duration (the
-    /// advisory lock in `repo_store::acquire_write` is connection-affine), so this
-    /// must exceed `max_concurrent_git_pushes` by `DB_POOL_APP_HEADROOM` or slow
-    /// pushes starve every other DB path — enforced by `Config::validate`.
+    /// max_connections, remembering admin tooling opens its own pool. Writes pin
+    /// the dedicated lock pool, not this one; this pool still needs enough headroom
+    /// for ordinary request handlers and metadata paths (`DB_POOL_APP_HEADROOM`).
     #[arg(
         long,
         env = "GITLAWB_DB_MAX_CONNECTIONS",
@@ -265,6 +263,46 @@ pub struct Config {
     )]
     pub db_max_connections: u32,
 
+    /// Maximum connections in the dedicated advisory-lock pool, which is separate
+    /// from the main pool above.
+    ///
+    /// Size this against the expected peak number of concurrent distinct-repo
+    /// writers, NOT small. Every in-flight repo write pins one connection here for
+    /// its whole duration (the write, its metadata tail, and the bounded archive
+    /// upload), so this value is a hard ceiling on simultaneous writes node-wide.
+    /// Keeping it separate from GITLAWB_DB_MAX_CONNECTIONS is what stops a push
+    /// burst from starving ordinary request handlers; the cost is that
+    /// (main pool + lock pool) must fit inside the database server's
+    /// max_connections, times the number of nodes, plus admin tooling.
+    #[arg(
+        long,
+        env = "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS",
+        default_value_t = 40,
+        value_parser = clap::value_parser!(u32).range(1..)
+    )]
+    pub db_lock_pool_max_connections: u32,
+
+    /// Upper bound, in seconds, on any single object-storage transfer that runs
+    /// while the per-repo advisory lock is HELD.
+    ///
+    /// Two bounded spans exist, and the bound applies per span. The acquire-side
+    /// refresh in `acquire_write` covers the existence HEAD and the download
+    /// together under ONE budget (it runs after the lock is taken and before the
+    /// guard is constructed), and the archive upload in `release` gets its own.
+    /// Worst-case slot occupancy is therefore about twice this value plus the git
+    /// work between them, not one times this value. Both used to be free, because the lock's
+    /// connection was returned to the pool immediately; now that a write guard
+    /// pins a lock-pool connection for its whole lifetime, an unbounded transfer
+    /// holds that slot, and enough stalled transfers deny every write on the node.
+    /// This is the bound that keeps a stall from becoming an outage.
+    #[arg(
+        long,
+        env = "GITLAWB_LOCK_HELD_TRANSFER_TIMEOUT_SECS",
+        default_value_t = 300,
+        value_parser = clap::value_parser!(u64).range(1..)
+    )]
+    pub lock_held_transfer_timeout_secs: u64,
+
     /// Maximum time a request waits for a pool connection before failing with
     /// 503, in seconds. Bounds queueing when the database is slow or down.
     #[arg(
@@ -359,15 +397,14 @@ pub struct Config {
     /// error rather than a boot-time panic).
     ///
     /// CONNECTION BUDGET. A push holds a Postgres connection from the node's separate
-    /// advisory-lock pool for the whole receive-pack, and that pool is sized from this
-    /// knob (this value + 8, clamped to 64 in `main.rs`). The node's total ceiling is
-    /// therefore `db_max_connections` (default 48) + the lock pool (default 40), i.e.
-    /// 88 by default, and at most `db_max_connections` + 64. Size BOTH against the
-    /// database server's `max_connections`: `db_max_connections`' own doc predates the
-    /// lock pool and no longer covers most of the node's connections. The +8 headroom
-    /// is shared with the three non-push `acquire_write` callers (`api/issues.rs` x2,
-    /// `api/pulls.rs`). Raising this knob past the clamp does NOT buy more lock-pool
-    /// connections; pushes beyond it wait briefly and then shed a 503 + Retry-After.
+    /// advisory-lock pool (`GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS`, default 40) for
+    /// the whole receive-pack. That pool must be at least this value plus
+    /// `DB_LOCK_POOL_NON_PUSH_HEADROOM` so admitted pushes and non-push mutations
+    /// can each pin a lock-pool connection for their whole duration. Size BOTH
+    /// pools against the database server's `max_connections`: the main pool
+    /// (`GITLAWB_DB_MAX_CONNECTIONS`, default 48) serves ordinary handlers, and the
+    /// lock pool serves writes. Raising this knob does not raise the lock pool;
+    /// set `GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS` explicitly.
     #[arg(
         long,
         env = "GITLAWB_MAX_CONCURRENT_GIT_PUSHES",
@@ -725,25 +762,37 @@ impl Config {
     /// the pool must clear the concurrent-write cap by at least this margin.
     pub const DB_POOL_APP_HEADROOM: u32 = 8;
 
+    /// Lock-pool slots reserved for non-push `acquire_write` callers (issue create,
+    /// close, merge) so a saturated push budget cannot turn ordinary mutations into
+    /// lock-pool exhaustion on unrelated repositories.
+    pub const DB_LOCK_POOL_NON_PUSH_HEADROOM: u32 = 8;
+
     /// Cross-field boot validation. Single-field ranges are enforced by clap; this
     /// catches combinations that ship a denial-of-service under otherwise-valid
     /// values. Call once at startup and fail fast on `Err`.
     pub fn validate(&self) -> Result<(), String> {
-        // A write pins one pooled connection for its whole duration (the
-        // connection-affine advisory lock in repo_store::acquire_write), and
-        // concurrent writes are capped at max_concurrent_git_pushes. If the pool
-        // does not exceed that cap by DB_POOL_APP_HEADROOM, a burst of slow pushes
-        // drains every connection and every other DB path 503s. (#174 F1)
-        let floor = (self.max_concurrent_git_pushes as u64) + (Self::DB_POOL_APP_HEADROOM as u64);
-        if (self.db_max_connections as u64) < floor {
+        // Concurrent git writes pin the dedicated lock pool for their whole
+        // duration (connection-affine advisory lock in acquire_write). The main
+        // pool no longer carries that occupancy.
+        let lock_floor = self
+            .max_concurrent_git_pushes
+            .saturating_add(Self::DB_LOCK_POOL_NON_PUSH_HEADROOM as usize);
+        if (self.db_lock_pool_max_connections as usize) < lock_floor {
             return Err(format!(
-                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least max_concurrent_git_pushes ({}) \
-                 + {} headroom = {}: each concurrent write pins one pooled connection for its whole \
-                 duration, so a smaller pool lets a burst of slow pushes starve every other DB path.",
-                self.db_max_connections,
+                "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS ({}) must be at least \
+                 max_concurrent_git_pushes ({}) plus {} for non-push mutations \
+                 that share the lock pool",
+                self.db_lock_pool_max_connections,
                 self.max_concurrent_git_pushes,
-                Self::DB_POOL_APP_HEADROOM,
-                floor
+                Self::DB_LOCK_POOL_NON_PUSH_HEADROOM
+            ));
+        }
+        let main_floor = Self::DB_POOL_APP_HEADROOM as u64;
+        if (self.db_max_connections as u64) < main_floor {
+            return Err(format!(
+                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least {} for non-git \
+                 database paths",
+                self.db_max_connections, main_floor
             ));
         }
         Ok(())
@@ -754,6 +803,25 @@ impl Config {
 mod tests {
     use super::*;
 
+    #[test]
+    fn lock_pool_size_defaults_to_40_and_rejects_zero() {
+        assert_eq!(
+            Config::parse_from(["gitlawb-node"]).db_lock_pool_max_connections,
+            40
+        );
+        assert_eq!(
+            Config::parse_from(["gitlawb-node", "--db-lock-pool-max-connections", "8"])
+                .db_lock_pool_max_connections,
+            8
+        );
+        // A zero-sized lock pool would deny every write, so clap must reject it
+        // rather than let a node boot into a state where no repo can be written.
+        assert!(
+            Config::try_parse_from(["gitlawb-node", "--db-lock-pool-max-connections", "0"])
+                .is_err()
+        );
+    }
+
     #[test]
     fn git_service_timeout_defaults_to_600_and_rejects_zero() {
         assert_eq!(
@@ -1363,41 +1431,61 @@ mod tests {
         );
     }
 
-    /// #174 F1: a connection-affine write lock pins a pooled connection per
-    /// concurrent write, so the pool must clear `max_concurrent_git_pushes` by
-    /// `DB_POOL_APP_HEADROOM` or a push burst starves every other DB path.
-    /// `validate()` must reject an under-sized pool at boot.
+    /// #174 F1: concurrent git writes pin the dedicated lock pool, not the main
+    /// pool. `validate()` must reject an under-sized lock pool at boot.
     #[test]
     fn db_pool_must_clear_the_git_push_cap() {
-        // Shipped defaults validate (48 >= 32 + 8).
+        // Shipped defaults validate (lock pool 40 >= pushes 32 + non-push headroom 8).
         Config::parse_from(["gitlawb-node"])
             .validate()
             .expect("default config must validate");
 
-        // An under-sized pool relative to the push cap is rejected (20 < 32 + 8).
-        let under = Config::parse_from([
+        // An under-sized lock pool relative to the push cap is rejected.
+        let under_lock = Config::parse_from([
             "gitlawb-node",
-            "--db-max-connections",
-            "20",
+            "--db-lock-pool-max-connections",
+            "16",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            under.validate().is_err(),
-            "db_max_connections 20 below max_concurrent_git_pushes 32 + headroom must be rejected"
+            under_lock.validate().is_err(),
+            "db_lock_pool_max_connections below max_concurrent_git_pushes + headroom must be rejected"
         );
 
-        // Exactly at the floor validates (40 == 32 + 8).
-        let at_floor = Config::parse_from([
+        // Exactly at the push cap without non-push headroom is rejected.
+        let push_only = Config::parse_from([
+            "gitlawb-node",
+            "--db-lock-pool-max-connections",
+            "32",
+            "--max-concurrent-git-pushes",
+            "32",
+        ]);
+        assert!(
+            push_only.validate().is_err(),
+            "db_lock_pool_max_connections equal to max_concurrent_git_pushes must be rejected"
+        );
+
+        // Main pool can be smaller than pushes + headroom when the lock pool carries writes.
+        let split = Config::parse_from([
             "gitlawb-node",
             "--db-max-connections",
+            "16",
+            "--db-lock-pool-max-connections",
             "40",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            at_floor.validate().is_ok(),
-            "db_max_connections at the floor (pushes + headroom) must validate"
+            split.validate().is_ok(),
+            "a small main pool with a large lock pool must validate"
+        );
+
+        // Main pool still needs the app headroom floor.
+        let under_main = Config::parse_from(["gitlawb-node", "--db-max-connections", "4"]);
+        assert!(
+            under_main.validate().is_err(),
+            "db_max_connections below DB_POOL_APP_HEADROOM must be rejected"
         );
     }
 

crates/gitlawb-node/src/db/mod.rs

@@ -265,11 +265,6 @@ pub struct Db {
 }
 
 impl Db {
-    /// Access the underlying Postgres connection pool.
-    pub fn pool(&self) -> &PgPool {
-        &self.pool
-    }
-
     #[cfg(test)]
     pub fn for_testing(pool: PgPool) -> Self {
         Self { pool }
@@ -310,6 +305,46 @@ impl Db {
         Ok(db)
     }
 
+    /// Build the dedicated pool that advisory-lock connections come from.
+    ///
+    /// Deliberately **lazy**: connections open on first use rather than at boot.
+    /// The main pool has to connect eagerly because it runs migrations, which is
+    /// why it needs `connect_db_with_retry`'s backoff and degraded-server
+    /// handoff. This pool has no startup work at all, so an eager connect would
+    /// only add a new way for the process to fail to boot, and would need a
+    /// second copy of that retry machinery to be safe. Being lazy removes the
+    /// failure mode instead of handling it: if Postgres is unreachable when the
+    /// first write arrives, that write fails on the pool's own acquire timeout,
+    /// the same way any other database-backed request already does.
+    ///
+    /// Kept separate from the main pool so a burst of lock-holding connections
+    /// cannot starve ordinary request handlers. See
+    /// `GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS` for the sizing tradeoff.
+    pub fn lock_pool(
+        database_url: &str,
+        max_connections: u32,
+        acquire_timeout: Duration,
+    ) -> Result<PgPool> {
+        info!(
+            max_connections,
+            acquire_timeout_secs = acquire_timeout.as_secs(),
+            "creating dedicated advisory-lock pool (lazy)"
+        );
+        PgPoolOptions::new()
+            .max_connections(max_connections)
+            // Explicit, and load-bearing rather than cosmetic: `RepoWriteGuard::Drop`
+            // has a no-runtime branch that calls `PoolConnection::leak()` and relies
+            // on the husk's own drop doing nothing. With `min_connections > 0` that
+            // drop still spawns a pool-replenish task, and spawning without a runtime
+            // panics inside Drop, which aborts the process during unwind. It is 0 by
+            // default, so this line exists to keep a future tuning change from
+            // silently re-arming that panic.
+            .min_connections(0)
+            .acquire_timeout(acquire_timeout)
+            .connect_lazy(database_url)
+            .context("creating advisory-lock pool")
+    }
+
     /// Cheap liveness probe against the pool, for readiness checks: one
     /// `SELECT 1` that fails fast when the database is unreachable.
     pub async fn ping(&self) -> Result<()> {

crates/gitlawb-node/src/error.rs

@@ -69,6 +69,15 @@ pub enum AppError {
     #[error("server overloaded: {0}")]
     Overloaded(String),
 
+    #[error("repository is busy")]
+    RepoBusy,
+
+    #[error("repository is temporarily unavailable")]
+    RepoUnavailable,
+
+    #[error("repository write was fenced by a concurrent publish")]
+    RepoWriteFenced,
+
     #[error("database error: {0}")]
     Db(#[from] sqlx::Error),
 
@@ -110,7 +119,31 @@ impl From<anyhow::Error> for AppError {
     fn from(err: anyhow::Error) -> Self {
         match err.downcast::<sqlx::Error>() {
             Ok(sql) => AppError::Db(sql),
-            Err(err) => AppError::Internal(err),
+            // Lock contention is transient and ordinary, so it must not land as a
+            // 500. The internal message names the owner slug and repo, so the
+            // variant carries nothing: the detail stays in the log at the raise
+            // site and the client gets a fixed retryable body.
+            Err(err) => match err.downcast::<crate::git::repo_store::LockPoolBusy>() {
+                Ok(_) => AppError::Overloaded("git write locks at capacity, retry shortly".into()),
+                Err(err) => match err.downcast::<crate::git::repo_store::RepoBusy>() {
+                    Ok(_) => AppError::RepoBusy,
+                    // Same reasoning one rung down: a refused under-lock refresh is a
+                    // transient storage condition, and its internal message names the
+                    // owner slug and repo, so the variant carries nothing.
+                    Err(err) => match err.downcast::<crate::git::repo_store::RepoUnavailable>() {
+                        Ok(_) => AppError::RepoUnavailable,
+                        // And one more rung: a publish the store refused twice is
+                        // transient in the same way, and the retry is the client's
+                        // to make. The variant carries nothing for the same reason
+                        // as the two above.
+                        Err(err) => match err.downcast::<crate::git::repo_store::RepoWriteFenced>()
+                        {
+                            Ok(_) => AppError::RepoWriteFenced,
+                            Err(err) => AppError::Internal(err),
+                        },
+                    },
+                },
+            },
         }
     }
 }
@@ -184,6 +217,29 @@ impl IntoResponse for AppError {
             // 504, distinct from the 500 git_error and from the read-gate's 404 /
             // the auth 401, so the client can tell a deadline from a failure.
             AppError::Timeout(msg) => (StatusCode::GATEWAY_TIMEOUT, "git_timeout", msg.clone()),
+            // 503 with a FIXED body: the caller should retry, and must not be told
+            // which repo is contended or for how long.
+            AppError::RepoBusy => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "repo_busy",
+                "repository is busy — retry".into(),
+            ),
+            // 503 with a FIXED body for the same reason: the caller should retry, and
+            // must not be told which repo could not be refreshed or why.
+            AppError::RepoUnavailable => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "repo_unavailable",
+                "repository is temporarily unavailable, retry".into(),
+            ),
+            // 503 with a FIXED body again, and its own code: the caller should
+            // retry, but the condition is not contention, so a client that
+            // distinguishes them should be able to. The body must not say which
+            // repo lost its publish or to whom.
+            AppError::RepoWriteFenced => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "repo_write_fenced",
+                "repository changed underneath this write, retry".into(),
+            ),
             AppError::Db(e) if db_unavailable(e) => (
                 StatusCode::SERVICE_UNAVAILABLE,
                 DB_UNAVAILABLE_CODE,
@@ -244,7 +300,11 @@ impl IntoResponse for AppError {
         // here rather than in bespoke early returns, keeping each variant handled once.
         if matches!(
             self,
-            AppError::Overloaded(_) | AppError::SearchIncomplete { .. }
+            AppError::Overloaded(_)
+                | AppError::SearchIncomplete { .. }
+                | AppError::RepoBusy
+                | AppError::RepoUnavailable
+                | AppError::RepoWriteFenced
         ) {
             resp.headers_mut().insert(
                 axum::http::header::RETRY_AFTER,
@@ -362,4 +422,31 @@ mod tests {
         let resp = err.into_response();
         assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
     }
+
+    #[test]
+    fn lock_pool_busy_via_anyhow_from_is_503_overloaded() {
+        let err: AppError = anyhow::Error::new(crate::git::repo_store::LockPoolBusy).into();
+        let resp = err.into_response();
+        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(
+            resp.headers().get("retry-after").unwrap().to_str().unwrap(),
+            "1"
+        );
+    }
+
+    #[test]
+    fn repo_busy_unavailable_and_fenced_advertise_retry_after() {
+        for err in [
+            AppError::RepoBusy,
+            AppError::RepoUnavailable,
+            AppError::RepoWriteFenced,
+        ] {
+            let resp = err.into_response();
+            assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+            assert_eq!(
+                resp.headers().get("retry-after").unwrap().to_str().unwrap(),
+                "1"
+            );
+        }
+    }
 }

crates/gitlawb-node/src/git/issues.rs

@@ -61,6 +61,27 @@ pub fn create_issue(repo_path: &Path, issue_id: &str, json: &str) -> Result<()>
     Ok(())
 }
 
+/// Remove a single issue ref while the write guard still holds the advisory lock,
+/// after a definite publish refusal. Best-effort: a failed delete leaves a
+/// local-only orphan, which is still better than telling the client to retry into a
+/// duplicate id. Not used on `UploadUnknowable`, where the PUT may still land.
+pub fn delete_issue_ref(
+    git_bin: &str,
+    repo_path: &Path,
+    issue_id: &str,
+    deadline: std::time::Instant,
+) -> Result<()> {
+    let ref_name = format!("refs/gitlawb/issues/{issue_id}");
+    crate::git::visibility_pack::run_bounded_git(
+        git_bin,
+        &["update-ref", "-d", &ref_name],
+        repo_path,
+        b"",
+        deadline,
+    )?;
+    Ok(())
+}
+
 /// List all issue refs and return their JSON content.
 pub fn list_issues(repo_path: &Path) -> Result<Vec<String>> {
     // List all refs under refs/gitlawb/issues/
@@ -278,6 +299,27 @@ mod tests {
         assert!(result.unwrap_err().to_string().contains("ambiguous"));
     }
 
+    #[test]
+    fn delete_issue_ref_removes_a_created_ref() {
+        let dir = TempDir::new().unwrap();
+        init_repo(&dir);
+        let full_id = "eee88888-0000-0000-0000-000000000000";
+        create_issue(
+            dir.path(),
+            full_id,
+            r#"{"id":"eee88888-0000-0000-0000-000000000000","status":"open"}"#,
+        )
+        .unwrap();
+        delete_issue_ref(
+            "git",
+            dir.path(),
+            full_id,
+            std::time::Instant::now() + std::time::Duration::from_secs(30),
+        )
+        .unwrap();
+        assert_eq!(resolve_issue_id(dir.path(), full_id).unwrap(), None);
+    }
+
     #[test]
     fn test_close_issue_via_prefix() {
         let dir = TempDir::new().unwrap();

crates/gitlawb-node/src/git/store.rs

@@ -870,6 +870,7 @@ pub fn merge_branch(
 }
 
 /// Resolve a repo disk path: {repos_dir}/{owner_slug}/{repo_name}.git
+#[allow(dead_code)] // exercised from test_support and state tests; production uses validated_repo_disk_path
 pub fn repo_disk_path(repos_dir: &Path, owner_did: &str, repo_name: &str) -> PathBuf {
     // Sanitize the DID for use as a directory name
     let owner_slug = owner_did.replace([':', '/'], "_");

crates/gitlawb-node/src/git/tigris.rs

@@ -5,12 +5,41 @@
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex, OnceLock};
 
 use anyhow::{Context, Result};
 use aws_sdk_s3::Client as S3Client;
 use tracing::{debug, info};
 
+/// The precondition an upload is fenced on.
+///
+/// Object storage is the only place a fence can hold. Dropping the future of an
+/// in-flight PUT does not cancel the request the server is already processing,
+/// so no amount of local locking stops an abandoned writer's bytes from landing
+/// after a successor has published. A conditional PUT the store itself refuses
+/// is what actually stops it.
+#[derive(Clone, Debug)]
+pub enum UploadPrecondition {
+    /// Publish only if the stored object is still the generation we observed.
+    IfMatch(String),
+    /// Publish only if nothing is stored under the key yet.
+    IfAbsent,
+    /// No fence. Last writer wins.
+    Unconditional,
+}
+
+/// Why an upload failed, split so a caller can tell "someone else already
+/// published this key" (expected, and dropping our bytes is the correct
+/// outcome) from a real storage failure.
+#[derive(Debug, thiserror::Error)]
+pub enum UploadError {
+    #[error("upload precondition lost (HTTP {status})")]
+    PreconditionLost { status: u16 },
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
 /// Wrapper around the S3 client with the configured bucket.
 #[derive(Clone)]
 pub struct TigrisClient {
@@ -31,21 +60,29 @@ impl TigrisClient {
         })
     }
 
-    /// Test-only constructor with an explicit S3 endpoint, region, and static
-    /// credentials — no env-var reads, so parallel tests cannot race each other's
-    /// `AWS_*` environment the way the env-based `new` would. Lets a test point
-    /// the client at a non-routable endpoint to exercise acquire-stall paths.
+    /// Build a client pointed at an arbitrary endpoint, for tests.
+    ///
+    /// The production constructor reads the endpoint and credentials from the
+    /// environment, which a test cannot steer without mutating process-global
+    /// state. This takes both explicitly so a test can aim the client at a
+    /// closed port and get a prompt transport error out of `exists()`.
+    ///
+    /// `RetryConfig::disabled()` is load-bearing, not tidiness: the SDK's default
+    /// policy retries a connection refusal with backoff, which turns each failing
+    /// call into seconds of waiting.
     #[cfg(test)]
-    pub(crate) async fn for_testing_with_endpoint(bucket: &str, endpoint_url: &str) -> Self {
-        let creds = aws_sdk_s3::config::Credentials::new("test", "test", None, None, "test");
-        let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
-            .endpoint_url(endpoint_url)
-            .region(aws_config::Region::new("auto"))
-            .credentials_provider(creds)
-            .load()
-            .await;
+    pub fn for_testing_with_endpoint(bucket: &str, endpoint: &str) -> Self {
+        use aws_sdk_s3::config::{retry::RetryConfig, Credentials, Region};
+
+        let config = aws_sdk_s3::config::Config::builder()
+            .endpoint_url(endpoint)
+            .credentials_provider(Credentials::new("test", "test", None, None, "test"))
+            .region(Region::new("auto"))
+            .retry_config(RetryConfig::disabled())
+            .behavior_version_latest()
+            .build();
         Self {
-            s3: S3Client::new(&config),
+            s3: S3Client::from_conf(config),
             bucket: bucket.to_string(),
         }
     }
@@ -77,8 +114,47 @@ impl TigrisClient {
         }
     }
 
-    /// Upload a local bare repo directory to Tigris as a tar.zst archive.
-    pub async fn upload(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()> {
+    /// Read the ETag of a repo archive, or `None` when nothing is stored under
+    /// the key. The ETag identifies the generation a later conditional upload
+    /// can fence itself on.
+    ///
+    /// Separate from `exists` rather than folded into it: `exists` has callers
+    /// that only want the boolean, and widening its return type would churn
+    /// every one of them for no benefit.
+    pub async fn head_etag(&self, owner_slug: &str, repo_name: &str) -> Result<Option<String>> {
+        let key = Self::repo_key(owner_slug, repo_name);
+        match self
+            .s3
+            .head_object()
+            .bucket(&self.bucket)
+            .key(&key)
+            .send()
+            .await
+        {
+            Ok(out) => Ok(Some(
+                out.e_tag()
+                    .context(format!("tigris HEAD {key}: hit carried no ETag"))?
+                    .to_string(),
+            )),
+            Err(e) => {
+                if e.as_service_error().is_some_and(|e| e.is_not_found()) {
+                    Ok(None)
+                } else {
+                    Err(anyhow::anyhow!("tigris HEAD {key}: {e}"))
+                }
+            }
+        }
+    }
+
+    /// Upload a local bare repo directory to Tigris as a tar.zst archive,
+    /// fenced by `precondition`.
+    pub async fn upload(
+        &self,
+        owner_slug: &str,
+        repo_name: &str,
+        local_path: &Path,
+        precondition: UploadPrecondition,
+    ) -> std::result::Result<(), UploadError> {
         let key = Self::repo_key(owner_slug, repo_name);
         debug!(key = %key, path = %local_path.display(), "uploading repo to tigris");
 
@@ -93,15 +169,57 @@ impl TigrisClient {
 
         let body = aws_sdk_s3::primitives::ByteStream::from(archive_bytes);
 
-        self.s3
+        let mut req = self
+            .s3
             .put_object()
             .bucket(&self.bucket)
             .key(&key)
             .body(body)
-            .content_type("application/zstd")
-            .send()
-            .await
-            .context(format!("tigris PUT {key}"))?;
+            .content_type("application/zstd");
+        match &precondition {
+            UploadPrecondition::IfMatch(etag) => req = req.if_match(etag),
+            UploadPrecondition::IfAbsent => req = req.if_none_match("*"),
+            UploadPrecondition::Unconditional => {}
+        }
+
+        if let Err(e) = req.send().await {
+            // `PutObjectError` models no PreconditionFailed variant (its arms are
+            // EncryptionTypeMismatch, InvalidRequest, InvalidWriteOffset,
+            // TooManyParts, Unhandled), so a refused precondition arrives as
+            // `Unhandled` and matching the enum would classify it as a generic
+            // failure. The raw HTTP status off the response is the only place the
+            // answer actually lives.
+            //
+            // Read it via `raw_response()`, not a `ServiceError`-only match: the
+            // SDK exposes the raw response for BOTH `ServiceError` and
+            // `ResponseError`, and a refused conditional PUT whose error body the
+            // SDK cannot parse (malformed XML, premature close) surfaces as
+            // `ResponseError`. Matching only `ServiceError` would classify that
+            // unparsable 409/412 as a generic failure, and `RepoWriteGuard::release`
+            // would log-and-succeed instead of taking the supersede retry,
+            // acknowledging a write that was definitively not published.
+            let status = e.raw_response().map(|raw| raw.status().as_u16());
+            // 412 is always a lost precondition. 409 is one only when we asked
+            // for create-only, which is how S3-compatible stores report "the key
+            // already exists". Everything else, 404 included, is a real failure:
+            // archive keys are never deleted (`delete` has no callers), so a 404
+            // here means something permanent like a missing bucket or a
+            // misrouted endpoint, and reporting that as a lost precondition
+            // would tell a caller to expect a successor that does not exist.
+            let lost = match status {
+                Some(412) => true,
+                Some(409) => matches!(precondition, UploadPrecondition::IfAbsent),
+                _ => false,
+            };
+            if lost {
+                return Err(UploadError::PreconditionLost {
+                    status: status.expect("a lost precondition came from a status"),
+                });
+            }
+            return Err(UploadError::Other(
+                anyhow::Error::new(e).context(format!("tigris PUT {key}")),
+            ));
+        }
 
         info!(key = %key, "uploaded repo to tigris");
         Ok(())
@@ -112,10 +230,32 @@ impl TigrisClient {
         &self,
         owner_slug: &str,
         repo_name: &str,
-        local_path: &Path,
+        local_path: &super::repo_store::ValidatedRepoDiskPath,
+        swap_authority: Option<Arc<AtomicBool>>,
     ) -> Result<()> {
+        self.download_to(owner_slug, repo_name, local_path, true, swap_authority)
+            .await
+            .map(|_| ())
+    }
+
+    /// Download a repo archive from Tigris and extract it, returning the
+    /// directory that was populated.
+    ///
+    /// `publish` controls whether the extract is swapped into `target` in place
+    /// (the live-path mutation used by writes; returns `target`) or unpacked
+    /// into a fresh temp directory under `target`'s parent (a non-mutating
+    /// snapshot read; returns the temp dir, which the caller owns and cleans
+    /// up). The snapshot form never touches the live repo path.
+    pub async fn download_to(
+        &self,
+        owner_slug: &str,
+        repo_name: &str,
+        target: &super::repo_store::ValidatedRepoDiskPath,
+        publish: bool,
+        swap_authority: Option<Arc<AtomicBool>>,
+    ) -> Result<DownloadExtract> {
         let key = Self::repo_key(owner_slug, repo_name);
-        debug!(key = %key, path = %local_path.display(), "downloading repo from tigris");
+        debug!(key = %key, path = %target.as_path().display(), "downloading repo from tigris");
 
         let resp = self
             .s3
@@ -133,21 +273,62 @@ impl TigrisClient {
             .context("reading tigris response body")?
             .into_bytes();
 
-        // Extract tar.zst to local path
-        tokio::task::spawn_blocking({
-            let local_path = local_path.to_path_buf();
-            move || decompress_repo(&data, &local_path)
+        // The snapshot temp dir is decided HERE, in the async layer, before the
+        // extraction runs. Cleanup ownership moves into the blocking task so a
+        // cancelled async future cannot drop it while extraction is still running.
+        let snapshot_tmp = if publish {
+            None
+        } else {
+            let parent = target.parent().context("snapshot path has no parent")?;
+            std::fs::create_dir_all(parent).context("creating parent dir")?;
+            let file_name = target
+                .file_name()
+                .context("snapshot path has no file name")?
+                .to_string_lossy();
+            Some(parent.join(format!(
+                ".{file_name}.tmp-snapshot.{}",
+                uuid::Uuid::new_v4()
+            )))
+        };
+
+        // Extract tar.zst to a directory.
+        let extracted = tokio::task::spawn_blocking({
+            let target = target.clone();
+            let snapshot_tmp = snapshot_tmp.clone();
+            move || -> Result<DownloadExtract> {
+                let result = (|| -> Result<DownloadExtract> {
+                    if publish {
+                        decompress_repo(&data, &target, swap_authority.as_ref())?;
+                        return Ok(DownloadExtract::Published(()));
+                    }
+                    // Non-mutating snapshot: unpack into the temp dir decided above.
+                    // The live repo path is never touched.
+                    let tmp_dir = snapshot_tmp.expect("snapshot path was decided above");
+                    std::fs::create_dir_all(&tmp_dir).context("creating temp extract dir")?;
+                    let unpack = (|| -> Result<()> {
+                        let decoder = zstd::stream::Decoder::new(&data[..])?;
+                        let mut archive = tar::Archive::new(decoder);
+                        archive.unpack(&tmp_dir).context("unpacking tar.zst")?;
+                        Ok(())
+                    })();
+                    if let Err(e) = unpack {
+                        let _ = std::fs::remove_dir_all(&tmp_dir);
+                        return Err(e);
+                    }
+                    Ok(DownloadExtract::Snapshot(TempSnapshotDir { path: tmp_dir }))
+                })();
+                result
+            }
         })
         .await
         .context("extract task panicked")?
         .context("extracting repo")?;
 
-        info!(key = %key, path = %local_path.display(), "downloaded repo from tigris");
-        Ok(())
+        info!(key = %key, path = %target.as_path().display(), "downloaded repo from tigris");
+        Ok(extracted)
     }
 
     /// Delete a repo archive from Tigris.
-    #[allow(dead_code)]
     pub async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()> {
         let key = Self::repo_key(owner_slug, repo_name);
         self.s3
@@ -180,7 +361,7 @@ fn compress_repo(repo_path: &Path) -> Result<Vec<u8>> {
 /// `decompress_repo`. Concurrent extractions unpack into isolated temp dirs in
 /// parallel, but the final `remove_dir_all` + `rename` must not interleave for
 /// the same `local_path`, or they race to a nondeterministic overwrite/failure.
-fn publish_lock(local_path: &Path) -> Arc<Mutex<()>> {
+pub(crate) fn publish_lock(local_path: &Path) -> Arc<Mutex<()>> {
     // KNOWN LIMITATION: this map is never evicted — one (PathBuf, Arc<Mutex>)
     // entry accrues per distinct repo path for the process lifetime. Bounded by
     // the number of repos a node hosts, so it's negligible for normal use, but
@@ -193,6 +374,35 @@ fn publish_lock(local_path: &Path) -> Arc<Mutex<()>> {
         .clone()
 }
 
+/// Owns a snapshot temp dir from extraction through handoff to [`RepoSnapshot`].
+/// Dropped when the `spawn_blocking` join result is abandoned, so cancellation
+/// before the outer future resumes still removes the directory.
+pub(crate) struct TempSnapshotDir {
+    path: PathBuf,
+}
+
+impl TempSnapshotDir {
+    pub(crate) fn into_repo_snapshot(self) -> super::repo_store::RepoSnapshot {
+        let path = self.path.clone();
+        std::mem::forget(self);
+        super::repo_store::RepoSnapshot::from_owned_path(path)
+    }
+}
+
+impl Drop for TempSnapshotDir {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_dir_all(&self.path);
+    }
+}
+
+/// What [`TigrisClient::download_to`] produced.
+pub enum DownloadExtract {
+    /// Published into the validated live repo path.
+    Published(()),
+    /// Unpacked into a throwaway temp dir that cleans up on drop until adopted.
+    Snapshot(TempSnapshotDir),
+}
+
 /// Decompress a tar.zst byte vector into a local directory.
 ///
 /// Extraction is atomic with respect to `local_path`: the archive is unpacked
@@ -200,11 +410,16 @@ fn publish_lock(local_path: &Path) -> Arc<Mutex<()>> {
 /// fully succeeds. A corrupt or truncated archive therefore can never clobber a
 /// good existing copy at `local_path` — on failure we discard the temp dir and
 /// leave `local_path` exactly as it was.
-fn decompress_repo(data: &[u8], local_path: &Path) -> Result<()> {
-    let parent = local_path.parent().context("repo path has no parent")?;
+fn decompress_repo(
+    data: &[u8],
+    local_path: &super::repo_store::ValidatedRepoDiskPath,
+    swap_authority: Option<&Arc<AtomicBool>>,
+) -> Result<()> {
+    let live = local_path.as_path();
+    let parent = live.parent().context("repo path has no parent")?;
     std::fs::create_dir_all(parent).context("creating parent dir")?;
 
-    let file_name = local_path
+    let file_name = live
         .file_name()
         .context("repo path has no file name")?
         .to_string_lossy();
@@ -229,17 +444,218 @@ fn decompress_repo(data: &[u8], local_path: &Path) -> Result<()> {
         return Err(e);
     }
 
-    // Swap the freshly-extracted repo into place. rename within the same parent
-    // is effectively atomic, but most platforms refuse to rename onto a
-    // non-empty dir, so remove the old copy first. Serialize this per repo path:
-    // concurrent extractions unpack into isolated temp dirs, but their swaps
-    // must not interleave or they race to a nondeterministic overwrite/failure.
-    let lock = publish_lock(local_path);
-    let _publish = lock.lock().expect("publish lock poisoned");
-    if local_path.exists() {
-        std::fs::remove_dir_all(local_path).context("removing stale repo dir")?;
-    }
-    std::fs::rename(&tmp_dir, local_path).context("swapping extracted repo into place")?;
+    // Swap through the validated-path helper so CodeQL sees the barrier before the
+    // remove/rename sink (`rust/path-injection`).
+    super::repo_store::swap_extracted_into_validated_repo(local_path, &tmp_dir, swap_authority)?;
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aws_sdk_s3::primitives::ByteStream;
+    use futures::FutureExt;
+
+    /// The envs the probe needs, all of them, or it does not run.
+    ///
+    /// `AWS_ENDPOINT_URL_S3` is included on purpose: without it the SDK resolves
+    /// to real AWS S3, and a probe that passed there would say nothing about
+    /// Tigris.
+    fn probe_env() -> Option<String> {
+        if std::env::var("GITLAWB_TIGRIS_PROBE").ok().as_deref() != Some("1") {
+            return None;
+        }
+        for name in [
+            "AWS_ACCESS_KEY_ID",
+            "AWS_SECRET_ACCESS_KEY",
+            "AWS_ENDPOINT_URL_S3",
+        ] {
+            if std::env::var(name).is_err() {
+                eprintln!("tigris conditional-write probe: {name} is unset, skipping");
+                return None;
+            }
+        }
+        match std::env::var("GITLAWB_TIGRIS_BUCKET") {
+            Ok(b) if !b.is_empty() => Some(b),
+            _ => {
+                eprintln!(
+                    "tigris conditional-write probe: GITLAWB_TIGRIS_BUCKET is unset, skipping"
+                );
+                None
+            }
+        }
+    }
+
+    /// One conditional PUT, reported as the status that REFUSED it, or `None`
+    /// when the store accepted the write.
+    ///
+    /// Accepted is the interesting answer here, not an error: it means the
+    /// endpoint ignored the header we fenced on.
+    async fn conditional_put(
+        s3: &S3Client,
+        bucket: &str,
+        key: &str,
+        body: &'static [u8],
+        if_match: Option<&str>,
+        if_none_match: Option<&str>,
+    ) -> Result<Option<u16>, String> {
+        let mut req = s3
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(body));
+        if let Some(v) = if_match {
+            req = req.if_match(v);
+        }
+        if let Some(v) = if_none_match {
+            req = req.if_none_match(v);
+        }
+        match req.send().await {
+            Ok(_) => Ok(None),
+            Err(e) => match e.raw_response() {
+                // Same raw-response rule as `upload`: a refused conditional PUT
+                // whose body the SDK cannot parse surfaces as `ResponseError`, and
+                // the status has to come off the raw response for both variants.
+                // Without it, an unparsable 409/412 would report "no HTTP
+                // response" here and skip the supersede retry.
+                Some(raw) => Ok(Some(raw.status().as_u16())),
+                None => Err(format!("conditional PUT {key}: no HTTP response: {e}")),
+            },
+        }
+    }
+
+    /// The probe body, written to RETURN its failures rather than panic on
+    /// them, so the caller's cleanup is reached on every arm.
+    async fn conditional_write_probe(s3: &S3Client, bucket: &str, key: &str) -> Result<(), String> {
+        // 1. A plain PUT under a throwaway key.
+        let seeded = s3
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(b"probe-one"))
+            .send()
+            .await
+            .map_err(|e| format!("seeding PUT {key}: {e}"))?;
+
+        // 2. Its ETag, which is the generation the next arm fences against.
+        let etag = seeded
+            .e_tag()
+            .ok_or_else(|| format!("seeding PUT {key} returned no ETag"))?
+            .to_string();
+
+        // 3. A deliberately wrong If-Match. A store honoring it answers 412.
+        let wrong = format!("\"{}\"", "0".repeat(32));
+        if etag.trim_matches('"') == wrong.trim_matches('"') {
+            return Err(format!(
+                "the seeded ETag {etag} collides with the deliberately wrong one, \
+                 so this arm would prove nothing"
+            ));
+        }
+        match conditional_put(s3, bucket, key, b"probe-two", Some(&wrong), None).await? {
+            Some(412) => {}
+            Some(status) => {
+                return Err(format!(
+                    "a stale If-Match must be refused with 412, the endpoint answered {status}"
+                ))
+            }
+            None => {
+                return Err(
+                    "a stale If-Match was ACCEPTED: this endpoint does not honor If-Match, so \
+                     the release fence cannot hold here"
+                        .to_string(),
+                )
+            }
+        }
+
+        // 4. If-None-Match `*` over the object that now exists. This arm matters
+        // MORE than the one above. An ignored If-Match eventually surfaces as
+        // odd behavior, because a stale writer overwrites and someone notices
+        // the lost tree. An ignored If-None-Match just returns 200, so a publish
+        // that should have been fenced lands with no error anywhere: the silent
+        // no-op the bucket-type caveat on this test describes.
+        match conditional_put(s3, bucket, key, b"probe-three", None, Some("*")).await? {
+            // Either status is a pass, and the asymmetry with the If-Match arm
+            // above mirrors `upload`'s classifier exactly: 412 is always a lost
+            // precondition, and 409 is one too when we asked for create-only.
+            // AWS documents 409 for a create-only conflict racing a delete, so a
+            // store answering it is enforcing the precondition and we already
+            // handle it. Pinning 412 alone here would fail the probe against a
+            // backend that is behaving correctly, which sends whoever runs it
+            // chasing a fault that is not there.
+            Some(412) | Some(409) => {}
+            Some(status) => {
+                return Err(format!(
+                    "create-only over an existing object must be refused with 412 or 409, \
+                     the endpoint answered {status}"
+                ))
+            }
+            None => {
+                return Err(
+                    "If-None-Match * was ACCEPTED over an existing object: this endpoint does \
+                     not honor create-only, so a fenced publish lands silently"
+                        .to_string(),
+                )
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Probe the REAL Tigris endpoint for the conditional-write semantics the
+    /// release fence depends on.
+    ///
+    /// UNTIL THIS IS RUN AGAINST REAL CREDENTIALS, the fence is verified against
+    /// vendor documentation and an in-process mock, not against the backend it
+    /// runs on. The mock implements the semantics we believe Tigris has; it
+    /// cannot tell us whether Tigris actually has them.
+    ///
+    /// The bucket matters, not just the endpoint. Tigris documents conditional
+    /// operations as supported on Single-region and Multi-region buckets only.
+    /// Global and Dual-region buckets are eventually consistent, and a
+    /// conditional PUT evaluated against a stale replica would make the fence a
+    /// silent no-op rather than an error. So point `GITLAWB_TIGRIS_BUCKET` at a
+    /// throwaway bucket of the SAME type production uses.
+    ///
+    /// Ignored by default and additionally gated on `GITLAWB_TIGRIS_PROBE=1`,
+    /// because it writes to a real bucket and costs real requests. Run with:
+    /// `GITLAWB_TIGRIS_PROBE=1 cargo test -p gitlawb-node --bin gitlawb-node
+    /// tigris_honors_conditional_writes -- --ignored --nocapture`
+    #[tokio::test]
+    #[ignore = "writes to a real Tigris bucket; needs GITLAWB_TIGRIS_PROBE=1 plus credentials"]
+    async fn tigris_honors_conditional_writes() {
+        let Some(bucket) = probe_env() else {
+            eprintln!(
+                "tigris conditional-write probe: skipped. Set GITLAWB_TIGRIS_PROBE=1, \
+                 GITLAWB_TIGRIS_BUCKET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and \
+                 AWS_ENDPOINT_URL_S3 to run it."
+            );
+            return;
+        };
+
+        let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
+        let s3 = S3Client::new(&config);
+        // A fresh key per run, so a probe that somehow orphaned an object on an
+        // earlier run cannot change what this one observes.
+        let key = format!("probe/conditional-write-{}.bin", uuid::Uuid::new_v4());
+
+        // CLEANUP MUST RUN ON EVERY ARM, and a failing assertion is precisely
+        // the case this probe exists to catch, so the delete cannot sit after
+        // the checks. The body returns its failures rather than panicking, and
+        // `catch_unwind` covers the panic an SDK call could still raise; either
+        // way the delete below is reached before the verdict is re-raised.
+        let outcome = std::panic::AssertUnwindSafe(conditional_write_probe(&s3, &bucket, &key))
+            .catch_unwind()
+            .await;
+
+        if let Err(e) = s3.delete_object().bucket(&bucket).key(&key).send().await {
+            eprintln!("tigris conditional-write probe: cleanup of {key} failed: {e}");
+        }
+
+        match outcome {
+            Ok(Ok(())) => {}
+            Ok(Err(msg)) => panic!("tigris conditional-write probe: {msg}"),
+            Err(payload) => std::panic::resume_unwind(payload),
+        }
+    }
+}

crates/gitlawb-node/src/ipfs_pin.rs

@@ -676,7 +676,7 @@ async fn warm_candidates(
                 &repo.owner_did,
                 &repo.name,
             ) {
-                Ok(p) if p.is_dir() => out.push((repo, created_at_key, p)),
+                Ok(p) if p.is_dir() => out.push((repo, created_at_key, p.into_path_buf())),
                 Ok(_) => {}
                 Err(e) => {
                     tracing::warn!(repo_id = %repo.id, err = %e, "sweep discovery: rejected unsafe repo path");

crates/gitlawb-node/src/main.rs

@@ -59,32 +59,6 @@ struct DbStartupStatus {
     next_retry_secs: AtomicU64,
 }
 
-/// Hard ceiling on the advisory-lock pool's `max_connections`.
-///
-/// `max_concurrent_git_pushes` is validated all the way up to 1_048_576, and the lock
-/// pool used to derive its size straight from that knob, so raising the push cap
-/// silently raised the node's Postgres connection ceiling with no CLI error and no
-/// relation to the server's own `max_connections` (#173 F4). The node's total budget is
-/// now bounded: `db_max_connections` (default 48) + at most this.
-const LOCK_POOL_MAX_CONNECTIONS: u32 = 64;
-
-/// Connections the lock pool keeps above the push cap. Covers the three non-push
-/// `acquire_write` callers (`api/issues.rs` x2, `api/pulls.rs`), which hold no
-/// concurrency permit, so a push never queues here for a connection where it did not
-/// before.
-const LOCK_POOL_PUSH_HEADROOM: u8 = 8;
-
-/// Size the advisory-lock pool for a given push cap: the cap plus
-/// [`LOCK_POOL_PUSH_HEADROOM`], clamped to [`LOCK_POOL_MAX_CONNECTIONS`]. Past the
-/// clamp a push may wait for a lock-pool connection, which is a bounded wait that sheds
-/// a clean 503 (see `LockPoolBusy`), not an unbounded hang.
-fn lock_pool_size(max_concurrent_git_pushes: usize) -> u32 {
-    u32::try_from(max_concurrent_git_pushes)
-        .unwrap_or(u32::MAX)
-        .saturating_add(u32::from(LOCK_POOL_PUSH_HEADROOM))
-        .min(LOCK_POOL_MAX_CONNECTIONS)
-}
-
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt()
@@ -116,8 +90,6 @@ async fn main() -> Result<()> {
 
     // Load or generate the node's identity keypair
     let keypair = load_or_create_keypair(&config)?;
-    // Sealing key for the legacy-scan continuation tokens, DERIVED from the identity
-    // just loaded so it is the same key after a restart (see `derive_scan_token_key`).
     let scan_token_key = AppState::derive_scan_token_key(&keypair);
     let node_did = keypair.did();
 
@@ -316,20 +288,20 @@ async fn main() -> Result<()> {
         None
     };
 
-    // Repo write locks run on their own pool, never the main query pool: each push
-    // holds its connection for the whole receive-pack, so a burst of concurrent
-    // pushes drawing from the main pool would park that many connections for the
-    // duration of their receive-packs and starve every other query. That holds
-    // whatever the two pools are sized at, which is why the separation is
-    // structural rather than a consequence of the defaults; config validate()
-    // separately requires db_max_connections >= max_concurrent_git_pushes + 8. See
-    // build_lock_pool for the cancellation semantics (#173).
-    let lock_pool = git::repo_store::build_lock_pool(
-        db.pool(),
-        lock_pool_size(config.max_concurrent_git_pushes),
+    // Advisory-lock connections come from their own pool: a write guard pins one
+    // for its whole lifetime, and sharing the application pool would let a push
+    // burst starve ordinary request handlers.
+    let lock_pool = db::Db::lock_pool(
+        &config.database_url,
+        config.db_lock_pool_max_connections,
         std::time::Duration::from_secs(config.db_acquire_timeout_secs),
+    )?;
+    let repo_store = git::repo_store::RepoStore::new(
+        config.repos_dir.clone(),
+        tigris,
+        lock_pool,
+        std::time::Duration::from_secs(config.lock_held_transfer_timeout_secs),
     );
-    let repo_store = git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, lock_pool);
 
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
     // DID (attacker-varied), so bound its key set to cap memory.
@@ -374,6 +346,23 @@ async fn main() -> Result<()> {
         tracing::warn!("GITLAWB_PUSH_RATE_LIMIT=0 — per-IP push rate limiting disabled");
     }
 
+    // close_issue drives a full archive download for non-owner author pre-checks.
+    // Keep it on its own bucket so it cannot drain the receive-pack limit.
+    let close_issue_limit = std::env::var("GITLAWB_CLOSE_ISSUE_RATE_LIMIT")
+        .ok()
+        .and_then(|v| v.trim().parse::<usize>().ok())
+        .unwrap_or(120);
+    let close_issue_rate_limiter = rate_limit::RateLimiter::new_bounded(
+        close_issue_limit,
+        std::time::Duration::from_secs(3600),
+        200_000,
+    );
+    if close_issue_limit == 0 {
+        tracing::warn!(
+            "GITLAWB_CLOSE_ISSUE_RATE_LIMIT=0 — per-IP close_issue rate limiting disabled"
+        );
+    }
+
     // Which forwarded header the edge is trusted to set. Default None (trust
     // nothing, key on the socket peer). Fly nodes set GITLAWB_TRUSTED_PROXY=fly;
     // a node behind Caddy/NGINX sets it to x-forwarded-for.
@@ -423,18 +412,10 @@ async fn main() -> Result<()> {
         rate_limiter,
         create_ip_rate_limiter,
         push_rate_limiter,
+        close_issue_rate_limiter,
         ipfs_max_history_walks: crate::api::ipfs::MAX_HISTORY_WALKS_PER_REQUEST,
-        // The legacy-probe budget is operator-tunable via GITLAWB_IPFS_MAX_LEGACY_PROBES
-        // (R5), which is what `ipfs_legacy_probe_budget` reads. NOT
-        // GITLAWB_IPFS_MAX_REPOS_WALKED, which this comment used to name: that is the
-        // separate cap on expensive visibility walks, so an operator following the old
-        // text tuned the walk cap and left this fan-out unchanged. The history-walk
-        // ceiling above stays constant (a smaller value false-503s a provenanced
-        // request). Default 256 preserves the shipped behaviour.
         ipfs_max_legacy_probes: AppState::ipfs_legacy_probe_budget(&config),
         ipfs_legacy_scan_page_rows: crate::api::ipfs::LEGACY_SCAN_PAGE_ROWS,
-        // Operator-tunable via GITLAWB_IPFS_MAX_LEGACY_SCAN_ROWS, read through the same
-        // helper shape as the probe budget so the knob cannot be a silent no-op.
         ipfs_max_legacy_scan_rows: AppState::ipfs_legacy_scan_row_budget(&config),
         ipfs_max_legacy_scan_rule_bytes: crate::api::ipfs::MAX_LEGACY_SCAN_RULE_BYTES_PER_REQUEST,
         ipfs_scan_token_key: Arc::new(scan_token_key),
@@ -513,10 +494,6 @@ async fn main() -> Result<()> {
             std::time::Duration::from_secs(3600),
             200_000,
         ),
-        // Separate WORK-budget bucket for the resolver's per-probe/per-walk charges (R6).
-        // Its capacity is DERIVED from the route limit (no new knob) and floored at the
-        // legacy-probe budget, so one full default-config legacy scan never self-throttles
-        // mid-request while the route brake above stays the pure once-per-request cap.
         ipfs_work_rate_limiter: rate_limit::RateLimiter::new_bounded(
             AppState::ipfs_work_budget(&config),
             std::time::Duration::from_secs(3600),
@@ -556,16 +533,14 @@ async fn main() -> Result<()> {
 
     // Periodic cleanup of expired rate limit entries + consumed-proof ledger
     {
-        let cleanup_state = state.clone();
+        let sweep_state = state.clone();
         let db = state.db.clone();
         let mut shutdown_rx = state.subscribe_shutdown();
         tokio::spawn(async move {
             loop {
                 tokio::select! {
                     _ = tokio::time::sleep(std::time::Duration::from_secs(300)) => {
-                        // Sweep every per-IP/DID limiter (incl. the ipfs walk brake)
-                        // so bounded maps shed stale keys instead of sitting at cap.
-                        cleanup_state.sweep_rate_limiters().await;
+                        sweep_state.sweep_rate_limiters().await;
                         let now = std::time::SystemTime::now()
                             .duration_since(std::time::UNIX_EPOCH)
                             .map(|d| d.as_secs() as i64)
@@ -708,44 +683,6 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-/// U4 (#173): spawn the periodic legacy provider-CID repair sweep. Releases before this
-/// version stored the PROVIDER CID (Kubo dag-pb / Pinata CIDv0) in `pinned_cids.cid`,
-/// and this version's `/ipfs/{cid}` resolver withholds any row whose stored key is not
-/// the raw-content CID. The opportunistic repair on the pin path only fires when a push
-/// re-carries the object, which normal git negotiation makes it not do, so those rows
-/// need a walk. DETACHED, never on the boot path: the caller keeps serving while this
-/// runs, and the sweep's own batch bound plus inter-batch delay keep it off the DB's
-/// critical path. Its cursor is durable, so a restart mid-walk resumes instead of
-/// rewinding. It re-arms after every run, including a failed one, so it never returns
-/// and there is no awaited value to log here; the shutdown watcher below is what ends
-/// it.
-///
-/// A named function rather than an inline block in `main` so the WIRING has a seam a
-/// test can call: that the task is spawned at all, that it reads its batch and delay
-/// from the config knobs rather than some other field, that the caller is not blocked
-/// on it, and that the shutdown watcher actually ends it mid-walk. The sweep's own
-/// behavior is covered elsewhere; this is the boot-path half.
-fn spawn_legacy_cid_sweep(state: &AppState, config: &Config) -> tokio::task::JoinHandle<()> {
-    let db = state.db.clone();
-    let repos_dir = config.repos_dir.clone();
-    let git_bin = state.git_bin.clone();
-    let git_timeout = std::time::Duration::from_secs(config.git_service_timeout_secs);
-    let batch = config.pin_repair_sweep_batch;
-    let delay = std::time::Duration::from_secs(config.pin_repair_sweep_delay_secs);
-    let mut shutdown_rx = state.subscribe_shutdown();
-    tokio::spawn(async move {
-        tokio::select! {
-            _ = ipfs_pin::run_sweep_rearmed(
-                &repos_dir, &git_bin, git_timeout, batch, delay,
-                ipfs_pin::SWEEP_REARM_DELAY, &db,
-            ) => {}
-            // Shutdown mid-walk simply drops the run; the persisted cursor means the
-            // next boot picks up where this one stopped.
-            _ = shutdown_rx.changed() => {}
-        }
-    })
-}
-
 fn spawn_shutdown_signal(tx: watch::Sender<bool>) {
     tokio::spawn(async move {
         #[cfg(unix)]
@@ -1187,20 +1124,20 @@ mod rate_limiter_sweep_tests {
         state.rate_limiter = RateLimiter::new(10, window);
         state.create_ip_rate_limiter = RateLimiter::new(10, window);
         state.push_rate_limiter = RateLimiter::new(10, window);
+        state.close_issue_rate_limiter = RateLimiter::new(10, window);
         state.sync_trigger_rate_limiter = RateLimiter::new(10, window);
         state.peer_write_rate_limiter = RateLimiter::new(10, window);
         state.ipfs_rate_limiter = RateLimiter::new(10, window);
-        state.ipfs_work_rate_limiter = RateLimiter::new(10, window);
 
         let limiters = |s: &crate::state::AppState| {
             [
                 s.rate_limiter.clone(),
                 s.create_ip_rate_limiter.clone(),
                 s.push_rate_limiter.clone(),
+                s.close_issue_rate_limiter.clone(),
                 s.sync_trigger_rate_limiter.clone(),
                 s.peer_write_rate_limiter.clone(),
                 s.ipfs_rate_limiter.clone(),
-                s.ipfs_work_rate_limiter.clone(),
             ]
         };
         for l in limiters(&state) {
@@ -1217,6 +1154,26 @@ mod rate_limiter_sweep_tests {
     }
 }
 
+/// it.
+fn spawn_legacy_cid_sweep(state: &AppState, config: &Config) -> tokio::task::JoinHandle<()> {
+    let db = state.db.clone();
+    let repos_dir = config.repos_dir.clone();
+    let git_bin = state.git_bin.clone();
+    let git_timeout = std::time::Duration::from_secs(config.git_service_timeout_secs);
+    let batch = config.pin_repair_sweep_batch;
+    let delay = std::time::Duration::from_secs(config.pin_repair_sweep_delay_secs);
+    let mut shutdown_rx = state.subscribe_shutdown();
+    tokio::spawn(async move {
+        tokio::select! {
+            _ = ipfs_pin::run_sweep_rearmed(
+                &repos_dir, &git_bin, git_timeout, batch, delay,
+                ipfs_pin::SWEEP_REARM_DELAY, &db,
+            ) => {}
+            _ = shutdown_rx.changed() => {}
+        }
+    })
+}
+
 async fn gossip_ping_round(
     db: &Db,
     client: &reqwest::Client,
@@ -1393,140 +1350,6 @@ fn load_or_create_keypair(config: &Config) -> Result<Keypair> {
     }
 }
 
-#[cfg(test)]
-mod legacy_cid_sweep_wiring_tests {
-    use super::spawn_legacy_cid_sweep;
-    use sqlx::PgPool;
-    use std::time::Duration;
-
-    /// Seed `count` `pinned_cids` rows whose keys are already canonical raw CIDv1, in a
-    /// known `sha256_hex` order. The sweep's own cost gate skips a raw-CIDv1 row without
-    /// reading bytes or resolving a repo, so each row is SCANNED (it advances the cursor)
-    /// and nothing else. That is what makes the cursor a clean readout of how far the
-    /// walk got, with no dependency on repos on disk.
-    async fn seed_scannable_rows(pool: &PgPool, count: usize) -> Vec<String> {
-        let mut shas = Vec::new();
-        for i in 1..=count {
-            let sha = format!("wire{i:02}");
-            let cid = gitlawb_core::cid::Cid::from_git_object_bytes(sha.as_bytes()).to_string();
-            assert!(
-                gitlawb_core::cid::is_raw_cidv1(&cid),
-                "the seeded key must hit the sweep's raw-CIDv1 skip, not a repair attempt"
-            );
-            sqlx::query("INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) VALUES ($1, $2, $3)")
-                .bind(&sha)
-                .bind(&cid)
-                .bind("2020-01-01T00:00:00Z")
-                .execute(pool)
-                .await
-                .unwrap();
-            shas.push(sha);
-        }
-        shas
-    }
-
-    /// Poll the persisted sweep cursor until it reaches `want`, or give up.
-    async fn cursor_reaches(db: &crate::db::Db, want: &str, within: Duration) -> String {
-        let deadline = std::time::Instant::now() + within;
-        loop {
-            let c = db.pin_repair_cursor().await.unwrap();
-            if c == want || std::time::Instant::now() >= deadline {
-                return c;
-            }
-            tokio::time::sleep(Duration::from_millis(20)).await;
-        }
-    }
-
-    /// #173 U4, the BOOT-PATH half. The sweep's own logic (batching, cursor resumption,
-    /// terminal vs retryable skips) is covered in `test_support`; what this covers is the
-    /// wiring `main` performs, which nothing else executes: the task is spawned at all,
-    /// it takes its batch and delay from the two `pin_repair_sweep_*` knobs rather than
-    /// some other config field, the caller is not blocked on the walk, and the shutdown
-    /// watcher ends the run mid-walk.
-    ///
-    /// Six scannable rows, batch 2, delay 30s. One pass must land the cursor on exactly
-    /// the second row and the task must then still be alive in its inter-batch sleep,
-    /// which pins both knobs at once: a different batch stops at a different row, and a
-    /// delay that did not come from the knob either finishes the table or leaves the task
-    /// gone. Shutdown must then end it while four rows are still unwalked.
-    #[sqlx::test]
-    async fn the_boot_path_spawns_the_sweep_detached_with_its_configured_knobs(pool: PgPool) {
-        let state = crate::test_support::test_state(pool.clone()).await;
-        let shas = seed_scannable_rows(&pool, 6).await;
-        let repos_dir = tempfile::TempDir::new().unwrap();
-
-        let mut config = (*state.config).clone();
-        config.repos_dir = repos_dir.path().to_path_buf();
-        config.pin_repair_sweep_batch = 2;
-        // Far longer than this test runs, so a task still alive after the first pass can
-        // only be one that is honoring the configured inter-batch delay.
-        config.pin_repair_sweep_delay_secs = 30;
-
-        let started = std::time::Instant::now();
-        let handle = spawn_legacy_cid_sweep(&state, &config);
-        let spawn_cost = started.elapsed();
-
-        let cursor = cursor_reaches(&state.db, &shas[1], Duration::from_secs(10)).await;
-        assert_eq!(
-            cursor, shas[1],
-            "the spawned sweep must run and stop its first pass at the CONFIGURED batch \
-             bound (2), leaving the cursor on the second row"
-        );
-        assert!(
-            spawn_cost < Duration::from_secs(1),
-            "the sweep must be detached, not awaited on the boot path; the spawn took \
-             {spawn_cost:?}"
-        );
-        assert!(
-            !handle.is_finished(),
-            "with a 30s inter-batch delay the task must still be sleeping between passes, \
-             not finished: a finished task means the delay was not the configured one"
-        );
-
-        state.shutdown();
-        tokio::time::timeout(Duration::from_secs(10), handle)
-            .await
-            .expect("the shutdown watcher must end the sweep, and not after its 30s delay")
-            .expect("the sweep task must not panic");
-
-        assert_eq!(
-            state.db.pin_repair_cursor().await.unwrap(),
-            shas[1],
-            "shutdown must have ended the run MID-walk, with the remaining rows unwalked"
-        );
-    }
-}
-
-#[cfg(test)]
-mod lock_pool_sizing_tests {
-    use super::{lock_pool_size, LOCK_POOL_MAX_CONNECTIONS, LOCK_POOL_PUSH_HEADROOM};
-
-    /// The default push cap gets its cap plus headroom, so no push ever queues for a
-    /// lock-pool connection where it did not before.
-    #[test]
-    fn default_push_cap_gets_headroom_over_the_cap() {
-        assert_eq!(lock_pool_size(32), 32 + u32::from(LOCK_POOL_PUSH_HEADROOM));
-        assert_eq!(lock_pool_size(1), 1 + u32::from(LOCK_POOL_PUSH_HEADROOM));
-    }
-
-    /// #173 F4: `max_concurrent_git_pushes` is validated all the way to 1_048_576, so an
-    /// operator raising it used to raise the node's Postgres connection ceiling with it,
-    /// silently and without bound. The lock pool is CLAMPED instead.
-    #[test]
-    fn an_oversized_push_cap_is_clamped_not_propagated() {
-        assert_eq!(lock_pool_size(1_048_576), LOCK_POOL_MAX_CONNECTIONS);
-        assert_eq!(lock_pool_size(usize::MAX), LOCK_POOL_MAX_CONNECTIONS);
-        // The largest cap that still fits under the clamp keeps its full headroom.
-        let widest = (LOCK_POOL_MAX_CONNECTIONS - u32::from(LOCK_POOL_PUSH_HEADROOM)) as usize;
-        assert_eq!(lock_pool_size(widest), LOCK_POOL_MAX_CONNECTIONS);
-        assert_eq!(
-            lock_pool_size(widest - 1),
-            LOCK_POOL_MAX_CONNECTIONS - 1,
-            "values below the clamp must not be rounded up to it"
-        );
-    }
-}
-
 #[cfg(test)]
 mod gossip_ssrf_tests {
     use super::{

crates/gitlawb-node/src/state.rs

@@ -79,6 +79,10 @@ pub struct AppState {
     /// brake a push flood from a DID farm (one throwaway DID per repo), so the
     /// push path throttles on the resolved client IP instead.
     pub push_rate_limiter: RateLimiter,
+    /// Per-client-IP rate limiter for `close_issue`'s pre-lock snapshot path.
+    /// Distinct from `push_rate_limiter` so a flood of close attempts cannot
+    /// drain the receive-pack budget for the same source IP.
+    pub close_issue_rate_limiter: RateLimiter,
     /// Per-client-IP ROUTE brake for `GET /ipfs/{cid}`: charged ONCE per request by the
     /// `rate_limit_by_ip` middleware (server.rs), never inside the handler. It bounds
     /// request RATE (the "requests per hour" contract of `GITLAWB_IPFS_RATE_LIMIT`) on
@@ -342,6 +346,7 @@ impl AppState {
         self.rate_limiter.cleanup().await;
         self.create_ip_rate_limiter.cleanup().await;
         self.push_rate_limiter.cleanup().await;
+        self.close_issue_rate_limiter.cleanup().await;
         self.ipfs_rate_limiter.cleanup().await;
         self.ipfs_work_rate_limiter.cleanup().await;
         self.sync_trigger_rate_limiter.cleanup().await;

crates/gitlawb-node/src/test_support.rs

@@ -103,6 +103,7 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
         create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
         push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
+        close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
         ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
         ipfs_work_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
         ipfs_max_history_walks: crate::api::ipfs::MAX_HISTORY_WALKS_PER_REQUEST,
@@ -557,6 +558,35 @@ mod tests {
         );
     }
 
+    /// Fork disk paths must go through `validated_repo_disk_path` so a user-supplied
+    /// name cannot reach `remove_dir_all` on an escaped path (CodeQL path-injection).
+    #[sqlx::test]
+    async fn fork_rejects_name_that_fails_validated_disk_path(pool: PgPool) {
+        let owner = "did:key:zFORKOWNERAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let state = test_state(pool).await;
+        let repo = seed_repo(owner, "fork-src");
+        state.db.create_repo(&repo).await.expect("seed repo");
+
+        let router = Router::new()
+            .route(
+                "/api/v1/repos/{owner}/{repo}/fork",
+                axum::routing::post(crate::api::repos::fork_repo),
+            )
+            .with_state(state.clone());
+        let too_long = "a".repeat(101);
+        let uri = format!("/api/v1/repos/{owner}/fork-src/fork");
+        let body = Body::from(format!(r#"{{"name":"{too_long}"}}"#));
+        let resp = router
+            .oneshot(signed_request_as(owner, Method::POST, &uri, body))
+            .await
+            .unwrap();
+        assert_eq!(
+            resp.status(),
+            StatusCode::BAD_REQUEST,
+            "fork names that fail validated_repo_disk_path must be rejected before clone"
+        );
+    }
+
     /// N13: the task handlers bind the acting DID to the signer. A caller signed
     /// as B claiming delegator_did A is rejected before any DB write (DB-free).
     #[sqlx::test]
@@ -15936,65 +15966,6 @@ mod tests {
             use super::*;
             use crate::api::repos::drain_faults;
 
-            /// Process-wide tracing capture so a test can assert the give-up is logged at
-            /// ERROR. A global default subscriber can only be installed once per process,
-            /// so it is shared by every test here and assertions filter on the repo id,
-            /// which is a fresh uuid per test.
-            mod logcap {
-                use std::sync::{Arc, Mutex, OnceLock};
-                use tracing::{Event, Level, Subscriber};
-                use tracing_subscriber::layer::{Context, Layer};
-                use tracing_subscriber::prelude::*;
-
-                type Lines = Arc<Mutex<Vec<(Level, String)>>>;
-
-                fn lines() -> &'static Lines {
-                    static LINES: OnceLock<Lines> = OnceLock::new();
-                    LINES.get_or_init(|| Arc::new(Mutex::new(Vec::new())))
-                }
-
-                struct Capture;
-                impl<S: Subscriber> Layer<S> for Capture {
-                    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
-                        struct V(String);
-                        impl tracing::field::Visit for V {
-                            fn record_debug(
-                                &mut self,
-                                field: &tracing::field::Field,
-                                value: &dyn std::fmt::Debug,
-                            ) {
-                                self.0.push_str(&format!(" {}={:?}", field.name(), value));
-                            }
-                        }
-                        let mut v = V(String::new());
-                        event.record(&mut v);
-                        lines()
-                            .lock()
-                            .unwrap()
-                            .push((*event.metadata().level(), v.0));
-                    }
-                }
-
-                pub(super) fn install() {
-                    static ONCE: OnceLock<()> = OnceLock::new();
-                    ONCE.get_or_init(|| {
-                        let _ = tracing::subscriber::set_global_default(
-                            tracing_subscriber::registry().with(Capture),
-                        );
-                    });
-                }
-
-                pub(super) fn errors_containing(needle: &str) -> Vec<String> {
-                    lines()
-                        .lock()
-                        .unwrap()
-                        .iter()
-                        .filter(|(lvl, msg)| *lvl == Level::ERROR && msg.contains(needle))
-                        .map(|(_, msg)| msg.clone())
-                        .collect()
-                }
-            }
-
             /// SCENARIO 1. The repo re-read fails once, then succeeds: the drain lap
             /// must still RUN, under the refreshed state, and pin the coalesced push's
             /// object. RED before the fix (the single `Err` returned `None`, the lap
@@ -16060,11 +16031,11 @@ mod tests {
             }
 
             /// SCENARIO 2. Every re-read attempt fails: the loop must give up on a BOUND
-            /// (asserted as a literal, so raising or removing the bound goes RED) and log
-            /// the give-up at ERROR so the residual loss is observable rather than silent.
+            /// (asserted as a literal, so raising or removing the bound goes RED) and hit
+            /// the give-up path that logs at ERROR in production (asserted here via the
+            /// drain_faults seam, which fires on the same branch as that log).
             #[sqlx::test]
             async fn u2_sustained_repo_reread_failure_is_bounded_and_logged(pool: PgPool) {
-                logcap::install();
                 let state = test_state(pool).await;
                 let owner = new_did();
                 let repo = seed_repo(&owner, "u2-bounded");
@@ -16115,11 +16086,10 @@ mod tests {
                     !state.db.is_pinned(&obj2).await.unwrap(),
                     "with the read never succeeding there is nothing fresh to act on"
                 );
-                let errs = logcap::errors_containing(&repo.id);
                 assert!(
-                    !errs.is_empty(),
-                    "the exhausted drain re-read is logged at ERROR with the repo id, so \
-                     the residual work loss is observable; captured: {errs:?}"
+                    drain_faults::reread_exhausted(&repo.id),
+                    "the exhausted drain re-read must hit the give-up path that logs at \
+                     ERROR in production"
                 );
                 assert!(
                     state.encrypt_inflight.is_empty(),

crates/gitlawb-node/tests/inv22_gates.rs

@@ -225,7 +225,8 @@ fn f4_release_keeps_conn_owned_until_unlock_resolves() {
     let repo_store = src("git/repo_store.rs");
 
     let rel_start = repo_store
-        .find("pub async fn release(mut self")
+        .find("pub async fn release(self")
+        .or_else(|| repo_store.find("pub async fn release(mut self"))
         .expect("F4 gate: repo_store.rs no longer defines RepoWriteGuard::release");
     let rel_end = repo_store[rel_start..]
         .find("impl Drop for RepoWriteGuard")