test(node): derive the first-OID expectation from every reachable object

sweep_persists_last_attempted_not_planned_tail_on_mid_batch_narrow sorted only
HEAD's commit, tree and three blobs to find the first object the sweep
dispatches. The fixture makes three commits, so the two earlier commits and
trees are reachable too, and an earlier commit id changes with its timestamp.
When one sorted below the other five the expected first OID was wrong and the
test failed.

Take the expectation from git rev-list --objects --all (nine objects). On an
unpatched main build the test failed 3 of 45 serial runs; with this change it
failed 0 of 45.

.env.example

@@ -365,6 +365,24 @@ GITLAWB_RATE_LIMIT_MAX=10
 # limiter. Default 3600.
 GITLAWB_RATE_LIMIT_WINDOW_SECS=3600
 
+# ── Write rate limiting (non-creation authenticated writes) ───────────────
+# Max requests per client IP per hour on the shared write bucket. It is debited
+# by: issue and PR comments, issue close, labels, stars, merges, protect and
+# unprotect, replicas, webhook create/delete, visibility (including the GET on
+# that group), tasks, bounties, profile, and every HTTP request to /graphql.
+# The /graphql brake wraps the whole route, so queries and the playground GET
+# consume this bucket too, not only mutations (GraphQL WebSocket subscriptions
+# are excluded). Issue comment and issue close debit it inside the handler,
+# after the read gate, rather than in router middleware. Issue close also draws
+# on its own GITLAWB_CLOSE_ISSUE_RATE_LIMIT bucket, and a close refused by
+# either cap spends nothing from the other. 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

@@ -2208,7 +2208,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -2701,7 +2701,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ab67060fc6b8ef687992d439ca0fa36e7ed17e9a0b16b25b601e8757df720de"
 dependencies = [
  "data-encoding",
- "syn 1.0.109",
+ "syn 2.0.117",
 ]
 
 [[package]]
@@ -3540,6 +3540,7 @@ dependencies = [
  "async-compression",
  "async-graphql",
  "async-graphql-axum",
+ "async-trait",
  "aws-config",
  "aws-sdk-s3",
  "axum",
@@ -4983,11 +4984,10 @@ dependencies = [
 
 [[package]]
 name = "multihash"
-version = "0.19.3"
+version = "0.19.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d"
+checksum = "577c63b00ad74d57e8c9aa870b5fccebf2fd64a308a5aee9f1bb88e4aea19447"
 dependencies = [
- "core2",
  "unsigned-varint",
 ]
 
@@ -7959,7 +7959,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]

crates/gitlawb-node/Cargo.toml

@@ -30,6 +30,7 @@ required-features = ["test-harness"]
 
 [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

@@ -164,6 +164,8 @@ pub async fn create_issue_comment(
     State(state): State<AppState>,
     Extension(auth): Extension<AuthenticatedDid>,
     Path((owner, repo, issue_id)): Path<(String, String, String)>,
+    headers: axum::http::HeaderMap,
+    crate::rate_limit::PeerAddr(peer): crate::rate_limit::PeerAddr,
     Json(req): Json<CreateIssueCommentRequest>,
 ) -> Result<(StatusCode, Json<IssueComment>)> {
     if req.body.trim().is_empty() {
@@ -176,6 +178,17 @@ pub async fn create_issue_comment(
     let (record, _rules) =
         crate::api::authorize_repo_read(&state, &owner, &repo, Some(auth.0.as_str()), "/").await?;
 
+    // Shared per-IP write brake, debited AFTER the read gate so a 429 cannot
+    // distinguish a hidden repo from a missing one (INV-12 — the same
+    // placement close_issue uses for its brake). The route group carries no
+    // outer brake for exactly this reason.
+    if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
+        if let Some(retry_after) = state.write_rate_limiter.check_retry(&key).await {
+            tracing::warn!(repo = %repo, key = %key, "create_issue_comment rate limited");
+            return Err(AppError::RateLimited(retry_after));
+        }
+    }
+
     let disk_path = state
         .repo_store
         .acquire(&record.owner_did, &record.name)
@@ -272,7 +285,22 @@ pub async fn close_issue(
     // close requests for arbitrary issue ids. Applied after the read gate so a
     // denied reader still sees 404, not 429.
     if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
+        // Both buckets debit here, after the read gate (INV-12): the shared
+        // per-IP write brake — the route group deliberately carries no outer
+        // brake — and this route's own close budget.
+        //
+        // A close refused by either cap spends nothing from the other bucket.
+        // The write bucket is debited first, and refunded when the close cap
+        // then refuses, so the guarantee holds even when a concurrent close
+        // takes the last close slot between the two checks. A write-bucket
+        // refusal advertises its window's Retry-After; the close-cap refusal
+        // keeps the plain 429 it has always returned.
+        if let Some(retry_after) = state.write_rate_limiter.check_retry(&key).await {
+            tracing::warn!(repo = %repo, key = %key, "close_issue rate limited");
+            return Err(AppError::RateLimited(retry_after));
+        }
         if !state.close_issue_rate_limiter.check(&key).await {
+            state.write_rate_limiter.refund(&key).await;
             tracing::warn!(repo = %repo, key = %key, "close_issue rate limited");
             return Err(AppError::TooManyRequests(
                 "rate limit exceeded — try again later".into(),
@@ -434,11 +462,47 @@ pub async fn close_issue(
         }
     }
 
+    // Snapshot the issue's pre-close JSON 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).
+    //
+    // Resolve first so the compensating rewrite below lands on the SAME ref the
+    // close mutates. The path id may be an 8-char prefix, and rewriting
+    // refs/gitlawb/issues/{prefix} with the pre-close content would leave the
+    // canonical {full_id} ref closed while the client saw a 5xx.
+    let full_issue_id = git_issues::resolve_issue_id(&disk_path, &issue_id)
+        .ok()
+        .flatten();
+    let prior_json = full_issue_id
+        .as_deref()
+        .and_then(|id| git_issues::get_issue(&disk_path, 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.
-    // Same short-circuit as create_issue, and before the 200 body below.
-    guard.release(close_result.is_ok()).await.into_result()?;
+    // Same short-circuit as create_issue, and before the 200 body below. With a
+    // snapshot in hand, a failed upload rewrites the ref back to the pre-close
+    // content instead of leaving the local repo closed while the client saw 5xx.
+    let release_result = match (close_result.is_ok(), prior_json, full_issue_id) {
+        (true, Some(json), Some(resolved_id)) => {
+            let git_bin = state.git_bin.clone();
+            let deadline = std::time::Instant::now()
+                + std::time::Duration::from_secs(state.config.git_service_timeout_secs);
+            guard
+                .release_compensating(true, move |path| {
+                    // Same bounded-write strategy create_issue's compensator
+                    // uses: the rollback rewrites the ref with the pre-close
+                    // content through the deadline-aware runner, so a hung git
+                    // cannot stall the guard past git_service_timeout.
+                    git_issues::create_issue_bounded(&git_bin, path, &resolved_id, &json, deadline)
+                })
+                .await
+                .into_result()
+        }
+        _ => guard.release(close_result.is_ok()).await.into_result(),
+    };
+    release_result?;
 
     let updated = close_result
         .map_err(|e| AppError::Git(e.to_string()))?
@@ -1233,6 +1297,198 @@ mod lock_pool_shed_tests {
         );
     }
 
+    /// A close refused by its own cap must not spend the shared write budget:
+    /// the 429 completed no close, so stars/comments/GraphQL keep their slot.
+    #[sqlx::test]
+    async fn close_issue_refused_by_close_cap_does_not_debit_write_bucket(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let owner = "did:key:zCLOSEDUALOWNERAAAAAAAAAAAAAAAAAAAAAAA";
+        let mut state = crate::test_support::test_state(pool).await;
+        state.close_issue_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.write_rate_limiter = crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+        state
+            .db
+            .create_repo(&seed_repo(owner, "dual-close"))
+            .await
+            .expect("seed repo");
+
+        let peer: SocketAddr = "203.0.113.77:7000".parse().unwrap();
+        let key = peer.ip().to_string();
+        assert!(
+            state.close_issue_rate_limiter.check(&key).await,
+            "exhaust only the close bucket"
+        );
+
+        let res = close_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), "dual-close".to_string(), "1".to_string())),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some(peer)),
+        )
+        .await;
+        assert!(
+            matches!(res, Err(AppError::TooManyRequests(_))),
+            "close cap exhausted must 429: {:?}",
+            res
+        );
+        assert!(
+            state.write_rate_limiter.check(&key).await,
+            "a close refused by the close cap must leave the write bucket untouched"
+        );
+    }
+
+    /// A close refused by the shared write cap must not spend the close budget.
+    #[sqlx::test]
+    async fn close_issue_refused_by_write_cap_does_not_debit_close_bucket(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let owner = "did:key:zCLOSEDUALOWNERBBBBBBBBBBBBBBBBBBBBBBB";
+        let mut state = crate::test_support::test_state(pool).await;
+        state.close_issue_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.write_rate_limiter = crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+        state
+            .db
+            .create_repo(&seed_repo(owner, "dual-write"))
+            .await
+            .expect("seed repo");
+
+        let peer: SocketAddr = "203.0.113.78:7000".parse().unwrap();
+        let key = peer.ip().to_string();
+        assert!(
+            state.write_rate_limiter.check(&key).await,
+            "exhaust only the write bucket"
+        );
+
+        let res = close_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), "dual-write".to_string(), "1".to_string())),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some(peer)),
+        )
+        .await;
+        let resp = res
+            .expect_err("write cap exhausted must refuse")
+            .into_response();
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "write cap exhausted must 429"
+        );
+        assert!(
+            state.close_issue_rate_limiter.check(&key).await,
+            "a close refused by the write cap must leave the close bucket untouched"
+        );
+    }
+
+    /// 429 from the shared write bucket must advertise a window-derived
+    /// Retry-After, as the outer-middleware 429s on the other write routes do.
+    fn assert_429_with_window_retry_after(err: AppError, what: &str) {
+        let resp = err.into_response();
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "{what}: an exhausted write bucket must 429"
+        );
+        let secs: u64 = resp
+            .headers()
+            .get("retry-after")
+            .unwrap_or_else(|| panic!("{what}: the write-bucket 429 must carry retry-after"))
+            .to_str()
+            .unwrap()
+            .parse()
+            .expect("retry-after must be integer seconds");
+        assert!(
+            (1..=3600).contains(&secs),
+            "{what}: retry-after {secs}s must fall inside the 3600s window, not a constant"
+        );
+    }
+
+    #[sqlx::test]
+    async fn close_issue_write_cap_429_carries_retry_after(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let owner = "did:key:zCLOSERETRYOWNERAAAAAAAAAAAAAAAAAAAAAA";
+        let mut state = crate::test_support::test_state(pool).await;
+        state.write_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(3600));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+        state
+            .db
+            .create_repo(&seed_repo(owner, "retry-close"))
+            .await
+            .expect("seed repo");
+
+        let peer: SocketAddr = "203.0.113.61:7000".parse().unwrap();
+        assert!(state.write_rate_limiter.check(&peer.ip().to_string()).await);
+
+        let res = close_issue(
+            State(state),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((
+                owner.to_string(),
+                "retry-close".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some(peer)),
+        )
+        .await;
+        assert_429_with_window_retry_after(
+            res.expect_err("write cap exhausted must refuse"),
+            "close_issue",
+        );
+    }
+
+    #[sqlx::test]
+    async fn create_issue_comment_write_cap_429_carries_retry_after(pool: PgPool) {
+        use std::net::SocketAddr;
+        use std::time::Duration;
+
+        let owner = "did:key:zCOMMENTRETRYOWNERAAAAAAAAAAAAAAAAAAAA";
+        let mut state = crate::test_support::test_state(pool).await;
+        state.write_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, Duration::from_secs(3600));
+        state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
+        state
+            .db
+            .create_repo(&seed_repo(owner, "retry-comment"))
+            .await
+            .expect("seed repo");
+
+        let peer: SocketAddr = "203.0.113.62:7000".parse().unwrap();
+        assert!(state.write_rate_limiter.check(&peer.ip().to_string()).await);
+
+        let res = create_issue_comment(
+            State(state),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((
+                owner.to_string(),
+                "retry-comment".to_string(),
+                "1".to_string(),
+            )),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(Some(peer)),
+            Json(CreateIssueCommentRequest {
+                body: "x".to_string(),
+            }),
+        )
+        .await;
+        assert_429_with_window_retry_after(
+            res.expect_err("write cap exhausted must refuse"),
+            "create_issue_comment",
+        );
+    }
+
     /// close_issue and receive-pack must not share one per-IP bucket.
     #[sqlx::test]
     async fn close_issue_rate_limit_does_not_drain_push_bucket(pool: PgPool) {
@@ -1320,4 +1576,251 @@ mod lock_pool_shed_tests {
 
         server.abort();
     }
+
+    /// The compensating rewrite after a refused publish must land on the ref the
+    /// close actually mutated. A close addressed by an 8-char prefix resolves to
+    /// the full id and closes `refs/gitlawb/issues/{full_id}`; a compensator
+    /// that rewrites the path id verbatim mints `refs/gitlawb/issues/{prefix}`
+    /// with the pre-close content and leaves the canonical ref closed while the
+    /// client saw a 5xx.
+    ///
+    /// Reaching the compensator observably takes one contrivance: a definite
+    /// refusal invalidates (deletes) the live tree BEFORE the closure runs, so
+    /// the ref it writes exists only when that deletion cannot complete. The
+    /// fixture makes the tree read-only except the two directories the close
+    /// and its rollback write (`refs/gitlawb/issues/` and its reflog), and
+    /// pre-stores the closed blob so the mutation's `hash-object -w` needs no
+    /// write either. Without the contrivance the compensator runs against a
+    /// path that is already gone and there is nothing to assert on.
+    #[sqlx::test]
+    #[cfg(unix)]
+    async fn close_via_prefix_restores_the_full_ref_when_publish_refuses(pool: PgPool) {
+        use axum::response::IntoResponse;
+        use std::os::unix::fs::PermissionsExt;
+
+        // Nothing is stored under the key (404 on HEAD: no under-lock download,
+        // a create-only publish fence) and the PUT is refused outright (a 4xx
+        // classifies as provably never dispatched), so release reaches the
+        // definite-refusal arm that invalidates the tree and runs the
+        // compensator.
+        let app = axum::Router::new().route(
+            "/{*key}",
+            axum::routing::any(|method: axum::http::Method| async move {
+                if method == axum::http::Method::HEAD {
+                    axum::http::StatusCode::NOT_FOUND.into_response()
+                } else {
+                    axum::http::StatusCode::FORBIDDEN.into_response()
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        let owner = "did:key:zISSUECLOSEROLLBACKBBBBBBBBBBBBBBB";
+        let repo_name = "rollback-close-prefix";
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        let disk = tempfile::TempDir::new().unwrap();
+        state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
+            disk.path().to_path_buf(),
+            crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
+        );
+        state
+            .db
+            .create_repo(&seed_repo(owner, repo_name))
+            .await
+            .expect("seed repo");
+
+        // Seed the live tree at the path the store resolves, with the issue
+        // under its full id so the route can be addressed by the 8-char prefix.
+        let live = state
+            .repo_store
+            .acquire(owner, repo_name)
+            .await
+            .expect("resolve live path");
+        crate::git::store::init_bare(&live).expect("init bare repo");
+        let full_id = "abc12345-0000-0000-0000-000000000000";
+        let prefix = "abc12345";
+        let open_json = serde_json::to_string(&IssueRecord {
+            id: full_id.to_string(),
+            title: "seeded".to_string(),
+            body: Some(String::new()),
+            author: Some(owner.to_string()),
+            created_at: chrono::Utc::now().to_rfc3339(),
+            status: "open".to_string(),
+            signed_payload: None,
+        })
+        .expect("serialize seeded issue");
+        git_issues::create_issue(&live, full_id, &open_json).expect("seed issue blob");
+
+        // Pre-store the blob `git_issues::close_issue` will write: the same
+        // parse/set-status/serialize sequence, so the bytes are identical and
+        // the mutation's `hash-object -w` is a no-op against read-only objects/.
+        let mut closed_value: serde_json::Value =
+            serde_json::from_str(&open_json).expect("parse seeded issue");
+        closed_value["status"] = serde_json::Value::String("closed".into());
+        let closed_json = serde_json::to_string(&closed_value).expect("closed issue json");
+        {
+            use std::io::Write;
+            let mut child = std::process::Command::new("git")
+                .args(["hash-object", "--stdin", "-w"])
+                .current_dir(&live)
+                .stdin(std::process::Stdio::piped())
+                .stdout(std::process::Stdio::null())
+                .spawn()
+                .expect("spawn hash-object");
+            child
+                .stdin
+                .take()
+                .expect("stdin")
+                .write_all(closed_json.as_bytes())
+                .expect("write blob");
+            assert!(
+                child.wait().expect("hash-object").success(),
+                "pre-storing the closed blob must succeed"
+            );
+        }
+
+        // Stand the tree through the refusal's remove_dir_all: read-only
+        // everywhere, then write restored only where the close mutation and its
+        // rollback actually land (the ref and its reflog).
+        fn make_readonly(path: &std::path::Path) -> std::io::Result<()> {
+            let mut p = path.metadata()?.permissions();
+            p.set_readonly(true);
+            std::fs::set_permissions(path, p)?;
+            if path.is_dir() {
+                for entry in std::fs::read_dir(path)? {
+                    make_readonly(&entry?.path())?;
+                }
+            }
+            Ok(())
+        }
+        make_readonly(&live).expect("make the live tree unremovable");
+        for dir in [
+            live.join("refs/gitlawb/issues"),
+            live.join("logs/refs/gitlawb/issues"),
+        ] {
+            std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755))
+                .expect("restore write where the rollback lands");
+        }
+
+        let shed = close_issue(
+            State(state.clone()),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), repo_name.to_string(), prefix.to_string())),
+            axum::http::HeaderMap::new(),
+            crate::rate_limit::PeerAddr(None),
+        )
+        .await;
+        let err = shed.expect_err("a refused publish must not report a closed issue");
+        assert!(
+            err.into_response().status().is_server_error(),
+            "a definite publish refusal must surface as a 5xx"
+        );
+
+        let restored = git_issues::get_issue(&live, full_id).expect("read the full ref");
+        assert_eq!(
+            restored.as_deref(),
+            Some(open_json.as_str()),
+            "the compensator must restore the pre-close content on the FULL ref, \
+             not on refs/gitlawb/issues/{prefix}"
+        );
+        assert_eq!(
+            git_issues::resolve_issue_id(&live, prefix).expect("resolve the prefix"),
+            Some(full_id.to_string()),
+            "no stray refs/gitlawb/issues/{{prefix}} may be minted beside the \
+             canonical ref: the prefix must still resolve to the full id"
+        );
+        let issues = git_issues::list_issues(&live).expect("list issues");
+        assert_eq!(
+            issues.len(),
+            1,
+            "exactly one issue ref may survive the rollback: {issues:?}"
+        );
+
+        // Restore writability so the TempDir can clean itself up.
+        let _ = std::process::Command::new("chmod")
+            .args(["-R", "u+w"])
+            .arg(&live)
+            .output();
+        server.abort();
+    }
+    /// INV-12 ordering: both issue write routes debit the shared write bucket
+    /// AFTER authorize_repo_read, so an exhausted bucket cannot answer 429 to a
+    /// caller the gate must 404 — that status difference would leak the repo's
+    /// existence. The route group carries no outer brake for this reason.
+    #[sqlx::test]
+    async fn an_exhausted_write_bucket_cannot_preempt_the_read_gate(pool: PgPool) {
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        state.write_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, std::time::Duration::from_secs(3600));
+        state
+            .db
+            .create_repo(&crate::db::RepoRecord {
+                id: "rid-write-brake-order".to_string(),
+                name: "priv".to_string(),
+                owner_did: "did:key:z6MkWriteBrakeOwner".to_string(),
+                description: None,
+                is_public: false,
+                default_branch: "main".to_string(),
+                created_at: chrono::Utc::now(),
+                updated_at: chrono::Utc::now(),
+                disk_path: "/tmp/write-brake-order".to_string(),
+                forked_from: None,
+                machine_id: None,
+            })
+            .await
+            .unwrap();
+
+        let headers = axum::http::HeaderMap::new();
+        let peer = || crate::rate_limit::PeerAddr(Some("203.0.113.77:5000".parse().unwrap()));
+        let key = crate::rate_limit::client_key(&headers, peer().0, state.push_limiter_trust)
+            .expect("a peer addr yields a brake key");
+        assert!(state.write_rate_limiter.check(&key).await);
+        assert!(
+            !state.write_rate_limiter.check(&key).await,
+            "the shared write bucket is empty for this caller"
+        );
+
+        let stranger = || crate::auth::AuthenticatedDid("did:key:z6MkStranger".to_string());
+        let path = || {
+            (
+                "did:key:z6MkWriteBrakeOwner".to_string(),
+                "priv".to_string(),
+                "1".to_string(),
+            )
+        };
+
+        let res = close_issue(
+            axum::extract::State(state.clone()),
+            axum::Extension(stranger()),
+            axum::extract::Path(path()),
+            headers.clone(),
+            peer(),
+        )
+        .await;
+        assert!(
+            matches!(res, Err(AppError::RepoNotFound(_))),
+            "close on an unreadable repo must reach the read gate, not the empty write bucket: {res:?}"
+        );
+
+        let res = create_issue_comment(
+            axum::extract::State(state),
+            axum::Extension(stranger()),
+            axum::extract::Path(path()),
+            headers,
+            peer(),
+            axum::Json(CreateIssueCommentRequest {
+                body: "x".to_string(),
+            }),
+        )
+        .await;
+        assert!(
+            matches!(res, Err(AppError::RepoNotFound(_))),
+            "comment on an unreadable repo must reach the read gate, not the empty write bucket: {res:?}"
+        );
+    }
 }

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

@@ -225,6 +225,19 @@ pub async fn merge_pr(
         .map_err(|e| crate::api::repos::acquire_write_app_error(&e, &name))?;
     let disk_path = guard.path().to_path_buf();
     let merger_did = auth.0;
+
+    // Snapshot the target ref's pre-merge OID so a refused durable upload can
+    // rewind the local merge, mirroring the close_issue / create_issue
+    // compensating pattern. A read failure or an absent ref (a first-commit
+    // merge onto an unborn branch) yields no rollback closure: the merge still
+    // applies and a refused publish still surfaces as 5xx.
+    let target_ref = format!("refs/heads/{}", pr.target_branch);
+    let git_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs);
+    let prior_tip = store::read_ref_bounded(&state.git_bin, &disk_path, &target_ref, git_timeout)
+        .await
+        .ok()
+        .flatten();
+
     let merge_result = store::merge_branch(
         &disk_path,
         &pr.target_branch,
@@ -237,7 +250,27 @@ pub async fn merge_pr(
     // Short-circuit on a refused publish before the PR is marked merged and
     // before the webhook fires. Both are irreversible announcements of a merge
     // commit that only exists on this node's disk.
-    guard.release(merge_result.is_ok()).await.into_result()?;
+    let release_result = match (merge_result.is_ok(), prior_tip) {
+        (true, Some(tip)) => {
+            let git_bin = state.git_bin.clone();
+            let deadline = std::time::Instant::now() + git_timeout;
+            guard
+                .release_compensating(true, move |path| {
+                    crate::git::visibility_pack::run_bounded_git(
+                        &git_bin,
+                        &["update-ref", &target_ref, &tip],
+                        path,
+                        b"",
+                        deadline,
+                    )
+                    .map(|_| ())
+                })
+                .await
+                .into_result()
+        }
+        _ => guard.release(merge_result.is_ok()).await.into_result(),
+    };
+    release_result?;
 
     let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
 
@@ -437,6 +470,136 @@ pub async fn list_comments(
     Ok(Json(serde_json::json!({ "comments": comments })))
 }
 
+#[cfg(test)]
+mod merge_rollback_tests {
+    use super::*;
+    use axum::response::IntoResponse;
+    use sqlx::PgPool;
+
+    fn git(dir: &std::path::Path, args: &[&str]) -> String {
+        let out = std::process::Command::new("git")
+            .args(["-C", &*dir.to_string_lossy()])
+            .args([
+                "-c",
+                "user.name=merge-rollback-test",
+                "-c",
+                "[email protected]",
+            ])
+            .args(args)
+            .output()
+            .unwrap();
+        assert!(
+            out.status.success(),
+            "git {args:?}: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+        String::from_utf8_lossy(&out.stdout).trim().to_string()
+    }
+
+    /// Build a bare repo whose `main` and `feature` refs diverge: one root
+    /// commit on main, one commit on feature, so merge_branch creates a real
+    /// merge commit that a refused publish must rewind.
+    fn seeded_merge_repo(dir: &std::path::Path) -> (String, String) {
+        crate::git::store::init_bare(dir).expect("bare repo");
+        let empty_tree = git(dir, &["mktree"]);
+        let root = git(dir, &["commit-tree", &empty_tree, "-m", "root"]);
+        git(dir, &["update-ref", "refs/heads/main", &root]);
+        let tip2 = git(
+            dir,
+            &["commit-tree", &empty_tree, "-m", "feature", "-p", &root],
+        );
+        git(dir, &["update-ref", "refs/heads/feature", &tip2]);
+        (root, tip2)
+    }
+
+    /// On a definite publish refusal, the local merge commit must be rewound:
+    /// the client got a 5xx, so the repo cannot show a merge the mirror never
+    /// learned about (same contract create_issue's rollback enforces).
+    #[sqlx::test]
+    async fn merge_pr_rewinds_local_merge_when_publish_refuses(pool: PgPool) {
+        // Endpoint that answers HEAD with an etag but refuses every write.
+        let app = axum::Router::new().route(
+            "/{*key}",
+            axum::routing::any(|method: axum::http::Method| async move {
+                if method == axum::http::Method::HEAD {
+                    let mut resp = axum::http::StatusCode::OK.into_response();
+                    resp.headers_mut()
+                        .insert("etag", axum::http::HeaderValue::from_static("\"gen-1\""));
+                    resp
+                } else {
+                    axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+        let owner = "did:key:zMERGEROLLBACKAAAAAAAAAAAAAAAAAAAAAAAA";
+        let repo_name = "rollback-merge";
+        let mut state = crate::test_support::test_state(pool.clone()).await;
+        let disk = tempfile::TempDir::new().unwrap();
+        let repo_path = disk.path().join("repo.git");
+        let (main_tip, _feature_tip) = seeded_merge_repo(&repo_path);
+
+        let mut seeded = crate::db::RepoRecord {
+            id: Uuid::new_v4().to_string(),
+            name: repo_name.to_string(),
+            owner_did: owner.to_string(),
+            description: None,
+            is_public: true,
+            default_branch: "main".to_string(),
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+            disk_path: String::new(),
+            forked_from: None,
+            machine_id: None,
+        };
+        seeded.disk_path = repo_path.to_string_lossy().to_string();
+        let repo_id = seeded.id.clone();
+        state.db.create_repo(&seeded).await.expect("seed repo");
+        let now = Utc::now().to_rfc3339();
+        state
+            .db
+            .create_pr(&PullRequest {
+                id: Uuid::new_v4().to_string(),
+                repo_id,
+                number: 1,
+                title: "rb".to_string(),
+                body: None,
+                author_did: owner.to_string(),
+                source_branch: "feature".to_string(),
+                target_branch: "main".to_string(),
+                status: "open".to_string(),
+                merged_by_did: None,
+                merged_at: None,
+                created_at: now.clone(),
+                updated_at: now,
+            })
+            .await
+            .expect("seed PR");
+        state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
+            disk.path().to_path_buf(),
+            crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
+        );
+
+        let res = merge_pr(
+            State(state),
+            Extension(AuthenticatedDid(owner.to_string())),
+            Path((owner.to_string(), repo_name.to_string(), 1)),
+        )
+        .await;
+        assert!(res.is_err(), "a refused publish must surface an error");
+
+        let tip_after = git(&repo_path, &["rev-parse", "refs/heads/main"]);
+        assert_eq!(
+            tip_after, main_tip,
+            "a refused publish must rewind the local merge to the pre-merge tip"
+        );
+    }
+}
+
 /// #173 F1 follow-up: merge holds no admission permit, so an exhausted write-lock
 /// pool must shed 503 + Retry-After.
 #[cfg(test)]

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

@@ -738,6 +738,12 @@ mod tests {
             task_tx.clone(),
             task_cursor_key.clone(),
         ));
+        let ws_schema = Arc::new(graphql::build_ws_schema(
+            db.clone(),
+            ref_tx.clone(),
+            task_tx.clone(),
+            task_cursor_key.clone(),
+        ));
         crate::state::AppState {
             config: Arc::new(Config::parse_from(["gitlawb-node"])),
             db,
@@ -748,11 +754,13 @@ mod tests {
             ref_update_tx: ref_tx,
             task_event_tx: task_tx,
             graphql_schema: schema,
+            graphql_ws_schema: ws_schema,
             task_cursor_key,
             machine_id: None,
             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)),
             close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
             ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),

crates/gitlawb-node/src/config.rs

@@ -1,4 +1,4 @@
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use std::path::{Path, PathBuf};
 
 /// Upper bound on `git_service_timeout_secs`, `ipfs_request_budget_secs`, and
@@ -28,18 +28,45 @@ use std::path::{Path, PathBuf};
 /// panicked are rejected.
 pub const GIT_SERVICE_TIMEOUT_SECS_MAX: u64 = 100 * 365 * 24 * 60 * 60;
 
+/// 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 = "Twigpine 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,
 
@@ -248,7 +275,7 @@ pub struct Config {
 
     /// Tigris (S3-compatible) bucket for repo storage.
     /// Leave empty to disable Tigris and use local-only storage.
-    #[arg(long, env = "GITLAWB_TIGRIS_BUCKET", default_value = "")]
+    #[arg(long, env = "GITLAWB_TIGRIS_BUCKET", default_value = "", global = true)]
     pub tigris_bucket: String,
 
     /// Maximum pack body size for git-receive-pack and git-upload-pack, in bytes.

crates/gitlawb-node/src/error.rs

@@ -50,6 +50,13 @@ pub enum AppError {
     #[error("too many requests: {0}")]
     TooManyRequests(String),
 
+    /// A 429 from a per-IP limiter that knows when its window frees a slot, so
+    /// the response advertises that delay as `Retry-After` the way the router
+    /// middleware's 429s do. `TooManyRequests` stays for limiters that have no
+    /// such delay to give.
+    #[error("too many requests: retry in {}s", .0.as_secs())]
+    RateLimited(std::time::Duration),
+
     #[error("incomplete: {0}")]
     Incomplete(String),
 
@@ -205,6 +212,11 @@ impl IntoResponse for AppError {
             AppError::TooManyRequests(msg) => {
                 (StatusCode::TOO_MANY_REQUESTS, "rate_limited", msg.clone())
             }
+            AppError::RateLimited(_) => (
+                StatusCode::TOO_MANY_REQUESTS,
+                "rate_limited",
+                crate::rate_limit::RATE_LIMIT_MESSAGE.to_string(),
+            ),
             AppError::Incomplete(msg) => {
                 (StatusCode::UNPROCESSABLE_ENTITY, "incomplete", msg.clone())
             }
@@ -221,6 +233,9 @@ 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.
             // 503 with a FIXED body: the caller should retry, and must not be told
             // which repo is contended or for how long.
             AppError::RepoBusy => (
@@ -315,6 +330,12 @@ impl IntoResponse for AppError {
                 axum::http::HeaderValue::from_static("1"),
             );
         }
+        if let AppError::RateLimited(retry_after) = &self {
+            resp.headers_mut().insert(
+                axum::http::header::RETRY_AFTER,
+                axum::http::HeaderValue::from(retry_after.as_secs().max(1)),
+            );
+        }
         resp
     }
 }

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

@@ -61,6 +61,37 @@ pub fn create_issue(repo_path: &Path, issue_id: &str, json: &str) -> Result<()>
     Ok(())
 }
 
+/// Deadline-bounded variant of [`create_issue`] for compensation paths: the
+/// same hash-object + update-ref write, but every git child runs through the
+/// shared bounded runner so a hung git cannot stall the write guard past
+/// `git_service_timeout`.
+pub fn create_issue_bounded(
+    git_bin: &str,
+    repo_path: &Path,
+    issue_id: &str,
+    json: &str,
+    deadline: std::time::Instant,
+) -> Result<()> {
+    let out = crate::git::visibility_pack::run_bounded_git(
+        git_bin,
+        &["hash-object", "--stdin", "-w"],
+        repo_path,
+        json.as_bytes(),
+        deadline,
+    )?;
+    let hash = String::from_utf8(out).context("git hash-object output is not valid UTF-8")?;
+    let hash = hash.trim();
+    let ref_name = format!("refs/gitlawb/issues/{issue_id}");
+    crate::git::visibility_pack::run_bounded_git(
+        git_bin,
+        &["update-ref", &ref_name, hash],
+        repo_path,
+        b"",
+        deadline,
+    )?;
+    Ok(())
+}
+
 /// Remove a single issue ref while the write guard still holds the advisory lock,
 /// after a definite publish refusal. Best-effort: a failed delete leaves a
 /// local-only orphan, which is still better than telling the client to retry into a

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

@@ -3,7 +3,8 @@
 //! Every handler that needs access to a git repo on disk goes through `RepoStore`:
 //!
 //! - `acquire()` — ensures the repo is on local disk (downloads from Tigris on cache miss).
-//! - `release_after_write()` — uploads the updated repo to Tigris after a write operation.
+//! - `upload_under_guard()`: uploads the updated repo to Tigris after a write, under a
+//!   per-repo advisory lock the caller already holds (the fork tail's durability step).
 //! - `init()` — creates a new bare repo locally and uploads to Tigris.
 //!
 //! When Tigris is disabled (bucket empty), this is a simple passthrough to local disk.
@@ -50,7 +51,7 @@ impl ConfirmedLiveTree {
     }
 }
 
-/// Centralized repo storage: local disk cache + optional Tigris backend.
+/// Centralized repo storage: local disk cache + optional object-store backend.
 #[derive(Clone)]
 pub struct RepoStore {
     repos_dir: PathBuf,
@@ -801,7 +802,22 @@ impl RepoStore {
     }
 
     /// Initialize a new bare repo on local disk and upload to Tigris.
+    /// Test-only convenience over `init_handle`; every
+    /// production caller keeps the upload inside its lock.
+    #[cfg(test)]
     pub async fn init(&self, owner_did: &str, repo_name: &str) -> Result<PathBuf> {
+        let (path, upload) = self.init_handle(owner_did, repo_name).await?;
+        drop(upload);
+        Ok(path)
+    }
+
+    /// [`init`](Self::init) plus the background archive upload's JoinHandle, so
+    /// callers holding a lock can keep the upload inside the lock's lifetime.
+    pub async fn init_handle(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<(PathBuf, Option<tokio::task::JoinHandle<Result<()>>>)> {
         let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?;
 
         store::init_bare(&local_path).context("initializing bare repo")?;
@@ -812,12 +828,13 @@ impl RepoStore {
         clear_quarantine(&local_path, repo_name);
 
         // Upload to Tigris in background
+        let mut upload_handle = None;
         if let Some(ref tigris) = self.tigris {
             let tigris = tigris.clone();
             let owner_slug = owner_slug.clone();
             let repo_name = repo_name.to_string();
             let path = local_path.clone();
-            tokio::spawn(async move {
+            upload_handle = Some(tokio::spawn(async move {
                 // Create-only, and load-bearing: this uploads a freshly
                 // initialized EMPTY repo, so a user who pushes immediately
                 // after creating one would have their archive replaced by this
@@ -828,20 +845,26 @@ impl RepoStore {
                     .upload(&owner_slug, &repo_name, &path, UploadPrecondition::IfAbsent)
                     .await
                 {
-                    Ok(_) => {}
-                    // Distinct from the warn arm: the fence refusing is the
-                    // design working, not a storage failure.
+                    Ok(_) => Ok(()),
+                    // Distinct from the error arm: the fence refusing means
+                    // someone else already published this key, so the empty
+                    // bytes dropped — but for a create caller the archive then
+                    // belongs to another writer, which IS a creation failure.
                     Err(UploadError::PreconditionLost { status }) => {
                         info!(repo = %repo_name, status, "dropped the empty-repo upload: another writer already published this repo");
+                        Err(anyhow::anyhow!(
+                            "empty-repo upload fenced: key already published (status {status})"
+                        ))
                     }
                     Err(e) => {
                         warn!(repo = %repo_name, err = %e, "failed to upload new repo to tigris");
+                        Err(anyhow::anyhow!("empty-repo upload failed: {e}"))
                     }
                 }
-            });
+            }));
         }
 
-        Ok(local_path.into_path_buf())
+        Ok((local_path.into_path_buf(), upload_handle))
     }
 
     /// Upload a repo to Tigris after a write operation (push, merge, fork, etc.).
@@ -893,6 +916,102 @@ impl RepoStore {
         Ok(())
     }
 
+    /// Whether an object store is configured. Used by purge-spam to decide
+    /// whether a repo with no local copy can be a remote-unverified candidate
+    /// (its emptiness verified under the lock after a refresh) versus fail-closed.
+    pub fn has_object_store(&self) -> bool {
+        self.tigris.is_some()
+    }
+
+    /// Try to take the per-repo advisory lock WITHOUT the write guard's
+    /// download/upload pipeline: a bare lock for callers (purge-spam, the
+    /// create/fork namespace serialization) that must serialize against the
+    /// write path on the same key but are not themselves publishing a tree.
+    ///
+    /// Returns `Some(guard)` if the lock was free, `None` if another writer holds
+    /// it (so the caller can skip rather than block). The guard holds a dedicated
+    /// pooled connection for its whole lifetime, so the lock lives on that one
+    /// connection and `release()` unlocks it on the SAME connection — a plain
+    /// pool query could unlock on a different connection and silently fail.
+    pub async fn try_lock_repo(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<Option<RepoLockGuard>> {
+        self.try_lock_repo_on(&self.lock_pool, owner_did, repo_name)
+            .await
+    }
+
+    /// [`try_lock_repo`](Self::try_lock_repo) against a caller-chosen pool. The
+    /// Postgres advisory lock is DATABASE-global, so the lock still mutually
+    /// excludes holders of the same key from any other pool — only the connection
+    /// the guard pins comes from `pool`. Backs the `lock_pool`-based delegators.
+    async fn try_lock_repo_on(
+        &self,
+        pool: &PgPool,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<Option<RepoLockGuard>> {
+        let (owner_slug, _local) = self.local_path(owner_did, repo_name)?;
+        let lock_key = advisory_lock_key(&owner_slug, repo_name);
+        let conn = pool.acquire().await.context("acquiring lock connection")?;
+        // Route through LockProbe so a cancellation racing the query cannot
+        // return a connection holding an acquired session lock to the pool:
+        // the probe's Drop closes the session unless take_conn hands it out.
+        let mut probe = LockProbe::new(conn);
+        if !probe
+            .try_lock(lock_key)
+            .await
+            .context("try advisory lock")?
+        {
+            return Ok(None);
+        }
+        let conn = probe
+            .take_conn()
+            .context("lock probe lost its connection after a confirmed acquire")?;
+        Ok(Some(RepoLockGuard {
+            conn: Some(conn),
+            lock_key,
+            repo_name: repo_name.to_string(),
+        }))
+    }
+
+    /// Bounded-wait lock-only acquire, without any object-store I/O. Retries
+    /// `try_lock_repo` with short backoff. Used by `create_repo` and `fork_repo`
+    /// to serialize their existence-check -> write -> row-insert spans against a
+    /// concurrent same-key purge (which holds this same lock across delete-row +
+    /// remove-dir). A key held past the bounded wait surfaces as `Ok(None)`,
+    /// which the create/fork handlers map to a retryable 503.
+    pub async fn lock_repo_blocking(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<Option<RepoLockGuard>> {
+        self.lock_repo_blocking_on(&self.lock_pool, owner_did, repo_name)
+            .await
+    }
+
+    /// [`lock_repo_blocking`](Self::lock_repo_blocking) against a caller-chosen
+    /// pool. The advisory lock is DB-global, so the guard mutually excludes the same
+    /// key across pools; only the pinned connection comes from `pool`. Backs the
+    /// `lock_pool`-based [`lock_repo_blocking`](Self::lock_repo_blocking) delegator.
+    pub async fn lock_repo_blocking_on(
+        &self,
+        pool: &PgPool,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<Option<RepoLockGuard>> {
+        for attempt in 0..30 {
+            if let Some(g) = self.try_lock_repo_on(pool, owner_did, repo_name).await? {
+                return Ok(Some(g));
+            }
+            if attempt < 29 {
+                tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+            }
+        }
+        Ok(None)
+    }
+
     /// Compute the local disk path and owner slug for a repo.
     ///
     /// Three-layer defence against path traversal:
@@ -1003,6 +1122,12 @@ pub(crate) fn claim_fork_disk_path(disk_path: &Path, attempt: &PublishAttemptId)
     let Some(path) = fork_attempt_path(disk_path) else {
         return;
     };
+    // The stamp may be claimed BEFORE the mirror clone starts, when the owner
+    // directory does not exist yet; create it so the claim is not silently lost
+    // (a missing stamp makes every later ownership check refuse to clean).
+    if let Some(parent) = path.parent() {
+        let _ = std::fs::create_dir_all(parent);
+    }
     if let Err(e) = std::fs::write(&path, attempt.as_str()) {
         warn!(
             path = %disk_path.display(),
@@ -1548,6 +1673,61 @@ fn invalidate_local_write_cache(local_path: &Path, repo_name: &str, reason: &str
     }
 }
 
+impl RepoStore {
+    /// Refresh the local copy from the authoritative object-store archive so an
+    /// emptiness recheck reflects remote state (used by purge-spam on a Tigris
+    /// deployment, where the admin node's local disk can be stale). Downloads
+    /// the archive over the local path when it exists; a no-op when no object
+    /// store is configured (single-machine). Errors propagate so the caller can
+    /// fail closed rather than delete on a stale-local view.
+    /// Returns the archive generation the refresh verified (its ETag), so the
+    /// caller can fence a later delete on exactly the view it checked.
+    pub async fn refresh_from_archive(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+    ) -> Result<Option<String>> {
+        let Some(ref store) = self.tigris else {
+            return Ok(None);
+        };
+        let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?;
+        let Some(generation) = store.head_generation(&owner_slug, repo_name).await? else {
+            return Ok(None);
+        };
+        store
+            .download(&owner_slug, repo_name, &local_path, None)
+            .await?;
+        Ok(generation.etag)
+    }
+
+    /// Delete the object-store archive for a repo. A no-op when no object store
+    /// is configured. Used by purge-spam so a deleted repo's archive cannot be
+    /// downloaded into a later repo created with the same owner/name. The
+    /// delete is fenced on the generation the store sees, so a cross-node
+    /// publish racing the purge makes it fail rather than erasing the
+    /// successor's archive.
+    /// `expected` is the generation the caller verified empty (an ETag from
+    /// [`refresh_from_archive`]); a moved generation means a PUT landed after
+    /// the check and the archive must be kept, so `Ok(false)` is a skip.
+    pub async fn delete_archive(
+        &self,
+        owner_did: &str,
+        repo_name: &str,
+        expected: Option<&str>,
+    ) -> Result<bool> {
+        let Some(ref store) = self.tigris else {
+            return Ok(true);
+        };
+        let (owner_slug, _local_path) = self.local_path(owner_did, repo_name)?;
+        store.delete_current(&owner_slug, repo_name, expected).await
+    }
+
+    /// The bound applied to transfers that run while a repo lock is held.
+    pub(crate) fn transfer_bound(&self) -> std::time::Duration {
+        self.lock_held_transfer_timeout
+    }
+}
+
 /// Strict allowlist validator for `owner_did` and `repo_name`.
 ///
 /// Rejects any character that isn't explicitly safe, plus length and
@@ -1704,7 +1884,7 @@ fn validate_owner_did(owner_did: &str) -> Result<()> {
     Ok(())
 }
 
-fn validate_repo_name(repo_name: &str) -> Result<()> {
+pub(crate) fn validate_repo_name(repo_name: &str) -> Result<()> {
     if repo_name.is_empty() {
         anyhow::bail!("repo_name is empty");
     }
@@ -2755,6 +2935,41 @@ pub fn build_lock_pool(source: &PgPool, max_connections: u32, acquire_timeout: D
         .connect_lazy_with((*source.connect_options()).clone())
 }
 
+/// Lock-only guard from [`RepoStore::try_lock_repo`]. Holds the Postgres advisory
+/// lock (and the dedicated connection it lives on) until `release()`. No Tigris
+/// I/O — unlike [`RepoWriteGuard`], releasing does NOT upload anything.
+pub struct RepoLockGuard {
+    conn: Option<sqlx::pool::PoolConnection<sqlx::Postgres>>,
+    lock_key: i64,
+    repo_name: String,
+}
+
+impl RepoLockGuard {
+    /// Release the advisory lock on the same connection that took it, then return
+    /// the connection to the pool.
+    pub async fn release(mut self) {
+        if let Some(mut conn) = self.conn.take() {
+            let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
+                .bind(self.lock_key)
+                .execute(&mut *conn)
+                .await;
+        }
+    }
+}
+
+impl Drop for RepoLockGuard {
+    fn drop(&mut self) {
+        // Same guarantee as RepoWriteGuard: a lock guard dropped without
+        // release() closes its connection so Postgres frees the advisory lock,
+        // rather than returning a lock-holding connection to the pool.
+        if let Some(mut conn) = self.conn.take() {
+            warn!(repo = %self.repo_name,
+                "RepoLockGuard dropped without release() — closing its connection to free the advisory lock");
+            conn.close_on_drop();
+        }
+    }
+}
+
 /// Compute a stable i64 hash for a Postgres advisory lock key.
 ///
 /// Uses SHA-256 (not `DefaultHasher`) so the same `(owner_slug, repo_name)`
@@ -2780,7 +2995,7 @@ pub(crate) fn advisory_lock_key(owner_slug: &str, repo_name: &str) -> i64 {
     i64::from_le_bytes(digest[..8].try_into().expect("sha256 output is >= 8 bytes"))
 }
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
     use crate::git::tigris::ATTEMPT_METADATA_KEY;
     use std::collections::HashMap;
@@ -4917,7 +5132,7 @@ mod tests {
     }
 
     /// Seed the minimum bare-repo shape so the upload has something to archive.
-    fn seed_bare_repo(path: &Path) {
+    pub(crate) fn seed_bare_repo(path: &Path) {
         std::fs::create_dir_all(path.join("objects/info")).unwrap();
         std::fs::create_dir_all(path.join("refs/heads")).unwrap();
         std::fs::write(path.join("HEAD"), "ref: refs/heads/main\n").unwrap();
@@ -5080,7 +5295,7 @@ mod tests {
     /// models the real S3 arm we care about (body fully transmitted, commit
     /// decided later) with no timing in it.
     #[derive(Clone, Debug)]
-    struct CapturedPut {
+    pub(crate) struct CapturedPut {
         key: String,
         body: Vec<u8>,
         if_match: Option<String>,
@@ -5090,7 +5305,7 @@ mod tests {
 
     /// One stored object.
     #[derive(Clone, Debug)]
-    struct MockObject {
+    pub(crate) struct MockObject {
         body: Vec<u8>,
         etag: String,
         /// The `x-amz-meta-gitlawb-attempt` it was written with. This is what
@@ -5103,14 +5318,14 @@ mod tests {
     /// `status` is `None` while a PUT is parked: it arrived and was logged, but
     /// no precondition has been evaluated for it yet.
     #[derive(Clone, Debug, PartialEq)]
-    struct PutAttempt {
+    pub(crate) struct PutAttempt {
         if_match: Option<String>,
         if_none_match: Option<String>,
         status: Option<u16>,
     }
 
     #[derive(Default)]
-    struct MockState {
+    pub(crate) struct MockState {
         /// Keyed by request path. A single-slot store was enough while every test
         /// drove one repo; fork creation touches the SOURCE key and the FORK key
         /// in one request, and collapsing them would make an assertion about one
@@ -5145,7 +5360,7 @@ mod tests {
     ///
     /// The fence tests downstream are only worth anything if a precondition can
     /// actually fail here, so this helper carries its own semantics tests below.
-    struct S3Mock {
+    pub(crate) struct S3Mock {
         endpoint: String,
         state: Arc<std::sync::Mutex<MockState>>,
         gate: Arc<tokio::sync::Notify>,
@@ -5215,7 +5430,7 @@ mod tests {
     }
 
     impl S3Mock {
-        async fn start() -> Self {
+        pub(crate) async fn start() -> Self {
             use axum::response::IntoResponse;
 
             let state = Arc::new(std::sync::Mutex::new(MockState::default()));
@@ -5424,7 +5639,7 @@ mod tests {
             }
         }
 
-        fn endpoint(&self) -> &str {
+        pub(crate) fn endpoint(&self) -> &str {
             &self.endpoint
         }
 
@@ -5445,7 +5660,7 @@ mod tests {
         }
 
         /// The stored object under a specific repo key.
-        fn object_for(&self, owner_slug: &str, repo_name: &str) -> Option<MockObject> {
+        pub(crate) fn object_for(&self, owner_slug: &str, repo_name: &str) -> Option<MockObject> {
             let key = format!("test-bucket/repos/v1/{owner_slug}/{repo_name}.tar.zst");
             self.state.lock().unwrap().objects.get(&key).cloned()
         }
@@ -5459,7 +5674,7 @@ mod tests {
         /// to pick up a precondition and then PUTs on it is therefore fencing
         /// on a generation that is already gone, which is the only way to drive
         /// two consecutive lost preconditions deterministically.
-        fn roll_generation_after_next_heads(&self, n: u32) {
+        pub(crate) fn roll_generation_after_next_heads(&self, n: u32) {
             self.state.lock().unwrap().roll_after_heads = n;
         }
 
@@ -5478,7 +5693,7 @@ mod tests {
         /// Deliver the next PUT in full and then fail it WITHOUT committing. The
         /// client's knowledge is identical to the arm above — that is the whole
         /// point — so only asking the store can tell the two apart.
-        fn fail_next_put_after_delivery(&self) {
+        pub(crate) fn fail_next_put_after_delivery(&self) {
             self.state.lock().unwrap().lose_next_response = Some(false);
         }
 
@@ -5489,7 +5704,7 @@ mod tests {
 
         /// Fail the next HEAD for ONE repo key with a 500, so a caller's
         /// reconciliation cannot be answered either way.
-        fn fail_next_head_for(&self, owner_slug: &str, repo_name: &str) {
+        pub(crate) fn fail_next_head_for(&self, owner_slug: &str, repo_name: &str) {
             let key = format!("test-bucket/repos/v1/{owner_slug}/{repo_name}.tar.zst");
             *self
                 .state
@@ -5501,7 +5716,7 @@ mod tests {
         }
 
         /// How many DELETEs the mock actually carried out.
-        fn deletes(&self) -> u32 {
+        pub(crate) fn deletes(&self) -> u32 {
             self.state.lock().unwrap().deletes
         }
 
@@ -5531,7 +5746,7 @@ mod tests {
             .0
         }
 
-        fn shutdown(&self) {
+        pub(crate) fn shutdown(&self) {
             self.server.abort();
         }
     }
@@ -6262,25 +6477,25 @@ mod tests {
 
     /// A client aimed at the same key the store under test publishes to, so a
     /// test can seed the archive or land an interfering publish of its own.
-    fn mock_tigris(mock: &S3Mock) -> TigrisClient {
+    pub(crate) fn mock_tigris(mock: &S3Mock) -> TigrisClient {
         TigrisClient::for_testing_with_endpoint("test-bucket", mock.endpoint())
     }
 
     /// The slug `local_path` derives from a DID, needed because a test seeds
     /// and reads the archive key directly.
-    fn owner_slug_of(owner_did: &str) -> String {
+    pub(crate) fn owner_slug_of(owner_did: &str) -> String {
         owner_did.replace([':', '/'], "_")
     }
 
     /// A bare-repo-shaped directory carrying `marker`, so a test can tell whose
     /// tree is stored without comparing compressed bytes.
-    fn marked_repo(path: &Path, marker: &str) {
+    pub(crate) fn marked_repo(path: &Path, marker: &str) {
         seed_bare_repo(path);
         std::fs::write(path.join("MARKER"), marker).unwrap();
     }
 
     /// The marker inside whatever archive is currently stored under the key.
-    async fn stored_marker(mock: &S3Mock, owner_slug: &str, repo_name: &str) -> String {
+    pub(crate) async fn stored_marker(mock: &S3Mock, owner_slug: &str, repo_name: &str) -> String {
         let out = TempDir::new().unwrap();
         let validated = super::validated_repo_disk_path(out.path(), "did:key:stored", "stored")
             .expect("test repo path must validate");

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

@@ -1817,6 +1817,10 @@ pub fn repo_disk_path(repos_dir: &Path, owner_did: &str, repo_name: &str) -> std
 
 #[cfg(test)]
 mod tests {
+    use super::{branch_diff_names, merge_branch};
+    use std::path::Path;
+    use std::process::Command;
+
     use super::{validate_git_ref, validate_git_ref_with_git, GitRefValidationError};
 
     #[test]
@@ -1883,10 +1887,6 @@ mod tests {
         );
     }
 
-    use super::{branch_diff_names, merge_branch};
-    use std::path::Path;
-    use std::process::Command;
-
     /// #26 split 1/4: a bare repo this node creates must KEEP REFLOGS, because
     /// the durable-outbox reconcile has no other way to prove that a stranded
     /// transition actually landed. `core.logAllRefUpdates` defaults to false for

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

@@ -522,6 +522,49 @@ impl TigrisClient {
         }
     }
 
+    /// Delete whatever archive currently exists, fenced on the generation read
+    /// at call time: a concurrent publish between the HEAD and the DELETE makes
+    /// the store refuse rather than erasing the successor's object. Used by
+    /// purge-spam, which deletes a repo it did not publish (so
+    /// [`delete_if_attempt_matches`](Self::delete_if_attempt_matches) cannot
+    /// apply) while still needing the same moved-generation protection.
+    /// Returns `true` when an archive was deleted.
+    /// Delete only if the object's generation still matches `expected` — the
+    /// generation the caller checked for emptiness. A PUT landing between that
+    /// check and this delete makes the generation move; deleting the newer
+    /// (possibly non-empty) archive is exactly the outcome the fence exists to
+    /// prevent, so a moved generation refuses rather than deletes.
+    pub async fn delete_current(
+        &self,
+        owner_slug: &str,
+        repo_name: &str,
+        expected: Option<&str>,
+    ) -> Result<bool> {
+        let Some(generation) = self.head_generation(owner_slug, repo_name).await? else {
+            return Ok(false);
+        };
+        if generation.etag.as_deref() != expected {
+            return Ok(false);
+        }
+        let key = Self::repo_key(owner_slug, repo_name);
+        let mut req = self.s3.delete_object().bucket(&self.bucket).key(&key);
+        if let Some(etag) = generation.etag.as_deref() {
+            req = req.if_match(etag);
+        }
+        match req.send().await {
+            Ok(_) => Ok(true),
+            Err(e) => {
+                if e.raw_response()
+                    .map(|raw| raw.status().as_u16())
+                    .is_some_and(|s| s == 412 || s == 409)
+                {
+                    anyhow::bail!("archive generation moved between head and delete; refusing");
+                }
+                Err(anyhow::anyhow!("tigris DELETE {key}: {e}"))
+            }
+        }
+    }
+
     /// Download a repo archive from Tigris and extract to local disk.
     pub async fn download(
         &self,

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

@@ -14,6 +14,11 @@ use subscription::SubscriptionRoot;
 
 pub type GitlawbSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;
 
+/// No mutations: the /graphql/ws transport keeps queries and subscriptions but
+/// has no mutation root, so the socket can never be used to run a write and
+/// sidestep the write-brake bucket the HTTP route pays.
+pub type GitlawbWsSchema = Schema<QueryRoot, async_graphql::EmptyMutation, SubscriptionRoot>;
+
 // Keep ordinary schema discovery and application queries usable while
 // bounding validation work and the number of resolver selections per request.
 const GRAPHQL_MAX_COMPLEXITY: usize = 400;
@@ -171,6 +176,30 @@ pub fn build_schema(
         .finish()
 }
 
+/// Build the subscriptions-only schema served over /graphql/ws. Carries the
+/// same broadcast senders the subscription resolvers consume. Mutations are
+/// disabled — the schema's mutation root is `EmptyMutation` — but queries still
+/// run via `QueryRoot` (they are reads and already ride the TaskReadBrake), so
+/// "subscriptions-only" refers to the write surface, not the operation set.
+pub fn build_ws_schema(
+    db: Arc<Db>,
+    ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
+    task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
+    task_cursor_key: crate::api::task_cursor::TaskCursorKey,
+) -> GitlawbWsSchema {
+    apply_query_limits(Schema::build(
+        QueryRoot,
+        async_graphql::EmptyMutation,
+        SubscriptionRoot,
+    ))
+    .data(db)
+    .data(ref_update_tx)
+    .data(task_event_tx)
+    .data(task_cursor_key)
+    .extension(TaskReadBrakeExtension)
+    .finish()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/gitlawb-node/src/ipfs_pin.rs

@@ -187,6 +187,11 @@ where
     F: std::future::Future<Output = Result<T>>,
 {
     let left = deadline.saturating_duration_since(Instant::now());
+    // `timeout` polls the future before it checks the timer, so a zero budget
+    // would still start the work. A spent deadline must not.
+    if left.is_zero() {
+        return Err(BoundedDbError::Elapsed);
+    }
     match tokio::time::timeout(left, fut).await {
         Ok(Ok(v)) => Ok(v),
         Ok(Err(e)) => Err(BoundedDbError::Db(e)),
@@ -3433,6 +3438,30 @@ mod tests {
         );
     }
 
+    /// A spent deadline must not start the work at all. `tokio::time::timeout`
+    /// polls the inner future before it checks the timer, so a future that is
+    /// ready on its first poll would otherwise slip through a zero budget and
+    /// return a value. Deterministic: the future has no await point.
+    #[tokio::test]
+    async fn db_bounded_spent_deadline_does_not_poll_a_ready_future() {
+        let spent = Instant::now() - Duration::from_secs(5);
+        let polled = std::sync::atomic::AtomicBool::new(false);
+        let out: Result<u8, BoundedDbError> = db_bounded(spent, async {
+            polled.store(true, std::sync::atomic::Ordering::SeqCst);
+            Ok(7u8)
+        })
+        .await;
+
+        assert!(
+            matches!(out, Err(BoundedDbError::Elapsed)),
+            "a spent deadline must yield the timeout arm even for a ready future: {out:?}"
+        );
+        assert!(
+            !polled.load(std::sync::atomic::Ordering::SeqCst),
+            "a spent deadline must not start the work"
+        );
+    }
+
     /// U5 (#173): the elapsed arm of the discovery record leaves neither row behind.
     ///
     /// What this proves and what it does NOT: it shows the wrapper composition

crates/gitlawb-node/src/lib.rs

@@ -3,6 +3,7 @@
 // migrations) so the out-of-crate deny-harness integration crate can spawn a
 // real node; `src/main.rs` is a thin shim over `run()`. Modules the harness
 // reaches are `pub`; the rest stay crate-private.
+mod admin;
 mod ans104;
 pub mod api;
 mod arweave;
@@ -84,6 +85,12 @@ pub async fn run() -> 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);
@@ -327,6 +334,12 @@ pub async fn run() -> Result<()> {
         task_event_tx.clone(),
         task_cursor_key.clone(),
     ));
+    let graphql_ws_schema = Arc::new(graphql::build_ws_schema(
+        Arc::clone(&db),
+        ref_update_tx.clone(),
+        task_event_tx.clone(),
+        task_cursor_key.clone(),
+    ));
 
     let machine_id = std::env::var("FLY_MACHINE_ID").ok();
     if let Some(ref mid) = machine_id {
@@ -382,36 +395,26 @@ pub async fn run() -> 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, issue close, labels, stars, merges, protect, replicas, webhook
+    // create/delete, visibility, tasks, bounties, profile, every HTTP /graphql
+    // request). Issue comment/close debit it in the handler, after the read
+    // gate; the rest via the router-level brake. 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");
 
     // close_issue drives a full archive download for non-owner author pre-checks.
     // Keep it on its own bucket so it cannot drain the receive-pack limit.
@@ -436,7 +439,7 @@ pub async fn run() -> 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
@@ -477,10 +480,12 @@ pub async fn run() -> Result<()> {
         ref_update_tx,
         task_event_tx,
         graphql_schema,
+        graphql_ws_schema,
         machine_id,
         repo_store,
         rate_limiter,
         create_ip_rate_limiter,
+        write_rate_limiter,
         push_rate_limiter,
         close_issue_rate_limiter,
         ipfs_max_history_walks: crate::api::ipfs::MAX_HISTORY_WALKS_PER_REQUEST,
@@ -630,14 +635,17 @@ pub async fn run() -> Result<()> {
 
     // Periodic cleanup of expired rate limit entries + consumed-proof ledger
     {
-        let sweep_state = state.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)) => {
-                        sweep_state.sweep_rate_limiters().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)
@@ -884,6 +892,110 @@ pub async fn run() -> 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<git::tigris::TigrisClient> = if config.tigris_bucket.is_empty()
+            {
+                None
+            } else {
+                match git::tigris::TigrisClient::new(&config.tigris_bucket).await {
+                    Ok(client) => Some(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::lock_pool(
+                &config.database_url,
+                config.db_lock_pool_max_connections,
+                acquire_timeout,
+            )?;
+            let repo_store = git::repo_store::RepoStore::new(
+                config.repos_dir.clone(),
+                object_store,
+                lock_pool,
+                std::time::Duration::from_secs(config.lock_held_transfer_timeout_secs),
+            );
+            admin::run_purge_spam(&db, &repo_store, &config.repos_dir, execute)
+                .await
+                .map(|_| ())
+        }
+    }
+}
+
 /// U4 (#173): spawn the periodic legacy provider-CID repair sweep. Releases before this
 /// version stored the PROVIDER CID (Kubo dag-pb / Pinata CIDv0) in `pinned_cids.cid`,
 /// and this version's `/ipfs/{cid}` resolver withholds any row whose stored key is not
@@ -1485,6 +1597,7 @@ mod rate_limiter_sweep_tests {
         let mut state = crate::test_support::test_state_lazy();
         state.rate_limiter = RateLimiter::new(10, window);
         state.create_ip_rate_limiter = RateLimiter::new(10, window);
+        state.write_rate_limiter = RateLimiter::new(10, window);
         state.push_rate_limiter = RateLimiter::new(10, window);
         state.close_issue_rate_limiter = RateLimiter::new(10, window);
         state.sync_trigger_rate_limiter = RateLimiter::new(10, window);
@@ -1498,6 +1611,7 @@ mod rate_limiter_sweep_tests {
             [
                 s.rate_limiter.clone(),
                 s.create_ip_rate_limiter.clone(),
+                s.write_rate_limiter.clone(),
                 s.push_rate_limiter.clone(),
                 s.close_issue_rate_limiter.clone(),
                 s.sync_trigger_rate_limiter.clone(),
@@ -1514,7 +1628,7 @@ mod rate_limiter_sweep_tests {
         }
 
         tokio::time::sleep(window * 3).await;
-        state.sweep_rate_limiters().await;
+        state.cleanup_rate_limiters().await;
 
         for (i, l) in limiters(&state).into_iter().enumerate() {
             assert_eq!(l.tracked_keys().await, 0, "limiter {i} was not swept");
@@ -1725,6 +1839,40 @@ async fn gossip_ping_round(
     }
 }
 
+#[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::{

crates/gitlawb-node/src/rate_limit.rs

@@ -70,18 +70,21 @@ impl RateLimiter {
         self.window.min(Duration::from_secs(1))
     }
 
-    /// The configured window, exposed so a 429 can carry an honest
-    /// `retry-after`: a hardcoded value lies when the operator retunes the
-    /// window, and a client that honors the header would retry into another
-    /// 429 until the real window elapses.
-    pub fn window(&self) -> Duration {
-        self.window
+    /// 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.
+    pub(crate) async fn check(&self, key: &str) -> bool {
+        self.check_retry(key).await.is_none()
     }
 
-    pub(crate) async fn check(&self, key: &str) -> bool {
+    /// 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;
@@ -93,10 +96,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
@@ -117,7 +135,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(
@@ -126,7 +148,23 @@ impl RateLimiter {
                 timestamps: vec![now],
             },
         );
-        true
+        None
+    }
+
+    /// Give back one slot a prior [`check`](Self::check) took for `key`, for a
+    /// caller that debits two buckets and must undo the first when the second
+    /// refuses. Drops the newest timestamp: the count is what the cap reads, so
+    /// that is equivalent to dropping the caller's own even if another request
+    /// landed in between. An unknown key is a no-op, and a key left with no
+    /// timestamps is removed so a refund never holds a map entry.
+    pub(crate) async fn refund(&self, key: &str) {
+        let mut state = self.state.lock().await;
+        if let Some(window) = state.get_mut(key) {
+            window.timestamps.pop();
+            if window.timestamps.is_empty() {
+                state.remove(key);
+            }
+        }
     }
 
     /// Non-consuming check: is this key ALREADY at its limit for the current window?
@@ -178,6 +216,19 @@ impl RateLimiter {
         });
     }
 
+    /// 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.
     /// Number of distinct keys currently tracked. Test-only introspection so a
     /// cross-module test can assert that a sweep actually evicted expired entries
     /// and observe what it reclaimed. There is no production reader.
@@ -185,6 +236,23 @@ impl RateLimiter {
     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 });
+    }
 }
 
 /// Per-source concurrency cap derived from the write-pool size: one resolved client
@@ -300,8 +368,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 too_many_requests(limiter.window().as_secs());
+        if let Some(retry_after) = limiter.check_retry(&did).await {
+            return too_many_requests(retry_after);
         }
     }
 
@@ -404,18 +472,21 @@ impl<S: Send + Sync> axum::extract::FromRequestParts<S> for PeerAddr {
     }
 }
 
+/// 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).
 /// The message every per-IP brake answers with, shared so the GraphQL surface
 /// (which cannot return a 429 status inside a 200 GraphQL envelope) says the
 /// same thing as the REST routes.
 pub const RATE_LIMIT_MESSAGE: &str = "rate limit exceeded — try again later";
 
-/// 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(retry_after_secs: u64) -> 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", retry_after_secs.to_string())],
+        [("retry-after", retry_after.as_secs().to_string())],
         RATE_LIMIT_MESSAGE,
     )
         .into_response()
@@ -482,9 +553,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(limiter.limiter.window().as_secs());
+                return too_many_requests(retry_after);
             }
         }
     }
@@ -870,4 +941,221 @@ 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}"
+        );
+    }
+
+    /// `refund` gives back exactly one slot a prior `check` took, so a caller
+    /// that debits two buckets can undo the first when the second refuses.
+    #[tokio::test]
+    async fn refund_returns_one_slot_and_is_a_no_op_for_an_unknown_key() {
+        let limiter = RateLimiter::new(2, Duration::from_secs(60));
+        assert!(limiter.check("k").await);
+        assert!(limiter.check("k").await);
+        assert!(!limiter.check("k").await, "bucket is full");
+
+        limiter.refund("k").await;
+        assert!(limiter.check("k").await, "one slot came back");
+        assert!(!limiter.check("k").await, "and only one");
+
+        // An unknown key must neither panic nor create an entry.
+        limiter.refund("never-seen").await;
+        assert!(limiter.state.lock().await.get("never-seen").is_none());
+
+        // Refunding the only slot drops the key, so a refunded key does not
+        // hold a map entry (the limiter's key cap bounds its own state).
+        let single = RateLimiter::new(1, Duration::from_secs(60));
+        assert!(single.check("solo").await);
+        single.refund("solo").await;
+        assert!(single.state.lock().await.get("solo").is_none());
+    }
 }

crates/gitlawb-node/src/reconciliation.rs

@@ -4343,10 +4343,18 @@ mod tests {
         let blob_a = repo_on_disk.git(&["rev-parse", "HEAD:a.txt"]);
         let blob_b = repo_on_disk.git(&["rev-parse", "HEAD:b.txt"]);
         let blob_c = repo_on_disk.git(&["rev-parse", "HEAD:c.txt"]);
-        let commit = repo_on_disk.git(&["rev-parse", "HEAD"]);
-        let tree = repo_on_disk.git(&["rev-parse", "HEAD^{tree}"]);
-        let mut all = [blob_a.clone(), blob_b.clone(), blob_c.clone(), commit, tree];
+        // Every object the sweep can plan, not just HEAD's: the three commits
+        // and three trees are all reachable, and an earlier commit's id changes
+        // with the commit timestamp. Sorting only HEAD's commit, tree and blobs
+        // picked the wrong "first" whenever an earlier commit sorted below them.
+        let mut all: Vec<String> = repo_on_disk
+            .git(&["rev-list", "--objects", "--all"])
+            .lines()
+            .filter_map(|l| l.split_whitespace().next())
+            .map(str::to_string)
+            .collect();
         all.sort();
+        assert_eq!(all.len(), 9, "3 commits + 3 trees + 3 blobs");
         let first = all[0].clone();
 
         let owner = "did:key:zAbortCursorOwner";

crates/gitlawb-node/src/server.rs

@@ -67,7 +67,7 @@ async fn graphql_ws_handler(
         key: rate_limit::client_key(&headers, peer, state.push_limiter_trust),
         request_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
     });
-    let schema = state.graphql_schema.as_ref().clone();
+    let schema = state.graphql_ws_schema.as_ref().clone();
     upgrade
         .protocols(ALL_WEBSOCKET_PROTOCOLS)
         .on_upgrade(move |stream| {
@@ -118,18 +118,57 @@ fn add_auth_layers(
         ))
 }
 
+/// 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 graphql_routes = Router::new()
+    // 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_http = 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 (mounted on
+        // its own router below, read-only subscriptions) gets no brake. The
+        // per-IP write brake covers /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 stays unbraked.
+        .layer(middleware::from_fn_with_state(
+            auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
+            auth::optional_signature,
+        ))
+        .write_braked(&write_ip_limiter);
+    let graphql_ws = Router::new()
         .route("/graphql/ws", get(graphql_ws_handler))
-        // Attach the verified DID to /graphql and /graphql/ws when a signature is present.
         .layer(middleware::from_fn_with_state(
             auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
             auth::optional_signature,
         ));
+    let graphql_routes = graphql_http.merge(graphql_ws);
 
-    // ── 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))
@@ -138,7 +177,8 @@ pub fn build_router(state: AppState) -> Router {
             .route("/api/v1/tasks/{id}/fail", post(tasks::fail_task)),
         state.clone(),
         SIGNED_BODY_LIMIT,
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // ── Task routes (read — open, but scoped) ──────────────────────────────
     // `optional_signature` attaches the verified DID when a signature is present
@@ -197,7 +237,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(
@@ -268,7 +314,8 @@ pub fn build_router(state: AppState) -> Router {
             ),
         state.clone(),
         SIGNED_BODY_LIMIT,
-    );
+    )
+    .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.
@@ -381,7 +428,8 @@ pub fn build_router(state: AppState) -> Router {
             ),
         state.clone(),
         SIGNED_BODY_LIMIT,
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
     // ── Bounty routes (read — open) ──────────────────────────────────────
     let bounty_read_routes = Router::new()
@@ -406,9 +454,15 @@ pub fn build_router(state: AppState) -> Router {
         Router::new().route("/api/v1/profile", axum::routing::put(profiles::set_profile)),
         state.clone(),
         SIGNED_BODY_LIMIT,
-    );
+    )
+    .write_braked(&write_ip_limiter);
 
-    // ── Issue routes (write — require HTTP Signature, no rate limit) ─────
+    // ── Issue routes (write — require HTTP Signature; NO outer write brake) ──
+    // Both handlers are read-gated before they mutate, so an outer per-IP
+    // brake would answer 429 before authorize_repo_read could return the
+    // opaque 404 a hidden repo owes the caller — the brake's status would
+    // distinguish a real repo from a missing one (INV-12). The write bucket
+    // is debited inside each handler, right after the read gate.
     let issue_write_routes = add_auth_layers(
         Router::new()
             .route(

crates/gitlawb-node/src/state.rs

@@ -72,6 +72,7 @@ pub struct AppState {
     pub task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
     /// GraphQL schema (queries + mutations + subscriptions)
     pub graphql_schema: Arc<crate::graphql::GitlawbSchema>,
+    pub graphql_ws_schema: Arc<crate::graphql::GitlawbWsSchema>,
     /// MAC key for task-list continuation tokens, derived from the node
     /// keypair seed. Held here (and handed to the GraphQL schema) so REST and
     /// GraphQL mint and accept the same tokens.
@@ -92,6 +93,21 @@ 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, webhook create/delete, visibility (its GET
+    /// shares the group), tasks, bounties, profile, and every HTTP `/graphql`
+    /// request (queries and the playground GET as well as mutations; the
+    /// WebSocket route is excluded). Issue comment and issue close debit it in
+    /// the handler after the read gate, not via router middleware; close also
+    /// draws on `close_issue_rate_limiter` and spends neither bucket when
+    /// either refuses. 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.
@@ -372,24 +388,6 @@ impl AppState {
         self.shutdown_tx.subscribe()
     }
 
-    /// Sweep expired entries from every per-IP/DID rate limiter. Driven by the
-    /// periodic cleanup task so a bounded limiter's key map sheds stale entries
-    /// instead of sitting near its cap until an inline capacity sweep reclaims
-    /// them. Every limiter on the state is swept here; adding a new limiter means
-    /// adding it to this list.
-    pub(crate) async fn sweep_rate_limiters(&self) {
-        self.rate_limiter.cleanup().await;
-        self.create_ip_rate_limiter.cleanup().await;
-        self.push_rate_limiter.cleanup().await;
-        self.close_issue_rate_limiter.cleanup().await;
-        self.ipfs_rate_limiter.cleanup().await;
-        self.ipfs_work_rate_limiter.cleanup().await;
-        self.sync_trigger_rate_limiter.cleanup().await;
-        self.peer_write_rate_limiter.cleanup().await;
-        self.task_read_rate_limiter.cleanup().await;
-        self.federated_rate_limiter.cleanup().await;
-    }
-
     /// Trigger graceful shutdown. Idempotent — calling more than once
     /// has no effect. Returns `true` if this call was the one that
     /// flipped the signal.
@@ -411,6 +409,26 @@ impl AppState {
         *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.close_issue_rate_limiter.cleanup().await;
+        self.ipfs_rate_limiter.cleanup().await;
+        self.ipfs_work_rate_limiter.cleanup().await;
+        self.sync_trigger_rate_limiter.cleanup().await;
+        self.peer_write_rate_limiter.cleanup().await;
+        self.task_read_rate_limiter.cleanup().await;
+        self.federated_rate_limiter.cleanup().await;
+    }
+
     /// Legacy-probe budget wired from the `GITLAWB_IPFS_MAX_LEGACY_PROBES` operator
     /// knob. The knob seeds `ipfs_max_legacy_probes` at construction so it controls the
     /// per-request legacy (NULL-provenance) probe fan-out it advertises. It deliberately
@@ -1273,6 +1291,35 @@ mod repo_identity_key_tests {
     }
 }
 
+#[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)"
+        );
+    }
+}
+
 #[cfg(test)]
 mod repo_write_lease_tests {
     use super::RepoWriteLeases;

crates/gitlawb-node/src/test_harness.rs

@@ -170,6 +170,12 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool, repos_dir: PathBuf) -> AppS
         task_tx.clone(),
         task_cursor_key.clone(),
     ));
+    let ws_schema = Arc::new(crate::graphql::build_ws_schema(
+        db.clone(),
+        ref_tx.clone(),
+        task_tx.clone(),
+        task_cursor_key.clone(),
+    ));
     AppState {
         config: Arc::new(Config::parse_from(["gitlawb-node"])),
         db,
@@ -180,11 +186,13 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool, repos_dir: PathBuf) -> AppS
         ref_update_tx: ref_tx,
         task_event_tx: task_tx,
         graphql_schema: schema,
+        graphql_ws_schema: ws_schema,
         task_cursor_key,
         machine_id: None,
         repo_store: crate::git::repo_store::RepoStore::for_testing(repos_dir, 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(600, Duration::from_secs(3600)),
         push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
         close_issue_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
         federated_rate_limiter: RateLimiter::new(12, Duration::from_secs(60)),

crates/gitlawb-node/src/test_support.rs

@@ -95,6 +95,13 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         task_tx.clone(),
         task_cursor_key.clone(),
     ));
+
+    let ws_schema = Arc::new(graphql::build_ws_schema(
+        db.clone(),
+        ref_tx.clone(),
+        task_tx.clone(),
+        task_cursor_key.clone(),
+    ));
     AppState {
         config: Arc::new(Config::parse_from(["gitlawb-node"])),
         db,
@@ -105,11 +112,13 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         ref_update_tx: ref_tx,
         task_event_tx: task_tx,
         graphql_schema: schema,
+        graphql_ws_schema: ws_schema,
         task_cursor_key,
         machine_id: None,
         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)),
         close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
         ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
@@ -3428,7 +3437,7 @@ mod tests {
         run(&["commit", "-q", "-m", "seed"], &src);
         // The tag that shadows the missing target branch: "release" resolves via
         // DWIM but refs/heads/release never exists.
-        run(&["tag", "release"], &src);
+        run(&["tag", "-m", "release", "release"], &src);
         run(&["checkout", "-q", "-b", "feature"], &src);
         std::fs::write(src.join("feature.txt"), b"feature-only").unwrap();
         run(&["add", "feature.txt"], &src);
@@ -3588,7 +3597,7 @@ mod tests {
         // moves past it. Revision DWIM prefers refs/tags/ over refs/heads/ for
         // an ambiguous short name, so this fixture is the strongest shadowing
         // shape: a naive resolution would merge onto the stale tag commit.
-        run(&["tag", "release"], &src);
+        run(&["tag", "-m", "release", "release"], &src);
         run(&["checkout", "-q", "-b", "release"], &src);
         std::fs::write(src.join("release.txt"), b"release-only").unwrap();
         run(&["add", "release.txt"], &src);
@@ -17954,12 +17963,12 @@ mod tests {
     }
 
     /// The periodic cleanup task must sweep the ipfs walk limiter, not only its
-    /// five siblings. Drives `AppState::sweep_rate_limiters` — the exact method the
+    /// five siblings. Drives `AppState::cleanup_rate_limiters` — the exact method the
     /// 300s loop calls — and asserts the ipfs limiter's expired entry is evicted.
     /// Dropping `ipfs_rate_limiter.cleanup()` from that method leaves the entry in
     /// place (`tracked_keys` stays 1): the RED proof that the sweep covers it.
     #[sqlx::test]
-    async fn sweep_rate_limiters_includes_ipfs_limiter(pool: PgPool) {
+    async fn cleanup_rate_limiters_includes_ipfs_limiter(pool: PgPool) {
         let mut state = test_state(pool).await;
         // Short window so a single recorded hit is already expired at sweep time.
         state.ipfs_rate_limiter = crate::rate_limit::RateLimiter::new(5, Duration::from_millis(50));
@@ -17976,7 +17985,7 @@ mod tests {
 
         // Expire the entry (still mapped — cleanup hasn't run), then sweep.
         tokio::time::sleep(Duration::from_millis(60)).await;
-        state.sweep_rate_limiters().await;
+        state.cleanup_rate_limiters().await;
 
         assert_eq!(
             state.ipfs_rate_limiter.tracked_keys().await,
@@ -18279,12 +18288,12 @@ mod tests {
 
     /// U5 (R6): the periodic cleanup task sweeps the NEW work-budget limiter too, not
     /// only the route limiter and its siblings. Mirrors
-    /// `sweep_rate_limiters_includes_ipfs_limiter`: drive `sweep_rate_limiters` and
+    /// `cleanup_rate_limiters_includes_ipfs_limiter`: drive `cleanup_rate_limiters` and
     /// assert the work limiter's expired entry is evicted. Dropping the
     /// `ipfs_work_rate_limiter.cleanup()` call from that method leaves the entry in place
     /// (`tracked_keys` stays 1): the RED proof the sweep covers it.
     #[sqlx::test]
-    async fn sweep_rate_limiters_includes_ipfs_work_limiter(pool: PgPool) {
+    async fn cleanup_rate_limiters_includes_ipfs_work_limiter(pool: PgPool) {
         let mut state = test_state(pool).await;
         state.ipfs_work_rate_limiter =
             crate::rate_limit::RateLimiter::new(5, Duration::from_millis(50));
@@ -18300,7 +18309,7 @@ mod tests {
         );
 
         tokio::time::sleep(Duration::from_millis(60)).await;
-        state.sweep_rate_limiters().await;
+        state.cleanup_rate_limiters().await;
 
         assert_eq!(
             state.ipfs_work_rate_limiter.tracked_keys().await,