fix(node): db_bounded returns Elapsed without polling when the deadline is spent

tokio::time::timeout polls the future before it checks the timer, so a zero
budget still started the work and could return a value. A spent deadline now
yields Elapsed at once. Makes discovery_record_elapsed_leaves_neither_row
deterministic.

Refs #585

.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/mod.rs

@@ -22,7 +22,6 @@ pub mod replicas;
 pub mod repos;
 pub mod resolve;
 pub mod stars;
-pub mod status;
 pub mod task_cursor;
 pub mod tasks;
 pub mod visibility;
@@ -69,12 +68,6 @@ pub(crate) async fn authorize_repo_read(
 /// representation only within `did:key`; never let a bare id match across methods —
 /// `did:web` / `did:gitlawb` share the base58 space with `did:key`, so a
 /// trailing-segment compare would treat `did:key:X` and `did:gitlawb:X` as equal.
-///
-/// The collapse itself lives in exactly one place, [`crate::db::normalize_owner_key`],
-/// which is also the function the stored `owner_did` / `authorizing_did` columns
-/// are normalized through and which `OWNER_KEY_CASE_SQL` mirrors byte for byte.
-/// Two identities match when they normalize to the same key; a second copy of the
-/// rule here is how the Rust gate and the SQL filters would drift apart.
 pub(crate) fn did_matches(a: &str, b: &str) -> bool {
     // One implementation, shared with `gl` and `git-remote-gitlawb` through core:
     // every boundary that compares a DID applies the same collapse rule.
@@ -212,7 +205,6 @@ mod authz_guard {
         let events = include_str!("events.rs");
         let tasks = include_str!("tasks.rs");
         let stars = include_str!("stars.rs");
-        let status = include_str!("status/mod.rs");
         let protect = include_str!("protect.rs");
         let visibility = include_str!("visibility.rs");
         let profiles = include_str!("profiles.rs");
@@ -238,13 +230,6 @@ mod authz_guard {
             // (existence hiding) and require_repo_owner guards the owner half.
             (webhooks, "list_webhooks", "authorize_repo_read("),
             (webhooks, "list_webhooks", "require_repo_owner("),
-            // Same two-half shape as list_webhooks: authorize_repo_read runs
-            // first (a quarantined or unreadable repo gets the missing-repo
-            // not-found, never a 403 that would confirm existence), then
-            // require_repo_owner 403s a non-owner of a readable repo. Both halves
-            // are pinned, because dropping either changes what a stranger learns.
-            (status, "create_status", "authorize_repo_read("),
-            (status, "create_status", "require_repo_owner("),
             (labels, "add_label", "authorize_repo_read("),
             (labels, "add_label", "require_repo_owner("),
             (labels, "remove_label", "authorize_repo_read("),
@@ -276,19 +261,6 @@ mod authz_guard {
             (protect, "list_protected_branches", "authorize_repo_read("),
             (labels, "list_labels", "authorize_repo_read("),
             (events, "list_repo_events", "authorize_repo_read("),
-            // The catch-up poll surface gates before any push-event row is read,
-            // so a stranger cannot use it to learn that a private repo exists or
-            // is being pushed to.
-            (events, "list_repo_push_events", "authorize_repo_read("),
-            // The status read gates before any claim data is loaded, so its deny
-            // is the repo's own not-found rather than an existence oracle over
-            // which commits were reported on.
-            (status, "commit_status", "authorize_repo_read("),
-            // The rollup gates on the same helper before the pull request row is
-            // loaded, so a caller who cannot read the repo cannot learn which
-            // pull request numbers exist on it — and the fallback's branch
-            // resolve and head persist sit behind that same gate.
-            (status, "pull_request_status", "authorize_repo_read("),
             // Bucket C — signer-self: the acting DID is matched/bound to auth.0
             (tasks, "create_task", "did_matches("),
             // #496: create_task is signer-self AND owner-when-repo-scoped — a
@@ -1199,7 +1171,6 @@ mod tests { async fn t() { let _ = did_matches(x, y); } }
             (include_str!("replicas.rs"), "replicas.rs"),
             (include_str!("repos.rs"), "repos.rs"),
             (include_str!("stars.rs"), "stars.rs"),
-            (include_str!("status/mod.rs"), "status/mod.rs"),
             (include_str!("visibility.rs"), "visibility.rs"),
             (include_str!("webhooks.rs"), "webhooks.rs"),
         ];

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

@@ -76,7 +76,6 @@ pub async fn create_pr(
         status: "open".to_string(),
         merged_by_did: None,
         merged_at: None,
-        head_commit: None,
         created_at: now.clone(),
         updated_at: now,
     };
@@ -227,15 +226,17 @@ pub async fn merge_pr(
     let disk_path = guard.path().to_path_buf();
     let merger_did = auth.0;
 
-    // The source head this merge is about to consume, read under the write lock.
-    // The PR row above was loaded before the lock, so a push landing in between
-    // already moved the branch and the row's own `head_commit` is stale; the
-    // value frozen on the merged PR has to be what was merged. `None` when the
-    // ref cannot be resolved, which leaves whatever the push path last stored.
-    let merged_source_head =
-        store::read_ref(&disk_path, &store::local_branch_ref(&pr.source_branch))
-            .ok()
-            .flatten();
+    // 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,
@@ -249,14 +250,31 @@ 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()))?;
 
-    state
-        .db
-        .merge_pr(&pr.id, &merger_did, merged_source_head.as_deref())
-        .await?;
+    state.db.merge_pr(&pr.id, &merger_did).await?;
     let _ = state.db.touch_repo(&record.id).await;
 
     webhooks::fire_event(
@@ -306,36 +324,7 @@ pub async fn close_pr(
         ));
     }
 
-    // Same refusal merge_pr gives, before the head is resolved: a repeated close
-    // must not store the branch's current tip on a historical pull request.
-    if pr.status != "open" {
-        return Err(AppError::BadRequest(format!("PR is already {}", pr.status)));
-    }
-
-    // Freeze the head this close decided against, the same snapshot merge_pr
-    // takes on its path: a closed pull request's rollup serves only the stored
-    // head, so a PR closed before its first status read would report
-    // head_resolved=false forever. The recorded push for the branch is the
-    // canonical answer; the disk read covers a push whose event write failed.
-    // An unresolvable head keeps whatever the push path last stored.
-    let branch_ref = store::local_branch_ref(&pr.source_branch);
-    let closed_head = if pr.head_commit.is_some() {
-        None
-    } else {
-        state
-            .db
-            .latest_push_sha_for_ref(&record.id, &branch_ref)
-            .await
-            .ok()
-            .flatten()
-            .or_else(|| {
-                store::read_ref(std::path::Path::new(&record.disk_path), &branch_ref)
-                    .ok()
-                    .flatten()
-            })
-            .map(|sha| sha.to_ascii_lowercase())
-    };
-    state.db.close_pr(&pr.id, closed_head.as_deref()).await?;
+    state.db.close_pr(&pr.id).await?;
 
     webhooks::fire_event(
         std::sync::Arc::clone(&state.db),
@@ -482,157 +471,131 @@ pub async fn list_comments(
 }
 
 #[cfg(test)]
-mod tests {
+mod merge_rollback_tests {
     use super::*;
-    use axum::body::Body;
-    use axum::http::{Method, StatusCode};
-    use axum::routing::post;
-    use axum::Router;
-    use std::process::Command;
-    use tower::ServiceExt;
+    use axum::response::IntoResponse;
+    use sqlx::PgPool;
 
-    const OWNER_DID: &str = "did:key:z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH";
-    const STALE_SHA: &str = "9999999999999999999999999999999999999999";
-
-    fn git(dir: &std::path::Path, args: &[&str]) {
-        let out = Command::new("git")
+    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)
-            .current_dir(dir)
-            .env("GIT_AUTHOR_NAME", "t")
-            .env("GIT_AUTHOR_EMAIL", "[email protected]")
-            .env("GIT_COMMITTER_NAME", "t")
-            .env("GIT_COMMITTER_EMAIL", "[email protected]")
-            .output()
-            .expect("git runs");
-        assert!(
-            out.status.success(),
-            "git {args:?} failed: {}",
-            String::from_utf8_lossy(&out.stderr)
-        );
-    }
-
-    /// A bare repo with `main` and a `feature` branch one commit ahead.
-    /// Returns the bare path and the real `feature` tip.
-    fn bare_repo_with_feature(
-        root: &std::path::Path,
-        owner_slug: &str,
-    ) -> (std::path::PathBuf, String) {
-        let work = root.join("work");
-        std::fs::create_dir_all(&work).unwrap();
-        git(&work, &["init", "-q", "-b", "main", "."]);
-        std::fs::write(work.join("a.txt"), "one").unwrap();
-        git(&work, &["add", "."]);
-        git(&work, &["commit", "-qm", "one"]);
-        git(&work, &["checkout", "-qb", "feature"]);
-        std::fs::write(work.join("b.txt"), "two").unwrap();
-        git(&work, &["add", "."]);
-        git(&work, &["commit", "-qm", "two"]);
-        git(&work, &["checkout", "-q", "main"]);
-
-        let bare_dir = root.join(owner_slug);
-        std::fs::create_dir_all(&bare_dir).unwrap();
-        let bare = bare_dir.join("demo.git");
-        git(
-            root,
-            &[
-                "clone",
-                "-q",
-                "--bare",
-                work.to_str().unwrap(),
-                bare.to_str().unwrap(),
-            ],
-        );
-
-        let tip = Command::new("git")
-            .args(["rev-parse", "refs/heads/feature"])
-            .current_dir(&bare)
             .output()
             .unwrap();
-        let tip = String::from_utf8_lossy(&tip.stdout).trim().to_string();
-        assert_eq!(tip.len(), 40, "feature tip must resolve");
-        (bare, tip)
+        assert!(
+            out.status.success(),
+            "git {args:?}: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+        String::from_utf8_lossy(&out.stdout).trim().to_string()
     }
 
-    /// Merging freezes the stored head at the source commit the merge actually
-    /// consumed. The PR row is loaded before the write lock is taken, so a push
-    /// that lands in between leaves the row's own `head_commit` stale; seeding a
-    /// value that matches nothing on disk stands in for that race, and the
-    /// merged PR must still come out pointing at the real `feature` tip.
+    /// 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_stamps_the_source_head_it_merged_over_a_racing_value(pool: sqlx::PgPool) {
-        let tmp = tempfile::tempdir().unwrap();
-        let owner_slug = OWNER_DID.replace([':', '/'], "_");
-        let (bare, feature_tip) = bare_repo_with_feature(tmp.path(), &owner_slug);
+    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;
-        state.repo_store =
-            crate::git::repo_store::RepoStore::for_testing(tmp.path().to_path_buf(), pool);
+        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 now = Utc::now();
-        let repo = crate::db::RepoRecord {
-            id: uuid::Uuid::new_v4().to_string(),
-            name: "demo".into(),
-            owner_did: OWNER_DID.into(),
+        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".into(),
-            created_at: now,
-            updated_at: now,
-            disk_path: bare.to_string_lossy().into_owned(),
+            default_branch: "main".to_string(),
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+            disk_path: String::new(),
             forked_from: None,
             machine_id: None,
         };
-        state.db.create_repo(&repo).await.unwrap();
-
-        let pr = PullRequest {
-            id: Uuid::new_v4().to_string(),
-            repo_id: repo.id.clone(),
-            number: 1,
-            title: "add b".into(),
-            body: None,
-            author_did: OWNER_DID.into(),
-            source_branch: "feature".into(),
-            target_branch: "main".into(),
-            status: "open".into(),
-            merged_by_did: None,
-            merged_at: None,
-            head_commit: None,
-            created_at: now.to_rfc3339(),
-            updated_at: now.to_rfc3339(),
-        };
-        state.db.create_pr(&pr).await.unwrap();
-        // The stale value the racing push scenario leaves on the row.
+        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
-            .set_open_pr_heads(&repo.id, "feature", STALE_SHA)
+            .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
-            .unwrap();
+            .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 db = state.db.clone();
-        let app = Router::new()
-            .route(
-                "/api/v1/repos/{owner}/{repo}/pulls/{number}/merge",
-                post(merge_pr),
-            )
-            .with_state(state);
+        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 response = app
-            .oneshot(crate::test_support::signed_request_as(
-                OWNER_DID,
-                Method::POST,
-                "/api/v1/repos/z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH/demo/pulls/1/merge",
-                Body::empty(),
-            ))
-            .await
-            .unwrap();
-        assert_eq!(response.status(), StatusCode::OK);
-
-        let merged = db.get_pr(&repo.id, 1).await.unwrap().unwrap();
-        assert_eq!(merged.status, "merged");
+        let tip_after = git(&repo_path, &["rev-parse", "refs/heads/main"]);
         assert_eq!(
-            merged.head_commit,
-            Some(feature_tip),
-            "the merge must stamp the source head it consumed, not the racing value"
+            tip_after, main_tip,
+            "a refused publish must rewind the local merge to the pre-merge tip"
         );
     }
 }
@@ -680,7 +643,6 @@ mod lock_pool_shed_tests {
         state
             .db
             .create_pr(&PullRequest {
-                head_commit: None,
                 id: Uuid::new_v4().to_string(),
                 repo_id: repo_id.clone(),
                 number: 1,

crates/gitlawb-node/src/api/status/mod.rs

@@ -1,665 +0,0 @@
-//! Commit status claims: the write path and the combined read.
-//!
-//! POST /api/v1/repos/:owner/:repo/statuses/:sha — append one claim (owner only)
-//! GET  /api/v1/repos/:owner/:repo/commits/:sha/status — the combined projection
-//!
-//! Claims are append-only: a producer reporting twice for the same context
-//! leaves both rows and the visible status is a projection over the history.
-//! Reporting twice means two REQUESTS, though. An exact repeat of one already
-//! accepted is answered with 200 and the row it wrote, not a second row: the
-//! projection elects the highest `seq`, so a replayed claim would not duplicate
-//! a verdict but overturn the one that superseded it.
-
-use axum::extract::{Extension, Path, State};
-use axum::http::StatusCode;
-use axum::Json;
-use chrono::Utc;
-use serde::{Deserialize, Serialize};
-use uuid::Uuid;
-
-use crate::auth::AuthenticatedDid;
-use crate::db::StatusClaim;
-use crate::error::{AppError, Result};
-use crate::state::AppState;
-
-/// Claims for one (repo, commit, producer, context). Generous for a real
-/// producer; a producer past it is misbehaving.
-const MAX_CLAIMS_PER_TUPLE: i64 = 100;
-/// Distinct contexts for one (repo, commit). The context string is caller-chosen
-/// and free-form, so the tuple cap alone bounds nothing.
-const MAX_CONTEXTS_PER_COMMIT: i64 = 50;
-/// Claim rows for one repo within the trailing window the db layer counts over.
-/// The commit SHA is caller-chosen and never existence-checked, so a writer at
-/// the two caps above can still fan out over fresh 40-hex strings without this
-/// one. It bounds the RATE of that fan-out rather than a lifetime total, because
-/// nothing prunes the table: a lifetime bound would close the surface on a repo
-/// permanently while answering with a status every client retries.
-const MAX_CLAIMS_PER_REPO_PER_WINDOW: i64 = 10_000;
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct CreateStatusRequest {
-    /// Payload format version. The signature covers these bytes, so the
-    /// field set is the signed format: `deny_unknown_fields` keeps a future
-    /// producer's claim from silently reading as this one, and `v` names the
-    /// version explicitly so a later revision has somewhere to live. Absent
-    /// means 1.
-    pub v: Option<u32>,
-    pub state: String,
-    pub context: String,
-    pub target_url: Option<String>,
-    pub description: Option<String>,
-}
-
-/// The wire state set (KTD-1): GitHub's four commit-status states, so absence
-/// stays distinguishable without a fifth value no client understands.
-const CLAIM_STATES: [&str; 4] = ["error", "failure", "pending", "success"];
-
-/// Bounds on the signature material this write path persists verbatim. Every one
-/// of the four is caller-influenced: the headers come off the request, the
-/// signing string grows with the component list the caller's Signature-Input
-/// chose, and the body is the caller's. Without a bound the claim log's row size
-/// is set by whoever writes to it, so these are explicit and generous rather than
-/// implicit and absent. A conforming `sign_request` produces roughly 100, 200 and
-/// 400 bytes for the first three.
-const MAX_SIGNATURE_CHARS: usize = 512;
-const MAX_SIGNATURE_INPUT_CHARS: usize = 1024;
-const MAX_SIGNING_STRING_CHARS: usize = 4096;
-/// The body is a `CreateStatusRequest`, whose own fields are already capped at
-/// roughly 3.3 KB in total by the three limits below; this leaves room for JSON
-/// framing and nothing more.
-///
-/// Enforced twice, at two different moments, because the handler's check alone
-/// is not the bound it advertises: `require_signature` buffers the whole body
-/// before any handler runs, so by the time [`create_status`] measures it the
-/// allocation has already happened. `build_router` applies the same number as a
-/// `RequestBodyLimitLayer` at the transport, outside the auth layers, which is
-/// what makes it a bound on what is ALLOCATED and not only on what is STORED.
-/// The handler keeps its own check regardless — it is the one that holds for a
-/// handler reached by any other route table, and it is what the material size
-/// tests drive.
-pub(crate) const MAX_REQUEST_BODY_BYTES: usize = 8192;
-
-const MAX_CONTEXT_CHARS: usize = 255;
-const MAX_TARGET_URL_CHARS: usize = 2048;
-const MAX_DESCRIPTION_CHARS: usize = 1024;
-
-/// The 201 body for an accepted claim: what the client needs to identify the row
-/// it just wrote, and nothing else.
-///
-/// Deliberately not the stored [`StatusClaim`]. That struct carries the signature
-/// material, and serializing it echoed the signature, the signing string and the
-/// whole request body back on every write — the body as a JSON array of integers.
-/// Write-time provenance belongs in the row, not on the wire, which is the same
-/// line [`StatusEntry`] draws on the read side.
-#[derive(Serialize)]
-pub struct CreatedStatus {
-    pub id: String,
-    pub repo_id: String,
-    pub commit_sha: String,
-    pub state: String,
-    pub context: String,
-    pub target_url: Option<String>,
-    pub description: Option<String>,
-    pub producer_did: String,
-    pub created_at: String,
-}
-
-impl CreatedStatus {
-    fn from_claim(claim: StatusClaim) -> Self {
-        Self {
-            id: claim.id,
-            repo_id: claim.repo_id,
-            commit_sha: claim.commit_sha,
-            state: claim.state,
-            context: claim.context,
-            target_url: claim.target_url,
-            description: claim.description,
-            producer_did: claim.producer_did,
-            created_at: claim.created_at,
-        }
-    }
-}
-
-/// POST /api/v1/repos/:owner/:repo/statuses/:sha
-pub async fn create_status(
-    State(state): State<AppState>,
-    Extension(auth): Extension<AuthenticatedDid>,
-    Path((owner, name, sha)): Path<(String, String, String)>,
-    material: Option<Extension<crate::auth::SignatureMaterial>>,
-    Json(req): Json<CreateStatusRequest>,
-) -> Result<(StatusCode, Json<CreatedStatus>)> {
-    // Read-visibility first, then owner, and the order is the security property.
-    // authorize_repo_read denies a quarantined repo before the visibility gate and
-    // answers with the repo's own not-found, byte-identical to a missing repo, so
-    // a caller who cannot read the repo cannot learn it exists. Loading the repo
-    // and comparing the owner would answer 403 there, turning this endpoint into
-    // an existence oracle. require_repo_owner then 403s a non-owner of a repo the
-    // caller can read, where existence is not secret.
-    let (record, _rules) =
-        crate::api::authorize_repo_read(&state, &owner, &name, Some(&auth.0), "/").await?;
-    crate::api::require_repo_owner(&record, &auth.0)?;
-
-    let commit_sha = normalize_sha(&sha)?;
-    if req.v.is_some_and(|v| v != 1) {
-        return Err(AppError::BadRequest(
-            "unsupported status payload version".into(),
-        ));
-    }
-    let claim_state = validate_state(&req.state)?;
-    let context = validate_context(&req.context)?;
-    let target_url = validate_target_url(req.target_url.as_deref())?;
-    let description = validate_description(req.description.as_deref())?;
-
-    // KTD-5: the signature material is captured here because the request carries
-    // the only copy, and `signing_string` is the exact byte sequence the
-    // middleware verified. Its absence means this handler was reached without
-    // `require_signature` — a server misconfiguration (a route group that lost
-    // its auth layer, a handler mounted somewhere new), not anything the client
-    // did, so it is a 500 and no row is written. Storing an empty payload instead
-    // would leave every claim from that moment on unverifiable with nothing going
-    // red, which is exactly the absence-renders-as-success shape.
-    let Some(Extension(material)) = material else {
-        return Err(AppError::Internal(anyhow::anyhow!(
-            "status claim write reached without verified signature material — \
-             the route is not behind require_signature"
-        )));
-    };
-    // The body is carried only behind the `PersistsSignedBody` marker, which
-    // this route's group applies outside the auth layers. Its absence means the
-    // marker did not reach the middleware, because the layer was dropped or
-    // reordered. That is the same class of misconfiguration as the branch above
-    // and gets the same answer. Storing an empty body instead would write claims
-    // that carry a signature over a digest of bytes nobody kept, which is
-    // exactly the case `stored_claim_re_verifies_and_a_tampered_row_does_not`
-    // exists to make impossible.
-    let Some(body) = material.body.clone() else {
-        return Err(AppError::Internal(anyhow::anyhow!(
-            "status claim write reached with signature material carrying no request body: \
-             the route lost its persist-body marker or applied it inside the auth layers"
-        )));
-    };
-    // Bounded before anything is written. The material is verified, which says
-    // the caller holds the key, not that what they signed is a reasonable size.
-    bound("signature", material.signature.len(), MAX_SIGNATURE_CHARS)?;
-    bound(
-        "signature_input",
-        material.signature_input.len(),
-        MAX_SIGNATURE_INPUT_CHARS,
-    )?;
-    bound(
-        "signing_string",
-        material.signing_string.len(),
-        MAX_SIGNING_STRING_CHARS,
-    )?;
-    bound("request body", body.len(), MAX_REQUEST_BODY_BYTES)?;
-    // After the bounds, so the scan below walks a signing string of known size.
-    require_body_bound_signature(&material, &body)?;
-
-    let digest = request_digest(&material, &body);
-    let (signature, signature_input, signing_string, request_body) = (
-        material.signature,
-        material.signature_input,
-        material.signing_string,
-        body.to_vec(),
-    );
-
-    let claim = StatusClaim {
-        id: Uuid::new_v4().to_string(),
-        // Ignored on insert: the database assigns the ordering key (KTD-3).
-        seq: 0,
-        repo_id: record.id.clone(),
-        commit_sha,
-        state: claim_state,
-        context,
-        target_url,
-        description,
-        // Both are the owner identity today (KTD-5); they split when delegated
-        // capabilities land.
-        producer_did: auth.0.clone(),
-        authorizing_did: auth.0,
-        signature,
-        signature_input,
-        signing_string,
-        request_body,
-        request_digest: digest,
-        created_at: Utc::now().to_rfc3339(),
-    };
-
-    let caps = crate::db::ClaimCaps {
-        per_tuple: MAX_CLAIMS_PER_TUPLE,
-        contexts_per_commit: MAX_CONTEXTS_PER_COMMIT,
-        per_repo_window: MAX_CLAIMS_PER_REPO_PER_WINDOW,
-    };
-    match state.db.insert_status_claim_capped(&claim, &caps).await? {
-        crate::db::ClaimInsert::Inserted(_) => {
-            Ok((StatusCode::CREATED, Json(CreatedStatus::from_claim(claim))))
-        }
-        // Already recorded, so 200 and the original row rather than 201 and a
-        // second one. The claim the caller gets back is the stored one, id
-        // included, which is what makes a retry safe to treat as success.
-        crate::db::ClaimInsert::AlreadyRecorded(existing) => {
-            Ok((StatusCode::OK, Json(CreatedStatus::from_claim(*existing))))
-        }
-        crate::db::ClaimInsert::CapExceeded(which) => Err(AppError::TooManyRequests(format!(
-            "claim limit reached for {which}"
-        ))),
-    }
-}
-
-/// Refuse a claim whose signature does not cover the body about to be stored.
-///
-/// A claim IS provenance, so the signature stored beside it has to bind the
-/// bytes stored beside it. It does not do that directly: `signing_string`
-/// reaches the body only through a `content-digest` line. And a caller can sign
-/// a content-digest the request then never carries. `require_signature` rebuilds
-/// the covered values from the request, so an ABSENT `Content-Digest` header
-/// makes the covered digest the empty string on the signing side and the
-/// verifying side alike; the Ed25519 check passes over method, path and an empty
-/// digest, and any body at all rides along. The row that write produces asserts
-/// a provenance that cannot be re-verified: the signing string covers a digest
-/// of bytes that are not the ones in `request_body`.
-///
-/// The predicate here is the write-time form of the one `re_verify` runs against
-/// a stored row in the tests — recompute the digest of the bytes about to be
-/// persisted and require the signing string to carry exactly that line — so what
-/// this route accepts and what a third party can later re-verify are the same
-/// question, asked at both ends.
-///
-/// This is a route property, checked where the claim becomes history, and it is
-/// deliberately not the only line of defence: requiring the header in
-/// `require_signature` closes the same hole one layer earlier and for every
-/// signed route, and that change belongs there rather than here because no other
-/// route's correctness should depend on the status handler. The two agree and
-/// neither substitutes for the other — without the middleware's requirement the
-/// empty digest is still accepted everywhere else, and without this one a later
-/// relaxation of the middleware would silently resume writing unverifiable
-/// history.
-///
-/// A 400, not a 500: the material is exactly what the client signed, and what
-/// they signed does not cover what they sent.
-fn require_body_bound_signature(
-    material: &crate::auth::SignatureMaterial,
-    body: &[u8],
-) -> Result<()> {
-    let covered = format!(
-        "\"content-digest\": {}",
-        gitlawb_core::http_sig::compute_content_digest(body)
-    );
-    // Line-exact rather than a substring search: `build_signing_string` emits one
-    // component per line, and a `contains` would also be satisfied by the digest
-    // appearing INSIDE some other covered component's value, which proves
-    // nothing about what the signature covered.
-    if material.signing_string.lines().any(|line| line == covered) {
-        return Ok(());
-    }
-    Err(AppError::BadRequest(
-        "the signature does not cover this request body: send a Content-Digest header \
-         computed over the body and name \"content-digest\" among the signature's covered \
-         components"
-            .into(),
-    ))
-}
-
-/// The identity of one signed write, as a hex sha-256.
-///
-/// All four inputs together, because a replay is the same bytes arriving twice:
-/// the signature, the input that names what it covers, the canonical string it
-/// was verified over, and the body that string covers through a content-digest.
-/// The signature alone would nearly do (it covers the other three transitively),
-/// but hashing what is actually stored means the digest describes the row rather
-/// than a claim about it.
-///
-/// Each field is length-prefixed so no two different requests can concatenate to
-/// the same bytes: without it, a signature ending in some prefix of the next
-/// field would collide with the shorter signature that absorbed it.
-///
-/// This is not a nonce, and deliberately so. A nonce table needs pruning and a
-/// pruning window is a second replay window; the claim row IS the record of what
-/// was accepted, so uniqueness on it is self-maintaining.
-fn request_digest(material: &crate::auth::SignatureMaterial, body: &[u8]) -> String {
-    use sha2::{Digest, Sha256};
-    let mut h = Sha256::new();
-    for part in [
-        material.signature.as_bytes(),
-        material.signature_input.as_bytes(),
-        material.signing_string.as_bytes(),
-        body,
-    ] {
-        h.update((part.len() as u64).to_be_bytes());
-        h.update(part);
-    }
-    format!("{:x}", h.finalize())
-}
-
-/// One reported context in the combined response. Signature material is
-/// deliberately absent: it is write-time provenance, not read-surface data.
-#[derive(Serialize)]
-pub struct StatusEntry {
-    pub state: String,
-    pub context: String,
-    pub target_url: Option<String>,
-    pub description: Option<String>,
-    pub producer_did: String,
-    pub created_at: String,
-}
-
-/// The combined commit status. `state` never leaves the four-value set (KTD-1),
-/// so absence is carried by `total_count` 0 with the pending state rather than a
-/// fifth value, and `reported_only` (R19) says out loud that the state covers the
-/// contexts that reported, not every check a caller expected.
-#[derive(Serialize)]
-pub struct CombinedStatus {
-    pub state: String,
-    pub sha: String,
-    pub total_count: usize,
-    pub statuses: Vec<StatusEntry>,
-    pub reported_only: bool,
-}
-
-/// GET /api/v1/repos/:owner/:repo/commits/:sha/status
-///
-/// The auth extension is optional and MUST be last in the extractor list: the
-/// route group carries `optional_signature`, so an unsigned caller reaches here
-/// with no `AuthenticatedDid` at all and a public repo still answers.
-pub async fn commit_status(
-    State(state): State<AppState>,
-    Path((owner, name, sha)): Path<(String, String, String)>,
-    auth: Option<Extension<AuthenticatedDid>>,
-) -> Result<Json<CombinedStatus>> {
-    let caller = auth.as_ref().map(|Extension(a)| a.0.as_str());
-    // The gate runs first and on the requested path, before any claim data is
-    // touched. Its deny is the repo's own not-found, byte-identical to a missing
-    // repo, so the status surface cannot answer "does this repo exist".
-    let (record, _rules) =
-        crate::api::authorize_repo_read(&state, &owner, &name, caller, "/").await?;
-
-    let commit_sha = normalize_sha(&sha)?;
-    let statuses = project_claims(&state, &record, &commit_sha).await?;
-
-    Ok(Json(CombinedStatus {
-        state: combined_state(&statuses).to_string(),
-        sha: commit_sha,
-        total_count: statuses.len(),
-        statuses,
-        reported_only: true,
-    }))
-}
-
-/// The projection, in one place. Both read surfaces call it, so R3's "the rollup
-/// is derived by the same projection as the commit read" holds by construction
-/// rather than by two implementations happening to agree.
-///
-/// KTD-5: current authorization, not write-time authorization. An ownership
-/// transfer drops the prior owner's claims from the projection while the
-/// append-only history keeps them.
-async fn project_claims(
-    state: &AppState,
-    record: &crate::db::RepoRecord,
-    commit_sha: &str,
-) -> Result<Vec<StatusEntry>> {
-    let claims = state
-        .db
-        .latest_status_claims(&record.id, commit_sha, &record.owner_did)
-        .await?;
-    Ok(claims
-        .into_iter()
-        .map(|c| StatusEntry {
-            state: c.state,
-            context: c.context,
-            target_url: c.target_url,
-            description: c.description,
-            producer_did: c.producer_did,
-            created_at: c.created_at,
-        })
-        .collect())
-}
-
-/// The pull request head rollup (R11). `state` stays inside the four wire values
-/// whatever happened to the head (KTD-1): an unresolvable head is carried by
-/// `head_resolved`, alongside the pull request's own state, so a client can tell
-/// "the head could not be resolved" from "the head resolved and nothing reported"
-/// without a fifth state value no client understands.
-#[derive(Serialize)]
-pub struct PullRequestStatus {
-    pub number: i64,
-    /// The pull request's own state: open, closed, or merged.
-    pub pull_request_state: String,
-    pub head_resolved: bool,
-    /// The target commit, absent exactly when `head_resolved` is false.
-    pub sha: Option<String>,
-    pub state: String,
-    pub total_count: usize,
-    pub statuses: Vec<StatusEntry>,
-    pub reported_only: bool,
-}
-
-/// GET /api/v1/repos/:owner/:repo/pulls/:number/status
-///
-/// Same optional-auth shape as the commit read, and the auth extension MUST stay
-/// last in the extractor list.
-pub async fn pull_request_status(
-    State(state): State<AppState>,
-    Path((owner, name, number)): Path<(String, String, i64)>,
-    auth: Option<Extension<AuthenticatedDid>>,
-) -> Result<Json<PullRequestStatus>> {
-    let caller = auth.as_ref().map(|Extension(a)| a.0.as_str());
-    // The gate runs first, on the requested path, before the pull request row is
-    // loaded. Its deny is the repo's own not-found, byte-identical to a missing
-    // repo. Once it passes, a missing pull request number is the plain not-found:
-    // existence is no longer secret.
-    let (record, _rules) =
-        crate::api::authorize_repo_read(&state, &owner, &name, caller, "/").await?;
-
-    let pr = state
-        .db
-        .get_pr(&record.id, number)
-        .await?
-        .ok_or_else(|| AppError::NotFound(format!("PR #{number} not found")))?;
-
-    let head = rollup_head(&state, &record, &pr).await?;
-
-    let statuses = match &head {
-        Some(sha) => project_claims(&state, &record, sha).await?,
-        None => Vec::new(),
-    };
-
-    Ok(Json(PullRequestStatus {
-        number: pr.number,
-        pull_request_state: pr.status,
-        head_resolved: head.is_some(),
-        sha: head,
-        state: combined_state(&statuses).to_string(),
-        total_count: statuses.len(),
-        statuses,
-        reported_only: true,
-    }))
-}
-
-/// Branch-head lookups performed by the rollup fallback, counted so a test can
-/// assert WORK DONE rather than only the answer returned. A fallback that
-/// re-resolved on every read and then discarded the answer would be invisible in
-/// the response and visible here.
-#[cfg(test)]
-pub(crate) static BRANCH_RESOLVES: std::sync::atomic::AtomicUsize =
-    std::sync::atomic::AtomicUsize::new(0);
-
-/// The rollup's target commit (KTD-4).
-///
-/// The stored head wins whenever it is set. It is frozen at close or merge, so a
-/// closed or merged pull request keeps the commit the decision was made against,
-/// and only an OPEN pull request with no stored head falls back to the branch.
-///
-/// The fallback is one database read: the latest push this node recorded for the
-/// pull request's source branch, from `repo_push_events`. It deliberately does
-/// NOT go through `git::store` ref listing, which needs an acquired repository
-/// path — on a cold node that downloads the whole repository from object storage,
-/// and this read group carries no rate limiter, so an anonymous caller could
-/// drive repeated downloads with a URL.
-///
-/// `repo_push_events` is the source rather than `branch_cids` because
-/// `record_push_events` writes it for every ref update unconditionally, whereas
-/// the sole writer of `branch_cids` sits behind a pin CID. A node with no object
-/// pinning configured never writes that table, so a resolve keyed on it could
-/// never answer there and every open pull request without a stored head reported
-/// `head_resolved: false` forever. The residual limit is that only pushes taken
-/// after this shipped have rows, so a branch last pushed before then still does
-/// not resolve.
-///
-/// The resolved head is persisted, which makes an unauthenticated GET write. That
-/// is only acceptable because the write is self-limiting: it fires exactly when
-/// `head_commit` is absent and it fills it, and the fill is conditioned on that
-/// same absence in SQL, so the PERSIST happens at most once per pull request.
-/// The resolve attempt itself is not once-only — a pull request whose branch does
-/// not resolve stores nothing, so there is nothing to short-circuit on and the
-/// lookup runs again on every read until it succeeds.
-async fn rollup_head(
-    state: &AppState,
-    record: &crate::db::RepoRecord,
-    pr: &crate::db::PullRequest,
-) -> Result<Option<String>> {
-    if let Some(stored) = pr.head_commit.clone() {
-        return Ok(Some(stored));
-    }
-    if pr.status != "open" {
-        return Ok(None);
-    }
-
-    // The push path records the full ref while a pull request stores the bare
-    // branch name, the same mismatch `crate::api::repos::branch_from_ref`
-    // reconciles on the write side; this is that mapping run backwards. The empty
-    // branch is excluded for the same reason it is there: `refs/heads/` is not a
-    // branch and nothing can have written it.
-    if pr.source_branch.is_empty() {
-        return Ok(None);
-    }
-    let ref_name = crate::git::store::local_branch_ref(&pr.source_branch);
-    #[cfg(test)]
-    BRANCH_RESOLVES.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-    let resolved = state
-        .db
-        .latest_push_sha_for_ref(&record.id, &ref_name)
-        .await?
-        // A recorded head that is not a commit SHA is no target at all, and
-        // storing it would leave the rollup pointing at nothing forever.
-        .and_then(|sha| normalize_sha(&sha).ok());
-
-    let Some(sha) = resolved else {
-        return Ok(None);
-    };
-    // Best effort, and deliberately not `?`. The head is already resolved and the
-    // response does not depend on this write landing, so a transient database
-    // failure here must not turn a served read into a 500; the next read resolves
-    // again and re-attempts the fill. Same catch-and-log the sibling writes on the
-    // push path use (`update_open_pr_heads`, `record_push_events`).
-    if let Err(e) = state.db.set_pr_head_if_absent(&pr.id, &sha).await {
-        tracing::warn!(
-            err = %e,
-            pr_id = %pr.id,
-            "failed to persist a resolved pull request head; serving the resolved head anyway"
-        );
-    }
-    Ok(Some(sha))
-}
-
-/// KTD-1's fold: any error or failure yields failure, else any pending yields
-/// pending, else success. An empty set is pending, never success — total absence
-/// of a verdict is its own state and must not read as a pass (R10).
-fn combined_state(statuses: &[StatusEntry]) -> &'static str {
-    if statuses.is_empty() {
-        return "pending";
-    }
-    if statuses
-        .iter()
-        .any(|s| s.state == "error" || s.state == "failure")
-    {
-        "failure"
-    } else if statuses.iter().any(|s| s.state == "pending") {
-        "pending"
-    } else if statuses.iter().all(|s| s.state == "success") {
-        "success"
-    } else {
-        // A stored state outside the vocabulary is not a pass: an
-        // unrecognized verdict folds toward failure, never silently to success.
-        "failure"
-    }
-}
-
-/// 40 hex characters, lowercased. The SHA is never existence-checked (KTD-7):
-/// a claim describes a commit object, and the object may not have arrived yet.
-fn normalize_sha(sha: &str) -> Result<String> {
-    if crate::api::repos::is_hex_sha(sha) {
-        Ok(sha.to_ascii_lowercase())
-    } else {
-        Err(AppError::BadRequest(
-            "commit sha must be exactly 40 hexadecimal characters".into(),
-        ))
-    }
-}
-
-/// One size bound on persisted signature material, measured in bytes.
-fn bound(what: &str, len: usize, max: usize) -> Result<()> {
-    if len > max {
-        return Err(AppError::BadRequest(format!(
-            "{what} must be at most {max} bytes"
-        )));
-    }
-    Ok(())
-}
-
-fn validate_state(state: &str) -> Result<String> {
-    if CLAIM_STATES.contains(&state) {
-        Ok(state.to_string())
-    } else {
-        Err(AppError::BadRequest(format!(
-            "state must be one of {}",
-            CLAIM_STATES.join(", ")
-        )))
-    }
-}
-
-/// The context is the projection key, so it is trimmed, bounded, and required to
-/// be free of control characters.
-fn validate_context(context: &str) -> Result<String> {
-    let trimmed = context.trim();
-    if trimmed.is_empty() || trimmed.chars().count() > MAX_CONTEXT_CHARS {
-        return Err(AppError::BadRequest(format!(
-            "context must be 1 to {MAX_CONTEXT_CHARS} characters"
-        )));
-    }
-    if trimmed.chars().any(char::is_control) {
-        return Err(AppError::BadRequest(
-            "context must not contain control characters".into(),
-        ));
-    }
-    Ok(trimmed.to_string())
-}
-
-fn validate_target_url(url: Option<&str>) -> Result<Option<String>> {
-    let Some(url) = url else { return Ok(None) };
-    if url.chars().count() > MAX_TARGET_URL_CHARS {
-        return Err(AppError::BadRequest(format!(
-            "target_url must be at most {MAX_TARGET_URL_CHARS} characters"
-        )));
-    }
-    let lower = url.to_ascii_lowercase();
-    if lower.starts_with("http://") || lower.starts_with("https://") {
-        Ok(Some(url.to_string()))
-    } else {
-        Err(AppError::BadRequest(
-            "target_url must be an http or https URL".into(),
-        ))
-    }
-}
-
-fn validate_description(description: Option<&str>) -> Result<Option<String>> {
-    match description {
-        Some(d) if d.chars().count() > MAX_DESCRIPTION_CHARS => Err(AppError::BadRequest(format!(
-            "description must be at most {MAX_DESCRIPTION_CHARS} characters"
-        ))),
-        other => Ok(other.map(str::to_string)),
-    }
-}
-
-#[cfg(test)]
-mod tests;

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

@@ -17,53 +17,6 @@ use crate::state::AppState;
 #[derive(Clone, Debug)]
 pub struct AuthenticatedDid(pub String);
 
-/// The RFC 9421 material this request was actually verified against, injected
-/// alongside [`AuthenticatedDid`] by `require_signature`.
-///
-/// A handler that records a signed claim as history has to store it: the header
-/// values and the canonical signing string are gone once the request ends, and a
-/// stored claim nobody can re-verify is not history. `signing_string` is the exact
-/// byte sequence the Ed25519 verification succeeded over, not a rebuild — a
-/// handler reconstructing it from headers would be verifying a different string
-/// than the middleware did.
-#[derive(Clone, Debug)]
-pub struct SignatureMaterial {
-    /// The `Signature` header value.
-    pub signature: String,
-    /// The `Signature-Input` header value.
-    pub signature_input: String,
-    /// The canonical bytes the signature covered.
-    pub signing_string: String,
-    /// The buffered request body, on the routes that persist it. The signing
-    /// string covers it only through the content-digest, so without these bytes
-    /// a stored claim can be shown to carry *a* valid signature over *some*
-    /// digest and nothing more. Carried here because the body is consumed
-    /// downstream by the extractor and cannot be recovered afterwards.
-    ///
-    /// `Option`, and populated only behind [`PersistsSignedBody`], because a
-    /// route that never stores the body has no use for a second handle to it.
-    /// The unlike-sized field is this one: the three above are header-derived
-    /// and bounded by the server's header limit, while a receive-pack POST
-    /// carries up to GITLAWB_MAX_PACK_BYTES (2 GB by default). `Bytes`, not
-    /// `Vec<u8>`, so the handle the marked routes do take is a refcount bump
-    /// over the buffer the middleware already holds rather than a copy.
-    pub body: Option<axum::body::Bytes>,
-}
-
-/// Marks a route group whose handler persists the signed request body.
-///
-/// Applied as an extension layer OUTSIDE the auth layers, so it is on the
-/// request before [`require_signature`] decides whether to carry the body. A
-/// marker applied inside them is never seen, and the failure is silent: the
-/// handler still compiles, tests that inject material by hand still pass, and
-/// production stores nothing. The status write handler refuses an absent body
-/// for that reason.
-///
-/// A marker rather than a path check because `require_signature` must not learn
-/// which URL persists claims; that knowledge belongs at the route table.
-#[derive(Clone, Copy, Debug)]
-pub struct PersistsSignedBody;
-
 /// A UCAN that passed full chain validation, with the root issuer the chain
 /// rests on. Inserted into request extensions by [`require_ucan_chain`] when
 /// `X-Ucan` is present; absent when the header is.
@@ -422,28 +375,10 @@ async fn verify_signature(limit: usize, request: Request, next: Next) -> Respons
 
     tracing::info!(did = %sig.key_id, "✓ authenticated request");
 
-    // The body travels only where it is stored. On every other signed route the
-    // middleware's handle would keep the buffer alive for the rest of the layer
-    // chain, until the handler's own extractor consumes the request.
-    let body = parts
-        .extensions
-        .get::<PersistsSignedBody>()
-        .map(|_| body_bytes.clone());
-
-    let material = SignatureMaterial {
-        signature: sig_header,
-        signature_input: sig_input,
-        signing_string,
-        body,
-    };
-
     let mut request = Request::from_parts(parts, Body::from(body_bytes));
     request
         .extensions_mut()
         .insert(AuthenticatedDid(sig.key_id.to_string()));
-    // Carry the verified material forward for handlers that persist a signed
-    // claim; none of it can be recovered once the request is gone.
-    request.extensions_mut().insert(material);
     next.run(request).await
 }
 
@@ -803,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,
@@ -813,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)),
@@ -855,109 +798,6 @@ mod tests {
         }
     }
 
-    /// Reports what the signature middleware put in `SignatureMaterial.body`,
-    /// so a test can tell "captured" from "deliberately not captured" apart
-    /// from "the middleware never ran".
-    async fn report_body(material: Option<axum::Extension<SignatureMaterial>>) -> String {
-        match material {
-            Some(axum::Extension(m)) => match m.body {
-                Some(bytes) => format!("captured:{}", bytes.len()),
-                None => "absent".to_string(),
-            },
-            None => "no-material".to_string(),
-        }
-    }
-
-    fn signed_post(path: &str, body: &[u8]) -> Request {
-        let kp = Keypair::generate();
-        let signed = gitlawb_core::http_sig::sign_request(&kp, "POST", path, body);
-        Request::builder()
-            .method("POST")
-            .uri(path)
-            .header("content-digest", signed.content_digest)
-            .header("signature-input", signed.signature_input)
-            .header("signature", signed.signature)
-            .body(axum::body::Body::from(body.to_vec()))
-            .unwrap()
-    }
-
-    /// The body handle travels only on a route that asked for it.
-    ///
-    /// Both arms drive a genuinely signed request through `require_signature`;
-    /// the only difference is the [`PersistsSignedBody`] marker layer. A
-    /// middleware that captured unconditionally would make both arms read
-    /// `captured`, and one that never read the marker would make both read
-    /// `absent`, so the pair pins the decision rather than the mechanism.
-    #[tokio::test]
-    async fn the_body_is_carried_only_when_the_route_marks_itself_as_persisting_it() {
-        let body = br#"{"state":"success"}"#;
-
-        // The marker layer is added AFTER the middleware layer, which is what
-        // makes it the outer of the two and therefore the one that runs first.
-        // This is the same relative position `build_router` uses on the status
-        // write group, where the marker sits outside `add_auth_layers`.
-        let marked = Router::new()
-            .route("/", axum::routing::post(report_body))
-            .layer(middleware::from_fn_with_state(
-                crate::auth::SignatureBodyLimit(65536),
-                require_signature,
-            ))
-            .layer(axum::Extension(PersistsSignedBody));
-        let resp = marked.oneshot(signed_post("/", body)).await.unwrap();
-        assert_eq!(resp.status(), StatusCode::OK);
-        let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
-        assert_eq!(
-            String::from_utf8(seen.to_vec()).unwrap(),
-            format!("captured:{}", body.len()),
-            "a route carrying the persist marker must receive the buffered body"
-        );
-
-        let unmarked = Router::new()
-            .route("/", axum::routing::post(report_body))
-            .layer(middleware::from_fn_with_state(
-                crate::auth::SignatureBodyLimit(65536),
-                require_signature,
-            ));
-        let resp = unmarked.oneshot(signed_post("/", body)).await.unwrap();
-        assert_eq!(resp.status(), StatusCode::OK);
-        let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
-        assert_eq!(
-            String::from_utf8(seen.to_vec()).unwrap(),
-            "absent",
-            "a signed route that never persists the body must not be handed a handle to it"
-        );
-    }
-
-    /// The marker is read from the request the middleware was given, and the
-    /// request it hands downstream still carries it. Both halves matter: the
-    /// first is the capture decision, the second is that rebuilding the request
-    /// from its parts does not drop extensions a later layer may want.
-    #[tokio::test]
-    async fn the_marker_survives_the_request_rebuild() {
-        async fn report_marker(marker: Option<axum::Extension<PersistsSignedBody>>) -> String {
-            match marker {
-                Some(_) => "marked".to_string(),
-                None => "unmarked".to_string(),
-            }
-        }
-
-        let app = Router::new()
-            .route("/", axum::routing::post(report_marker))
-            .layer(middleware::from_fn_with_state(
-                crate::auth::SignatureBodyLimit(65536),
-                require_signature,
-            ))
-            .layer(axum::Extension(PersistsSignedBody));
-        let resp = app.oneshot(signed_post("/", b"{}")).await.unwrap();
-        assert_eq!(resp.status(), StatusCode::OK);
-        let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
-        assert_eq!(
-            String::from_utf8(seen.to_vec()).unwrap(),
-            "marked",
-            "the marker must reach the handler, so the middleware saw it too"
-        );
-    }
-
     #[tokio::test]
     async fn require_ucan_chain_no_header_passes_through() {
         let state = make_test_state(Keypair::generate().did());

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/durable_outbox.rs

@@ -1285,38 +1285,6 @@ async fn run_effect_bundle(
         }
     }
 
-    // 7a. Commit-status surfaces: the stored pull-request heads and the
-    //     catch-up poll feed are durable accounting like everything else
-    //     in this bundle, so they ride the same proven-landing authority —
-    //     derived from the accepted transition rows, never from a live
-    //     parse, and a landing reconcile proves only later still emits
-    //     them here on the drain pass. Head writes check each branch's
-    //     current tip before moving it: this executor runs after the repo
-    //     write lock is released, so a newer push may already have moved
-    //     the branch and only the still-current landing may win the head.
-    let landed: Vec<crate::api::repos::RefUpdate> = accepted_children
-        .iter()
-        .map(|c| crate::api::repos::RefUpdate {
-            old_sha: c.old_sha.clone(),
-            new_sha: c.new_sha.clone(),
-            ref_name: c.ref_name.clone(),
-        })
-        .collect();
-    let heads =
-        crate::api::repos::heads_still_at_tip(std::path::Path::new(&repo.disk_path), &landed)?;
-    if let Err(e) = crate::api::repos::update_open_pr_heads(&state.db, &req.repo_id, &heads).await {
-        return Ok(EffectsOutcome::Retry {
-            last_error: format!("stored pr heads: {e}"),
-        });
-    }
-    if let Err(e) =
-        crate::api::repos::record_push_events(&state.db, &req.id, accepted_children).await
-    {
-        return Ok(EffectsOutcome::Retry {
-            last_error: format!("push events: {e}"),
-        });
-    }
-
     // 8. Per-ref certs and anchor jobs. Each accepted child gets one
     //    of each. Failures are accumulated; the first one is
     //    returned as the Retry reason. Anchor identity is by occurrence

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
     }
 }
@@ -358,16 +379,6 @@ mod tests {
         );
     }
 
-    /// A 429 does not advertise a retry interval: the status-claim caps behind
-    /// it are per-commit and append-only, so a client honoring a 1s hint would
-    /// retry a request that can never succeed.
-    #[test]
-    fn too_many_requests_carries_no_retry_after() {
-        let resp = AppError::TooManyRequests("x".into()).into_response();
-        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
-        assert!(resp.headers().get("retry-after").is_none());
-    }
-
     /// #226: raw sqlx/DB detail must never appear in the Internal 500 body.
     #[tokio::test]
     async fn internal_error_body_is_opaque() {

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

@@ -683,14 +683,12 @@ pub fn list_refs(repo_path: &Path) -> Result<Vec<(String, String)>> {
 /// a 40-char SHA-1 hex string. The reconcile compares two hex
 /// strings.
 ///
-/// `show-ref --verify` matches the exact ref name — unlike `rev-parse`,
-/// which peels expressions, so it is the right primitive when the name is
-/// caller-adjacent. Most production callers use [`read_ref_bounded`]; this
-/// unbounded form also serves test harnesses that stage markers without a
-/// timeout budget.
+/// Production callers use [`read_ref_bounded`]; this unbounded form is kept
+/// for test harnesses that stage markers without a timeout budget.
+#[allow(dead_code)]
 pub fn read_ref(repo_path: &Path, ref_name: &str) -> Result<Option<String>> {
-    // allow-unbounded-git: exact-name resolution; callers that need a timeout
-    // budget use read_ref_bounded.
+    // allow-unbounded-git: #[allow(dead_code)] harness form; production
+    // callers use read_ref_bounded.
     let output = Command::new("git")
         .args(["show-ref", "--verify", "--hash", ref_name])
         .current_dir(repo_path)
@@ -1579,7 +1577,7 @@ fn guard_refs(target_branch: &str, source_branch: &str) -> Result<()> {
 
 /// Stored short branch names are passed to git as explicit local-branch refs so a
 /// syntax-valid name cannot be reinterpreted as a revision shorthand at the sink.
-pub(crate) fn local_branch_ref(short_name: &str) -> String {
+fn local_branch_ref(short_name: &str) -> String {
     format!("refs/heads/{short_name}")
 }
 
@@ -1819,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]
@@ -1885,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/server.rs

@@ -15,7 +15,7 @@ use tracing::Level;
 
 use crate::api::{
     agents, arweave, bounties, certs, changelog, events, ipfs, issues, labels, peers, profiles,
-    protect, pulls, register, replicas, repos, resolve, stars, status, tasks, visibility, webhooks,
+    protect, pulls, register, replicas, repos, resolve, stars, tasks, visibility, webhooks,
 };
 use crate::auth;
 use crate::rate_limit;
@@ -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,69 +314,8 @@ pub fn build_router(state: AppState) -> Router {
             ),
         state.clone(),
         SIGNED_BODY_LIMIT,
-    );
-
-    // ── Status claim writes — signature, plus the per-DID throttle AND the
-    // per-IP flood brake `creation_routes` carries. Not in `write_routes`, which
-    // has neither: every call here appends a row, so an unthrottled writer grows
-    // the claim log on demand and the three per-repo caps would be the only bound.
-    let status_write_routes = add_auth_layers(
-        Router::new()
-            .route(
-                "/api/v1/repos/{owner}/{repo}/statuses/{sha}",
-                post(status::create_status),
-            )
-            .layer(middleware::from_fn(rate_limit::rate_limit_by_did))
-            .layer(axum::Extension(state.rate_limiter.clone())),
-        state.clone(),
-        SIGNED_BODY_LIMIT,
     )
-    // The route's advertised body bound, applied at the transport. It sits
-    // OUTSIDE the auth layers because `require_signature` collects and hashes
-    // the whole body before any handler runs: the handler's own check measures
-    // an allocation that has already happened, so without this layer the 8 KiB
-    // figure bounded what gets stored and not what gets buffered, and
-    // concurrent requests multiplied exactly the cost it was written to
-    // prevent. It sits INSIDE the rate limiters so an oversized request is
-    // still counted by the per-IP brake before it is refused.
-    //
-    // A declared Content-Length over the limit is answered 413 without reading
-    // a byte; a body that declares nothing (or understates) is truncated at the
-    // limit and fails where it is read, which `require_signature` reports as an
-    // unreadable body.
-    .layer(RequestBodyLimitLayer::new(status::MAX_REQUEST_BODY_BYTES))
-    // The one route that persists the signed request body, so the one route
-    // whose signature material carries it. Applied OUTSIDE `add_auth_layers`
-    // (outermost = runs first) because `require_signature` reads this marker to
-    // decide whether to carry the body; a layer added inside the auth pair runs
-    // after the middleware and is never seen. `create_status` refuses an absent
-    // body rather than storing an empty column, so getting this order wrong
-    // fails loudly instead of quietly writing unverifiable claims.
-    .layer(axum::Extension(auth::PersistsSignedBody))
-    .layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
-    .layer(axum::Extension(rate_limit::IpRateLimiter {
-        limiter: state.create_ip_rate_limiter.clone(),
-        trust: state.push_limiter_trust,
-    }));
-
-    // ── Status claim reads — open, behind `optional_signature` so the handler
-    // sees the caller when one signs and still answers a public repo for an
-    // anonymous reader. Without the layer the handler's optional auth extension
-    // is None for EVERY caller, including the owner, and every private repo
-    // reads as not-found.
-    let status_read_routes = Router::new()
-        .route(
-            "/api/v1/repos/{owner}/{repo}/commits/{sha}/status",
-            get(status::commit_status),
-        )
-        .route(
-            "/api/v1/repos/{owner}/{repo}/pulls/{number}/status",
-            get(status::pull_request_status),
-        )
-        .layer(middleware::from_fn_with_state(
-            auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
-            auth::optional_signature,
-        ));
+    .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.
@@ -443,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()
@@ -468,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(
@@ -586,10 +578,6 @@ pub fn build_router(state: AppState) -> Router {
             "/api/v1/repos/{owner}/{repo}/events",
             get(events::list_repo_events),
         )
-        .route(
-            "/api/v1/repos/{owner}/{repo}/push-events",
-            get(events::list_repo_push_events),
-        )
         .route("/api/v1/agents", get(agents::list_agents))
         .route("/api/v1/agents/{did}", get(agents::show_agent))
         .route("/api/v1/agents/{did}/trust", get(agents::get_trust))
@@ -695,8 +683,6 @@ pub fn build_router(state: AppState) -> Router {
         .merge(profile_write_routes)
         .merge(creation_routes)
         .merge(write_routes)
-        .merge(status_write_routes)
-        .merge(status_read_routes)
         .merge(git_write_routes)
         .merge(git_read_routes)
         .merge(issue_write_routes)

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)),
@@ -390,7 +398,6 @@ impl TestNode {
             status: "open".to_string(),
             merged_by_did: None,
             merged_at: None,
-            head_commit: None,
             created_at: now.clone(),
             updated_at: now,
         };

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)),
@@ -175,153 +184,6 @@ pub(crate) fn signed_request_as(did: &str, method: Method, uri: &str, body: Body
         .expect("request builder")
 }
 
-/// Slice a region out of a module's own source (obtained with `include_str!`)
-/// and drop full-line `//` comments from it.
-///
-/// Several guards in `api/` assert that a piece of PRODUCTION code does or does
-/// not contain something: that the receive-pack handler really calls the
-/// push-event recorder, that the status module never acquires the repository.
-/// Each guard carried its own copy of this parse, and three copies of a parser
-/// is three places for the slicing to drift apart while every guard still reads
-/// green.
-///
-/// `start` and `end` are literal anchors. The region runs from the first
-/// occurrence of `start` (inclusive; the top of the file when `None`) to the
-/// next occurrence of `end` (exclusive; end of file when `None`).
-///
-/// Returns `None` when an anchor is missing, rather than silently falling back
-/// to a larger or empty region. Each caller supplies its own message for that
-/// case and keeps its own assertion that the region really covers what it claims
-/// to, which is what stops a bug in this one helper from turning every guard
-/// that uses it into a vacuous pass.
-///
-/// Comment lines are stripped so that a doc comment naming the very call a guard
-/// searches for cannot stand in for the call itself.
-pub(crate) fn scrape_source_region(
-    src: &str,
-    start: Option<&str>,
-    end: Option<&str>,
-) -> Option<String> {
-    let from = match start {
-        Some(anchor) => src.find(anchor)?,
-        None => 0,
-    };
-    let rest = &src[from..];
-    let to = match end {
-        // Search past the first CHARACTER, so an `end` anchor that also matches
-        // at the very start of the region cannot collapse it to nothing. A fixed
-        // one-byte skip does the same job on ASCII and lands mid-character on a
-        // multibyte first character, where the slice fails and the helper answers
-        // `None`, which is indistinguishable, to every caller, from a missing
-        // anchor.
-        Some(anchor) => {
-            let skip = rest.chars().next().map_or(0, char::len_utf8);
-            rest.get(skip..)?.find(anchor)? + skip
-        }
-        None => rest.len(),
-    };
-    Some(
-        rest[..to]
-            .lines()
-            .filter(|l| !l.trim_start().starts_with("//"))
-            .collect::<Vec<_>>()
-            .join("\n"),
-    )
-}
-
-#[cfg(test)]
-mod scrape_source_region_tests {
-    use super::scrape_source_region;
-
-    const SRC: &str = "prelude line\nfn target() {\n    // calls_the_thing()\n    calls_the_thing();\n}\nfn later() {\n    calls_the_thing();\n}\n";
-
-    #[test]
-    fn stops_at_the_end_anchor_so_a_later_function_cannot_satisfy_a_guard() {
-        let body = scrape_source_region(SRC, Some("fn target()"), Some("\n}")).unwrap();
-        assert!(body.contains("calls_the_thing();"));
-        assert!(
-            !body.contains("fn later()"),
-            "the region must stop at the first column-zero brace, got: {body}"
-        );
-    }
-
-    #[test]
-    fn strips_full_line_comments_so_a_doc_comment_cannot_stand_in_for_code() {
-        let commented = "fn target() {\n    // calls_the_thing();\n}\n";
-        let body = scrape_source_region(commented, Some("fn target()"), Some("\n}")).unwrap();
-        assert!(
-            !body.contains("calls_the_thing"),
-            "a commented-out call must not survive the scrape, got: {body}"
-        );
-    }
-
-    #[test]
-    fn no_start_anchor_reads_from_the_top_of_the_file() {
-        let body = scrape_source_region(SRC, None, Some("fn later()")).unwrap();
-        assert!(body.starts_with("prelude line"));
-        assert!(!body.contains("fn later()"));
-    }
-
-    #[test]
-    fn no_end_anchor_reads_to_the_end_of_the_file() {
-        let body = scrape_source_region(SRC, Some("fn later()"), None).unwrap();
-        assert!(body.trim_end().ends_with('}'));
-    }
-
-    /// A missing anchor is None, never a silently wider or empty region: that is
-    /// what lets each caller keep its own "the thing I meant to scan is gone"
-    /// message instead of asserting against whatever the fallback produced.
-    #[test]
-    fn a_missing_anchor_is_none_rather_than_a_fallback_region() {
-        assert!(scrape_source_region(SRC, Some("fn absent()"), Some("\n}")).is_none());
-        assert!(scrape_source_region(SRC, Some("fn target()"), Some("~absent~")).is_none());
-    }
-
-    /// A region whose first character is multibyte is scraped, not reported
-    /// missing.
-    ///
-    /// The end-anchor search skips the region's first CHARACTER. Skipping a
-    /// fixed byte instead lands mid-character here, the slice fails, and the
-    /// helper answers `None`, which every caller reads as "the anchor is gone"
-    /// and reports as a missing region. A guard that says "not found"
-    /// for input that is present is worse than one that fails loudly, and the
-    /// five callers sharing this helper all inherit it.
-    #[test]
-    fn a_multibyte_first_character_does_not_read_as_a_missing_anchor() {
-        let src = "é prelude\nfn target() {\n    body();\n}\nfn later() {}\n";
-
-        let body = scrape_source_region(src, None, Some("\n}"))
-            .expect("a region starting on a multibyte character must be scraped");
-        assert!(
-            body.starts_with("é prelude") && body.contains("body();"),
-            "the region must cover the multibyte start through the end anchor, got: {body}"
-        );
-        assert!(
-            !body.contains("fn later()"),
-            "the region must still stop at the end anchor, got: {body}"
-        );
-
-        // The same through a multibyte START anchor, so the skip is measured
-        // from the region rather than from the top of the file.
-        let body = scrape_source_region(src, Some("é"), Some("\n}"))
-            .expect("a multibyte start anchor must be scraped");
-        assert!(body.starts_with('é'), "got: {body}");
-    }
-
-    /// The end anchor is searched past the region's first character, so a
-    /// region whose own start also matches the end anchor is not collapsed to
-    /// nothing.
-    #[test]
-    fn an_end_anchor_matching_at_the_start_does_not_collapse_the_region() {
-        let src = "\n}\nfn target() {\n    body();\n}\n";
-        let body = scrape_source_region(src, None, Some("\n}")).unwrap();
-        assert!(
-            body.contains("fn target()") && body.contains("body();"),
-            "got: {body}"
-        );
-    }
-}
-
 /// A local endpoint whose TCP accept succeeds instantly but that never writes an
 /// HTTP response, so any request against it stalls deterministically until the
 /// caller's own timeout. (A non-routable address hangs only if the network
@@ -329,6 +191,7 @@ mod scrape_source_region_tests {
 /// test pass for the wrong reason.) The accepted sockets are parked in the
 /// spawned task, which dies with the test's runtime, so the peer never sees a
 /// close mid-test.
+#[allow(dead_code)]
 pub(crate) async fn silent_http_endpoint() -> String {
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let endpoint = format!("http://{}", listener.local_addr().unwrap());
@@ -341,71 +204,6 @@ pub(crate) async fn silent_http_endpoint() -> String {
     endpoint
 }
 
-/// The response body decoded as JSON.
-#[allow(dead_code)] // shared response helper for api test mods
-pub(crate) async fn body_json(resp: axum::response::Response) -> serde_json::Value {
-    let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
-        .await
-        .expect("body bytes");
-    serde_json::from_slice(&bytes).expect("json body")
-}
-
-/// Status plus the full response body, for the byte-identical deny comparison.
-#[allow(dead_code)] // shared response helper for api test mods
-pub(crate) async fn status_and_bytes(
-    resp: axum::response::Response,
-) -> (axum::http::StatusCode, Vec<u8>) {
-    let status = resp.status();
-    let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
-        .await
-        .unwrap()
-        .to_vec();
-    (status, bytes)
-}
-
-/// A `PendingRefTransition` test row in `applied` state: the fields the
-/// commit-status effect writers read (`ref_name`, `new_sha`, `ordinal`,
-/// `repo_id`) are the parameters, the rest is inert filler matching the
-/// shape `insert_pending_ref_transitions` writes. `applied_at` stays None
-/// so the recorded event falls back to the per-push timestamp the writer
-/// computes once for the whole batch — the shape a real push produces.
-#[allow(dead_code)] // commit-status surface tests across api modules
-pub(crate) fn pending_child(
-    request_id: &str,
-    repo_id: &str,
-    ref_name: &str,
-    old_sha: &str,
-    new_sha: &str,
-    ordinal: i32,
-) -> crate::db::PendingRefTransition {
-    crate::db::PendingRefTransition {
-        id: crate::db::deterministic_id(&[
-            "pending_ref_transition",
-            request_id,
-            repo_id,
-            ref_name,
-            old_sha,
-            new_sha,
-        ]),
-        request_id: request_id.to_string(),
-        repo_id: repo_id.to_string(),
-        ref_name: ref_name.to_string(),
-        old_sha: old_sha.to_string(),
-        new_sha: new_sha.to_string(),
-        pusher_did: "did:key:z6pusher".to_string(),
-        node_did: "did:key:z6node".to_string(),
-        signature_header: String::new(),
-        signature_input: String::new(),
-        content_digest: String::new(),
-        state: crate::db::pending_state::APPLIED.to_string(),
-        created_at: "2026-01-01T00:00:00Z".to_string(),
-        applied_at: None,
-        cancelled_at: None,
-        ordinal,
-        git_target_kind: Some("update".to_string()),
-    }
-}
-
 /// Body of the `for-each-ref` arm of a fake-git fixture in the
 /// COLUMN SHAPE `blob_paths` phase 2 parses. The caller is
 /// responsible for wrapping this in a full `case "$1" in ... esac`
@@ -2484,7 +2282,6 @@ mod tests {
 
         // Insert the poisoned row DIRECTLY, as a pre-fix row or an ungated writer would.
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -2843,7 +2640,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -2962,7 +2758,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -3087,7 +2882,6 @@ mod tests {
             state.db.create_repo(&repo).await.expect("seed repo");
 
             let pr = crate::db::PullRequest {
-                head_commit: None,
                 id: uuid::Uuid::new_v4().to_string(),
                 repo_id: repo.id.clone(),
                 number: 1,
@@ -3204,7 +2998,6 @@ mod tests {
             state.db.create_repo(&repo).await.expect("seed repo");
 
             let pr = crate::db::PullRequest {
-                head_commit: None,
                 id: uuid::Uuid::new_v4().to_string(),
                 repo_id: repo.id.clone(),
                 number: 1,
@@ -3389,7 +3182,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -3516,7 +3308,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -3646,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);
@@ -3685,7 +3476,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -3807,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);
@@ -3850,7 +3640,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -4041,7 +3830,6 @@ mod tests {
         state.db.create_repo(&repo).await.expect("seed repo");
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -4194,7 +3982,6 @@ mod tests {
         let _ = std::fs::remove_file(&pwn);
 
         let pr = crate::db::PullRequest {
-            head_commit: None,
             id: uuid::Uuid::new_v4().to_string(),
             repo_id: repo.id.clone(),
             number: 1,
@@ -18176,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));
@@ -18198,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,
@@ -18501,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));
@@ -18522,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,

crates/gitlawb-node/tests/support/probe.rs

@@ -915,7 +915,6 @@ fn fill(path: &str, fixture: &Fixture, repo: &str, id_source: IdSource) -> Strin
     };
     path.replace("{owner}", &fixture.owner_did)
         .replace("{repo}", repo)
-        .replace("{sha}", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
         .replace("{number}", "1")
         .replace("{id}", id)
         .replace("{label}", "bug")

crates/gitlawb-node/tests/support/routes.rs

@@ -234,23 +234,6 @@ pub fn deny_bearing_routes() -> &'static [Row] {
             id_source: IdSource::Fixed,
             anon_read: AnonRead::Deny404,
         },
-        // create_status: signed non-owner on a public repo passes the read
-        // gate and hits `require_repo_owner` — a plain owner-gate 403. The
-        // body is a minimal valid claim (the request is body-bound signed, so
-        // the probe's RFC-9421 content-digest coverage is load-bearing: a
-        // signer that did not cover the body would 400 before the gate).
-        Row {
-            method: "POST",
-            path: "/api/v1/repos/{owner}/{repo}/statuses/{sha}",
-            gate: GateClass::OwnerGate,
-            handler: "status::create_status",
-            body: Some(r#"{"state":"pending","context":"ci/build"}"#),
-            needs: NO_ENTITY,
-            reach: Reach::None,
-            principals: NO_PRINCIPAL,
-            id_source: IdSource::Fixed,
-            anon_read: AnonRead::Deny404,
-        },
         // #195 (F1): close_pr / close_issue are owner-OR-author gates (pulls.rs:276,
         // issues.rs:255). A plain OwnerGate tests only the owner arm, so reverting
         // the author arm is invisible. MultiPrincipalGate drives BOTH arms.
@@ -801,50 +784,6 @@ pub fn deny_bearing_routes() -> &'static [Row] {
             id_source: IdSource::Fixed,
             anon_read: AnonRead::Deny404,
         },
-        // pull_request_status: the seeded PRIVATE PR #1 is the reach target, so
-        // the owner twin gets a 200 rollup, not the absent-entity 404 the gate's
-        // own existence-hiding would make indistinguishable.
-        Row {
-            method: "GET",
-            path: "/api/v1/repos/{owner}/{repo}/pulls/{number}/status",
-            gate: GateClass::ReadGate,
-            handler: "status::pull_request_status",
-            body: None,
-            needs: &["priv_pr"],
-            reach: Reach::ReaderReads,
-            principals: NO_PRINCIPAL,
-            id_source: IdSource::Fixed,
-            anon_read: AnonRead::Deny404,
-        },
-        // commit_status / list_repo_push_events: repo-scoped optional-auth reads
-        // whose deny is the repo's own existence-hiding 404. Neither needs a
-        // seeded entity — a status rollup over zero claims and an empty
-        // push-event page are both valid 2xx responses, so the owner twin
-        // proving the gate is the repo lookup, not the data.
-        Row {
-            method: "GET",
-            path: "/api/v1/repos/{owner}/{repo}/commits/{sha}/status",
-            gate: GateClass::ReadGate,
-            handler: "status::commit_status",
-            body: None,
-            needs: NO_ENTITY,
-            reach: Reach::ReaderReads,
-            principals: NO_PRINCIPAL,
-            id_source: IdSource::Fixed,
-            anon_read: AnonRead::Deny404,
-        },
-        Row {
-            method: "GET",
-            path: "/api/v1/repos/{owner}/{repo}/push-events",
-            gate: GateClass::ReadGate,
-            handler: "events::list_repo_push_events",
-            body: None,
-            needs: NO_ENTITY,
-            reach: Reach::ReaderReads,
-            principals: NO_PRINCIPAL,
-            id_source: IdSource::Fixed,
-            anon_read: AnonRead::Deny404,
-        },
         // get_cert: id-keyed by a real ref-certificate issued by an owner push to
         // the private repo. No cert-content marker is seeded: get_cert read-gates
         // the repo on "/" and 404s a non-reader BEFORE fetching the cert, so cert