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/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/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,117 +61,25 @@ pub fn create_issue(repo_path: &Path, issue_id: &str, json: &str) -> Result<()>
     Ok(())
 }
 
-/// Issue ids are UUIDs; refs under `refs/gitlawb/issues/` only ever carry
-/// `[0-9a-zA-Z_-]` names. Anything else is not a value we pass to git.
-fn issue_ref_name_is_safe(ref_name: &str) -> bool {
-    let Some(id) = ref_name.strip_prefix("refs/gitlawb/issues/") else {
-        return false;
-    };
-    !id.is_empty()
-        && id
-            .chars()
-            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
-}
-
-/// Resolve a ref name without putting it on a command line: enumerate refs
-/// (constant argv) and match in memory. A nonzero exit propagates, so "cannot
-/// determine" is an error rather than a quiet `false`.
-fn ref_name_resolves(repo_path: &Path, ref_name: &str) -> Result<bool> {
-    // allow-unbounded-git: failure-path existence recheck, module convention.
-    let out = Command::new("git")
-        .args(["for-each-ref", "--format=%(refname)"])
-        .current_dir(repo_path)
-        .output()
-        .context("failed to run git for-each-ref")?;
-    if !out.status.success() {
-        let stderr = String::from_utf8_lossy(&out.stderr);
-        anyhow::bail!("git for-each-ref failed: {}", stderr.trim());
-    }
-    let stdout = String::from_utf8_lossy(&out.stdout);
-    Ok(stdout.lines().any(|line| line.trim() == ref_name))
-}
-
-/// Read the blob behind an issue ref.
-///
-/// `Ok(None)` means the ref itself is absent. A ref that resolves but whose
-/// object cannot be read (corrupt object, unreadable store, a mid-read gc) is
-/// an error, not an absence (#426).
-///
-/// The ref name reaches git on stdin, never argv: `cat-file --batch` answers
-/// `<oid> <type> <size>\n<body>\n` for a readable object and
-/// `<spec> missing\n` for an absent ref or an unreadable object alike, so the
-/// `missing` case is re-resolved by name enumeration to tell them apart.
-fn read_issue_blob(repo_path: &Path, ref_name: &str) -> Result<Option<String>> {
-    if !issue_ref_name_is_safe(ref_name) {
-        anyhow::bail!("refusing issue ref outside the issues namespace: {ref_name}");
-    }
-
-    // allow-unbounded-git: this module's spawns predate the bounded runner and
-    // its callers run inside issue handlers that already hold the repo guard.
-    let mut child = Command::new("git")
-        .args(["cat-file", "--batch"])
-        .current_dir(repo_path)
-        .stdin(std::process::Stdio::piped())
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped())
-        .spawn()
-        .context("failed to spawn git cat-file --batch")?;
-
-    use std::io::Write;
-    // Reap the child even when the write fails so a stdin error cannot drop an
-    // unwaited Child and leak a zombie (#53).
-    let write_result = match child.stdin.take() {
-        Some(mut stdin) => stdin
-            .write_all(ref_name.as_bytes())
-            .and_then(|()| stdin.write_all(b"\n")),
-        None => Err(std::io::Error::new(
-            std::io::ErrorKind::BrokenPipe,
-            "git cat-file --batch stdin unavailable",
-        )),
-    };
-    let output = child
-        .wait_with_output()
-        .context("git cat-file --batch failed")?;
-    write_result.context("failed to write ref name to git cat-file stdin")?;
-
-    if !output.status.success() {
-        let stderr = String::from_utf8_lossy(&output.stderr);
-        anyhow::bail!("git cat-file --batch failed: {}", stderr.trim());
-    }
-
-    // "<spec> missing\n" covers an absent ref and an unreadable object; only a
-    // ref that still resolves is the error case.
-    let missing = format!("{ref_name} missing\n");
-    if output.stdout.as_slice() == missing.as_bytes() {
-        return match ref_name_resolves(repo_path, ref_name)? {
-            true => anyhow::bail!("issue ref resolves but its object is unreadable: {ref_name}"),
-            false => Ok(None),
-        };
-    }
-
-    // Header: "<oid> <type> <size>\n" then exactly <size> bytes plus a newline.
-    let header_end = output
-        .stdout
-        .iter()
-        .position(|b| *b == b'\n')
-        .context("unexpected git cat-file --batch output")?;
-    let header = String::from_utf8_lossy(&output.stdout[..header_end]);
-    let mut parts = header.split(' ');
-    parts.next();
-    let obj_type = parts.next().unwrap_or("");
-    if obj_type != "blob" {
-        anyhow::bail!("issue ref {ref_name} resolves to a non-blob object");
-    }
-    let size: usize = parts
-        .next()
-        .and_then(|s| s.parse().ok())
-        .context("unexpected git cat-file --batch header")?;
-    let body = &output.stdout[header_end + 1..];
-    anyhow::ensure!(
-        body.len() >= size,
-        "truncated git cat-file --batch body for {ref_name}"
-    );
-    Ok(Some(String::from_utf8_lossy(&body[..size]).to_string()))
+/// 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.
@@ -187,11 +95,9 @@ pub fn list_issues(repo_path: &Path) -> Result<Vec<String>> {
         .output()
         .context("failed to run git for-each-ref")?;
 
-    // for-each-ref exits 0 with empty output when no refs match, so a nonzero
-    // exit is a real enumeration failure, not "no issues yet".
     if !list_output.status.success() {
-        let stderr = String::from_utf8_lossy(&list_output.stderr);
-        anyhow::bail!("git for-each-ref failed: {}", stderr.trim());
+        // No issues yet
+        return Ok(vec![]);
     }
 
     let refs_str = String::from_utf8_lossy(&list_output.stdout);
@@ -203,12 +109,16 @@ pub fn list_issues(repo_path: &Path) -> Result<Vec<String>> {
             continue;
         }
 
-        // Read the blob content. A ref that vanished between for-each-ref and
-        // here is skipped; a ref that resolves but fails to read propagates so
-        // a degraded object store errors the listing instead of under-reporting.
-        match read_issue_blob(repo_path, ref_name)? {
-            Some(content) => issues.push(content),
-            None => continue,
+        // Read the blob content
+        let cat_output = Command::new("git")
+            .args(["cat-file", "blob", ref_name])
+            .current_dir(repo_path)
+            .output()
+            .context("failed to run git cat-file")?;
+
+        if cat_output.status.success() {
+            let content = String::from_utf8_lossy(&cat_output.stdout).to_string();
+            issues.push(content);
         }
     }
 
@@ -238,12 +148,8 @@ pub fn resolve_issue_id(repo_path: &Path, id_or_prefix: &str) -> Result<Option<S
         .output()
         .context("failed to run git for-each-ref")?;
 
-    // for-each-ref exits 0 with empty output when nothing matches, so a
-    // nonzero exit is a real enumeration failure (e.g. not a repo), not
-    // "issue not found".
     if !list.status.success() {
-        let stderr = String::from_utf8_lossy(&list.stderr);
-        anyhow::bail!("git for-each-ref failed: {}", stderr.trim());
+        return Ok(None);
     }
 
     let output = String::from_utf8_lossy(&list.stdout);
@@ -276,12 +182,8 @@ pub fn close_issue(repo_path: &Path, issue_id: &str) -> Result<Option<String>> {
         None => return Ok(None),
     };
 
-    // A ref that resolved and is now absent was deleted between the two calls;
-    // any other read failure propagates as an error rather than panic (#426).
-    let raw = match get_issue(repo_path, &full_id)? {
-        Some(raw) => raw,
-        None => return Ok(None),
-    };
+    let raw = get_issue(repo_path, &full_id)?
+        .expect("ref existed in resolve but not in get — should be impossible");
 
     let mut issue: serde_json::Value =
         serde_json::from_str(&raw).context("invalid issue JSON in git ref")?;
@@ -300,7 +202,18 @@ pub fn get_issue(repo_path: &Path, issue_id: &str) -> Result<Option<String>> {
     };
 
     let ref_name = format!("refs/gitlawb/issues/{full_id}");
-    read_issue_blob(repo_path, &ref_name)
+    let cat_output = Command::new("git")
+        .args(["cat-file", "blob", &ref_name])
+        .current_dir(repo_path)
+        .output()
+        .context("failed to run git cat-file")?;
+
+    if !cat_output.status.success() {
+        return Ok(None);
+    }
+
+    let content = String::from_utf8_lossy(&cat_output.stdout).to_string();
+    Ok(Some(content))
 }
 
 #[cfg(test)]
@@ -386,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();
@@ -402,109 +336,4 @@ mod tests {
         let v: serde_json::Value = serde_json::from_str(&updated).unwrap();
         assert_eq!(v["status"], "closed");
     }
-
-    /// Overwrite the loose object behind an issue ref with garbage, so
-    /// `cat-file` fails while `rev-parse` still resolves the ref (#426).
-    fn corrupt_issue_object(dir: &TempDir, full_id: &str) {
-        let oid_out = Command::new("git")
-            .args(["rev-parse", &format!("refs/gitlawb/issues/{full_id}")])
-            .current_dir(dir.path())
-            .output()
-            .unwrap();
-        let oid = String::from_utf8(oid_out.stdout)
-            .unwrap()
-            .trim()
-            .to_string();
-        let obj = dir
-            .path()
-            .join(".git/objects")
-            .join(&oid[..2])
-            .join(&oid[2..]);
-        std::fs::remove_file(&obj).unwrap();
-        std::fs::write(&obj, b"garbage").unwrap();
-    }
-
-    #[test]
-    fn test_get_and_close_missing_issue_return_none() {
-        let dir = TempDir::new().unwrap();
-        init_repo(&dir);
-        assert_eq!(get_issue(dir.path(), "nosuch00").unwrap(), None);
-        assert_eq!(close_issue(dir.path(), "nosuch00").unwrap(), None);
-    }
-
-    // #426: a directory that is not a repo makes `cat-file --batch` exit
-    // nonzero; that failure is an error, not a quiet None.
-    #[test]
-    fn test_issue_read_on_non_repo_errors_instead_of_none() {
-        let dir = TempDir::new().unwrap();
-        let not_a_repo = dir.path().join("not-a-repo");
-        std::fs::create_dir_all(&not_a_repo).unwrap();
-        assert!(read_issue_blob(&not_a_repo, "refs/gitlawb/issues/deadbeef").is_err());
-    }
-
-    // #426: the ref name is constrained to the issues namespace and a safe
-    // charset before it is fed to git, even on stdin (a newline would smuggle
-    // a second `--batch` spec).
-    #[test]
-    fn test_issue_read_rejects_ref_outside_issues_namespace() {
-        let dir = TempDir::new().unwrap();
-        init_repo(&dir);
-        assert!(read_issue_blob(dir.path(), "refs/heads/main").is_err());
-        assert!(read_issue_blob(dir.path(), "refs/gitlawb/issues/../x").is_err());
-    }
-
-    // #426: a failed for-each-ref enumeration is an error, not an empty list.
-    #[test]
-    fn test_list_issues_on_non_repo_errors_instead_of_empty() {
-        let dir = TempDir::new().unwrap();
-        let not_a_repo = dir.path().join("not-a-repo");
-        std::fs::create_dir_all(&not_a_repo).unwrap();
-        assert!(list_issues(&not_a_repo).is_err());
-    }
-
-    // #426: the same enumeration failure inside resolve_issue_id must reach
-    // get_issue/close_issue as an error, not a "not found" None.
-    #[test]
-    fn test_get_and_close_on_non_repo_error_instead_of_none() {
-        let dir = TempDir::new().unwrap();
-        let not_a_repo = dir.path().join("not-a-repo");
-        std::fs::create_dir_all(&not_a_repo).unwrap();
-        assert!(get_issue(&not_a_repo, "nosuch00").is_err());
-        assert!(close_issue(&not_a_repo, "nosuch00").is_err());
-    }
-
-    // #426: a ref that resolves but whose object is corrupt is a read error,
-    // not an absent issue. `close_issue` used to `expect()` on that None and
-    // panic inside a request handler holding the write guard.
-    #[test]
-    fn test_corrupt_issue_object_errors_instead_of_panicking() {
-        let dir = TempDir::new().unwrap();
-        init_repo(&dir);
-        let full_id = "bad00000-0000-0000-0000-000000000000";
-        create_issue(dir.path(), full_id, r#"{"status":"open"}"#).unwrap();
-        corrupt_issue_object(&dir, full_id);
-
-        assert!(get_issue(dir.path(), full_id).is_err());
-        assert!(close_issue(dir.path(), full_id).is_err());
-        assert!(list_issues(dir.path()).is_err());
-    }
-
-    #[test]
-    fn test_list_issues_returns_all() {
-        let dir = TempDir::new().unwrap();
-        init_repo(&dir);
-        create_issue(
-            dir.path(),
-            "aaa00000-0000-0000-0000-000000000000",
-            r#"{"status":"open"}"#,
-        )
-        .unwrap();
-        create_issue(
-            dir.path(),
-            "bbb00000-0000-0000-0000-000000000000",
-            r#"{"status":"open"}"#,
-        )
-        .unwrap();
-        assert_eq!(list_issues(dir.path()).unwrap().len(), 2);
-    }
 }

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")