fix(node): close review residuals: coalesce cold downloads to one in-flight task, compensate a timed-out upload with the rolled-back tree, aggregate partial ref restores, race-test the anchor lock, fail-closed snapshot test

Auto-deref (&mut tx) on the lock_repo_slug calls and rustfmt of the new test
lines. No behavior change.

.env.example

@@ -13,9 +13,16 @@ GITLAWB_PUBLIC_URL=https://your-node.example.com
 # ── Server ────────────────────────────────────────────────────────────────
 GITLAWB_HOST=0.0.0.0
 GITLAWB_PORT=7545
+# Optional address to bind a Prometheus /metrics exposition endpoint on (e.g.
+# 127.0.0.1:9091). Leave empty (default) to disable. Bind to localhost or a
+# private interface — the metrics endpoint is unauthenticated.
+GITLAWB_METRICS_ADDR=
 
 # ── Storage ───────────────────────────────────────────────────────────────
 GITLAWB_REPOS_DIR=/data/repos
+# Tigris (S3-compatible) bucket for repo storage. Leave empty (default) to
+# disable Tigris and use local-only storage.
+GITLAWB_TIGRIS_BUCKET=
 
 # PostgreSQL connection URL. Required.
 # When using the bundled docker-compose, this is wired automatically.
@@ -26,6 +33,12 @@ DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb
 # connections open lazily. Size against the DB server's max_connections,
 # remembering admin tooling opens its own pool.
 GITLAWB_DB_MAX_CONNECTIONS=20
+# Maximum connections in the dedicated advisory-lock pool (separate from the
+# pool above). Each in-flight repo write pins one connection here for its whole
+# lifetime, so size it to peak concurrent distinct-repo writers — keeping it
+# separate is what stops a push burst from starving request handlers. Keep
+# (main pool + lock pool) within the DB server's max_connections.
+GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS=32
 # 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
@@ -39,6 +52,9 @@ GITLAWB_DB_RETRY_INITIAL_SECS=5
 GITLAWB_DB_RETRY_MAX_SECS=60
 
 # ── IPFS pinning (Pinata) ─────────────────────────────────────────────────
+# URL of a local IPFS/Kubo node HTTP API (e.g. http://127.0.0.1:5001). Leave
+# empty (default) to disable local IPFS.
+GITLAWB_IPFS_API=
 # Get a JWT at https://app.pinata.cloud/developers/api-keys
 GITLAWB_PINATA_JWT=
 GITLAWB_PINATA_UPLOAD_URL=https://uploads.pinata.cloud/v3/files
@@ -127,6 +143,20 @@ GITLAWB_PUSH_RATE_LIMIT=600
 # the client IP. 0 disables. Default 120.
 GITLAWB_CREATE_RATE_LIMIT=120
 
+# ── Write rate limiting (non-creation authenticated writes) ───────────────
+# Max non-creation write requests per client IP per hour: issue/PR comments,
+# labels, stars, merges, protect/unprotect, replicas, visibility, tasks,
+# bounties, profile, and all GraphQL HTTP requests. The brake wraps the whole
+# /graphql route, so queries and the playground GET consume this bucket too, not
+# only mutations (GraphQL WebSocket subscriptions are excluded). Its own bucket,
+# separate from the creation and push brakes. Uses GITLAWB_TRUSTED_PROXY to
+# resolve the client IP.
+# NOTE: this is a per-IP aggregate across ALL those write actions, so behind a
+# shared NAT/egress IP (or with GITLAWB_TRUSTED_PROXY unset) many users collapse
+# onto one bucket — raise this for automation-heavy or multi-user single-IP
+# deployments. 0 disables. Default 600.
+GITLAWB_WRITE_RATE_LIMIT=600
+
 # ── Peer-sync rate limiting (per client IP, uses GITLAWB_TRUSTED_PROXY below) ─
 # /api/v1/peers/announce and /api/v1/sync/notify accept unsigned requests from
 # known peers and run at higher frequency, so a generous bucket. Separate from

Cargo.lock

@@ -3300,7 +3300,7 @@ dependencies = [
 
 [[package]]
 name = "git-remote-gitlawb"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "anyhow",
  "gitlawb-core",
@@ -3312,7 +3312,7 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-attest"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "base64",
  "ed25519-dalek",
@@ -3329,7 +3329,7 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-core"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "anyhow",
  "base64",
@@ -3356,13 +3356,14 @@ dependencies = [
 
 [[package]]
 name = "gitlawb-node"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "alloy",
  "anyhow",
  "async-compression",
  "async-graphql",
  "async-graphql-axum",
+ "async-trait",
  "aws-config",
  "aws-sdk-s3",
  "axum",
@@ -3412,10 +3413,11 @@ dependencies = [
 
 [[package]]
 name = "gl"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "alloy",
  "anyhow",
+ "base64",
  "chrono",
  "clap",
  "dirs",

crates/gitlawb-node/Cargo.toml

@@ -12,6 +12,7 @@ path = "src/main.rs"
 
 [dependencies]
 gitlawb-core = { path = "../gitlawb-core" }
+async-trait = "0.1"
 ed25519-dalek = { workspace = true }
 base64 = { workspace = true }
 tokio = { workspace = true }

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

@@ -70,11 +70,37 @@ pub async fn create_issue(
 
     let create_result = git_issues::create_issue(&disk_path, &issue_id, &json_str);
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(create_result.is_ok()).await;
+    // Always release the advisory lock, even on error; upload to Tigris only on
+    // success. A false return means the durable upload failed/timed out: the
+    // issue ref applied to local disk but never reached object storage, so the
+    // next acquire_write would revert it from the stale archive. Report 5xx so
+    // the client retries rather than trusting an issue that never landed durably,
+    // and skip the trust bump below. (Same P1 data-loss guard as receive-pack.)
+    // On that failure, roll back the local ref BEFORE the lock releases — an
+    // orphan ref would make the retry a duplicate, and in an unlock-to-delete
+    // window a concurrent same-repo write could upload an archive still
+    // carrying it. Best-effort: a failed delete logs and the 5xx still returns.
+    let upload_ok = guard
+        .release_with_failure_cleanup(create_result.is_ok(), |path| {
+            if let Err(e) = git_issues::delete_issue_ref(path, &issue_id) {
+                tracing::warn!(issue = %issue_id, err = %e,
+                    "failed to roll back local issue ref after failed durable upload; \
+                     a retry may duplicate this issue");
+            }
+        })
+        .await;
 
     create_result.map_err(|e| AppError::Git(e.to_string()))?;
 
+    if !upload_ok {
+        tracing::error!(repo = %record.name,
+            "durable upload failed after issue create; reporting failure so the client retries");
+        return Err(AppError::Git(format!(
+            "durable storage upload failed for {}",
+            record.name
+        )));
+    }
+
     // Bump trust score for the issue author — increment current score by 0.05
     // (avoids the push_count=0 stuck-at-0.05 bug for agents who only file issues)
     if let Some(ref author_did) = issue.author {
@@ -244,11 +270,14 @@ pub async fn close_issue(
             .ok()
             .and_then(|i| i.author),
         Ok(None) => {
-            guard.release(false).await;
+            // release(false): no upload attempted, so the bool is always true
+            // (nothing to fail on). Ignore it on this pre-write error path.
+            let _ = guard.release(false).await;
             return Err(AppError::NotFound(format!("issue {issue_id} not found")));
         }
         Err(e) => {
-            guard.release(false).await;
+            // release(false): no upload attempted, nothing to fail on.
+            let _ = guard.release(false).await;
             return Err(AppError::Git(e.to_string()));
         }
     };
@@ -257,21 +286,57 @@ pub async fn close_issue(
         .as_deref()
         .is_some_and(|a| crate::api::did_matches(&auth.0, a));
     if !is_owner && !is_author {
-        guard.release(false).await;
+        // release(false): no upload attempted, nothing to fail on.
+        let _ = guard.release(false).await;
         return Err(AppError::Forbidden(
             "only the repo owner or the issue author can close this issue".into(),
         ));
     }
 
+    // Snapshot the issue ref's pre-close OID so a failed durable upload can roll
+    // the close back on local disk, mirroring create_issue. Best-effort: if the
+    // snapshot read fails we proceed without a rollback closure (the close still
+    // applies and a failed upload still surfaces as a 5xx below).
+    let prior_ref = git_issues::issue_ref_oid(&disk_path, &issue_id)
+        .ok()
+        .flatten();
+
     let close_result = git_issues::close_issue(&disk_path, &issue_id);
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(close_result.is_ok()).await;
+    // Always release the advisory lock, even on error; upload to Tigris only on
+    // success. A false return means the durable upload failed: the close applied
+    // to local disk but never reached object storage, so a later acquire_write
+    // reverts it. Report 5xx so the client retries. (P1 data-loss guard.)
+    // On that failure, roll the local close BACK to the captured open ref BEFORE
+    // the lock releases — otherwise the local fast path in RepoStore::acquire
+    // serves the "closed" state until a later write refreshes the archive, even
+    // though the client got a 5xx. The cleanup runs only when an upload was
+    // attempted and failed, while the advisory lock is still held.
+    let upload_ok = guard
+        .release_with_failure_cleanup(close_result.is_ok(), |path| {
+            if let Some((full_id, oid)) = &prior_ref {
+                if let Err(e) = git_issues::restore_issue_ref(path, full_id, oid) {
+                    tracing::warn!(issue = %issue_id, err = %e,
+                        "failed to roll back local issue close after failed durable upload; \
+                         a local-fast-path read may serve a stale closed state");
+                }
+            }
+        })
+        .await;
 
     let updated = close_result
         .map_err(|e| AppError::Git(e.to_string()))?
         .ok_or_else(|| AppError::RepoNotFound(format!("issue {issue_id} not found")))?;
 
+    if !upload_ok {
+        tracing::error!(repo = %repo,
+            "durable upload failed after issue close; reporting failure so the client retries");
+        return Err(AppError::Git(format!(
+            "durable storage upload failed for {}",
+            record.name
+        )));
+    }
+
     let issue: serde_json::Value = serde_json::from_str(&updated)
         .map_err(|e| AppError::BadRequest(format!("invalid issue data: {e}")))?;
 
@@ -279,3 +344,421 @@ pub async fn close_issue(
 
     Ok(Json(issue))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::extract::{Path as AxPath, State};
+    use axum::Extension;
+    use std::path::Path as StdPath;
+    use std::time::Duration;
+
+    // Object store whose upload() parks forever, so release()'s bounded timeout
+    // is the only thing that completes it, modeling a stalled durable PUT. exists()
+    // is false so acquire_write never downloads over the fresh local bare repo.
+    struct StallStore;
+    #[async_trait::async_trait]
+    impl crate::git::tigris::ObjectStore for StallStore {
+        async fn exists(&self, _o: &str, _r: &str) -> anyhow::Result<bool> {
+            Ok(false)
+        }
+        async fn upload(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            std::future::pending::<()>().await;
+            Ok(())
+        }
+        async fn download(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn delete(&self, _o: &str, _r: &str) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
+    // Object store whose upload succeeds immediately. exists() is false so
+    // acquire_write never performs a reverting refresh over the local dir —
+    // the retry test below pins that contract explicitly.
+    struct OkStore;
+    #[async_trait::async_trait]
+    impl crate::git::tigris::ObjectStore for OkStore {
+        async fn exists(&self, _o: &str, _r: &str) -> anyhow::Result<bool> {
+            Ok(false)
+        }
+        async fn upload(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn download(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn delete(&self, _o: &str, _r: &str) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
+    // Shared setup for the failed-upload rollback tests: a state whose repo
+    // store stalls uploads (200ms release timeout), a public repo row for
+    // `owner/name`, and an initialized bare repo at the returned path.
+    async fn stall_state_with_repo(
+        pool: sqlx::PgPool,
+        repos_dir: &tempfile::TempDir,
+        owner: &str,
+        name: &str,
+    ) -> (crate::state::AppState, std::path::PathBuf) {
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(std::sync::Arc::new(StallStore)),
+            pool.clone(),
+        )
+        .with_release_upload_timeout(Duration::from_millis(200));
+
+        let owner_slug = owner.replace([':', '/'], "_");
+        let bare = repos_dir
+            .path()
+            .join(&owner_slug)
+            .join(format!("{name}.git"));
+
+        state
+            .db
+            .upsert_mirror_repo(owner, name, &bare.to_string_lossy(), None, false)
+            .await
+            .unwrap();
+
+        std::fs::create_dir_all(&bare).unwrap();
+        let out = std::process::Command::new("git")
+            .args(["init", "--bare", "-q", &bare.to_string_lossy()])
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "git init --bare failed");
+
+        (state, bare)
+    }
+
+    async fn create_issue_status(
+        state: &crate::state::AppState,
+        owner: &str,
+        name: &str,
+    ) -> axum::http::StatusCode {
+        use axum::response::IntoResponse;
+        let resp = super::create_issue(
+            State(state.clone()),
+            Extension(crate::auth::AuthenticatedDid(owner.to_string())),
+            AxPath((owner.to_string(), name.to_string())),
+            axum::Json(CreateIssueRequest {
+                title: "t".into(),
+                body: None,
+                signed_payload: None,
+            }),
+        )
+        .await;
+        match resp {
+            Ok((code, _)) => code,
+            Err(e) => e.into_response().status(),
+        }
+    }
+
+    async fn close_issue_status(
+        state: &crate::state::AppState,
+        owner: &str,
+        name: &str,
+        issue_id: &str,
+    ) -> axum::http::StatusCode {
+        use axum::response::IntoResponse;
+        let resp = super::close_issue(
+            State(state.clone()),
+            Extension(crate::auth::AuthenticatedDid(owner.to_string())),
+            AxPath((owner.to_string(), name.to_string(), issue_id.to_string())),
+        )
+        .await;
+        match resp {
+            Ok(_) => axum::http::StatusCode::OK,
+            Err(e) => e.into_response().status(),
+        }
+    }
+
+    // Seed an OPEN issue directly on local disk, bypassing the (stalling) upload
+    // path so the close test starts from a durably-consistent open issue.
+    fn seed_open_issue(bare: &StdPath, owner: &str, issue_id: &str) {
+        let issue = IssueRecord {
+            id: issue_id.to_string(),
+            title: "t".into(),
+            body: None,
+            author: Some(owner.to_string()),
+            created_at: Utc::now().to_rfc3339(),
+            status: "open".to_string(),
+            signed_payload: None,
+        };
+        let json = serde_json::to_string(&issue).unwrap();
+        git_issues::create_issue(bare, issue_id, &json).unwrap();
+    }
+
+    fn issue_status(bare: &StdPath, issue_id: &str) -> String {
+        let raw = git_issues::get_issue(bare, issue_id).unwrap().unwrap();
+        let v: serde_json::Value = serde_json::from_str(&raw).unwrap();
+        v["status"].as_str().unwrap().to_string()
+    }
+
+    // U2 [F2, P1] — the failed-upload rollback on the close path. close_issue
+    // applies the "closed" ref to LOCAL disk, then release() uploads to durable
+    // storage. When that upload times out the local close is NOT persisted, so a
+    // later acquire_write reverts it from the stale archive — but until then, a
+    // local-fast-path read via RepoStore::acquire serves the stale "closed"
+    // state even though the client got a 5xx. The fix rolls the close back to the
+    // captured open ref while the advisory lock is held. RED with the cleanup
+    // closure reverted to `|_| {}`: the read sees "closed".
+    #[sqlx::test]
+    async fn issue_close_failed_upload_rolls_back_to_open(pool: sqlx::PgPool) {
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        let owner = "z6issuecloserollbackowner";
+        let name = "issuecloserollback";
+        let (state, bare) = stall_state_with_repo(pool, &repos_dir, owner, name).await;
+
+        let issue_id = "aaaa1111-0000-0000-0000-000000000000";
+        seed_open_issue(&bare, owner, issue_id);
+        assert_eq!(issue_status(&bare, issue_id), "open", "seed must be open");
+
+        // Close via the handler — the durable upload stalls and times out (5xx).
+        let status = close_issue_status(&state, owner, name, issue_id).await;
+        assert!(
+            status.is_server_error(),
+            "a stalled durable upload on issue-close must 500, got {status}"
+        );
+
+        // The local fast path must still see the issue OPEN: the failed upload
+        // rolled the close back while the lock was held.
+        assert_eq!(
+            issue_status(&bare, issue_id),
+            "open",
+            "a failed durable upload must roll the local close back to open; a \
+             local-fast-path read served the stale closed state"
+        );
+    }
+
+    // Success path: a working store closes the issue (2xx) and the read reflects
+    // the close. Confirms the rollback wiring does not fire on success.
+    #[sqlx::test]
+    async fn issue_close_success_persists_close(pool: sqlx::PgPool) {
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        let owner = "z6issueclosesuccessowner";
+        let name = "issueclosesuccess";
+        let (mut state, bare) = stall_state_with_repo(pool.clone(), &repos_dir, owner, name).await;
+        // Swap the stalling store for one whose upload succeeds immediately.
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(std::sync::Arc::new(OkStore)),
+            pool.clone(),
+        )
+        .with_release_upload_timeout(Duration::from_millis(200));
+
+        let issue_id = "bbbb2222-0000-0000-0000-000000000000";
+        seed_open_issue(&bare, owner, issue_id);
+
+        let status = close_issue_status(&state, owner, name, issue_id).await;
+        assert_eq!(status, StatusCode::OK, "working store must close (200)");
+        assert_eq!(
+            issue_status(&bare, issue_id),
+            "closed",
+            "a successful close must persist as closed"
+        );
+    }
+
+    // Pre-write error path: closing a nonexistent issue returns 404 and never
+    // attempts an upload (release(false)), so the stalling store cannot hang it.
+    #[sqlx::test]
+    async fn issue_close_not_found_releases_without_upload(pool: sqlx::PgPool) {
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        let owner = "z6issueclosemissingowner";
+        let name = "issueclosemissing";
+        let (state, _bare) = stall_state_with_repo(pool, &repos_dir, owner, name).await;
+
+        let status =
+            close_issue_status(&state, owner, name, "cccc3333-0000-0000-0000-000000000000").await;
+        assert_eq!(
+            status,
+            StatusCode::NOT_FOUND,
+            "closing a missing issue must 404 without attempting an upload"
+        );
+    }
+
+    fn issue_refs(bare: &StdPath) -> Vec<String> {
+        let out = std::process::Command::new("git")
+            .args([
+                "for-each-ref",
+                "--format=%(refname)",
+                "refs/gitlawb/issues/",
+            ])
+            .current_dir(bare)
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "git for-each-ref failed");
+        String::from_utf8_lossy(&out.stdout)
+            .lines()
+            .filter(|l| !l.trim().is_empty())
+            .map(|l| l.trim().to_string())
+            .collect()
+    }
+
+    // Rollback half of the #196 round-four issue-create fix. When the durable
+    // upload fails/times out, the handler 500s (covered below) — but the issue
+    // ref it wrote to LOCAL disk must also be rolled back, while the advisory
+    // lock is still held. An orphan ref makes the 500 dishonest: a retry
+    // re-creates the issue under a new uuid while the orphan is still listed,
+    // and a concurrent same-repo write can upload an archive carrying it.
+    // RED pre-fix: the orphan ref survives the failed create.
+    #[sqlx::test]
+    async fn issue_create_failed_upload_rolls_back_local_ref(pool: sqlx::PgPool) {
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        let owner = "z6issuerollbackowner";
+        let name = "issuerollback";
+        let (state, bare) = stall_state_with_repo(pool, &repos_dir, owner, name).await;
+
+        let status = create_issue_status(&state, owner, name).await;
+        assert!(
+            status.is_server_error(),
+            "stalled upload must 500, got {status}"
+        );
+
+        let refs = issue_refs(&bare);
+        assert!(
+            refs.is_empty(),
+            "a failed durable upload must roll back the local issue ref while \
+             the advisory lock is held; orphan ref(s) survived: {refs:?}"
+        );
+    }
+
+    // Retry half: after the failed-upload 500, a retry with a WORKING store
+    // must yield exactly one issue. Pre-fix the orphan ref from the failed
+    // attempt makes the retry a duplicate (two issues in list_issues).
+    #[sqlx::test]
+    async fn issue_create_failed_upload_retry_does_not_duplicate(pool: sqlx::PgPool) {
+        use crate::git::tigris::ObjectStore as _;
+
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        let owner = "z6issueretryowner";
+        let name = "issueretry";
+        let (state, bare) = stall_state_with_repo(pool.clone(), &repos_dir, owner, name).await;
+
+        let status = create_issue_status(&state, owner, name).await;
+        assert!(
+            status.is_server_error(),
+            "stalled upload must 500, got {status}"
+        );
+
+        // PIN the double's contract: exists() stays false after the failed
+        // upload, so the retry's acquire_write performs no reverting refresh
+        // from a stale archive. Without this pin, a download could revert the
+        // orphan ref and the assertion below would pass without the rollback
+        // fix — it must rest on the rollback, not a stale-archive revert.
+        let owner_slug = owner.replace([':', '/'], "_");
+        assert!(
+            !OkStore.exists(&owner_slug, name).await.unwrap(),
+            "test double contract broken: exists() must stay false"
+        );
+
+        // Retry with a store whose upload succeeds.
+        let mut retry_state = state.clone();
+        retry_state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(std::sync::Arc::new(OkStore)),
+            pool.clone(),
+        )
+        .with_release_upload_timeout(Duration::from_millis(200));
+
+        let status = create_issue_status(&retry_state, owner, name).await;
+        assert_eq!(
+            status,
+            StatusCode::CREATED,
+            "retry with working store must 201"
+        );
+
+        let issues = crate::git::issues::list_issues(&bare).unwrap();
+        assert_eq!(
+            issues.len(),
+            1,
+            "retry after a failed-upload 500 must yield exactly ONE issue; the \
+             failed attempt's orphan ref duplicated it (got {} issues)",
+            issues.len()
+        );
+    }
+
+    // P1-class data-loss regression, issue-create path. create_issue writes the
+    // issue ref to local disk, then release(true) uploads the repo to durable
+    // storage. When that upload times out the local write is NOT persisted, so
+    // the next acquire_write reverts it from the stale archive. The pre-fix
+    // handler ignored release()'s bool and still returned 201 AND bumped the
+    // author's trust score, so the client trusted an issue that never landed
+    // durably. The fix surfaces a failed/timed-out upload as a 5xx and skips the
+    // success tail (the trust bump). Stall the upload past a tiny release timeout
+    // and assert (a) the handler returns 5xx and (b) the trust bump did not run.
+    // RED pre-fix: 201 + trust score bumped to 0.05.
+    #[sqlx::test]
+    async fn issue_create_durable_upload_timeout_fails_and_skips_side_effects(pool: sqlx::PgPool) {
+        use axum::response::IntoResponse;
+
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(std::sync::Arc::new(StallStore)),
+            pool.clone(),
+        )
+        .with_release_upload_timeout(Duration::from_millis(200));
+
+        let owner = "z6issueownerdurablefail";
+        let name = "issuedurablefail";
+        let owner_slug = owner.replace([':', '/'], "_");
+        let bare = repos_dir
+            .path()
+            .join(&owner_slug)
+            .join(format!("{name}.git"));
+
+        // Public mirror repo so the owner clears authorize_repo_read, and a real
+        // agent row so the post-write trust bump is an observable durable side
+        // effect (register_agent seeds trust_score = 0.0).
+        state
+            .db
+            .upsert_mirror_repo(owner, name, &bare.to_string_lossy(), None, false)
+            .await
+            .unwrap();
+        state.db.register_agent(owner, &[]).await.unwrap();
+
+        std::fs::create_dir_all(&bare).unwrap();
+        let out = std::process::Command::new("git")
+            .args(["init", "--bare", "-q", &bare.to_string_lossy()])
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "git init --bare failed");
+
+        // acquire_write → git_issues::create_issue (writes the ref) → release
+        // (upload stalls, times out after 200ms). The client MUST see a failure.
+        let resp = super::create_issue(
+            State(state.clone()),
+            Extension(crate::auth::AuthenticatedDid(owner.to_string())),
+            AxPath((owner.to_string(), name.to_string())),
+            axum::Json(CreateIssueRequest {
+                title: "t".into(),
+                body: None,
+                signed_payload: None,
+            }),
+        )
+        .await;
+        let status = match resp {
+            Ok((code, _)) => code,
+            Err(e) => e.into_response().status(),
+        };
+        assert!(
+            status.is_server_error(),
+            "a timed-out durable upload on issue-create must fail (5xx) so the \
+             client retries (got {status}), which the client trusts as a landed \
+             issue that a later acquire_write would silently revert"
+        );
+
+        // The success tail (trust bump) must NOT have run: the issue was never
+        // durably accepted. register_agent seeded 0.0; a bump would make it 0.05.
+        let score = state.db.get_trust_score(owner).await.unwrap();
+        assert_eq!(
+            score, 0.0,
+            "the trust bump must NOT run when the durable upload failed; the \
+             issue was not durably accepted (score bumped to {score})"
+        );
+    }
+}

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

@@ -216,6 +216,14 @@ pub async fn merge_pr(
         .map_err(|e| AppError::Git(e.to_string()))?;
     let disk_path = guard.path().to_path_buf();
     let merger_did = auth.0;
+
+    // Snapshot the target branch's pre-merge tip so a failed durable upload can
+    // roll the merge commit back on local disk (mirrors create_issue). Without
+    // this the local fast path in RepoStore::acquire serves the merged tree until
+    // a later write refreshes the archive, even though the client got a 5xx.
+    let target_ref = format!("refs/heads/{}", pr.target_branch);
+    let prior_target = store::ref_oid(&disk_path, &target_ref).ok().flatten();
+
     let merge_result = store::merge_branch(
         &disk_path,
         &pr.target_branch,
@@ -224,11 +232,37 @@ pub async fn merge_pr(
         &pr.title,
     );
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(merge_result.is_ok()).await;
+    // Always release the advisory lock, even on error; upload to Tigris only on
+    // success. A false return means the durable upload failed/timed out: the merge
+    // commit applied to local disk but never reached object storage, so the next
+    // acquire_write reverts it from the stale archive. Report 5xx so the client
+    // retries rather than marking the PR merged in the DB while the merged tree
+    // was silently lost. Fail BEFORE merge_pr / touch_repo / webhooks. (P1
+    // data-loss guard, same as receive-pack.) On that failure, roll the local
+    // merge back to the captured tip BEFORE the lock releases.
+    let upload_ok = guard
+        .release_with_failure_cleanup(merge_result.is_ok(), |path| {
+            if let Some(oid) = &prior_target {
+                if let Err(e) = store::set_ref(path, &target_ref, oid) {
+                    tracing::warn!(repo = %record.name, err = %e,
+                        "failed to roll back local merge commit after failed durable upload; \
+                         a local-fast-path read may serve the un-uploaded merged tree");
+                }
+            }
+        })
+        .await;
 
     let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
 
+    if !upload_ok {
+        tracing::error!(repo = %record.name,
+            "durable upload failed after merge; reporting failure so the client retries");
+        return Err(AppError::Git(format!(
+            "durable storage upload failed for {}",
+            record.name
+        )));
+    }
+
     state.db.merge_pr(&pr.id, &merger_did).await?;
     let _ = state.db.touch_repo(&record.id).await;
 
@@ -424,3 +458,169 @@ pub async fn list_comments(
     let comments = state.db.list_pr_comments(&pr.id).await?;
     Ok(Json(serde_json::json!({ "comments": comments })))
 }
+
+#[cfg(test)]
+mod tests {
+    // super::* brings in the handler plus the axum Extension/Path/State
+    // extractors and the Utc/Uuid/PullRequest types the seed uses.
+    use super::*;
+    use std::path::Path as StdPath;
+    use std::time::Duration;
+
+    // Object store whose upload() parks forever, so release()'s bounded timeout
+    // is the only thing that completes it, modeling a stalled durable PUT.
+    // exists() is false so acquire_write never downloads over the local bare repo.
+    struct StallStore;
+    #[async_trait::async_trait]
+    impl crate::git::tigris::ObjectStore for StallStore {
+        async fn exists(&self, _o: &str, _r: &str) -> anyhow::Result<bool> {
+            Ok(false)
+        }
+        async fn upload(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            std::future::pending::<()>().await;
+            Ok(())
+        }
+        async fn download(&self, _o: &str, _r: &str, _p: &StdPath) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn delete(&self, _o: &str, _r: &str) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
+    // The failed-upload rollback on the merge path. merge_pr writes the merge
+    // commit to LOCAL disk, then release() uploads to durable storage. When
+    // that upload times out the handler must 5xx (before merge_pr/webhooks)
+    // AND roll the target branch back to its pre-merge tip while the advisory
+    // lock is held, so a local-fast-path read never serves the un-uploaded
+    // merged tree. RED with the cleanup closure at ~pulls.rs:244 reverted to
+    // `|_| {}`: main stays at the merge commit.
+    #[sqlx::test]
+    async fn merge_failed_upload_rolls_back_target_ref(pool: sqlx::PgPool) {
+        use axum::response::IntoResponse;
+        use std::process::Command;
+
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        let repos_dir = tempfile::TempDir::new().unwrap();
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir.path().to_path_buf(),
+            Some(std::sync::Arc::new(StallStore)),
+            pool.clone(),
+        )
+        .with_release_upload_timeout(Duration::from_millis(200));
+
+        let owner = "z6mergerollbackowner";
+        let name = "mergerollback";
+        let owner_slug = owner.replace([':', '/'], "_");
+        let bare = repos_dir
+            .path()
+            .join(&owner_slug)
+            .join(format!("{name}.git"));
+
+        state
+            .db
+            .upsert_mirror_repo(owner, name, &bare.to_string_lossy(), None, false)
+            .await
+            .unwrap();
+
+        fn git(args: &[&str], dir: &StdPath) -> String {
+            let out = Command::new("git")
+                .args(args)
+                .current_dir(dir)
+                .output()
+                .unwrap();
+            assert!(
+                out.status.success(),
+                "git {args:?} failed: {}",
+                String::from_utf8_lossy(&out.stderr)
+            );
+            String::from_utf8_lossy(&out.stdout).trim().to_string()
+        }
+
+        // Scratch repo: c1 on main, c2 on feature; the bare clone carries both
+        // branches, so the merge of feature into main has real work to do.
+        let work = tempfile::TempDir::new().unwrap();
+        git(&["init", "-q", "-b", "main", "."], work.path());
+        git(&["config", "user.email", "t@t"], work.path());
+        git(&["config", "user.name", "t"], work.path());
+        std::fs::write(work.path().join("f.txt"), "one").unwrap();
+        git(&["add", "f.txt"], work.path());
+        git(&["commit", "-q", "-m", "c1"], work.path());
+        git(&["checkout", "-q", "-b", "feature"], work.path());
+        std::fs::write(work.path().join("g.txt"), "two").unwrap();
+        git(&["add", "g.txt"], work.path());
+        git(&["commit", "-q", "-m", "c2"], work.path());
+
+        std::fs::create_dir_all(bare.parent().unwrap()).unwrap();
+        let out = Command::new("git")
+            .args([
+                "clone",
+                "--bare",
+                "-q",
+                &work.path().to_string_lossy(),
+                &bare.to_string_lossy(),
+            ])
+            .output()
+            .unwrap();
+        assert!(
+            out.status.success(),
+            "clone --bare failed: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+
+        let pre_merge_tip = store::ref_oid(&bare, "refs/heads/main")
+            .unwrap()
+            .expect("main exists before the merge");
+
+        // Seed the open PR row the handler looks up.
+        let record = state.db.get_repo(owner, name).await.unwrap().unwrap();
+        let now = Utc::now().to_rfc3339();
+        state
+            .db
+            .create_pr(&PullRequest {
+                id: Uuid::new_v4().to_string(),
+                repo_id: record.id.clone(),
+                number: 1,
+                title: "t".into(),
+                body: None,
+                author_did: owner.to_string(),
+                source_branch: "feature".into(),
+                target_branch: "main".into(),
+                status: "open".to_string(),
+                merged_by_did: None,
+                merged_at: None,
+                created_at: now.clone(),
+                updated_at: now,
+            })
+            .await
+            .unwrap();
+
+        // Merge as the owner: the merge commit applies locally, the durable
+        // upload stalls and times out, and the handler must report 5xx.
+        let resp = super::merge_pr(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), name.to_string(), 1)),
+        )
+        .await;
+        let status = match resp {
+            Ok(_) => StatusCode::OK,
+            Err(e) => e.into_response().status(),
+        };
+        assert!(
+            status.is_server_error(),
+            "a stalled durable upload on merge must 5xx, got {status}"
+        );
+
+        // The target branch must still point at the pre-merge tip: the failed
+        // upload rolled the merge commit back while the lock was held.
+        let tip = store::ref_oid(&bare, "refs/heads/main")
+            .unwrap()
+            .expect("main still exists after the rollback");
+        assert_eq!(
+            tip, pre_merge_tip,
+            "a failed durable upload must roll the target branch back to its \
+             pre-merge tip; a local-fast-path read served the un-uploaded merge"
+        );
+    }
+}

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

@@ -515,6 +515,7 @@ mod tests {
             repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
             rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
             create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
+            write_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
             push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
             push_limiter_trust: crate::rate_limit::TrustedProxy::None,
             sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)),

crates/gitlawb-node/src/config.rs

@@ -1,18 +1,45 @@
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use std::path::PathBuf;
 
+/// Optional admin subcommands. When none is given, the binary runs the node
+/// daemon as before — the default (no-subcommand) startup path is unchanged.
+#[derive(Subcommand, Debug, Clone)]
+pub enum Command {
+    /// Dry-run (default) or, with --execute, delete empty spam-burst repos owned
+    /// by the known burst DID. Never touches the hard-excluded DIDs, and verifies
+    /// each repo is empty (zero git refs) per repo before selecting it.
+    PurgeSpam {
+        /// Actually delete the candidates. Omit for a dry-run that prints the
+        /// candidate list and deletes nothing.
+        #[arg(long, default_value_t = false)]
+        execute: bool,
+    },
+}
+
 #[derive(Parser, Debug, Clone)]
 #[command(name = "gitlawb-node", about = "gitlawb node daemon", version)]
 pub struct Config {
-    /// Directory where bare git repositories are stored
-    #[arg(long, env = "GITLAWB_REPOS_DIR", default_value = "./data/repos")]
+    /// Admin subcommand to run instead of the node daemon. Absent = run the node.
+    #[command(subcommand)]
+    pub command: Option<Command>,
+
+    /// Directory where bare git repositories are stored. `global` so it can
+    /// follow a subcommand (e.g. `gitlawb-node purge-spam --repos-dir …`).
+    #[arg(
+        long,
+        env = "GITLAWB_REPOS_DIR",
+        default_value = "./data/repos",
+        global = true
+    )]
     pub repos_dir: PathBuf,
 
-    /// PostgreSQL connection URL (Supabase or any Postgres instance)
+    /// PostgreSQL connection URL (Supabase or any Postgres instance). `global` so
+    /// admin subcommands accept it in either position.
     #[arg(
         long,
         env = "DATABASE_URL",
-        default_value = "postgresql://localhost/gitlawb"
+        default_value = "postgresql://localhost/gitlawb",
+        global = true
     )]
     pub database_url: String,
 
@@ -194,6 +221,22 @@ pub struct Config {
     )]
     pub db_max_connections: u32,
 
+    /// Maximum connections in the dedicated advisory-lock pool, separate from
+    /// the main pool above. Every in-flight repo write (git-receive-pack, fork,
+    /// merge) pins one connection here for its whole lifetime — the write, its
+    /// metadata/fan-out tail, and the bounded post-write archive upload — so
+    /// size it to the expected peak number of concurrent distinct-repo writers,
+    /// NOT small. Keeping it separate from GITLAWB_DB_MAX_CONNECTIONS is what
+    /// stops a push burst from starving normal request handlers. Keep
+    /// (main pool + lock pool) within the database server's max_connections.
+    #[arg(
+        long,
+        env = "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS",
+        default_value_t = 32,
+        value_parser = clap::value_parser!(u32).range(1..)
+    )]
+    pub db_lock_pool_max_connections: u32,
+
     /// 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(

crates/gitlawb-node/src/error.rs

@@ -35,9 +35,6 @@ pub enum AppError {
     #[error("invalid request: {0}")]
     BadRequest(String),
 
-    #[error("too many requests: {0}")]
-    TooManyRequests(String),
-
     #[error("incomplete: {0}")]
     Incomplete(String),
 
@@ -47,6 +44,9 @@ pub enum AppError {
     #[error("git service timed out: {0}")]
     Timeout(String),
 
+    #[error("service unavailable: {0}")]
+    Unavailable(String),
+
     #[error("database error: {0}")]
     Db(#[from] sqlx::Error),
 
@@ -132,9 +132,6 @@ impl IntoResponse for AppError {
             // IcaptchaProofRequired is handled above (it carries extra headers/fields).
             AppError::IcaptchaProofRequired { .. } => unreachable!("handled before this match"),
             AppError::BadRequest(msg) => (StatusCode::BAD_REQUEST, "bad_request", msg.clone()),
-            AppError::TooManyRequests(msg) => {
-                (StatusCode::TOO_MANY_REQUESTS, "rate_limited", msg.clone())
-            }
             AppError::Incomplete(msg) => {
                 (StatusCode::UNPROCESSABLE_ENTITY, "incomplete", msg.clone())
             }
@@ -142,6 +139,14 @@ 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, retryable: a transient resource contention (e.g. the repo lock
+            // pool is pinned by a write burst or a purge holds the key), distinct
+            // from the 500 git_error so the client knows to back off and retry.
+            AppError::Unavailable(msg) => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "service_unavailable",
+                msg.clone(),
+            ),
             AppError::Db(e) if db_unavailable(e) => (
                 StatusCode::SERVICE_UNAVAILABLE,
                 DB_UNAVAILABLE_CODE,
@@ -182,4 +187,20 @@ mod tests {
             StatusCode::INTERNAL_SERVER_ERROR
         );
     }
+
+    #[test]
+    fn unavailable_maps_to_503_distinct_from_git_500() {
+        // The create-repo lock-acquire failure returns Unavailable so a client sees
+        // a retryable 503, not a terminal 500 git_error.
+        assert_eq!(
+            AppError::Unavailable("lock pool pinned".into())
+                .into_response()
+                .status(),
+            StatusCode::SERVICE_UNAVAILABLE
+        );
+        assert_eq!(
+            AppError::Git("x".into()).into_response().status(),
+            StatusCode::INTERNAL_SERVER_ERROR
+        );
+    }
 }

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

@@ -61,6 +61,51 @@ pub fn create_issue(repo_path: &Path, issue_id: &str, json: &str) -> Result<()>
     Ok(())
 }
 
+/// Delete an issue ref. Used to roll back a locally written issue whose
+/// durable upload failed, so a retry does not duplicate it. The dangling
+/// blob object is harmless.
+pub fn delete_issue_ref(repo_path: &Path, issue_id: &str) -> Result<()> {
+    let ref_name = format!("refs/gitlawb/issues/{issue_id}");
+    let output = Command::new("git")
+        .args(["update-ref", "-d", &ref_name])
+        .current_dir(repo_path)
+        .output()
+        .context("failed to run git update-ref -d")?;
+
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        anyhow::bail!("git update-ref -d failed: {stderr}");
+    }
+
+    Ok(())
+}
+
+/// Read the object id an issue ref currently points to (resolving an 8-char
+/// prefix), returning `(full_id, oid)`. Returns None if no such issue exists.
+/// Used to snapshot an issue ref's pre-mutation state so a failed durable
+/// upload can restore it, mirroring `create_issue`'s rollback.
+pub fn issue_ref_oid(repo_path: &Path, issue_id: &str) -> Result<Option<(String, String)>> {
+    let full_id = match resolve_issue_id(repo_path, issue_id)? {
+        Some(id) => id,
+        None => return Ok(None),
+    };
+    let ref_name = format!("refs/gitlawb/issues/{full_id}");
+    let oid = match crate::git::store::ref_oid(repo_path, &ref_name)? {
+        Some(oid) => oid,
+        None => return Ok(None),
+    };
+    Ok(Some((full_id, oid)))
+}
+
+/// Reset an issue ref back to a captured object id — rolls back a local
+/// mutation (e.g. a close) whose durable upload failed, so a local-fast-path
+/// read does not serve the un-uploaded state. `full_id` must be the resolved
+/// UUID (from [`issue_ref_oid`]). The now-dangling object is harmless.
+pub fn restore_issue_ref(repo_path: &Path, full_id: &str, oid: &str) -> Result<()> {
+    let ref_name = format!("refs/gitlawb/issues/{full_id}");
+    crate::git::store::set_ref(repo_path, &ref_name, oid)
+}
+
 /// List all issue refs and return their JSON content.
 pub fn list_issues(repo_path: &Path) -> Result<Vec<String>> {
     // List all refs under refs/gitlawb/issues/

crates/gitlawb-node/src/git/repo_store/download.rs

@@ -0,0 +1,431 @@
+//! Read-path download coordination (KTD-3) for [`RepoStore`].
+//!
+//! The read path's cache-miss handling lives here: the per-repo async mutex
+//! that serializes concurrent readers of the same repo, the holder's fetch +
+//! extract + under-lock publish, and the RAII temp-dir cleanup. `acquire` and
+//! `acquire_fresh` stay in the parent module and call into `download_published`.
+//! These methods reach the parent's private fields (`download_locks`) and
+//! private methods (`try_lock_repo`) directly, because a child module can see
+//! its parent's private items.
+
+use super::*;
+
+/// Outcome of a coordinated read-path download (KTD-3).
+pub(super) enum DownloadOutcome {
+    /// The local dir is present: this reader downloaded and published it under
+    /// the advisory lock, or a concurrent reader did and this one served it.
+    Published,
+    /// The download was discarded without publishing: the advisory lock was
+    /// contended (a live writer or purge holds it), the lock pool errored, or
+    /// the archive vanished under the lock (purged mid-download). The caller
+    /// degrades to its missing-path or serve-local-copy outcome.
+    Skipped,
+}
+
+/// RAII cleanup for a read-path download temp dir (finding 1). Its `Drop`
+/// removes the dir (best effort), so a handler future cancelled mid-download
+/// (axum drops it on client disconnect, the same hazard `RepoWriteGuard` /
+/// `RepoLockGuard` Drop impls exist for) cannot leak a fully-populated repo
+/// dir. `disarm` forgets the dir once the publish rename has consumed it.
+struct TempDownloadDir {
+    path: PathBuf,
+    armed: bool,
+}
+
+impl TempDownloadDir {
+    fn new(path: PathBuf) -> Self {
+        Self { path, armed: true }
+    }
+
+    fn path(&self) -> &Path {
+        &self.path
+    }
+
+    /// Disarm: the temp has been consumed (renamed into place), so `Drop` must
+    /// not try to remove it.
+    fn disarm(mut self) {
+        self.armed = false;
+    }
+}
+
+impl Drop for TempDownloadDir {
+    fn drop(&mut self) {
+        if self.armed {
+            let _ = std::fs::remove_dir_all(&self.path);
+        }
+    }
+}
+
+/// RAII removal of a download-coordination map entry (finding 4). Its `Drop`
+/// removes the entry from `download_locks` whenever the map still holds THIS
+/// Arc (ptr_eq-guarded so a newer entry inserted after an earlier removal is
+/// never clobbered). Removal is guaranteed on every path (a normal return or
+/// `?` drops it in the holder; once a download is in flight it travels with
+/// the spawned task and drops at settle, covering holder timeout and a
+/// handler future cancelled mid cold-read, which axum triggers on client
+/// disconnect), which is what keeps the map from growing per arbitrary
+/// requested name: the pre-fix explicit removals ran only at the labelled
+/// return points, which a cancelled future skipped, leaking the entry. The
+/// outer map is a std::sync::Mutex so
+/// this Drop can prune it synchronously (Drop cannot be async); it is only ever
+/// held briefly for a get/insert/remove, never across an await.
+struct DownloadEntryGuard {
+    locks: Arc<std::sync::Mutex<HashMap<String, Arc<Mutex<()>>>>>,
+    key: String,
+    entry: Arc<Mutex<()>>,
+}
+
+impl Drop for DownloadEntryGuard {
+    fn drop(&mut self) {
+        // A poisoned map is still safe to prune — recover the guard and remove.
+        let mut map = self.locks.lock().unwrap_or_else(|p| p.into_inner());
+        if map
+            .get(&self.key)
+            .is_some_and(|cur| Arc::ptr_eq(cur, &self.entry))
+        {
+            map.remove(&self.key);
+        }
+    }
+}
+
+impl RepoStore {
+    /// Coordinated read-path download (KTD-3). Serializes concurrent readers
+    /// of the same repo on a per-repo async mutex: the first one in becomes
+    /// the holder and runs [`download_and_publish`](Self::download_and_publish);
+    /// a contended reader awaits the holder, re-checks the local dir on wake,
+    /// and serves what the holder published instead of re-downloading. Only
+    /// callers that already confirmed the archive exists reach this, and the
+    /// map entry is removed on completion, so the map cannot grow per
+    /// arbitrary requested name.
+    pub(super) async fn download_published(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+        owner_slug: &str,
+        local_path: &Path,
+        store: &Arc<dyn ObjectStore>,
+        local_existed_at_entry: bool,
+    ) -> Result<DownloadOutcome> {
+        let map_key = format!("{owner_slug}/{repo_name}");
+        let entry = {
+            let mut map = self.download_locks.lock().unwrap();
+            Arc::clone(
+                map.entry(map_key.clone())
+                    .or_insert_with(|| Arc::new(Mutex::new(()))),
+            )
+        };
+        let held = match Arc::clone(&entry).try_lock_owned() {
+            Ok(g) => g,
+            Err(_) => {
+                // Bounded coalescing wait: once a holder's download is in
+                // flight, the spawned task owns this lock (and the map entry)
+                // until it settles, whatever happens to the holder (timeout or
+                // cancellation), so an unbounded lock().await here would park
+                // this reader for as long as a stall lasts. On elapse, report
+                // the same timeout error the holder would. Do NOT touch the
+                // map entry on this path: pruning it while the download is in
+                // flight would let the next arrival start a duplicate download.
+                let waited = tokio::time::timeout(
+                    self.release_upload_timeout,
+                    Arc::clone(&entry).lock_owned(),
+                )
+                .await;
+                let g = match waited {
+                    Ok(g) => g,
+                    Err(_) => {
+                        if local_path.exists() {
+                            // A concurrent reader published while we waited.
+                            return Ok(DownloadOutcome::Published);
+                        }
+                        return Err(anyhow::anyhow!("read download timed out"));
+                    }
+                };
+                if local_path.exists() {
+                    // A concurrent reader published while we waited — serve it.
+                    // The holder's own guard prunes the map entry on its exit.
+                    return Ok(DownloadOutcome::Published);
+                }
+                // The holder degraded without publishing; this reader takes over.
+                // The prior holder removed the map entry on its way out, so
+                // re-register THIS entry before downloading, so readers arriving
+                // now coalesce onto it instead of starting a duplicate download
+                // (finding 3). Narrows the duplicate-download window; publishes
+                // stay serialized by the advisory lock regardless.
+                {
+                    let mut map = self.download_locks.lock().unwrap();
+                    map.entry(map_key.clone())
+                        .or_insert_with(|| Arc::clone(&entry));
+                }
+                g
+            }
+        };
+        // RAII: remove the map entry when this holder's work is done (finding
+        // 4). Created only once THIS reader holds the per-repo lock, so a
+        // parked or timed-out waiter never prunes the entry out from under a
+        // live holder; its Drop is ptr_eq-guarded, so it never clobbers a
+        // newer entry. Ownership moves into download_and_publish and from
+        // there into the spawned download task itself, so on EVERY settle path
+        // (success, download error, holder timeout, holder cancellation) the
+        // entry stays registered and the lock stays held until the in-flight
+        // download settles: waiters queue on the ONE download instead of each
+        // timeout or cancel wave starting a fresh full download of the same
+        // object.
+        let entry_guard = DownloadEntryGuard {
+            locks: Arc::clone(&self.download_locks),
+            key: map_key,
+            entry: Arc::clone(&entry),
+        };
+        self.download_and_publish(
+            owner_did,
+            repo_name,
+            owner_slug,
+            local_path,
+            store,
+            local_existed_at_entry,
+            entry_guard,
+            held,
+        )
+        .await
+    }
+
+    /// The holder's half of the coordinated download: fetch and extract the
+    /// archive into a temp sibling with NO advisory lock held (the network
+    /// phase must never pin a lock-pool connection — a cold-read burst across
+    /// distinct repos must not drain the writer-sized pool), then take the
+    /// per-repo advisory lock only around the publish: re-check the archive
+    /// still exists under the lock (a purge deletes it under this same lock,
+    /// so a post-purge downloader discards rather than resurrecting), swap
+    /// the temp copy into place, release. Advisory contention or a lock-pool
+    /// error discards the temp copy and degrades — no blocking lock or pool
+    /// wait exists anywhere on the read path.
+    ///
+    /// `local_existed_at_entry` selects the stale-swap guard (finding 2): a cold
+    /// `acquire` passes `false`, an `acquire_fresh` refresh passes `true`. Under
+    /// the lock, before the swap, we detect a writer that published a fresher
+    /// copy during our unlocked download window and serve theirs instead of
+    /// clobbering it with our now-stale temp.
+    ///
+    /// `entry_guard` and `held` are this holder's per-repo coordination (the
+    /// download_locks map entry and the locked per-repo mutex). Both move into
+    /// the spawned download task, which returns them through its JoinHandle on
+    /// success; on any other settle path (download error, task panic, holder
+    /// timeout, holder cancellation) they drop when the task settles, so
+    /// waiters keep queueing on the SAME in-flight download until it settles
+    /// rather than each starting a duplicate one.
+    #[allow(clippy::too_many_arguments)]
+    async fn download_and_publish(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+        owner_slug: &str,
+        local_path: &Path,
+        store: &Arc<dyn ObjectStore>,
+        local_existed_at_entry: bool,
+        entry_guard: DownloadEntryGuard,
+        held: tokio::sync::OwnedMutexGuard<()>,
+    ) -> Result<DownloadOutcome> {
+        let parent = local_path.parent().context("repo path has no parent")?;
+        std::fs::create_dir_all(parent).context("creating repo parent dir")?;
+        let file_name = local_path
+            .file_name()
+            .context("repo path has no file name")?
+            .to_string_lossy();
+
+        // Best-effort sweep of leftover temp siblings from a prior download whose
+        // cleanup never ran (a process crash or kill mid-download, where no Drop
+        // and no owning task survive to remove the temp). Serialized per
+        // repo by the download mutex, so no concurrent same-repo download owns a
+        // live temp here; scoped to THIS repo's prefix, so other repos are
+        // untouched.
+        let tmp_prefix = format!(".{file_name}.tmp-download.");
+        if let Ok(entries) = std::fs::read_dir(parent) {
+            for entry in entries.flatten() {
+                if entry.file_name().to_string_lossy().starts_with(&tmp_prefix) {
+                    let _ = std::fs::remove_dir_all(entry.path());
+                }
+            }
+        }
+
+        // Stale-swap signal (finding 2): capture the local dir's mtime BEFORE the
+        // unlocked download. Under the lock, a changed mtime (acquire_fresh,
+        // refreshing over an existing copy) or the dir's mere appearance (cold
+        // acquire) means a writer published a fresher copy during our download
+        // window, so we serve theirs rather than swap our stale temp over it.
+        let entry_mtime = std::fs::metadata(local_path)
+            .and_then(|m| m.modified())
+            .ok();
+
+        // Unique per-download temp target (same parent as local_path so the
+        // publish rename stays on one filesystem); mirrors the extract temp
+        // naming in tigris.rs. Wrapped in an RAII guard so a handler future
+        // cancelled mid-download (axum drops it on client disconnect) cannot
+        // leak the populated temp dir (the same hazard the guard Drop impls
+        // below exist for, finding 1).
+        let temp = TempDownloadDir::new(parent.join(format!(
+            ".{file_name}.tmp-download.{}",
+            uuid::Uuid::new_v4()
+        )));
+
+        // Ownership of the download AND the temp-dir guard moves into a spawned
+        // task, so timeout and handler cancellation share ONE cleanup path
+        // (finding F3 and its cancellation twin). `store.download` runs its tar
+        // extraction in `spawn_blocking`, and neither a tokio timeout nor a
+        // dropped handler future cancels a spawn_blocking task: pre-fix,
+        // dropping the download future (a timed-out await, or axum dropping the
+        // whole handler on client disconnect) detached the extraction, whose
+        // later rename resurrected `temp.path()` AFTER the guard's Drop had
+        // removed it, leaving an orphan only a later same-repo download sweeps.
+        // The spawned task is never dropped mid-flight, so the download future
+        // always runs to settle. On success it returns the still-armed guard
+        // through the JoinHandle: the caller consumes it and publishes below.
+        // If the caller timed out or was cancelled, the JoinHandle is gone, so
+        // the runtime drops the returned guard when the task settles and its
+        // Drop removes the temp dir; cleanup is tied to the extraction's
+        // completion, never racing it (this subsumes the previous detached
+        // janitor). A download error drops the guard inside the task with the
+        // same at-settle timing.
+        let store_dl = Arc::clone(store);
+        let (dl_owner, dl_repo, dl_target) = (
+            owner_slug.to_string(),
+            repo_name.to_string(),
+            temp.path().to_path_buf(),
+        );
+        // The per-repo coordination (`held`, `entry_guard`) moves into the task
+        // alongside the temp guard: whatever happens to THIS caller (timeout
+        // below, or a handler future cancelled by axum on client disconnect),
+        // the lock stays held and the map entry stays registered until the
+        // in-flight download settles, so waiters and new arrivals queue on the
+        // ONE download instead of starting duplicates. On success the tuple
+        // returns through the JoinHandle: the caller publishes under `held` as
+        // before. If the caller timed out or was cancelled, the runtime drops
+        // the returned tuple when the task settles, in field order: `temp`
+        // (dir removed), then `held` (lock released), then `entry_guard` (map
+        // entry pruned), the same unlock-before-prune order as a normal
+        // return.
+        let dl_task = tokio::spawn(async move {
+            // The guards travel as one `(held, entry_guard)` tuple so that
+            // wherever it drops (here on error, in the runtime at settle after
+            // a caller timeout or cancellation, or in the caller after a
+            // normal publish), its field order gives unlock-before-prune.
+            match store_dl.download(&dl_owner, &dl_repo, &dl_target).await {
+                Ok(()) => Ok((temp, (held, entry_guard))),
+                // The download settled, so cleanup is safe. Explicit drops pin
+                // the order (async-block captures have no guaranteed one):
+                // temp removed, lock released, map entry pruned last.
+                Err(e) => {
+                    drop(temp);
+                    drop(held);
+                    drop(entry_guard);
+                    Err(e)
+                }
+            }
+        });
+        // Bound the wait, not the work: a stalled GET would park this reader
+        // indefinitely. On timeout the caller gets the same error as before,
+        // but the coordination does NOT unwind: it lives in the task (see
+        // above) and is released only at settle. Waiters therefore stay queued
+        // on the SAME in-flight download (each bounded by its own timeout in
+        // download_published) instead of each timeout wave starting a fresh
+        // full download of the same object; the task keeps running and cleans
+        // up its temp when the download settles.
+        // `_coordination` is `(held, entry_guard)`, kept alive to the end of
+        // this function exactly as the separate guards were before: the
+        // publish below still runs with the per-repo lock held, and the map
+        // entry is pruned only on exit (unlock first, prune second, per the
+        // tuple's field order).
+        let (temp, _coordination) =
+            match tokio::time::timeout(self.release_upload_timeout, dl_task).await {
+                Ok(Ok(Ok(returned))) => returned,
+                Ok(Ok(Err(e))) => return Err(e),
+                // The task panicked; its unwind dropped the guards, removing
+                // the dir, releasing the lock, and pruning the map entry.
+                Ok(Err(join_err)) => {
+                    return Err(anyhow::Error::from(join_err).context("read download task failed"))
+                }
+                Err(_) => {
+                    warn!(repo = %repo_name, timeout_secs = self.release_upload_timeout.as_secs(),
+                        "read download timed out — discarding");
+                    // Dropping the JoinHandle detaches the task; the runtime
+                    // frees temp dir, lock, and map entry at settle (see the
+                    // spawn comment above).
+                    return Err(anyhow::anyhow!("read download timed out"));
+                }
+            };
+
+        let guard = match self.try_lock_repo(owner_did, repo_name).await {
+            Ok(Some(g)) => g,
+            Ok(None) => {
+                debug!(repo = %repo_name,
+                    "read download discarded — repo lock held by a live writer or purge");
+                return Ok(DownloadOutcome::Skipped);
+            }
+            Err(e) => {
+                warn!(repo = %repo_name, err = %e,
+                    "read download discarded — could not acquire repo lock");
+                return Ok(DownloadOutcome::Skipped);
+            }
+        };
+        // Re-check under the lock: the archive gone here means a purge won the
+        // key mid-download — discard, never publish (the read-side twin of the
+        // upload paths' under-lock dir re-check). Bounded so a stalled HEAD
+        // cannot pin the advisory lock + its lock-pool connection. Both a
+        // timeout and an exists() error collapse to "not present" and take the
+        // discard arm: fail closed rather than publish unconfirmed state.
+        let archive_present = matches!(
+            tokio::time::timeout(
+                self.release_upload_timeout,
+                store.exists(owner_slug, repo_name),
+            )
+            .await,
+            Ok(Ok(true))
+        );
+        if !archive_present {
+            warn!(repo = %repo_name,
+                "read download discarded — archive gone under lock (purged?)");
+            guard.release().await;
+            return Ok(DownloadOutcome::Skipped);
+        }
+
+        // Stale-swap guard (finding 2): a writer that locked, published a fresher
+        // copy, and released during our unlocked download window must not have
+        // its work clobbered by our now-stale temp. The only concurrent mutator
+        // of local_path is a writer or a purge (readers of the same repo
+        // serialize on the download mutex), so the signals below unambiguously
+        // mean "a writer published here."
+        let writer_published = match (local_existed_at_entry, entry_mtime) {
+            // acquire_fresh over an existing copy: presence is always true, so it
+            // cannot signal a writer; a CHANGED mtime can (the writer replaced
+            // the dir via rename, minting a fresh mtime).
+            (true, Some(entry_m)) => std::fs::metadata(local_path)
+                .and_then(|m| m.modified())
+                .map(|now| now != entry_m)
+                .unwrap_or(false),
+            // Cold acquire, or acquire_fresh whose dir was absent at entry: the
+            // dir being present now means a writer published it during the window.
+            _ => local_path.exists(),
+        };
+        if writer_published {
+            warn!(repo = %repo_name,
+                "read download discarded: a writer published a fresher copy during the download window");
+            guard.release().await;
+            // temp guard's Drop removes the stale temp; serve the fresher local copy.
+            return Ok(DownloadOutcome::Published);
+        }
+
+        let swapped = (|| -> std::io::Result<()> {
+            if local_path.exists() {
+                std::fs::remove_dir_all(local_path)?;
+            }
+            std::fs::rename(temp.path(), local_path)
+        })();
+        guard.release().await;
+        match swapped {
+            Ok(()) => {
+                temp.disarm(); // the rename consumed the temp; nothing to clean up
+                Ok(DownloadOutcome::Published)
+            }
+            // temp guard's Drop removes the temp if the rename left it behind.
+            Err(e) => Err(e).context("publishing downloaded repo into place"),
+        }
+    }
+}

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

@@ -63,6 +63,105 @@ pub fn list_refs(repo_path: &Path) -> Result<Vec<(String, String)>> {
     Ok(refs)
 }
 
+/// Read the object id a single ref currently points to. Returns None if the
+/// ref does not exist. Used to snapshot a ref's pre-write state so a failed
+/// durable upload can roll the local write back.
+pub fn ref_oid(repo_path: &Path, ref_name: &str) -> Result<Option<String>> {
+    let output = Command::new("git")
+        .args(["rev-parse", "--verify", ref_name])
+        .current_dir(repo_path)
+        .output()
+        .context("failed to run git rev-parse")?;
+    if !output.status.success() {
+        return Ok(None);
+    }
+    let oid = String::from_utf8_lossy(&output.stdout).trim().to_string();
+    if oid.is_empty() {
+        return Ok(None);
+    }
+    Ok(Some(oid))
+}
+
+/// Force a single ref back to a captured object id — rolls back a local write
+/// (e.g. a merge commit) whose durable upload failed, so a local-fast-path
+/// read does not serve the un-uploaded state. The now-dangling objects are
+/// harmless.
+pub fn set_ref(repo_path: &Path, ref_name: &str, oid: &str) -> Result<()> {
+    let output = Command::new("git")
+        .args(["update-ref", ref_name, oid])
+        .current_dir(repo_path)
+        .output()
+        .context("failed to run git update-ref")?;
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        bail!("git update-ref failed: {stderr}");
+    }
+    Ok(())
+}
+
+/// Restore a repo's refs to a previously captured snapshot: reset every
+/// snapshot ref to its captured object id, and delete any ref that exists now
+/// but was absent from the snapshot (created by the intervening write). Used
+/// to roll back a receive-pack whose durable upload failed, so the next
+/// local-fast-path read does not serve refs that never reached object storage.
+/// Best-effort per ref: an individual failure is collected and reported, but
+/// the remaining refs are still attempted.
+pub fn restore_refs(repo_path: &Path, snapshot: &[(String, String)]) -> Result<()> {
+    use std::collections::HashSet;
+
+    let mut errors: Vec<String> = Vec::new();
+
+    // Reset every snapshot ref to its captured oid (recreates refs the write
+    // deleted, and rewinds refs the write advanced).
+    for (ref_name, oid) in snapshot {
+        if let Err(e) = set_ref(repo_path, ref_name, oid) {
+            errors.push(format!("{ref_name}: {e}"));
+        }
+    }
+
+    // Delete refs that exist now but were not in the snapshot (created by the
+    // write being rolled back). A failure to LIST the current refs is collected
+    // like any per-ref failure, never returned early: the snapshot-reset half
+    // above already ran, and bailing here would discard its collected errors.
+    let snapshot_names: HashSet<&str> = snapshot.iter().map(|(r, _)| r.as_str()).collect();
+    match list_refs(repo_path) {
+        Ok(current) => {
+            for (ref_name, _) in &current {
+                if !snapshot_names.contains(ref_name.as_str()) {
+                    match Command::new("git")
+                        .args(["update-ref", "-d", ref_name])
+                        .current_dir(repo_path)
+                        .output()
+                    {
+                        Ok(output) if output.status.success() => {}
+                        Ok(output) => {
+                            let stderr = String::from_utf8_lossy(&output.stderr);
+                            errors.push(format!("{ref_name} (delete): {stderr}"));
+                        }
+                        Err(e) => {
+                            errors.push(format!(
+                                "{ref_name} (delete): failed to run git update-ref -d: {e}"
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+        Err(e) => {
+            errors.push(format!("(extra-ref sweep) listing current refs: {e}"));
+        }
+    }
+
+    if !errors.is_empty() {
+        bail!(
+            "failed to restore {} ref(s): {}",
+            errors.len(),
+            errors.join("; ")
+        );
+    }
+    Ok(())
+}
+
 /// Read the current HEAD commit hash of a repository.
 /// Returns None if the repo is empty (no commits yet).
 pub fn head_commit(repo_path: &Path) -> Result<Option<String>> {
@@ -455,10 +554,176 @@ pub fn repo_disk_path(repos_dir: &Path, owner_did: &str, repo_name: &str) -> Pat
 
 #[cfg(test)]
 mod tests {
-    use super::branch_diff_names;
+    use super::{branch_diff_names, list_refs, restore_refs};
     use std::path::Path;
     use std::process::Command;
 
+    fn run_git(dir: &Path, args: &[&str]) -> String {
+        let out = Command::new("git")
+            .args(args)
+            .current_dir(dir)
+            .output()
+            .unwrap();
+        assert!(
+            out.status.success(),
+            "git {args:?} failed: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+        String::from_utf8_lossy(&out.stdout).trim().to_string()
+    }
+
+    /// Bare repo with main at c2 and keeper at c2, whose "pre-push" state was
+    /// main and keeper both at c1. Returns (workdir, baredir, bare_path, c1, c2).
+    fn advanced_bare_repo() -> (
+        tempfile::TempDir,
+        tempfile::TempDir,
+        std::path::PathBuf,
+        String,
+        String,
+    ) {
+        let work = tempfile::TempDir::new().unwrap();
+        run_git(work.path(), &["init", "-q", "-b", "main", "."]);
+        run_git(work.path(), &["config", "user.email", "t@t"]);
+        run_git(work.path(), &["config", "user.name", "t"]);
+        std::fs::write(work.path().join("f.txt"), "one").unwrap();
+        run_git(work.path(), &["add", "f.txt"]);
+        run_git(work.path(), &["commit", "-qm", "c1"]);
+        let c1 = run_git(work.path(), &["rev-parse", "HEAD"]);
+        std::fs::write(work.path().join("f.txt"), "two").unwrap();
+        run_git(work.path(), &["add", "f.txt"]);
+        run_git(work.path(), &["commit", "-qm", "c2"]);
+        let c2 = run_git(work.path(), &["rev-parse", "HEAD"]);
+
+        let dir = tempfile::TempDir::new().unwrap();
+        let bare = dir.path().join("repo.git");
+        let out = Command::new("git")
+            .args([
+                "clone",
+                "--bare",
+                "-q",
+                &work.path().to_string_lossy(),
+                &bare.to_string_lossy(),
+            ])
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "clone --bare failed");
+        // "The push" advanced main to c2 (already there from the clone), moved
+        // keeper c1 -> c2, and created feature at c2.
+        run_git(&bare, &["update-ref", "refs/heads/keeper", &c2]);
+        run_git(&bare, &["update-ref", "refs/heads/feature", &c2]);
+        (work, dir, bare, c1, c2)
+    }
+
+    fn sorted_refs(bare: &Path) -> Vec<(String, String)> {
+        let mut refs = list_refs(bare).unwrap();
+        refs.sort();
+        refs
+    }
+
+    /// A single unrestorable snapshot entry (invalid oid, `set_ref` fails) must
+    /// not abort the restore: every OTHER snapshot ref is still reset and the
+    /// push-created extra ref is still deleted, and the fn returns Err carrying
+    /// the failure. RED with the reset loop reverted to `set_ref(..)?`: the
+    /// broken entry is first in the snapshot, so an early return leaves keeper
+    /// at c2 and feature alive.
+    #[test]
+    fn restore_refs_partial_failure_still_resets_rest_and_deletes_extras() {
+        let (_work, _dir, bare, c1, c2) = advanced_bare_repo();
+
+        let snapshot = vec![
+            (
+                "refs/heads/broken".to_string(),
+                "not-a-valid-oid".to_string(),
+            ),
+            ("refs/heads/main".to_string(), c1.clone()),
+            ("refs/heads/keeper".to_string(), c1.clone()),
+        ];
+        let err = restore_refs(&bare, &snapshot).unwrap_err();
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("refs/heads/broken"),
+            "the aggregated error names the failed ref: {msg}"
+        );
+
+        let refs = sorted_refs(&bare);
+        assert_eq!(
+            refs,
+            vec![
+                ("refs/heads/keeper".to_string(), c1.clone()),
+                ("refs/heads/main".to_string(), c1.clone()),
+            ],
+            "keeper and main are reset to c1 and feature is deleted despite the \
+             broken entry (c2 was {c2})"
+        );
+    }
+
+    /// Happy path unchanged by the hardening: a fully valid snapshot restores
+    /// exactly (refs rewound, created ref deleted) and returns Ok.
+    #[test]
+    fn restore_refs_happy_path_restores_snapshot_exactly() {
+        let (_work, _dir, bare, c1, _c2) = advanced_bare_repo();
+
+        let snapshot = vec![
+            ("refs/heads/main".to_string(), c1.clone()),
+            ("refs/heads/keeper".to_string(), c1.clone()),
+        ];
+        restore_refs(&bare, &snapshot).unwrap();
+
+        let refs = sorted_refs(&bare);
+        assert_eq!(
+            refs,
+            vec![
+                ("refs/heads/keeper".to_string(), c1.clone()),
+                ("refs/heads/main".to_string(), c1),
+            ],
+            "restore must reproduce the snapshot exactly"
+        );
+    }
+
+    /// When the internal `list_refs` (extra-ref sweep) fails, the snapshot-reset
+    /// half must still have run and its collected failures must surface in the
+    /// aggregated error, not be discarded by an early `?` return. The repo is a
+    /// real bare repo declaring repositoryformatversion=999, so both `set_ref`
+    /// and `list_refs` fail deterministically. RED with the sweep reverted to
+    /// `let current = list_refs(repo_path)?;`: the raw for-each-ref error
+    /// propagates without the "failed to restore" aggregate or the ref name.
+    #[test]
+    fn restore_refs_aggregates_when_internal_list_refs_fails() {
+        let dir = tempfile::TempDir::new().unwrap();
+        let bare = dir.path().join("repo.git");
+        let out = Command::new("git")
+            .args(["init", "--bare", "-q", &bare.to_string_lossy()])
+            .output()
+            .unwrap();
+        assert!(out.status.success(), "git init --bare failed");
+        std::fs::write(
+            bare.join("config"),
+            "[core]\n\trepositoryformatversion = 999\n\tbare = true\n",
+        )
+        .unwrap();
+        assert!(
+            list_refs(&bare).is_err(),
+            "precondition: list_refs must fail on the corrupted repo"
+        );
+
+        let snapshot = vec![("refs/heads/main".to_string(), "a".repeat(40))];
+        let err = restore_refs(&bare, &snapshot).unwrap_err();
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("failed to restore"),
+            "must return the aggregated error, not the raw list_refs error: {msg}"
+        );
+        assert!(
+            msg.contains("refs/heads/main"),
+            "the set_ref failure from the reset half must survive the sweep \
+             failure: {msg}"
+        );
+        assert!(
+            msg.contains("listing current refs"),
+            "the sweep failure itself is also collected: {msg}"
+        );
+    }
+
     #[test]
     fn branch_diff_names_lists_changed_paths() {
         let td = tempfile::TempDir::new().unwrap();

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

@@ -11,6 +11,21 @@ use anyhow::{Context, Result};
 use aws_sdk_s3::Client as S3Client;
 use tracing::{debug, info};
 
+/// Object storage for git bare-repo archives. The seam that lets `RepoStore`
+/// (and tests) depend on storage behavior without a live bucket. `TigrisClient`
+/// is the production implementation.
+#[async_trait::async_trait]
+pub trait ObjectStore: Send + Sync {
+    /// Whether an archive exists for this repo.
+    async fn exists(&self, owner_slug: &str, repo_name: &str) -> Result<bool>;
+    /// Upload a local bare repo directory as an archive.
+    async fn upload(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()>;
+    /// Download and extract a repo archive to local disk.
+    async fn download(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()>;
+    /// Delete a repo archive.
+    async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()>;
+}
+
 /// Wrapper around the S3 client with the configured bucket.
 #[derive(Clone)]
 pub struct TigrisClient {
@@ -35,9 +50,12 @@ impl TigrisClient {
     fn repo_key(owner_slug: &str, repo_name: &str) -> String {
         format!("repos/v1/{owner_slug}/{repo_name}.tar.zst")
     }
+}
 
+#[async_trait::async_trait]
+impl ObjectStore for TigrisClient {
     /// Check if a repo archive exists in Tigris.
-    pub async fn exists(&self, owner_slug: &str, repo_name: &str) -> Result<bool> {
+    async fn exists(&self, owner_slug: &str, repo_name: &str) -> Result<bool> {
         let key = Self::repo_key(owner_slug, repo_name);
         match self
             .s3
@@ -59,7 +77,7 @@ 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<()> {
+    async fn upload(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()> {
         let key = Self::repo_key(owner_slug, repo_name);
         debug!(key = %key, path = %local_path.display(), "uploading repo to tigris");
 
@@ -89,12 +107,7 @@ impl TigrisClient {
     }
 
     /// Download a repo archive from Tigris and extract to local disk.
-    pub async fn download(
-        &self,
-        owner_slug: &str,
-        repo_name: &str,
-        local_path: &Path,
-    ) -> Result<()> {
+    async fn download(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()> {
         let key = Self::repo_key(owner_slug, repo_name);
         debug!(key = %key, path = %local_path.display(), "downloading repo from tigris");
 
@@ -128,8 +141,7 @@ impl TigrisClient {
     }
 
     /// Delete a repo archive from Tigris.
-    #[allow(dead_code)]
-    pub async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()> {
+    async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()> {
         let key = Self::repo_key(owner_slug, repo_name);
         self.s3
             .delete_object()

crates/gitlawb-node/src/main.rs

@@ -1,3 +1,4 @@
+mod admin;
 mod api;
 mod arweave;
 mod auth;
@@ -70,6 +71,12 @@ async fn main() -> Result<()> {
 
     let mut config = Config::parse();
 
+    // Admin subcommands run out-of-band and exit, never starting the node. The
+    // no-subcommand path below is the unchanged daemon startup.
+    if let Some(command) = config.command.clone() {
+        return run_admin_command(command, &config).await;
+    }
+
     // Merge the embedded seed list of public network nodes into the runtime
     // bootstrap peers. Operators can opt out via GITLAWB_BOOTSTRAP_DISABLE_SEEDS.
     bootstrap::merge_seeds(&mut config);
@@ -279,8 +286,22 @@ async fn main() -> Result<()> {
         None
     };
 
+    let object_store =
+        tigris.map(|t| std::sync::Arc::new(t) as std::sync::Arc<dyn git::tigris::ObjectStore>);
+    // Advisory-lock guards get their own pool, separate from the app pool
+    // (db.pool()) that serves normal handlers, so a concurrent-push burst pins
+    // lock connections here instead of starving request handlers. Sized to peak
+    // writers via GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS.
+    let lock_pool_acquire_timeout = std::time::Duration::from_secs(config.db_acquire_timeout_secs);
+    let lock_pool = Db::connect_lock_pool(
+        &config.database_url,
+        config.db_lock_pool_max_connections,
+        lock_pool_acquire_timeout,
+    )
+    .await
+    .context("connecting advisory-lock pool")?;
     let repo_store =
-        git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, db.pool().clone());
+        git::repo_store::RepoStore::new(config.repos_dir.clone(), object_store, lock_pool);
 
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
     // DID (attacker-varied), so bound its key set to cap memory.
@@ -294,36 +315,24 @@ async fn main() -> Result<()> {
     // capped regardless of how many identities it mints. Sized well above any
     // legitimate per-IP creation rate; GITLAWB_CREATE_RATE_LIMIT overrides, 0
     // disables. Bounded key set — the key is a client-influenced IP.
-    let create_limit = std::env::var("GITLAWB_CREATE_RATE_LIMIT")
-        .ok()
-        .and_then(|v| v.trim().parse::<usize>().ok())
-        .unwrap_or(120);
-    let create_ip_rate_limiter = rate_limit::RateLimiter::new_bounded(
-        create_limit,
-        std::time::Duration::from_secs(3600),
-        200_000,
-    );
-    if create_limit == 0 {
-        tracing::warn!("GITLAWB_CREATE_RATE_LIMIT=0 — per-IP creation rate limiting disabled");
-    }
+    let create_ip_rate_limiter = build_ip_limiter("GITLAWB_CREATE_RATE_LIMIT", 120, "creation");
+
+    // Per-IP brake for the authenticated non-creation write surface (issue/PR
+    // comments, labels, stars, merges, protect, replicas, visibility, tasks,
+    // bounties, profile, GraphQL mutations). Its own bucket, separate from the
+    // creation brake, so a write flood cannot drain the creation budget and vice
+    // versa (same rationale as the sync_trigger / peer_write split). Sized above
+    // any legitimate per-IP write rate — real agent automation makes many small
+    // writes per hour. GITLAWB_WRITE_RATE_LIMIT overrides; 0 disables. Bounded
+    // key set — the key is a client-influenced IP.
+    let write_rate_limiter = build_ip_limiter("GITLAWB_WRITE_RATE_LIMIT", 600, "write");
 
     // Push-path flood brake: max git-receive-pack requests per client IP per
     // hour (counts both the info/refs advertisement and the push POST). Sized
     // for heavy agent automation while still stopping flood traffic (the June
     // 2026 attack pushed several times per second per IP). GITLAWB_PUSH_RATE_LIMIT
     // overrides; 0 disables. Bounded key set — the key is a client-influenced IP.
-    let push_limit = std::env::var("GITLAWB_PUSH_RATE_LIMIT")
-        .ok()
-        .and_then(|v| v.trim().parse::<usize>().ok())
-        .unwrap_or(600);
-    let push_rate_limiter = rate_limit::RateLimiter::new_bounded(
-        push_limit,
-        std::time::Duration::from_secs(3600),
-        200_000,
-    );
-    if push_limit == 0 {
-        tracing::warn!("GITLAWB_PUSH_RATE_LIMIT=0 — per-IP push rate limiting disabled");
-    }
+    let push_rate_limiter = build_ip_limiter("GITLAWB_PUSH_RATE_LIMIT", 600, "push");
 
     // 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;
@@ -331,7 +340,7 @@ async fn main() -> Result<()> {
     let push_limiter_trust = rate_limit::TrustedProxy::from_env_value(
         &std::env::var("GITLAWB_TRUSTED_PROXY").unwrap_or_default(),
     );
-    tracing::info!(trust = ?push_limiter_trust, push_limit, "push rate limiter configured");
+    tracing::info!(trust = ?push_limiter_trust, "push rate limiter trust configured");
 
     // Peer-sync flood brakes, keyed on the resolved client IP (per-DID is useless
     // here — a did:key farm self-registers). Two buckets so an unsigned notify
@@ -373,6 +382,7 @@ async fn main() -> Result<()> {
         repo_store,
         rate_limiter,
         create_ip_rate_limiter,
+        write_rate_limiter,
         push_rate_limiter,
         push_limiter_trust,
         sync_trigger_rate_limiter,
@@ -408,22 +418,17 @@ async fn main() -> Result<()> {
 
     // Periodic cleanup of expired rate limit entries + consumed-proof ledger
     {
-        let rl = state.rate_limiter.clone();
-        let create_ip_rl = state.create_ip_rate_limiter.clone();
-        let push_rl = state.push_rate_limiter.clone();
-        let sync_trigger_rl = state.sync_trigger_rate_limiter.clone();
-        let peer_write_rl = state.peer_write_rate_limiter.clone();
+        // Clone the whole state so the sweep uses AppState::cleanup_rate_limiters
+        // (the single source of truth that reaps EVERY limiter, including
+        // write_rate_limiter) rather than a hand-maintained list that can omit one.
+        let 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)) => {
-                        rl.cleanup().await;
-                        create_ip_rl.cleanup().await;
-                        push_rl.cleanup().await;
-                        sync_trigger_rl.cleanup().await;
-                        peer_write_rl.cleanup().await;
+                        state.cleanup_rate_limiters().await;
                         let now = std::time::SystemTime::now()
                             .duration_since(std::time::UNIX_EPOCH)
                             .map(|d| d.as_secs() as i64)
@@ -564,6 +569,110 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+/// Pure resolver for a usize rate-limit env value. Returns `(resolved, invalid)`:
+/// an absent or empty value resolves to `default` and is NOT invalid; a non-empty
+/// value that fails to parse resolves to `default` and IS invalid (so the caller
+/// warns rather than silently disabling the intended, usually stricter, brake).
+fn resolve_rate_limit(raw: Option<&str>, default: usize) -> (usize, bool) {
+    match raw.map(str::trim) {
+        None | Some("") => (default, false),
+        Some(t) => match t.parse::<usize>() {
+            Ok(n) => (n, false),
+            Err(_) => (default, true),
+        },
+    }
+}
+
+/// Read a rate-limit env var, warning on a present-but-unparseable value. A typo
+/// like `GITLAWB_WRITE_RATE_LIMIT=5O` must not silently revert to the default and
+/// leave the operator believing a stricter cap is in force.
+fn rate_limit_from_env(var: &str, default: usize) -> usize {
+    let raw = std::env::var(var).ok();
+    let (value, invalid) = resolve_rate_limit(raw.as_deref(), default);
+    if invalid {
+        tracing::warn!(
+            var,
+            value = raw.as_deref().unwrap_or(""),
+            default,
+            "invalid rate-limit value — using default; the intended brake is NOT applied"
+        );
+    }
+    value
+}
+
+/// Build a bounded per-client-IP rate limiter from an env var: resolve the limit
+/// (honoring the present-but-unparseable warning in `rate_limit_from_env`), build
+/// the limiter with the shared 1-hour window and 200k-key cap, and warn when the
+/// limit is 0 (disabled). Collapses the identical create/write/push setup.
+fn build_ip_limiter(var: &str, default: usize, label: &str) -> rate_limit::RateLimiter {
+    let limit = rate_limit_from_env(var, default);
+    let limiter =
+        rate_limit::RateLimiter::new_bounded(limit, std::time::Duration::from_secs(3600), 200_000);
+    if limit == 0 {
+        tracing::warn!("{var}=0 — per-IP {label} rate limiting disabled");
+    } else {
+        tracing::info!(%var, limit, "per-IP {label} rate limiter configured");
+    }
+    limiter
+}
+
+/// Dispatch an admin subcommand. Connects the database directly (no server, no
+/// p2p, no metrics) and runs the requested tool to completion, then exits.
+async fn run_admin_command(command: config::Command, config: &Config) -> Result<()> {
+    match command {
+        config::Command::PurgeSpam { execute } => {
+            let acquire_timeout = std::time::Duration::from_secs(config.db_acquire_timeout_secs);
+            let db = Db::connect(
+                &config.database_url,
+                config.db_max_connections,
+                acquire_timeout,
+            )
+            .await
+            .context("connecting to database for purge-spam")?;
+            // Build the object store so purge is authoritative against remote
+            // state: the emptiness recheck consults the archive (not just
+            // possibly-stale local disk) and a successful purge deletes the
+            // archive too. When no bucket is configured this stays None and purge
+            // is local-only, exactly as before.
+            let object_store: Option<std::sync::Arc<dyn git::tigris::ObjectStore>> = if config
+                .tigris_bucket
+                .is_empty()
+            {
+                None
+            } else {
+                match git::tigris::TigrisClient::new(&config.tigris_bucket).await {
+                    Ok(client) => Some(std::sync::Arc::new(client)),
+                    Err(e) => {
+                        // Fail closed: without the configured store we cannot
+                        // do the authoritative recheck, and a local-only purge
+                        // on a Tigris deployment is the stale-view delete this
+                        // guards against. Refuse to run rather than fall back.
+                        anyhow::bail!(
+                                "purge-spam: GITLAWB_TIGRIS_BUCKET is set but the object-store client failed to initialize ({e}); refusing to run a local-only purge on a Tigris deployment"
+                            );
+                    }
+                }
+            };
+            // Give the purge-spam store a dedicated advisory-lock pool, separate
+            // from `db`'s app pool. The guard then holds a lock-pool connection
+            // while `delete_repo_by_id` runs on the app pool, so a purge at
+            // GITLAWB_DB_MAX_CONNECTIONS=1 does not self-deadlock (KTD3).
+            let lock_pool = Db::connect_lock_pool(
+                &config.database_url,
+                config.db_lock_pool_max_connections,
+                acquire_timeout,
+            )
+            .await
+            .context("connecting advisory-lock pool for purge-spam")?;
+            let repo_store =
+                git::repo_store::RepoStore::new(config.repos_dir.clone(), object_store, lock_pool);
+            admin::run_purge_spam(&db, &repo_store, &config.repos_dir, execute)
+                .await
+                .map(|_| ())
+        }
+    }
+}
+
 fn spawn_shutdown_signal(tx: watch::Sender<bool>) {
     tokio::spawn(async move {
         #[cfg(unix)]
@@ -1024,6 +1133,40 @@ fn load_or_create_keypair(config: &Config) -> Result<Keypair> {
     }
 }
 
+#[cfg(test)]
+mod rate_limit_env_tests {
+    use super::resolve_rate_limit;
+
+    #[test]
+    fn absent_uses_default_without_warning() {
+        assert_eq!(resolve_rate_limit(None, 600), (600, false));
+    }
+
+    #[test]
+    fn empty_or_whitespace_uses_default_without_warning() {
+        assert_eq!(resolve_rate_limit(Some(""), 600), (600, false));
+        assert_eq!(resolve_rate_limit(Some("   "), 600), (600, false));
+    }
+
+    #[test]
+    fn valid_value_parses() {
+        assert_eq!(resolve_rate_limit(Some("50"), 600), (50, false));
+        assert_eq!(resolve_rate_limit(Some(" 50 "), 600), (50, false));
+        // 0 is a VALID value (disables the brake); the ==0 warning is handled
+        // separately at the call site. It must NOT be flagged invalid here.
+        assert_eq!(resolve_rate_limit(Some("0"), 600), (0, false));
+    }
+
+    #[test]
+    fn present_but_unparseable_defaults_and_flags_invalid() {
+        // The L8 case: a fat-fingered stricter cap ("5O", letter O) must default
+        // AND be flagged so the caller warns, not silently disable the brake.
+        assert_eq!(resolve_rate_limit(Some("5O"), 600), (600, true));
+        assert_eq!(resolve_rate_limit(Some("abc"), 600), (600, true));
+        assert_eq!(resolve_rate_limit(Some("-1"), 600), (600, true));
+    }
+}
+
 #[cfg(test)]
 mod gossip_ssrf_tests {
     use super::ping_peer_health;

crates/gitlawb-node/src/rate_limit.rs

@@ -70,10 +70,23 @@ impl RateLimiter {
         self.window.min(Duration::from_secs(1))
     }
 
+    /// Returns `true` if the request is allowed. Thin wrapper over
+    /// [`check_retry`](Self::check_retry) for callers that only need the
+    /// allow/deny decision, not the rejection's Retry-After delay. Test-only now:
+    /// production 429 sites use `check_retry` so they can advertise Retry-After.
+    #[cfg(test)]
     pub(crate) async fn check(&self, key: &str) -> bool {
+        self.check_retry(key).await.is_none()
+    }
+
+    /// Record a request against `key`. Returns `None` when it is allowed, or
+    /// `Some(retry_after)` when it is rejected — the delay a client should wait
+    /// before retrying, so a 429 can advertise a value consistent with the
+    /// actual window instead of a constant.
+    pub(crate) async fn check_retry(&self, key: &str) -> Option<Duration> {
         // max_requests == 0 means the limiter is disabled, not "block all".
         if self.max_requests == 0 {
-            return true;
+            return None;
         }
         let now = Instant::now();
         let mut state = self.state.lock().await;
@@ -85,10 +98,25 @@ impl RateLimiter {
                 .timestamps
                 .retain(|t| now.duration_since(*t) < self.window);
             if window.timestamps.len() >= self.max_requests {
-                return false;
+                // Insertion order is preserved, so after the retain the first
+                // element is the oldest live request. It ages out (freeing a
+                // slot) one window after it landed, so that remaining time is
+                // the delay to advertise. Clamp to >= 1 so a 429 never tells a
+                // client to retry immediately.
+                let retry = window
+                    .timestamps
+                    .first()
+                    .map(|oldest| {
+                        self.window
+                            .as_secs()
+                            .saturating_sub(now.duration_since(*oldest).as_secs())
+                    })
+                    .unwrap_or(0)
+                    .max(1);
+                return Some(Duration::from_secs(retry));
             }
             window.timestamps.push(now);
-            return true;
+            return None;
         }
 
         // New key. Enforce the cap BEFORE inserting so a flood of distinct keys
@@ -109,7 +137,11 @@ impl RateLimiter {
             }
             drop(last_sweep);
             if state.len() >= self.max_keys {
-                return false;
+                // The key is untracked (never inserted), so there is no oldest
+                // entry to derive a delay from. Advertise the full window as a
+                // conservative bound: capacity frees as other keys expire, at
+                // the latest one window from now.
+                return Some(self.window.max(Duration::from_secs(1)));
             }
         }
         state.insert(
@@ -118,7 +150,7 @@ impl RateLimiter {
                 timestamps: vec![now],
             },
         );
-        true
+        None
     }
 
     pub async fn cleanup(&self) {
@@ -130,6 +162,41 @@ impl RateLimiter {
             !w.timestamps.is_empty()
         });
     }
+
+    /// Test-only: insert a fully-expired entry (no live timestamps) that the next
+    /// `cleanup()` must reclaim. Models a key whose window has aged out, without
+    /// depending on wall-clock sleeps or a short window. Used cross-module by the
+    /// AppState cleanup guard test (H2).
+    #[cfg(test)]
+    pub(crate) async fn insert_reclaimable_for_test(&self, key: &str) {
+        self.state
+            .lock()
+            .await
+            .insert(key.to_string(), Window { timestamps: vec![] });
+    }
+
+    /// Test-only: number of distinct keys currently tracked.
+    #[cfg(test)]
+    pub(crate) async fn tracked_keys(&self) -> usize {
+        self.state.lock().await.len()
+    }
+
+    /// Test-only: seed a key's bucket with entries of the given ages (how long
+    /// ago each request landed), so a test can drive the Retry-After computation
+    /// off a controlled oldest-entry age without wall-clock sleeps. Ages that
+    /// would underflow the monotonic clock collapse to `now`.
+    #[cfg(test)]
+    pub(crate) async fn seed_aged_entry_for_test(&self, key: &str, ages: &[Duration]) {
+        let now = Instant::now();
+        let timestamps = ages
+            .iter()
+            .map(|a| now.checked_sub(*a).unwrap_or(now))
+            .collect();
+        self.state
+            .lock()
+            .await
+            .insert(key.to_string(), Window { timestamps });
+    }
 }
 
 pub async fn rate_limit_by_did(request: Request, next: Next) -> Response {
@@ -141,13 +208,8 @@ pub async fn rate_limit_by_did(request: Request, next: Next) -> Response {
         .map(|a| a.0.clone());
 
     if let (Some(limiter), Some(did)) = (limiter, did) {
-        if !limiter.check(&did).await {
-            return (
-                StatusCode::TOO_MANY_REQUESTS,
-                [("retry-after", "60")],
-                "rate limit exceeded — try again later",
-            )
-                .into_response();
+        if let Some(retry_after) = limiter.check_retry(&did).await {
+            return too_many_requests(retry_after);
         }
     }
 
@@ -250,13 +312,13 @@ impl<S: Send + Sync> axum::extract::FromRequestParts<S> for PeerAddr {
     }
 }
 
-/// The shared 429 response for the per-IP flood brakes. Route-agnostic: this
-/// middleware now serves the push path AND the peer-sync routes, so the message
-/// stays generic (the offending path is recorded in the warn log below).
-pub fn too_many_requests() -> Response {
+/// The shared 429 response for the rate-limit middlewares (per-DID and per-IP
+/// alike). Route-agnostic: the message stays generic (the offending path, when
+/// logged, is recorded at the call site).
+pub fn too_many_requests(retry_after: Duration) -> Response {
     (
         StatusCode::TOO_MANY_REQUESTS,
-        [("retry-after", "60")],
+        [("retry-after", retry_after.as_secs().to_string())],
         "rate limit exceeded — try again later",
     )
         .into_response()
@@ -274,9 +336,9 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response {
 
     if let Some(limiter) = limiter {
         if let Some(key) = client_key(request.headers(), peer, limiter.trust) {
-            if !limiter.limiter.check(&key).await {
+            if let Some(retry_after) = limiter.limiter.check_retry(&key).await {
                 tracing::warn!(key = %key, path = %request.uri().path(), "per-IP rate limit exceeded");
-                return too_many_requests();
+                return too_many_requests(retry_after);
             }
         }
     }
@@ -605,4 +667,196 @@ mod tests {
             "an exhausted IP must be braked with 429 before auth runs, not leak to 401"
         );
     }
+
+    // ── Adversarial TrustedProxy verification through the middleware (U5) ──
+    // These complement the client_key unit tests above by driving hostile
+    // headers through rate_limit_by_ip on a real router: the property under test
+    // is that the *bucket key* cannot be rotated by a spoofer.
+
+    async fn post_with(
+        router: &axum::Router,
+        peer: SocketAddr,
+        hdrs: &[(&str, &str)],
+    ) -> StatusCode {
+        use tower::ServiceExt;
+        let mut b = axum::http::Request::builder()
+            .method(axum::http::Method::POST)
+            .uri("/api/v1/repos");
+        for (k, v) in hdrs {
+            b = b.header(*k, *v);
+        }
+        let mut req = b.body(axum::body::Body::empty()).unwrap();
+        req.extensions_mut().insert(ConnectInfo(peer));
+        router.clone().oneshot(req).await.unwrap().status()
+    }
+
+    #[tokio::test]
+    async fn xff_spoofed_leftmost_hop_cannot_rotate_bucket() {
+        // XForwardedFor mode, budget 1. A spoofer varies the client-controlled
+        // leftmost hop every request but the trusted proxy always appends the
+        // same rightmost hop. All requests must key to the rightmost hop, so the
+        // second is braked despite a fresh leftmost value.
+        let router = ip_limited_over_auth_router(IpRateLimiter {
+            limiter: RateLimiter::new(1, Duration::from_secs(60)),
+            trust: TrustedProxy::XForwardedFor,
+        });
+        let p1: SocketAddr = "10.0.0.1:1".parse().unwrap();
+        let p2: SocketAddr = "10.0.0.2:1".parse().unwrap();
+        // First request passes the brake, reaches (failing) auth.
+        assert_eq!(
+            post_with(&router, p1, &[("x-forwarded-for", "9.9.9.1, 203.0.113.5")]).await,
+            StatusCode::UNAUTHORIZED
+        );
+        // Different leftmost + different socket peer, SAME rightmost trusted hop:
+        // braked, because the key is the rightmost hop, not the spoofed value.
+        assert_eq!(
+            post_with(&router, p2, &[("x-forwarded-for", "9.9.9.2, 203.0.113.5")]).await,
+            StatusCode::TOO_MANY_REQUESTS,
+            "a spoofer varying the leftmost XFF hop must not rotate its bucket key"
+        );
+    }
+
+    #[tokio::test]
+    async fn xff_distinct_real_clients_get_distinct_buckets() {
+        // Distinct trusted (rightmost) hops are distinct clients: neither throttles
+        // the other, so a busy legitimate client cannot starve a different one.
+        let router = ip_limited_over_auth_router(IpRateLimiter {
+            limiter: RateLimiter::new(1, Duration::from_secs(60)),
+            trust: TrustedProxy::XForwardedFor,
+        });
+        let peer: SocketAddr = "10.0.0.9:1".parse().unwrap();
+        assert_eq!(
+            post_with(
+                &router,
+                peer,
+                &[("x-forwarded-for", "1.1.1.1, 203.0.113.5")]
+            )
+            .await,
+            StatusCode::UNAUTHORIZED
+        );
+        assert_eq!(
+            post_with(
+                &router,
+                peer,
+                &[("x-forwarded-for", "1.1.1.1, 203.0.113.6")]
+            )
+            .await,
+            StatusCode::UNAUTHORIZED,
+            "a different trusted client hop must get its own bucket"
+        );
+    }
+
+    #[tokio::test]
+    async fn absent_trusted_header_collapses_to_socket_peer() {
+        // The documented fallback: in a trusted-proxy mode, a request with no
+        // forwarded header keys on the socket peer. Behind a real edge every
+        // request shares the proxy's socket, so they collapse onto one bucket —
+        // asserted here by name so any change to this behavior is deliberate.
+        let router = ip_limited_over_auth_router(IpRateLimiter {
+            limiter: RateLimiter::new(1, Duration::from_secs(60)),
+            trust: TrustedProxy::XForwardedFor,
+        });
+        let proxy: SocketAddr = "172.16.0.1:1".parse().unwrap();
+        assert_eq!(
+            post_with(&router, proxy, &[]).await,
+            StatusCode::UNAUTHORIZED
+        );
+        assert_eq!(
+            post_with(&router, proxy, &[]).await,
+            StatusCode::TOO_MANY_REQUESTS,
+            "with no trusted header, requests fall back to the socket peer key (collapse)"
+        );
+    }
+
+    // ── Retry-After derived from the window, not a constant 60 (U5) ───────
+    // A 429 must advertise the delay until the oldest live request ages out of
+    // the limiter's window, so the value tracks the real window instead of the
+    // hardcoded 60 that contradicted every limiter's 3600s window.
+
+    async fn resp_from(router: &axum::Router, peer: SocketAddr) -> Response {
+        use tower::ServiceExt;
+        let mut req = axum::http::Request::builder()
+            .method(axum::http::Method::POST)
+            .uri("/o/r/git-receive-pack")
+            .body(axum::body::Body::empty())
+            .unwrap();
+        req.extensions_mut().insert(ConnectInfo(peer));
+        router.clone().oneshot(req).await.unwrap()
+    }
+
+    fn retry_after_secs(resp: &Response) -> u64 {
+        resp.headers()
+            .get("retry-after")
+            .expect("429 must carry a retry-after header")
+            .to_str()
+            .unwrap()
+            .parse()
+            .expect("retry-after must be an integer number of seconds")
+    }
+
+    #[tokio::test]
+    async fn retry_after_reflects_full_window_for_freshly_filled_bucket() {
+        // A just-filled bucket's oldest entry is ~now, so the advertised delay
+        // must be close to the whole window (100s), NOT the old constant 60.
+        let router = ip_limited_router(IpRateLimiter {
+            limiter: RateLimiter::new(1, Duration::from_secs(100)),
+            trust: TrustedProxy::None,
+        });
+        let peer: SocketAddr = "203.0.113.50:6000".parse().unwrap();
+        assert_eq!(resp_from(&router, peer).await.status(), StatusCode::OK);
+        let resp = resp_from(&router, peer).await;
+        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
+        let retry = retry_after_secs(&resp);
+        assert!(
+            (95..=100).contains(&retry),
+            "freshly-filled bucket must advertise ~window (95..=100s), got {retry} (constant-60 bug if 60)"
+        );
+    }
+
+    #[tokio::test]
+    async fn retry_after_shrinks_as_oldest_entry_ages() {
+        // Seed the sole slot with an entry already 95s into a 100s window: the
+        // next request is rejected and must advertise only the ~5s remaining,
+        // not 60.
+        let limiter = RateLimiter::new(1, Duration::from_secs(100));
+        let peer: SocketAddr = "203.0.113.51:6000".parse().unwrap();
+        limiter
+            .seed_aged_entry_for_test(&peer.ip().to_string(), &[Duration::from_secs(95)])
+            .await;
+        let router = ip_limited_router(IpRateLimiter {
+            limiter,
+            trust: TrustedProxy::None,
+        });
+        let resp = resp_from(&router, peer).await;
+        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
+        let retry = retry_after_secs(&resp);
+        assert!(
+            (1..=10).contains(&retry),
+            "an aged bucket must advertise the small remaining delay (1..=10s), got {retry}"
+        );
+    }
+
+    #[tokio::test]
+    async fn retry_after_clamps_to_at_least_one_second() {
+        // Oldest entry is 99s into a 100s window: the raw remainder rounds toward
+        // 1s and must never be advertised as 0 (a 0/blank retry lets a rejected
+        // client hammer immediately).
+        let limiter = RateLimiter::new(1, Duration::from_secs(100));
+        let peer: SocketAddr = "203.0.113.52:6000".parse().unwrap();
+        limiter
+            .seed_aged_entry_for_test(&peer.ip().to_string(), &[Duration::from_secs(99)])
+            .await;
+        let router = ip_limited_router(IpRateLimiter {
+            limiter,
+            trust: TrustedProxy::None,
+        });
+        let resp = resp_from(&router, peer).await;
+        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
+        let retry = retry_after_secs(&resp);
+        assert!(retry >= 1, "retry-after must clamp to >= 1, got {retry}");
+        assert!(
+            retry <= 2,
+            "with 99s of a 100s window elapsed, the remaining delay is ~1s, got {retry}"
+        );
+    }
 }

crates/gitlawb-node/src/server.rs

@@ -55,18 +55,50 @@ fn add_auth_layers(router: Router<AppState>, state: AppState) -> Router<AppState
         .layer(middleware::from_fn(auth::require_signature))
 }
 
+/// Attach the shared per-IP write brake to an authenticated write-route group,
+/// outermost so a flood is rejected before signature verification. Every
+/// authenticated non-creation write group MUST call this or it silently ships
+/// unbraked with no compile error. Deliberately a separate call, NOT folded into
+/// `add_auth_layers`: the creation routes use a different bucket
+/// (`create_ip_rate_limiter`) and the graphql group is not built through
+/// `add_auth_layers`, so a helper keyed on auth would attach the wrong limiter.
+trait WriteBraked {
+    fn write_braked(self, limiter: &rate_limit::IpRateLimiter) -> Self;
+}
+
+impl WriteBraked for Router<AppState> {
+    fn write_braked(self, limiter: &rate_limit::IpRateLimiter) -> Self {
+        self.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
+            .layer(axum::Extension(limiter.clone()))
+    }
+}
+
 pub fn build_router(state: AppState) -> Router {
     // ── GraphQL routes ─────────────────────────────────────────────────────
     let schema = state.graphql_schema.as_ref().clone();
+
+    // Per-IP write brake, shared across every authenticated non-creation write
+    // group (see `AppState::write_rate_limiter`). Built once here so each group
+    // clones the same bucket; attached outermost on each group so a flood is
+    // rejected before auth runs. Separate bucket from the creation brake.
+    let write_ip_limiter = rate_limit::IpRateLimiter {
+        limiter: state.write_rate_limiter.clone(),
+        trust: state.push_limiter_trust,
+    };
+
     let graphql_routes = Router::new()
         .route("/graphql", get(graphql_playground).post(graphql_handler))
         // Attach the verified DID to /graphql when a signature is present. The
         // layer covers only routes added before it, so /graphql/ws (added after,
-        // read-only subscriptions) stays open.
+        // read-only subscriptions) stays open. The per-IP write brake covers the
+        // same /graphql (KTD-5: an HTTP-layer brake cannot tell a query POST from
+        // a mutation POST, so it counts every /graphql request — acceptable, both
+        // cost work); /graphql/ws subscriptions stay unbraked.
         .layer(middleware::from_fn(auth::optional_signature))
+        .write_braked(&write_ip_limiter)
         .route_service("/graphql/ws", GraphQLSubscription::new(schema));
 
-    // ── Task routes (write — require HTTP Signature) ───────────────────────
+    // ── Task routes (write — require HTTP Signature + per-IP write brake) ──
     let task_write_routes = add_auth_layers(
         Router::new()
             .route("/api/v1/tasks", post(tasks::create_task))
@@ -74,7 +106,8 @@ pub fn build_router(state: AppState) -> Router {
             .route("/api/v1/tasks/{id}/complete", post(tasks::complete_task))
             .route("/api/v1/tasks/{id}/fail", post(tasks::fail_task)),
         state.clone(),
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // ── Task routes (read — open) ──────────────────────────────────────────
     let task_read_routes = Router::new()
@@ -111,7 +144,13 @@ pub fn build_router(state: AppState) -> Router {
     .layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
     .layer(axum::Extension(create_ip_limiter));
 
-    // ── Write routes — require HTTP Signature (no rate limit) ─────────────
+    // ── Write routes — require HTTP Signature + per-IP write brake ────────
+    // Same per-IP flood defense as the creation routes, on its own `write`
+    // bucket (see `AppState::write_rate_limiter`). Per-DID is deliberately not
+    // paired (a DID farm never trips it; busy legitimate agents would false-
+    // positive). The bundled visibility GET is a read that also draws from the
+    // write bucket — harmless, since the bucket is sized far above any legit
+    // per-IP read rate.
     let write_routes = add_auth_layers(
         Router::new()
             .route(
@@ -181,7 +220,8 @@ pub fn build_router(state: AppState) -> Router {
                 axum::routing::delete(agents::deregister_agent),
             ),
         state.clone(),
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // Body limit is raised to GITLAWB_MAX_PACK_BYTES (default 2 GB) for git
     // routes only — all other API routes keep axum's default 2 MB cap.
@@ -247,7 +287,8 @@ pub fn build_router(state: AppState) -> Router {
                 post(bounties::dispute_bounty),
             ),
         state.clone(),
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // ── Bounty routes (read — open) ──────────────────────────────────────
     let bounty_read_routes = Router::new()
@@ -268,9 +309,10 @@ pub fn build_router(state: AppState) -> Router {
     let profile_write_routes = add_auth_layers(
         Router::new().route("/api/v1/profile", axum::routing::put(profiles::set_profile)),
         state.clone(),
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
-    // ── Issue routes (write — require HTTP Signature, no rate limit) ─────
+    // ── Issue routes (write — require HTTP Signature + per-IP write brake) ─
     let issue_write_routes = add_auth_layers(
         Router::new()
             .route(
@@ -282,7 +324,8 @@ pub fn build_router(state: AppState) -> Router {
                 post(issues::create_issue_comment),
             ),
         state.clone(),
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // ── Peer discovery routes ─────────────────────────────────────────────
     // Peer writes accept signatures when present and can require them after a

crates/gitlawb-node/src/state.rs

@@ -62,6 +62,16 @@ pub struct AppState {
     /// resolved client IP is what actually stops a single-source flood (same
     /// rationale as `push_rate_limiter`). Keyed by `push_limiter_trust`.
     pub create_ip_rate_limiter: RateLimiter,
+    /// Per-client-IP rate limiter for the authenticated write surface that is
+    /// not repo/agent creation: issue/PR comments, labels, stars, merges,
+    /// protect/unprotect, replicas, visibility, tasks, bounties, profile, and
+    /// the GraphQL `MutationRoot`. Separate bucket from `create_ip_rate_limiter`
+    /// (its own budget, per the `sync_trigger`/`peer_write` precedent) so a
+    /// write flood cannot drain the creation budget and vice versa. Per-DID is
+    /// deliberately NOT paired here: a DID farm never trips it, and busy
+    /// legitimate agents would false-positive. `GITLAWB_WRITE_RATE_LIMIT`
+    /// overrides the default, 0 disables. Keyed by `push_limiter_trust`.
+    pub write_rate_limiter: RateLimiter,
     /// Per-client-IP rate limiter for git-receive-pack. Per-DID limits cannot
     /// brake a push flood from a DID farm (one throwaway DID per repo), so the
     /// push path throttles on the resolved client IP instead.
@@ -118,4 +128,48 @@ impl AppState {
     pub fn is_shutting_down(&self) -> bool {
         *self.shutdown_tx.borrow()
     }
+
+    /// Reclaim expired entries from EVERY rate limiter. Single source of truth for
+    /// the periodic cleanup loop (main.rs), co-located with the limiter fields so a
+    /// newly-added limiter is cleaned here rather than silently omitted from a
+    /// hand-maintained list in main() (H2: `write_rate_limiter` was the omitted
+    /// one). Every limiter field above MUST be swept here; the guard test
+    /// `cleanup_rate_limiters_reaps_the_write_limiter` fails if it is not.
+    pub(crate) async fn cleanup_rate_limiters(&self) {
+        self.rate_limiter.cleanup().await;
+        self.create_ip_rate_limiter.cleanup().await;
+        self.write_rate_limiter.cleanup().await;
+        self.push_rate_limiter.cleanup().await;
+        self.sync_trigger_rate_limiter.cleanup().await;
+        self.peer_write_rate_limiter.cleanup().await;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    /// H2 guard: the write-surface limiter must be reaped by the shared cleanup
+    /// routine. Seeds a reclaimable entry into `write_rate_limiter` and asserts
+    /// `cleanup_rate_limiters` removes it. Goes RED if `write_rate_limiter` is
+    /// dropped from `cleanup_rate_limiters` — the exact H2 omission, now guarded.
+    #[tokio::test]
+    async fn cleanup_rate_limiters_reaps_the_write_limiter() {
+        let state = crate::test_support::test_state_lazy();
+        state
+            .write_rate_limiter
+            .insert_reclaimable_for_test("some-ip")
+            .await;
+        assert_eq!(
+            state.write_rate_limiter.tracked_keys().await,
+            1,
+            "precondition: the reclaimable entry is tracked"
+        );
+
+        state.cleanup_rate_limiters().await;
+
+        assert_eq!(
+            state.write_rate_limiter.tracked_keys().await,
+            0,
+            "write_rate_limiter must be reaped by cleanup_rate_limiters (H2)"
+        );
+    }
 }

crates/gitlawb-node/src/test_support.rs

@@ -77,6 +77,7 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
         rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
         create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
+        write_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
         push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
         push_limiter_trust: crate::rate_limit::TrustedProxy::None,
         sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)),