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

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

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

.env.example

@@ -17,31 +17,6 @@ GITLAWB_PORT=7545
 # ── Storage ───────────────────────────────────────────────────────────────
 GITLAWB_REPOS_DIR=/data/repos
 
-# ── Object storage (durable repo archives) ────────────────────────────────
-# Backend for whole-repo archives: s3 | fs | ipfs. Empty = auto-detect:
-# s3 when a bucket is set, else fs when GITLAWB_STORAGE_FS_DIR is set, else
-# local-only (repos live only on this node's disk). `ipfs` is never
-# auto-selected — setting GITLAWB_IPFS_API alone keeps its pinning-only
-# meaning; opt in explicitly with GITLAWB_STORAGE_BACKEND=ipfs. NOTE: the
-# ipfs backend stores archives in the Kubo daemon's LOCAL MFS namespace, so
-# every node must point at the SAME Kubo instance.
-GITLAWB_STORAGE_BACKEND=
-# Bucket for the s3 backend (Tigris, R2, AWS S3, MinIO, B2).
-# GITLAWB_TIGRIS_BUCKET is honored as a legacy alias.
-GITLAWB_S3_BUCKET=
-# Endpoint URL override for the s3 backend (R2/MinIO). On Tigris/Fly the
-# endpoint arrives via AWS_ENDPOINT_URL_S3 — leave empty.
-GITLAWB_S3_ENDPOINT=
-# Force path-style addressing (required by MinIO and some S3-compatibles).
-GITLAWB_S3_FORCE_PATH_STYLE=false
-# Directory for the fs (local filesystem) backend.
-GITLAWB_STORAGE_FS_DIR=
-# Ack pushes before the durable upload finishes (write-back). Lower latency;
-# opt-in durability tradeoff — see --help for the full semantics.
-GITLAWB_ASYNC_UPLOAD=false
-# Dedicated DB pool for per-repo write locks; a push pins one connection for
-# its lifetime, so this bounds per-node push concurrency.
-
 # PostgreSQL connection URL. Required.
 # When using the bundled docker-compose, this is wired automatically.
 DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb
@@ -49,11 +24,26 @@ DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb
 # ── Database pool & startup resilience ────────────────────────────────────
 # Maximum connections in the PostgreSQL pool. A cap, not a floor —
 # connections open lazily. Size against the DB server's max_connections,
-# remembering admin tooling opens its own pool. Each concurrent write pins one
-# connection for its whole duration (the connection-affine advisory lock), so the
-# node REJECTS at boot any value below GITLAWB_MAX_CONCURRENT_GIT_PUSHES + 8
-# headroom — keep this comfortably above that (default 48 for pushes 32).
+# remembering admin tooling opens its own pool.
 GITLAWB_DB_MAX_CONNECTIONS=48
+# Maximum connections in the DEDICATED advisory-lock pool, separate from the
+# pool above. Every in-flight repo write pins one connection here for its whole
+# duration, so this is a hard ceiling on simultaneous writes node-wide: size it
+# to expected peak concurrent writers, not small. Keeping it separate is what
+# stops a push burst from starving ordinary request handlers. Budget
+# (GITLAWB_DB_MAX_CONNECTIONS + this) per node against the server's
+# max_connections, times node count, plus admin tooling.
+GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS=40
+# Upper bound, in seconds, on any object-storage transfer that runs while a
+# per-repo write lock is HELD (the archive download inside acquire_write and the
+# upload inside release). These were free before the lock's connection was
+# pinned to the guard; now an unbounded stall holds a lock-pool slot, and enough
+# stalls deny every write on the node. The bound applies PER SPAN and there are
+# two (the acquire-side refresh, which covers the existence check and download
+# together, and the release-side upload), so worst-case slot occupancy is about
+# twice this value plus the git work between them. Read it together with the pool
+# size above and with GITLAWB_GIT_SERVICE_TIMEOUT_SECS.
+GITLAWB_LOCK_HELD_TRANSFER_TIMEOUT_SECS=300
 # Seconds a request waits for a pool connection before failing with 503.
 GITLAWB_DB_ACQUIRE_TIMEOUT_SECS=5
 # Upper bound on each startup connect+migrate attempt, in seconds. Keep it

Cargo.lock

@@ -3441,7 +3441,6 @@ dependencies = [
  "async-compression",
  "async-graphql",
  "async-graphql-axum",
- "async-trait",
  "aws-config",
  "aws-sdk-s3",
  "axum",
@@ -3466,7 +3465,6 @@ dependencies = [
  "libp2p-kad",
  "libp2p-quic",
  "libp2p-swarm",
- "md-5",
  "mockito",
  "multiaddr",
  "prometheus",

README.md

@@ -397,7 +397,7 @@ Important node settings:
 | `GITLAWB_AUTO_SYNC` | Enable automatic sync from known peers. |
 | `GITLAWB_MAX_PACK_BYTES` | Max git pack body size for smart-HTTP routes. |
 | `GITLAWB_GIT_SERVICE_TIMEOUT_SECS` | Max seconds a served git upload-pack, receive-pack, or `info/refs` advertisement may run before it is aborted (504). Default 600. Also bounds the withheld-blob classification walk (on both the upload-pack serve and receive-pack replication paths) and the push-side pin-candidate discovery (`rev-list` / `cat-file`), each reaped via process-group teardown at the deadline. On the path-scoped upload-pack path the classification walk and the pack serve share ONE deadline, so this value bounds their combined duration rather than granting each stage a full budget: a walk that consumes it leaves the serve nothing and the clone gets a 504. Serving large path-scoped repos may therefore need a higher value than they did when each stage was budgeted separately. Accepted range is 1 to 3153600000 (100 years), since the node derives deadlines from this value and a larger one cannot be represented. |
-| `GITLAWB_GIT_ACQUIRE_TIMEOUT_SECS` | Max seconds the storage-acquisition phase (storage-backend HEAD/GET, push advisory-lock) of a served git op may run before the request is shed with a 503, separate from the git-run timeout. The concurrency permit is released on expiry so a stalled backend cannot pin the pool. Default 30. |
+| `GITLAWB_GIT_ACQUIRE_TIMEOUT_SECS` | Max seconds the storage-acquisition phase (Tigris HEAD/GET, push advisory-lock) of a served git op may run before the request is shed with a 503, separate from the git-run timeout. The concurrency permit is released on expiry so a stalled backend cannot pin the pool. Default 30. |
 | `GITLAWB_MAX_CONCURRENT_GIT_OPS` | Max concurrent served git READ ops (upload-pack and its `info/refs` advertisement) across all callers; over-cap sheds a 503 + Retry-After. Anonymous reads draw from this pool, so pair it with `GITLAWB_MAX_CONCURRENT_READS_PER_CALLER`. Pushes and the receive-pack advertisement have their own pools, so a read flood cannot shed an authenticated push. Default 128. |
 | `GITLAWB_MAX_CONCURRENT_GIT_PUSHES` | Max concurrent `git-receive-pack` POST operations, in a pool separate from the read pool. The anon receive-pack `info/refs` advertisement runs in a third pool of the same size, disjoint from both, so an advertisement flood cannot shed a push either. Two per-source push caps are derived from this value (`/8`, floor 1) and have no env var of their own. Over-cap sheds a 503 + Retry-After. Default 32. |
 | `GITLAWB_MAX_CONCURRENT_READS_PER_CALLER` | Max concurrent read ops a single caller may hold, so one caller cannot monopolize the read pool. Keyed on the resolved source IP, never the DID, and only as granular as `GITLAWB_TRUSTED_PROXY`: left unset, a node behind an edge or NAT keys every caller on the edge IP and this collapses to one global cap. Default 16. |
@@ -412,13 +412,7 @@ Important node settings:
 | `GITLAWB_IPFS_REQUEST_BUDGET_SECS` | Absolute wall-clock budget for one admitted `/ipfs/{cid}` request's acquire+walk lifetime. Per-stage clamps bound the acquire and walk stages to the remaining budget, and no stage starts once it is exhausted; the scan then stops with a retryable 503. The object-type probe and content-read `cat-file` subprocesses are budget-checked before starting and each also run under their own deadline (the lesser of `GITLAWB_GIT_SERVICE_TIMEOUT_SECS` and the remaining budget), reaped via process-group teardown, so a hung `cat-file` cannot hold the request's walk slot past it. One hang path is still unbounded: the probe's object-store readability check is a plain filesystem sweep with nothing to reap, so a wedged filesystem can hold the slot past the deadline. Default 600. Accepted range is 1 to 3153600000 (100 years), since the node derives a deadline from this value and a larger one cannot be represented. |
 | `GITLAWB_IPFS_RESOLVE_BUDGET_SECS` | Shorter budget for the pre-walk CID resolve inside an admitted `/ipfs/{cid}` request: the lookup that maps the requested CID to its git oid(s), which runs while the scarce walk admission is already held. A well-formed CID with no pin row does no probe and no walk work, so without this it could hold a walk slot for the whole request budget while nothing walked, and enough such requests shed every real retrieval at admission. The effective deadline is the lesser of this and the remaining request budget, so a value above `GITLAWB_IPFS_REQUEST_BUDGET_SECS` degrades to the request budget. Only the resolve is on this clock; walk and probe work stay on the request budget, so a slow but progressing scan is never shed by it. Default 10. Accepted range is 1 to 3153600000 (100 years). |
 | `GITLAWB_IPFS_RATE_LIMIT` | Max `/ipfs/{cid}` requests per client IP per hour (route flood brake). 0 disables. Default 600. |
-| `GITLAWB_STORAGE_BACKEND` | Object-storage backend for repo archives: `s3`, `fs`, or `ipfs`. Empty = auto-detect (`s3` if a bucket is set, else `fs` if a dir is set, else local-only). `ipfs` is never auto-selected and requires all nodes to share one Kubo instance (MFS is daemon-local). |
-| `GITLAWB_S3_BUCKET` | Bucket for the `s3` backend (Tigris, R2, AWS S3, MinIO, B2). |
-| `GITLAWB_S3_ENDPOINT` | Endpoint URL override for the `s3` backend (R2/MinIO; empty on Tigris/Fly). |
-| `GITLAWB_S3_FORCE_PATH_STYLE` | Force path-style S3 addressing (MinIO and some S3-compatibles). |
-| `GITLAWB_STORAGE_FS_DIR` | Directory for the `fs` (local filesystem) backend. |
-| `GITLAWB_ASYNC_UPLOAD` | Ack pushes before the durable storage upload (write-back). Lower latency, opt-in durability tradeoff. Default `false`. |
-| `GITLAWB_TIGRIS_BUCKET` | Legacy alias for `GITLAWB_S3_BUCKET` (selects the `s3` backend). |
+| `GITLAWB_TIGRIS_BUCKET` | Optional S3/Tigris shared repo storage bucket. |
 | `GITLAWB_PINATA_JWT` | Optional Pinata/IPFS warm-storage pinning. |
 | `GITLAWB_IRYS_URL` | Optional Irys/Arweave permanent anchoring. |
 

crates/gitlawb-node/Cargo.toml

@@ -33,15 +33,9 @@ sqlx = { version = "0.8", features = ["postgres", "runtime-tokio-rustls", "chron
 clap = { version = "4", features = ["derive", "env"] }
 bytes = "1"
 libc = "0.2"
-async-trait = "0.1"
 cid = { workspace = true }
 hex = { workspace = true }
 sha2 = { workspace = true }
-# Content etags for repo archives: S3 single-part ETags are the body MD5, so a
-# client-side MD5 lets the uploader record its *intended* etag before the PUT
-# (crash-recovery provenance for the pending-upload marker). Not used for
-# anything security-sensitive.
-md-5 = "0.10"
 hmac = { workspace = true }
 http-body-util = "0.1"
 tokio-util = { version = "0.7", features = ["io"] }
@@ -63,7 +57,7 @@ aws-sdk-s3 = { version = "1", default-features = false, features = ["sigv4a", "d
 aws-config = { version = "1", features = ["behavior-version-latest"] }
 async-compression = { version = "0.4", features = ["tokio", "zstd"] }
 tar = "0.4"
-zstd = { version = "0.13", features = ["zstdmt"] }
+zstd = "0.13"
 # Prometheus metrics. Used to expose a /metrics endpoint for ops/observability
 # on the opt-in GITLAWB_METRICS_ADDR listener. The crate is also the de-facto
 # exposition format encoder in the Rust ecosystem.

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

@@ -2333,7 +2333,7 @@ pub(crate) enum MarkerQuery {
 
 // Test-only fault-injection seam for the `needs_scan` marker pair
 // (`pin_sources_at_cap`, `pin_sources_incomplete`), same idea as
-// `RepoStore::storage_stall`: hold one specific await open so the clamp around it is
+// `RepoStore::tigris_stall`: hold one specific await open so the clamp around it is
 // the one observed to fire.
 //
 // A `LOCK TABLE` fixture cannot isolate these two. `pin_sources_at_cap` reads
@@ -3757,12 +3757,14 @@ mod tests {
         // parallel test run); the silent local endpoint stalls the HEAD
         // deterministically.
         let endpoint = crate::test_support::silent_http_endpoint().await;
-        let blob: std::sync::Arc<dyn crate::storage::BlobStore> = std::sync::Arc::new(
-            crate::storage::s3::S3BlobStore::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await,
+        let tigris =
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
         );
-        let archive = crate::storage::archive::RepoArchive::new(blob);
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(archive), pool);
         state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
         let mut cfg = (*state.config).clone();
         cfg.git_acquire_timeout_secs = 1;
@@ -3832,12 +3834,14 @@ mod tests {
         // consults the silent local endpoint and stalls to the 1s timeout
         // (endpoint-pinned test client, no AWS_* env reads).
         let endpoint = crate::test_support::silent_http_endpoint().await;
-        let blob: std::sync::Arc<dyn crate::storage::BlobStore> = std::sync::Arc::new(
-            crate::storage::s3::S3BlobStore::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await,
+        let tigris =
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
         );
-        let archive = crate::storage::archive::RepoArchive::new(blob);
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(archive), pool);
         let mut cfg = (*state.config).clone();
         cfg.git_acquire_timeout_secs = 1;
         state.config = Arc::new(cfg);
@@ -7959,12 +7963,19 @@ mod tests {
         // consults the silent local endpoint and stalls past the budget
         // (endpoint-pinned test client, no AWS_* env reads).
         let endpoint = crate::test_support::silent_http_endpoint().await;
-        let blob: std::sync::Arc<dyn crate::storage::BlobStore> = std::sync::Arc::new(
-            crate::storage::s3::S3BlobStore::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await,
+        let tigris =
+            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint);
+        state.repo_store = crate::git::repo_store::RepoStore::new(
+            repos_dir,
+            Some(tigris),
+            pool,
+            std::time::Duration::from_secs(300),
         );
-        let archive = crate::storage::archive::RepoArchive::new(blob);
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(archive), pool);
+        state
+            .db
+            .upsert_mirror_repo("z6f3budget", "ghost", "/unused-ghost", None, false)
+            .await
+            .unwrap();
         let mut cfg = (*state.config).clone();
         cfg.ipfs_request_budget_secs = 1;
         cfg.git_acquire_timeout_secs = 2;

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

@@ -209,9 +209,6 @@ pub async fn merge_pr(
         return Err(AppError::BadRequest(format!("PR is already {}", pr.status)));
     }
 
-    // Shed 503 + Retry-After on an exhausted write-lock POOL instead of a generic
-    // git 500 (#173 F1). Merging holds no admission permit, so it reaches the pool
-    // unthrottled; reuse the push handler's mapping so the two cannot drift.
     let guard = state
         .repo_store
         .acquire_write(&record.owner_did, &record.name)
@@ -227,31 +224,13 @@ pub async fn merge_pr(
         &pr.title,
     );
 
-    // Always release the advisory lock — even on error; upload to storage only on success.
-    let release_result = guard.release(merge_result.is_ok()).await;
+    // Always release the advisory lock — even on error; upload to Tigris only on success.
+    // Short-circuit on a refused publish before the PR is marked merged and
+    // before the webhook fires. Both are irreversible announcements of a merge
+    // commit that only exists on this node's disk.
+    guard.release(merge_result.is_ok()).await.into_result()?;
 
     let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
-    // A durable-upload failure is recoverable ONLY while the pending-upload
-    // marker protects the merge commit; verify it exists before choosing to
-    // proceed (proceeding keeps the DB status consistent with the already
-    // merged ref, which a retry cannot un-merge). Without the marker nothing
-    // protects the merge from a stale-archive rollback — fail the request so
-    // the inconsistency is surfaced instead of silently losable.
-    if let Err(e) = release_result {
-        if state
-            .repo_store
-            .pending_marker_exists(&record.owner_did, &record.name)
-        {
-            tracing::error!(repo = %record.name, pr = %pr.id, err = %e,
-                "merge committed locally but durable upload failed — storage re-syncs on next upload");
-        } else {
-            tracing::error!(repo = %record.name, pr = %pr.id, err = %e,
-                "merge committed locally with NO durable protection — failing the request");
-            return Err(AppError::Git(format!(
-                "merge applied locally but durability could not be guaranteed: {e}"
-            )));
-        }
-    }
 
     state.db.merge_pr(&pr.id, &merger_did).await?;
     let _ = state.db.touch_repo(&record.id).await;
@@ -449,12 +428,8 @@ pub async fn list_comments(
     Ok(Json(serde_json::json!({ "comments": comments })))
 }
 
-/// #173 F1 follow-up: `merge_pr` reaches `acquire_write` holding NO admission permit
-/// (unlike the push handler, which is capped by the git-push semaphore), so it is one
-/// of the callers most likely to meet an exhausted write-lock POOL under load. An
-/// exhausted pool is a capacity signal, so the merge must shed 503 + Retry-After
-/// (`AppError::Overloaded`) the way the push handler does, not report the generic
-/// 500 git error that says nothing about retrying.
+/// #173 F1 follow-up: merge holds no admission permit, so an exhausted write-lock
+/// pool must shed 503 + Retry-After.
 #[cfg(test)]
 mod lock_pool_shed_tests {
     use super::*;
@@ -478,20 +453,15 @@ mod lock_pool_shed_tests {
         }
     }
 
-    /// RED-before/GREEN-after for `merge_pr`. Both directions: the shed while the only
-    /// lock-pool connection is held by a guard on a DIFFERENT repo (so this is pool
-    /// capacity, not advisory-lock contention on this repo), and the must-not case
-    /// once that connection is back.
     #[sqlx::test]
     async fn merge_pr_lock_pool_exhaustion_sheds_503_not_500(pool: PgPool) {
         let owner = "did:key:zMERGELOCKPOOLOWNERAAAAAAAAAAAAAAAAAAAAA";
         let mut state = crate::test_support::test_state(pool.clone()).await;
-        // One lock-pool connection with a short checkout timeout, so a single held
-        // guard exhausts it promptly rather than at the pool default.
         state.repo_store = crate::git::repo_store::RepoStore::new(
             std::path::PathBuf::from("/tmp/gitlawb-pulls-lockpool"),
             None,
             crate::git::repo_store::build_lock_pool(&pool, 1, std::time::Duration::from_secs(1)),
+            std::time::Duration::from_secs(300),
         );
 
         let repo = seed_repo(owner, "lp-merge");
@@ -545,9 +515,7 @@ mod lock_pool_shed_tests {
             "merge_pr: a capacity shed must tell the client when to retry"
         );
 
-        // MUST-NOT: with the pool free again the merge is not shed as capacity (it
-        // fails later on the nonexistent on-disk repo, which is a git 500).
-        held.release(false).await.ok();
+        let _ = held.release(false).await;
         let admitted = merge_pr(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),

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

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

crates/gitlawb-node/src/config.rs

@@ -172,50 +172,11 @@ pub struct Config {
     #[arg(long, env = "GITLAWB_HEARTBEAT_INTERVAL_HOURS", default_value_t = 20)]
     pub heartbeat_interval_hours: u64,
 
-    /// Tigris (S3-compatible) bucket for repo storage. Legacy alias for
-    /// `s3_bucket` — still honoured so existing deployments keep working.
+    /// 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 = "")]
     pub tigris_bucket: String,
 
-    /// Object-storage backend: `s3` (any S3-compatible service), `fs` (local
-    /// directory), or `ipfs` (Kubo MFS). Empty = auto-detect: `s3` when a bucket
-    /// is set, else `fs` when a storage dir is set, else local-only. `ipfs` is
-    /// never auto-selected (`GITLAWB_IPFS_API` alone keeps its pinning-only
-    /// meaning) — set this explicitly to `ipfs` to opt in, and note the ipfs
-    /// backend requires ALL nodes to share one Kubo instance (MFS is
-    /// daemon-local, not shared network storage).
-    #[arg(long, env = "GITLAWB_STORAGE_BACKEND", default_value = "")]
-    pub storage_backend: String,
-
-    /// Bucket for the `s3` backend (Tigris, R2, AWS S3, MinIO, B2). Falls back to
-    /// `tigris_bucket` when empty.
-    #[arg(long, env = "GITLAWB_S3_BUCKET", default_value = "")]
-    pub s3_bucket: String,
-
-    /// Endpoint URL override for the `s3` backend (e.g. R2/MinIO). On Tigris/Fly
-    /// the endpoint is auto-provided via `AWS_ENDPOINT_URL_S3`, so leave empty.
-    #[arg(long, env = "GITLAWB_S3_ENDPOINT", default_value = "")]
-    pub s3_endpoint: String,
-
-    /// Force path-style S3 addressing (required by MinIO and some S3-compatibles).
-    #[arg(long, env = "GITLAWB_S3_FORCE_PATH_STYLE", default_value_t = false)]
-    pub s3_force_path_style: bool,
-
-    /// Directory for the `fs` (local filesystem) storage backend.
-    #[arg(long, env = "GITLAWB_STORAGE_FS_DIR", default_value = "")]
-    pub storage_fs_dir: String,
-
-    /// Acknowledge a push to the client before the durable upload to object
-    /// storage finishes (write-back). Lowers push latency, but opens a
-    /// durability window: if the upload fails or the node stops first, storage
-    /// stays stale until the next successful upload. A persisted pending-upload
-    /// marker keeps the local copy authoritative on this node in that window
-    /// (no rollback of the acked push), but other nodes serve the stale archive
-    /// until the re-upload lands. Off by default (strict upload-before-ack).
-    #[arg(long, env = "GITLAWB_ASYNC_UPLOAD", default_value_t = false)]
-    pub async_upload: bool,
-
     /// Maximum pack body size for git-receive-pack and git-upload-pack, in bytes.
     /// Applies only to git smart-HTTP routes — all other API routes keep the 2 MB default.
     /// Default: 2 GB.  Set lower on resource-constrained nodes.
@@ -291,11 +252,9 @@ pub struct Config {
 
     /// Maximum connections in the PostgreSQL pool. This is a cap, not a floor
     /// (connections open lazily). Size against the database server's
-    /// max_connections, remembering admin tooling opens its own pool. Each
-    /// concurrent write pins one pooled connection for its whole duration (the
-    /// advisory lock in `repo_store::acquire_write` is connection-affine), so this
-    /// must exceed `max_concurrent_git_pushes` by `DB_POOL_APP_HEADROOM` or slow
-    /// pushes starve every other DB path — enforced by `Config::validate`.
+    /// max_connections, remembering admin tooling opens its own pool. Writes pin
+    /// the dedicated lock pool, not this one; this pool still needs enough headroom
+    /// for ordinary request handlers and metadata paths (`DB_POOL_APP_HEADROOM`).
     #[arg(
         long,
         env = "GITLAWB_DB_MAX_CONNECTIONS",
@@ -304,6 +263,46 @@ pub struct Config {
     )]
     pub db_max_connections: u32,
 
+    /// Maximum connections in the dedicated advisory-lock pool, which is separate
+    /// from the main pool above.
+    ///
+    /// Size this against the expected peak number of concurrent distinct-repo
+    /// writers, NOT small. Every in-flight repo write pins one connection here for
+    /// its whole duration (the write, its metadata tail, and the bounded archive
+    /// upload), so this value is a hard ceiling on simultaneous writes node-wide.
+    /// Keeping it separate from GITLAWB_DB_MAX_CONNECTIONS is what stops a push
+    /// burst from starving ordinary request handlers; the cost is that
+    /// (main pool + lock pool) must fit inside the database server's
+    /// max_connections, times the number of nodes, plus admin tooling.
+    #[arg(
+        long,
+        env = "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS",
+        default_value_t = 40,
+        value_parser = clap::value_parser!(u32).range(1..)
+    )]
+    pub db_lock_pool_max_connections: u32,
+
+    /// Upper bound, in seconds, on any single object-storage transfer that runs
+    /// while the per-repo advisory lock is HELD.
+    ///
+    /// Two bounded spans exist, and the bound applies per span. The acquire-side
+    /// refresh in `acquire_write` covers the existence HEAD and the download
+    /// together under ONE budget (it runs after the lock is taken and before the
+    /// guard is constructed), and the archive upload in `release` gets its own.
+    /// Worst-case slot occupancy is therefore about twice this value plus the git
+    /// work between them, not one times this value. Both used to be free, because the lock's
+    /// connection was returned to the pool immediately; now that a write guard
+    /// pins a lock-pool connection for its whole lifetime, an unbounded transfer
+    /// holds that slot, and enough stalled transfers deny every write on the node.
+    /// This is the bound that keeps a stall from becoming an outage.
+    #[arg(
+        long,
+        env = "GITLAWB_LOCK_HELD_TRANSFER_TIMEOUT_SECS",
+        default_value_t = 300,
+        value_parser = clap::value_parser!(u64).range(1..)
+    )]
+    pub lock_held_transfer_timeout_secs: u64,
+
     /// Maximum time a request waits for a pool connection before failing with
     /// 503, in seconds. Bounds queueing when the database is slow or down.
     #[arg(
@@ -398,15 +397,14 @@ pub struct Config {
     /// error rather than a boot-time panic).
     ///
     /// CONNECTION BUDGET. A push holds a Postgres connection from the node's separate
-    /// advisory-lock pool for the whole receive-pack, and that pool is sized from this
-    /// knob (this value + 8, clamped to 64 in `main.rs`). The node's total ceiling is
-    /// therefore `db_max_connections` (default 48) + the lock pool (default 40), i.e.
-    /// 88 by default, and at most `db_max_connections` + 64. Size BOTH against the
-    /// database server's `max_connections`: `db_max_connections`' own doc predates the
-    /// lock pool and no longer covers most of the node's connections. The +8 headroom
-    /// is shared with the three non-push `acquire_write` callers (`api/issues.rs` x2,
-    /// `api/pulls.rs`). Raising this knob past the clamp does NOT buy more lock-pool
-    /// connections; pushes beyond it wait briefly and then shed a 503 + Retry-After.
+    /// advisory-lock pool (`GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS`, default 40) for
+    /// the whole receive-pack. That pool must be at least this value plus
+    /// `DB_LOCK_POOL_NON_PUSH_HEADROOM` so admitted pushes and non-push mutations
+    /// can each pin a lock-pool connection for their whole duration. Size BOTH
+    /// pools against the database server's `max_connections`: the main pool
+    /// (`GITLAWB_DB_MAX_CONNECTIONS`, default 48) serves ordinary handlers, and the
+    /// lock pool serves writes. Raising this knob does not raise the lock pool;
+    /// set `GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS` explicitly.
     #[arg(
         long,
         env = "GITLAWB_MAX_CONCURRENT_GIT_PUSHES",
@@ -764,25 +762,37 @@ impl Config {
     /// the pool must clear the concurrent-write cap by at least this margin.
     pub const DB_POOL_APP_HEADROOM: u32 = 8;
 
+    /// Lock-pool slots reserved for non-push `acquire_write` callers (issue create,
+    /// close, merge) so a saturated push budget cannot turn ordinary mutations into
+    /// lock-pool exhaustion on unrelated repositories.
+    pub const DB_LOCK_POOL_NON_PUSH_HEADROOM: u32 = 8;
+
     /// Cross-field boot validation. Single-field ranges are enforced by clap; this
     /// catches combinations that ship a denial-of-service under otherwise-valid
     /// values. Call once at startup and fail fast on `Err`.
     pub fn validate(&self) -> Result<(), String> {
-        // A write pins one pooled connection for its whole duration (the
-        // connection-affine advisory lock in repo_store::acquire_write), and
-        // concurrent writes are capped at max_concurrent_git_pushes. If the pool
-        // does not exceed that cap by DB_POOL_APP_HEADROOM, a burst of slow pushes
-        // drains every connection and every other DB path 503s. (#174 F1)
-        let floor = (self.max_concurrent_git_pushes as u64) + (Self::DB_POOL_APP_HEADROOM as u64);
-        if (self.db_max_connections as u64) < floor {
+        // Concurrent git writes pin the dedicated lock pool for their whole
+        // duration (connection-affine advisory lock in acquire_write). The main
+        // pool no longer carries that occupancy.
+        let lock_floor = self
+            .max_concurrent_git_pushes
+            .saturating_add(Self::DB_LOCK_POOL_NON_PUSH_HEADROOM as usize);
+        if (self.db_lock_pool_max_connections as usize) < lock_floor {
             return Err(format!(
-                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least max_concurrent_git_pushes ({}) \
-                 + {} headroom = {}: each concurrent write pins one pooled connection for its whole \
-                 duration, so a smaller pool lets a burst of slow pushes starve every other DB path.",
-                self.db_max_connections,
+                "GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS ({}) must be at least \
+                 max_concurrent_git_pushes ({}) plus {} for non-push mutations \
+                 that share the lock pool",
+                self.db_lock_pool_max_connections,
                 self.max_concurrent_git_pushes,
-                Self::DB_POOL_APP_HEADROOM,
-                floor
+                Self::DB_LOCK_POOL_NON_PUSH_HEADROOM
+            ));
+        }
+        let main_floor = Self::DB_POOL_APP_HEADROOM as u64;
+        if (self.db_max_connections as u64) < main_floor {
+            return Err(format!(
+                "GITLAWB_DB_MAX_CONNECTIONS ({}) must be at least {} for non-git \
+                 database paths",
+                self.db_max_connections, main_floor
             ));
         }
         Ok(())
@@ -793,6 +803,25 @@ impl Config {
 mod tests {
     use super::*;
 
+    #[test]
+    fn lock_pool_size_defaults_to_40_and_rejects_zero() {
+        assert_eq!(
+            Config::parse_from(["gitlawb-node"]).db_lock_pool_max_connections,
+            40
+        );
+        assert_eq!(
+            Config::parse_from(["gitlawb-node", "--db-lock-pool-max-connections", "8"])
+                .db_lock_pool_max_connections,
+            8
+        );
+        // A zero-sized lock pool would deny every write, so clap must reject it
+        // rather than let a node boot into a state where no repo can be written.
+        assert!(
+            Config::try_parse_from(["gitlawb-node", "--db-lock-pool-max-connections", "0"])
+                .is_err()
+        );
+    }
+
     #[test]
     fn git_service_timeout_defaults_to_600_and_rejects_zero() {
         assert_eq!(
@@ -1402,41 +1431,61 @@ mod tests {
         );
     }
 
-    /// #174 F1: a connection-affine write lock pins a pooled connection per
-    /// concurrent write, so the pool must clear `max_concurrent_git_pushes` by
-    /// `DB_POOL_APP_HEADROOM` or a push burst starves every other DB path.
-    /// `validate()` must reject an under-sized pool at boot.
+    /// #174 F1: concurrent git writes pin the dedicated lock pool, not the main
+    /// pool. `validate()` must reject an under-sized lock pool at boot.
     #[test]
     fn db_pool_must_clear_the_git_push_cap() {
-        // Shipped defaults validate (48 >= 32 + 8).
+        // Shipped defaults validate (lock pool 40 >= pushes 32 + non-push headroom 8).
         Config::parse_from(["gitlawb-node"])
             .validate()
             .expect("default config must validate");
 
-        // An under-sized pool relative to the push cap is rejected (20 < 32 + 8).
-        let under = Config::parse_from([
+        // An under-sized lock pool relative to the push cap is rejected.
+        let under_lock = Config::parse_from([
             "gitlawb-node",
-            "--db-max-connections",
-            "20",
+            "--db-lock-pool-max-connections",
+            "16",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            under.validate().is_err(),
-            "db_max_connections 20 below max_concurrent_git_pushes 32 + headroom must be rejected"
+            under_lock.validate().is_err(),
+            "db_lock_pool_max_connections below max_concurrent_git_pushes + headroom must be rejected"
         );
 
-        // Exactly at the floor validates (40 == 32 + 8).
-        let at_floor = Config::parse_from([
+        // Exactly at the push cap without non-push headroom is rejected.
+        let push_only = Config::parse_from([
+            "gitlawb-node",
+            "--db-lock-pool-max-connections",
+            "32",
+            "--max-concurrent-git-pushes",
+            "32",
+        ]);
+        assert!(
+            push_only.validate().is_err(),
+            "db_lock_pool_max_connections equal to max_concurrent_git_pushes must be rejected"
+        );
+
+        // Main pool can be smaller than pushes + headroom when the lock pool carries writes.
+        let split = Config::parse_from([
             "gitlawb-node",
             "--db-max-connections",
+            "16",
+            "--db-lock-pool-max-connections",
             "40",
             "--max-concurrent-git-pushes",
             "32",
         ]);
         assert!(
-            at_floor.validate().is_ok(),
-            "db_max_connections at the floor (pushes + headroom) must validate"
+            split.validate().is_ok(),
+            "a small main pool with a large lock pool must validate"
+        );
+
+        // Main pool still needs the app headroom floor.
+        let under_main = Config::parse_from(["gitlawb-node", "--db-max-connections", "4"]);
+        assert!(
+            under_main.validate().is_err(),
+            "db_max_connections below DB_POOL_APP_HEADROOM must be rejected"
         );
     }
 

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

@@ -265,11 +265,6 @@ pub struct Db {
 }
 
 impl Db {
-    /// Access the underlying Postgres connection pool.
-    pub fn pool(&self) -> &PgPool {
-        &self.pool
-    }
-
     #[cfg(test)]
     pub fn for_testing(pool: PgPool) -> Self {
         Self { pool }
@@ -310,6 +305,46 @@ impl Db {
         Ok(db)
     }
 
+    /// Build the dedicated pool that advisory-lock connections come from.
+    ///
+    /// Deliberately **lazy**: connections open on first use rather than at boot.
+    /// The main pool has to connect eagerly because it runs migrations, which is
+    /// why it needs `connect_db_with_retry`'s backoff and degraded-server
+    /// handoff. This pool has no startup work at all, so an eager connect would
+    /// only add a new way for the process to fail to boot, and would need a
+    /// second copy of that retry machinery to be safe. Being lazy removes the
+    /// failure mode instead of handling it: if Postgres is unreachable when the
+    /// first write arrives, that write fails on the pool's own acquire timeout,
+    /// the same way any other database-backed request already does.
+    ///
+    /// Kept separate from the main pool so a burst of lock-holding connections
+    /// cannot starve ordinary request handlers. See
+    /// `GITLAWB_DB_LOCK_POOL_MAX_CONNECTIONS` for the sizing tradeoff.
+    pub fn lock_pool(
+        database_url: &str,
+        max_connections: u32,
+        acquire_timeout: Duration,
+    ) -> Result<PgPool> {
+        info!(
+            max_connections,
+            acquire_timeout_secs = acquire_timeout.as_secs(),
+            "creating dedicated advisory-lock pool (lazy)"
+        );
+        PgPoolOptions::new()
+            .max_connections(max_connections)
+            // Explicit, and load-bearing rather than cosmetic: `RepoWriteGuard::Drop`
+            // has a no-runtime branch that calls `PoolConnection::leak()` and relies
+            // on the husk's own drop doing nothing. With `min_connections > 0` that
+            // drop still spawns a pool-replenish task, and spawning without a runtime
+            // panics inside Drop, which aborts the process during unwind. It is 0 by
+            // default, so this line exists to keep a future tuning change from
+            // silently re-arming that panic.
+            .min_connections(0)
+            .acquire_timeout(acquire_timeout)
+            .connect_lazy(database_url)
+            .context("creating advisory-lock pool")
+    }
+
     /// Cheap liveness probe against the pool, for readiness checks: one
     /// `SELECT 1` that fails fast when the database is unreachable.
     pub async fn ping(&self) -> Result<()> {
@@ -1239,17 +1274,6 @@ impl Db {
         Ok(())
     }
 
-    /// Compensating delete for creation flows: remove the row a failed
-    /// create/fork just inserted, keyed strictly by our own generated id so a
-    /// concurrent same-name winner's row can never be affected.
-    pub async fn delete_repo_by_id(&self, id: &str) -> Result<()> {
-        sqlx::query("DELETE FROM repos WHERE id = $1")
-            .bind(id)
-            .execute(&self.pool)
-            .await?;
-        Ok(())
-    }
-
     /// Register a mirrored repo from a peer in the local DB so git smart HTTP can serve it.
     /// Uses INSERT OR IGNORE (SQLite) / ON CONFLICT DO NOTHING (Postgres) so it's idempotent.
     pub async fn upsert_mirror_repo(

crates/gitlawb-node/src/error.rs

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

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

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

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

@@ -3,4 +3,5 @@ pub mod push_delta;
 pub mod repo_store;
 pub mod smart_http;
 pub mod store;
+pub mod tigris;
 pub mod visibility_pack;

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

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

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

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

crates/gitlawb-node/src/ipfs_pin.rs

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

crates/gitlawb-node/src/main.rs

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

crates/gitlawb-node/src/metrics.rs

@@ -15,8 +15,6 @@
 //!     `gitlawb_pack_size_bytes`
 //!   * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards
 //!   * currently-connected peer count — `gitlawb_peers_connected`
-//!   * repos whose local copy is ahead of durable storage —
-//!     `gitlawb_pending_upload_markers`
 //!
 //! All metrics live in a single process-wide registry initialized by
 //! [`init`]. Increment helpers (`record_push`, `record_auth_failure`, ...)
@@ -53,7 +51,6 @@ static SYNC_PROCESSED: OnceLock<IntCounterVec> = OnceLock::new();
 static WEBHOOK_DELIVERIES: OnceLock<IntCounterVec> = OnceLock::new();
 static PACK_SIZE: OnceLock<Histogram> = OnceLock::new();
 static PEERS_CONNECTED: OnceLock<IntGauge> = OnceLock::new();
-static PENDING_UPLOAD_MARKERS: OnceLock<IntGauge> = OnceLock::new();
 
 /// One-time initializer. Builds the registry, registers every metric,
 /// and sets the constant `gitlawb_info` gauge. Idempotent — calling
@@ -205,18 +202,6 @@ fn init_inner(version: &str, node_did: &str) {
         .set(peers_connected)
         .expect("set PEERS_CONNECTED once");
 
-    let pending_markers = IntGauge::with_opts(Opts::new(
-        "gitlawb_pending_upload_markers",
-        "Repos whose local copy is ahead of durable storage (pending-upload markers outstanding)",
-    ))
-    .expect("gitlawb_pending_upload_markers definition");
-    registry
-        .register(Box::new(pending_markers.clone()))
-        .expect("register gitlawb_pending_upload_markers");
-    PENDING_UPLOAD_MARKERS
-        .set(pending_markers)
-        .expect("set PENDING_UPLOAD_MARKERS once");
-
     REGISTRY
         .set(registry)
         .expect("set REGISTRY once (init must be called exactly once)");
@@ -299,24 +284,6 @@ pub fn set_peers_connected(count: i64) {
     }
 }
 
-/// Update the outstanding pending-upload marker gauge (repos whose local copy
-/// is ahead of durable storage).
-pub fn set_pending_upload_markers(count: i64) {
-    if let Some(g) = PENDING_UPLOAD_MARKERS.get() {
-        g.set(count);
-    }
-}
-
-/// Adjust the pending-upload marker gauge by a delta (marker created = +1,
-/// marker removed = -1). The startup sweep seeds the absolute count via
-/// [`set_pending_upload_markers`]; runtime marker churn keeps it current
-/// through this.
-pub fn add_pending_upload_markers(delta: i64) {
-    if let Some(g) = PENDING_UPLOAD_MARKERS.get() {
-        g.add(delta);
-    }
-}
-
 /// Encode the registry as the standard Prometheus text exposition format.
 /// Returns an error if `init` was never called.
 pub fn encode() -> Result<String, prometheus::Error> {

crates/gitlawb-node/src/state.rs

@@ -61,7 +61,7 @@ pub struct AppState {
     pub graphql_schema: Arc<crate::graphql::GitlawbSchema>,
     /// Fly.io machine ID — used for fly-replay routing in multi-machine deployments
     pub machine_id: Option<String>,
-    /// Centralized repo storage: local disk cache + optional object-store backend
+    /// Centralized repo storage: local disk cache + optional Tigris backend
     pub repo_store: RepoStore,
     /// Per-DID rate limiter for creation endpoints (repos, issues, PRs)
     pub rate_limiter: RateLimiter,
@@ -79,6 +79,10 @@ pub struct AppState {
     /// brake a push flood from a DID farm (one throwaway DID per repo), so the
     /// push path throttles on the resolved client IP instead.
     pub push_rate_limiter: RateLimiter,
+    /// Per-client-IP rate limiter for `close_issue`'s pre-lock snapshot path.
+    /// Distinct from `push_rate_limiter` so a flood of close attempts cannot
+    /// drain the receive-pack budget for the same source IP.
+    pub close_issue_rate_limiter: RateLimiter,
     /// Per-client-IP ROUTE brake for `GET /ipfs/{cid}`: charged ONCE per request by the
     /// `rate_limit_by_ip` middleware (server.rs), never inside the handler. It bounds
     /// request RATE (the "requests per hour" contract of `GITLAWB_IPFS_RATE_LIMIT`) on
@@ -342,6 +346,7 @@ impl AppState {
         self.rate_limiter.cleanup().await;
         self.create_ip_rate_limiter.cleanup().await;
         self.push_rate_limiter.cleanup().await;
+        self.close_issue_rate_limiter.cleanup().await;
         self.ipfs_rate_limiter.cleanup().await;
         self.ipfs_work_rate_limiter.cleanup().await;
         self.sync_trigger_rate_limiter.cleanup().await;

crates/gitlawb-node/src/storage/archive.rs

@@ -1,446 +0,0 @@
-//! Repo-archive layer: stores a bare git repo as a single
-//! `repos/v1/{owner_slug}/{repo_name}.tar.zst` object on top of any
-//! [`BlobStore`] backend. Backend-agnostic replacement for the old
-//! single-backend (Tigris-only) client.
-
-use std::collections::HashMap;
-use std::path::{Path, PathBuf};
-use std::sync::{Arc, Mutex, OnceLock};
-
-use anyhow::{Context, Result};
-use bytes::Bytes;
-use tracing::{debug, info};
-
-use super::BlobStore;
-
-#[derive(Clone)]
-pub struct RepoArchive {
-    store: Arc<dyn BlobStore>,
-}
-
-impl RepoArchive {
-    pub fn new(store: Arc<dyn BlobStore>) -> Self {
-        Self { store }
-    }
-
-    /// Object key for a repo archive.
-    fn key(owner_slug: &str, repo_name: &str) -> String {
-        format!("repos/v1/{owner_slug}/{repo_name}.tar.zst")
-    }
-
-    /// Current archive etag, or `None` if the repo isn't in storage yet.
-    pub async fn head_etag(&self, owner_slug: &str, repo_name: &str) -> Result<Option<String>> {
-        let key = Self::key(owner_slug, repo_name);
-        Ok(self
-            .store
-            .head(&key)
-            .await?
-            .map(|m| m.etag.unwrap_or_else(|| format!("size:{}", m.size))))
-    }
-
-    /// Whether the repo archive exists in storage.
-    pub async fn exists(&self, owner_slug: &str, repo_name: &str) -> Result<bool> {
-        Ok(self
-            .store
-            .head(&Self::key(owner_slug, repo_name))
-            .await?
-            .is_some())
-    }
-
-    /// Compress the bare repo and upload it. Returns the new etag (for the
-    /// skip-redundant-download cache).
-    pub async fn upload(
-        &self,
-        owner_slug: &str,
-        repo_name: &str,
-        local_path: &Path,
-    ) -> Result<Option<String>> {
-        self.upload_with_intent(owner_slug, repo_name, local_path, |_| Ok(()))
-            .await
-    }
-
-    /// Like [`upload`](Self::upload), but calls `record_intent` with the
-    /// archive's content MD5 after compressing and BEFORE the PUT. On backends
-    /// whose etag is the body MD5 (S3-compatibles for single-part puts; the fs
-    /// backend by construction), that value equals the etag the PUT will
-    /// produce, letting crash recovery recognize "storage holds exactly what
-    /// this node was uploading" — its own completed upload, not divergence.
-    /// `record_intent` returning `Err` ABORTS the upload before the PUT: an
-    /// intent that could not be durably recorded must not be outrun by the
-    /// upload it describes, or a crash after the PUT reads as external
-    /// divergence. Recovery validates a candidate by fetching the remote
-    /// bytes and comparing their MD5, so the mechanism is backend-independent
-    /// (it does not assume the backend's etag is a content hash).
-    pub async fn upload_with_intent<F>(
-        &self,
-        owner_slug: &str,
-        repo_name: &str,
-        local_path: &Path,
-        record_intent: F,
-    ) -> Result<Option<String>>
-    where
-        F: FnOnce(&str) -> Result<()> + Send,
-    {
-        let key = Self::key(owner_slug, repo_name);
-        let archive_bytes = tokio::task::spawn_blocking({
-            let local_path = local_path.to_path_buf();
-            move || compress_repo(&local_path)
-        })
-        .await
-        .context("tar task panicked")?
-        .context("compressing repo")?;
-
-        record_intent(&content_md5_hex(&archive_bytes))
-            .context("recording upload intent before PUT")?;
-
-        let meta = self
-            .store
-            .put(&key, Bytes::from(archive_bytes))
-            .await
-            .context("uploading repo archive")?;
-        info!(key = %key, backend = self.store.backend_name(), "uploaded repo archive");
-        Ok(meta.etag.or_else(|| Some(format!("size:{}", meta.size))))
-    }
-
-    /// Download the repo archive and extract it to `local_path` (atomic swap).
-    pub async fn download(
-        &self,
-        owner_slug: &str,
-        repo_name: &str,
-        local_path: &Path,
-    ) -> Result<()> {
-        let key = Self::key(owner_slug, repo_name);
-        debug!(key = %key, "downloading repo archive");
-        let data = self
-            .store
-            .get(&key)
-            .await
-            .context("fetching repo archive")?
-            .ok_or_else(|| anyhow::anyhow!("repo archive missing: {key}"))?;
-
-        tokio::task::spawn_blocking({
-            let local_path = local_path.to_path_buf();
-            move || decompress_repo(&data, &local_path)
-        })
-        .await
-        .context("extract task panicked")?
-        .context("extracting repo")?;
-        info!(key = %key, path = %local_path.display(), "downloaded repo archive");
-        Ok(())
-    }
-
-    /// Fetch the raw archive object bytes (no extraction). Recovery-path
-    /// helper: lets the caller validate a heal candidate by hashing the
-    /// actual remote content instead of trusting backend etag semantics.
-    pub async fn fetch_raw(&self, owner_slug: &str, repo_name: &str) -> Result<Option<Bytes>> {
-        self.store.get(&Self::key(owner_slug, repo_name)).await
-    }
-
-    /// Delete a repo archive. No production caller yet (creation flows are
-    /// claim-first, so they never need to delete an archive); kept for the
-    /// repo-deletion path that will need it.
-    #[allow(dead_code)]
-    pub async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()> {
-        self.store.delete(&Self::key(owner_slug, repo_name)).await
-    }
-}
-
-/// Hex MD5 of a byte body — the etag an S3-compatible single-part PUT (and
-/// the fs backend, by construction) will assign to it.
-pub(crate) fn content_md5_hex(bytes: &[u8]) -> String {
-    use md5::{Digest, Md5};
-    format!("{:x}", Md5::digest(bytes))
-}
-
-/// Remove orphaned swap-phase work dirs (`.{repo}.tmp-extract.{uuid}` and
-/// `.{repo}.bak-{uuid}`) left under `repos_dir/{owner_slug}/` by a process
-/// killed mid-`decompress_repo`. Safe to run only at startup, before any
-/// extraction is in flight: a live swap owns exactly these names. Deleting a
-/// `.bak-` orphan loses no data — these dirs only exist on the download path,
-/// so storage still holds the archive and the next `acquire()` re-downloads.
-/// Returns the number of directories removed.
-pub fn sweep_orphaned_swap_dirs(repos_dir: &Path) -> usize {
-    let mut removed = 0;
-    let Ok(owners) = std::fs::read_dir(repos_dir) else {
-        return 0;
-    };
-    for owner in owners.flatten() {
-        let owner_path = owner.path();
-        if !owner_path.is_dir() {
-            continue;
-        }
-        let Ok(entries) = std::fs::read_dir(&owner_path) else {
-            continue;
-        };
-        for entry in entries.flatten() {
-            let name = entry.file_name();
-            let name = name.to_string_lossy();
-            let is_swap_litter =
-                name.starts_with('.') && (name.contains(".tmp-extract.") || name.contains(".bak-"));
-            if is_swap_litter && entry.path().is_dir() {
-                match std::fs::remove_dir_all(entry.path()) {
-                    Ok(()) => {
-                        info!(dir = %entry.path().display(), "removed orphaned swap dir");
-                        removed += 1;
-                    }
-                    Err(e) => {
-                        tracing::warn!(dir = %entry.path().display(), err = %e,
-                            "failed to remove orphaned swap dir");
-                    }
-                }
-            }
-        }
-    }
-    removed
-}
-
-/// Compress a bare repo directory into a tar.zst byte vector.
-fn compress_repo(repo_path: &Path) -> Result<Vec<u8>> {
-    let buf = Vec::new();
-    // Level 3 = fast + decent ratio. Compression dominates the post-push upload
-    // for larger repos; zstd's multithreaded mode splits the stream across
-    // worker threads for a near-linear speedup at the same ratio. Capped so a
-    // single push can't monopolize a small node's cores.
-    let mut encoder = zstd::stream::Encoder::new(buf, 3)?;
-    let workers = std::thread::available_parallelism()
-        .map(|n| n.get().min(4) as u32)
-        .unwrap_or(1);
-    if workers > 1 {
-        encoder
-            .multithread(workers)
-            .context("enabling multithreaded zstd")?;
-    }
-    let mut tar = tar::Builder::new(encoder);
-    tar.append_dir_all(".", repo_path)
-        .context("building tar archive")?;
-    let encoder = tar.into_inner().context("finishing tar")?;
-    let compressed = encoder.finish().context("finishing zstd")?;
-    Ok(compressed)
-}
-
-/// Per-repo-path lock serializing the publish (swap-into-place) step of
-/// `decompress_repo`. Concurrent extractions unpack into isolated temp dirs in
-/// parallel, but the final `remove_dir_all` + `rename` must not interleave for
-/// the same `local_path`, or they race to a nondeterministic overwrite/failure.
-fn publish_lock(local_path: &Path) -> Arc<Mutex<()>> {
-    // KNOWN LIMITATION: this map is never evicted — one (PathBuf, Arc<Mutex>)
-    // entry accrues per distinct repo path for the process lifetime. Bounded by
-    // the number of repos a node hosts, so it's negligible for normal use, but
-    // high-volume/churning deployments may want LRU or weak-ref eviction here.
-    static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
-    let locks = LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
-    let mut map = locks.lock().expect("publish lock map poisoned");
-    map.entry(local_path.to_path_buf())
-        .or_insert_with(|| Arc::new(Mutex::new(())))
-        .clone()
-}
-
-/// Decompress a tar.zst byte vector into a local directory.
-///
-/// Extraction is atomic with respect to `local_path`: the archive is unpacked
-/// into a sibling temp directory first, and only swapped into place once it
-/// fully succeeds. A corrupt or truncated archive therefore can never clobber a
-/// good existing copy at `local_path` — on failure we discard the temp dir and
-/// leave `local_path` exactly as it was.
-fn decompress_repo(data: &[u8], local_path: &Path) -> Result<()> {
-    let parent = local_path.parent().context("repo path has no parent")?;
-    std::fs::create_dir_all(parent).context("creating parent dir")?;
-
-    let file_name = local_path
-        .file_name()
-        .context("repo path has no file name")?
-        .to_string_lossy();
-    // Unique per-extraction temp dir: a fixed name would let two concurrent
-    // extractions of the same repo share one dir and clobber each other's
-    // in-progress unpack. A fresh UUID also means it can't collide with a
-    // leftover dir from a previously-interrupted run.
-    //
-    // Built through the shared sibling barrier rather than a bare `join`: the
-    // name embeds `file_name`, which carries user-provided data from the repo
-    // name, and this path is handed straight to `create_dir_all`, `unpack` and
-    // `remove_dir_all`. See `repo_store::validated_sibling_path`.
-    let tmp_dir = crate::git::repo_store::validated_sibling_path(
-        local_path,
-        &format!(".{file_name}.tmp-extract.{}", uuid::Uuid::new_v4()),
-    )?;
-
-    std::fs::create_dir_all(&tmp_dir).context("creating temp extract dir")?;
-
-    let unpack = (|| -> Result<()> {
-        let decoder = zstd::stream::Decoder::new(data)?;
-        let mut archive = tar::Archive::new(decoder);
-        archive.unpack(&tmp_dir).context("unpacking tar.zst")?;
-        Ok(())
-    })();
-    if let Err(e) = unpack {
-        let _ = std::fs::remove_dir_all(&tmp_dir);
-        return Err(e);
-    }
-
-    // Swap the freshly-extracted repo into place. rename within the same parent
-    // is effectively atomic, but most platforms refuse to rename onto a
-    // non-empty dir, so remove the old copy first. Serialize this per repo path:
-    // concurrent extractions unpack into isolated temp dirs, but their swaps
-    // must not interleave or they race to a nondeterministic overwrite/failure.
-    let lock = publish_lock(local_path);
-    let _publish = lock.lock().expect("publish lock poisoned");
-    let swap = (|| -> Result<()> {
-        // Move any existing repo aside to a backup first, rather than deleting
-        // it up front: if the rename of the new copy then fails, we restore the
-        // backup so `local_path` is never left without a valid repo. (Most
-        // platforms refuse to rename onto a non-empty dir, hence the move-aside.)
-        let backup = if local_path.exists() {
-            // Same sibling barrier as the temp-extract dir above: the name
-            // embeds the user-provided repo file name, and this path is renamed
-            // onto and later removed recursively.
-            let b = crate::git::repo_store::validated_sibling_path(
-                local_path,
-                &format!(".{file_name}.bak-{}", uuid::Uuid::new_v4()),
-            )?;
-            std::fs::rename(local_path, &b).context("moving existing repo to backup")?;
-            Some(b)
-        } else {
-            None
-        };
-        match std::fs::rename(&tmp_dir, local_path).context("swapping extracted repo into place") {
-            Ok(()) => {
-                if let Some(b) = backup {
-                    let _ = std::fs::remove_dir_all(&b);
-                }
-                Ok(())
-            }
-            Err(e) => {
-                // Restore the previous copy so the repo isn't left missing.
-                if let Some(b) = backup {
-                    let _ = std::fs::rename(&b, local_path);
-                }
-                Err(e)
-            }
-        }
-    })();
-    if swap.is_err() {
-        // Don't leak the extracted temp dir if the swap failed.
-        let _ = std::fs::remove_dir_all(&tmp_dir);
-    }
-    swap
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::fs;
-
-    fn seed_repo(dir: &std::path::Path) {
-        fs::create_dir_all(dir.join("refs/heads")).unwrap();
-        fs::write(dir.join("HEAD"), b"ref: refs/heads/main\n").unwrap();
-        fs::write(dir.join("refs/heads/main"), b"abc123\n").unwrap();
-        fs::write(dir.join("config"), b"[core]\n\tbare = true\n").unwrap();
-    }
-
-    #[test]
-    fn compress_decompress_round_trip_preserves_files() {
-        let src = tempfile::tempdir().unwrap();
-        seed_repo(src.path());
-
-        let bytes = compress_repo(src.path()).unwrap();
-        assert!(!bytes.is_empty());
-
-        let out_parent = tempfile::tempdir().unwrap();
-        let out = out_parent.path().join("restored.git");
-        decompress_repo(&bytes, &out).unwrap();
-
-        assert_eq!(
-            fs::read(out.join("HEAD")).unwrap(),
-            b"ref: refs/heads/main\n"
-        );
-        assert_eq!(fs::read(out.join("refs/heads/main")).unwrap(), b"abc123\n");
-        assert_eq!(
-            fs::read(out.join("config")).unwrap(),
-            b"[core]\n\tbare = true\n"
-        );
-    }
-
-    #[test]
-    fn decompress_swap_replaces_existing_dir_atomically() {
-        let src = tempfile::tempdir().unwrap();
-        fs::write(src.path().join("HEAD"), b"new\n").unwrap();
-        let bytes = compress_repo(src.path()).unwrap();
-
-        // Pre-existing copy with stale junk that the swap must fully replace.
-        let out_parent = tempfile::tempdir().unwrap();
-        let out = out_parent.path().join("repo.git");
-        fs::create_dir_all(&out).unwrap();
-        fs::write(out.join("STALE"), b"old\n").unwrap();
-
-        decompress_repo(&bytes, &out).unwrap();
-        assert_eq!(fs::read(out.join("HEAD")).unwrap(), b"new\n");
-        assert!(
-            !out.join("STALE").exists(),
-            "stale content must be gone after the swap"
-        );
-    }
-
-    #[test]
-    fn decompress_corrupt_archive_leaves_existing_copy_untouched() {
-        let out_parent = tempfile::tempdir().unwrap();
-        let out = out_parent.path().join("repo.git");
-        fs::create_dir_all(&out).unwrap();
-        fs::write(out.join("HEAD"), b"good\n").unwrap();
-
-        // Garbage is not a valid tar.zst: unpack fails before the swap, so the
-        // existing copy is preserved (atomicity claim).
-        assert!(decompress_repo(b"not a real archive", &out).is_err());
-        assert_eq!(fs::read(out.join("HEAD")).unwrap(), b"good\n");
-    }
-
-    #[test]
-    fn sweep_removes_only_orphaned_swap_dirs() {
-        let repos = tempfile::tempdir().unwrap();
-        let owner = repos.path().join("did_key_z6MkAlice");
-
-        // Litter from an interrupted swap...
-        let tmp_extract = owner.join(".repo.git.tmp-extract.1234-uuid");
-        let bak = owner.join(".repo.git.bak-5678-uuid");
-        // ...alongside things the sweep must not touch.
-        let live_repo = owner.join("repo.git");
-        let dotfile = owner.join(".keep");
-        fs::create_dir_all(&tmp_extract).unwrap();
-        fs::create_dir_all(&bak).unwrap();
-        fs::create_dir_all(&live_repo).unwrap();
-        fs::write(&dotfile, b"").unwrap();
-
-        assert_eq!(sweep_orphaned_swap_dirs(repos.path()), 2);
-        assert!(!tmp_extract.exists());
-        assert!(!bak.exists());
-        assert!(live_repo.exists());
-        assert!(dotfile.exists());
-
-        // Idempotent on a clean tree.
-        assert_eq!(sweep_orphaned_swap_dirs(repos.path()), 0);
-    }
-
-    #[tokio::test]
-    async fn upload_download_round_trip_over_fs_backend() {
-        let store_dir = tempfile::tempdir().unwrap();
-        let store: Arc<dyn BlobStore> =
-            Arc::new(crate::storage::fs::FsBlobStore::new(store_dir.path()).unwrap());
-        let archive = RepoArchive::new(store);
-
-        let src = tempfile::tempdir().unwrap();
-        seed_repo(src.path());
-
-        assert!(!archive.exists("owner", "repo").await.unwrap());
-        let etag = archive.upload("owner", "repo", src.path()).await.unwrap();
-        assert!(etag.is_some());
-        assert!(archive.exists("owner", "repo").await.unwrap());
-
-        let out_parent = tempfile::tempdir().unwrap();
-        let out = out_parent.path().join("repo.git");
-        archive.download("owner", "repo", &out).await.unwrap();
-        assert_eq!(
-            fs::read(out.join("HEAD")).unwrap(),
-            b"ref: refs/heads/main\n"
-        );
-        assert_eq!(fs::read(out.join("refs/heads/main")).unwrap(), b"abc123\n");
-    }
-}

crates/gitlawb-node/src/storage/fs.rs

@@ -1,374 +0,0 @@
-//! Local filesystem blob backend.
-//!
-//! Stores each object as a file under a configured root directory, using the
-//! object key as a relative path. For self-hosters without S3 and for tests of
-//! the storage abstraction. The etag is the body's MD5 (matching S3 semantics
-//! for single-part puts) persisted in a `.etag` sidecar at write time, so the
-//! skip-redundant-download optimization can rely on "etag unchanged ⇒ content
-//! unchanged" even on filesystems with coarse timestamps, and an uploader can
-//! predict the etag its PUT will produce (crash-recovery provenance). Objects
-//! predating the sidecar fall back to a size-mtime fingerprint.
-
-use std::path::{Path, PathBuf};
-
-use anyhow::{Context, Result};
-use async_trait::async_trait;
-use bytes::Bytes;
-
-use super::{validate_key, BlobStore, ObjectMeta};
-
-#[derive(Clone)]
-pub struct FsBlobStore {
-    root: PathBuf,
-}
-
-impl FsBlobStore {
-    pub fn new(root: impl AsRef<Path>) -> Result<Self> {
-        let root = root.as_ref().to_path_buf();
-        std::fs::create_dir_all(&root)
-            .with_context(|| format!("creating storage dir {}", root.display()))?;
-        Ok(Self { root })
-    }
-
-    fn path_for(&self, key: &str) -> Result<PathBuf> {
-        validate_key(key)?;
-        let path = self.root.join(key);
-        // Defence in depth: the resolved path must stay under root.
-        if !path.starts_with(&self.root) {
-            anyhow::bail!("blob key escaped storage root: {key}");
-        }
-        Ok(path)
-    }
-
-    /// Sidecar file persisting the object's etag: the body's MD5, computed at
-    /// write time. RepoStore treats etag equality as proof the local copy is
-    /// current, so different content must always yield a different token — a
-    /// `size-mtime` fingerprint cannot guarantee that on mounted filesystems
-    /// with coarse timestamp precision (two different same-size writes in one
-    /// tick collide); a content hash can. Identical content re-written yields
-    /// the same etag, which is semantically exact for a freshness token.
-    fn sidecar_of(path: &Path) -> PathBuf {
-        let mut os = path.as_os_str().to_owned();
-        os.push(".etag");
-        PathBuf::from(os)
-    }
-
-    /// Striped lock serializing the sidecar+blob publish step of `put`.
-    /// The two renames are only pairwise-consistent when same-key puts don't
-    /// interleave: unserialized, put A's content can land under put B's etag,
-    /// and "etag unchanged ⇒ content unchanged" breaks in the direction that
-    /// serves stale data as current. A fixed stripe array (same-key puts
-    /// always hash to the same stripe) bounds memory regardless of how many
-    /// distinct keys the process ever writes; cross-key collisions only cost
-    /// a moment of extra serialization on the cheap rename pair.
-    fn put_publish_lock(path: &Path) -> &'static std::sync::Mutex<()> {
-        use std::hash::{Hash, Hasher};
-        use std::sync::{Mutex, OnceLock};
-        const STRIPES: usize = 64;
-        static LOCKS: OnceLock<Vec<Mutex<()>>> = OnceLock::new();
-        let locks = LOCKS.get_or_init(|| (0..STRIPES).map(|_| Mutex::new(())).collect());
-        let mut hasher = std::collections::hash_map::DefaultHasher::new();
-        path.hash(&mut hasher);
-        &locks[(hasher.finish() as usize) % STRIPES]
-    }
-
-    /// Fallback fingerprint for objects written before the sidecar existed.
-    fn legacy_etag(md: &std::fs::Metadata) -> String {
-        let mtime = md
-            .modified()
-            .ok()
-            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
-            .map(|d| d.as_nanos())
-            .unwrap_or(0);
-        format!("{}-{}", md.len(), mtime)
-    }
-}
-
-#[async_trait]
-impl BlobStore for FsBlobStore {
-    fn backend_name(&self) -> &'static str {
-        "fs"
-    }
-
-    async fn get(&self, key: &str) -> Result<Option<Bytes>> {
-        let path = self.path_for(key)?;
-        match tokio::fs::read(&path).await {
-            Ok(data) => Ok(Some(Bytes::from(data))),
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
-            Err(e) => Err(e).context(format!("reading {}", path.display())),
-        }
-    }
-
-    async fn put(&self, key: &str, body: Bytes) -> Result<ObjectMeta> {
-        let path = self.path_for(key)?;
-        let parent = path
-            .parent()
-            .context("blob path has no parent")?
-            .to_path_buf();
-        // Unique temp name per write: a fixed suffix would let concurrent puts
-        // to the same key overwrite each other's temp file and corrupt the blob.
-        let tmp = path.with_extension(format!("{}.tmp-put", uuid::Uuid::new_v4()));
-        let path2 = path.clone();
-        // Atomic write: temp file in the same dir, then rename into place. On any
-        // failure, remove the temp file so a failed write can't leak it. The
-        // trailing stat + etag-sidecar write run inside the same blocking task —
-        // no synchronous fs call ever touches the async runtime.
-        tokio::task::spawn_blocking(move || -> Result<ObjectMeta> {
-            std::fs::create_dir_all(&parent).context("creating blob parent dir")?;
-            let etag = crate::storage::archive::content_md5_hex(&body);
-            let sidecar = Self::sidecar_of(&path2);
-            let sidecar_tmp = sidecar.with_extension(format!("{}.tmp-put", uuid::Uuid::new_v4()));
-            // Remember the published etag so a failed blob rename can restore
-            // it: leaving the NEW etag over the OLD content is no longer a
-            // harmless redundant download — under the pending-marker rules an
-            // unexplained remote etag is terminal divergence.
-            let prev_etag = std::fs::read_to_string(&sidecar).ok();
-            let write_and_swap = (|| -> Result<()> {
-                std::fs::write(&tmp, &body).context("writing temp blob")?;
-                // Publish the new etag BEFORE the blob (each via its own
-                // tmp+rename): a crash between the two renames then yields
-                // new-etag/old-content — a redundant download, or at worst a
-                // loud-and-safe pending-marker divergence — instead of
-                // old-etag/new-content, which etag-equality consumers would
-                // silently serve as current while stale. The publish pair is
-                // serialized per key so concurrent same-key puts can't
-                // interleave one put's etag with another's content.
-                let _guard = Self::put_publish_lock(&path2)
-                    .lock()
-                    .expect("put publish lock poisoned");
-                std::fs::write(&sidecar_tmp, &etag).context("writing etag sidecar")?;
-                std::fs::rename(&sidecar_tmp, &sidecar).context("publishing etag sidecar")?;
-                std::fs::rename(&tmp, &path2).context("renaming blob into place")?;
-                Ok(())
-            })();
-            if let Err(e) = write_and_swap {
-                let _ = std::fs::remove_file(&tmp);
-                let _ = std::fs::remove_file(&sidecar_tmp);
-                // Roll the sidecar back to describe the content actually on
-                // disk (the sidecar may have been published before the blob
-                // rename failed).
-                match prev_etag {
-                    Some(prev) => {
-                        let _ = std::fs::write(&sidecar, prev);
-                    }
-                    None => {
-                        let _ = std::fs::remove_file(&sidecar);
-                    }
-                }
-                return Err(e);
-            }
-            let md = std::fs::metadata(&path2).context("stat blob after write")?;
-            Ok(ObjectMeta {
-                size: md.len(),
-                etag: Some(etag),
-            })
-        })
-        .await
-        .context("fs put task panicked")?
-    }
-
-    async fn head(&self, key: &str) -> Result<Option<ObjectMeta>> {
-        let path = self.path_for(key)?;
-        // Probe existence by io error kind, not path.exists(): a permission/IO
-        // error must surface, not be silently reported as "not found".
-        let md = match tokio::fs::metadata(&path).await {
-            Ok(md) => md,
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
-            Err(e) => return Err(e).context(format!("stat {}", path.display())),
-        };
-        let etag = match tokio::fs::read_to_string(Self::sidecar_of(&path)).await {
-            Ok(tag) => tag.trim().to_string(),
-            // Object written before the sidecar existed — legacy fingerprint.
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Self::legacy_etag(&md),
-            Err(e) => {
-                return Err(e).context(format!("reading etag sidecar for {}", path.display()))
-            }
-        };
-        Ok(Some(ObjectMeta {
-            size: md.len(),
-            etag: Some(etag),
-        }))
-    }
-
-    async fn delete(&self, key: &str) -> Result<()> {
-        let path = self.path_for(key)?;
-        match tokio::fs::remove_file(&path).await {
-            Ok(()) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-            Err(e) => return Err(e).context(format!("deleting {}", path.display())),
-        }
-        match tokio::fs::remove_file(Self::sidecar_of(&path)).await {
-            Ok(()) => Ok(()),
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
-            Err(e) => Err(e).context(format!("deleting etag sidecar for {}", path.display())),
-        }
-    }
-}
-
-/// Test-only helper: enumerate stored keys. `list` was dropped from the
-/// `BlobStore` trait until a production consumer (GC/admin/migration) exists;
-/// the tests here still need it to assert on stored state.
-#[cfg(test)]
-impl FsBlobStore {
-    async fn list(&self, prefix: &str) -> Result<Vec<String>> {
-        let root = self.root.clone();
-        let prefix = prefix.to_string();
-        tokio::task::spawn_blocking(move || -> Result<Vec<String>> {
-            let mut keys = Vec::new();
-            let mut stack = vec![root.clone()];
-            while let Some(dir) = stack.pop() {
-                // Propagate read errors rather than skipping: a partial listing
-                // reported as success would mislead GC/admin/migration callers.
-                let rd = std::fs::read_dir(&dir)
-                    .with_context(|| format!("listing {}", dir.display()))?;
-                for entry in rd {
-                    // Propagate per-entry errors rather than dropping them via
-                    // flatten(): a partial listing must not look like success.
-                    let entry =
-                        entry.with_context(|| format!("reading entry under {}", dir.display()))?;
-                    let path = entry.path();
-                    if path.is_dir() {
-                        stack.push(path);
-                    } else if let Ok(rel) = path.strip_prefix(&root) {
-                        let key = rel.to_string_lossy().replace('\\', "/");
-                        // Etag sidecars are backend metadata, not objects.
-                        if key.starts_with(&prefix) && !key.ends_with(".etag") {
-                            keys.push(key);
-                        }
-                    }
-                }
-            }
-            Ok(keys)
-        })
-        .await
-        .context("fs list task panicked")?
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[tokio::test]
-    async fn put_get_head_delete_list_roundtrip() {
-        let dir = tempfile::tempdir().unwrap();
-        let store = FsBlobStore::new(dir.path()).unwrap();
-
-        // Absent key
-        assert!(store.get("repos/v1/a/x.tar.zst").await.unwrap().is_none());
-        assert!(store.head("repos/v1/a/x.tar.zst").await.unwrap().is_none());
-
-        // Put then get
-        let body = Bytes::from_static(b"hello blob");
-        let meta = store
-            .put("repos/v1/a/x.tar.zst", body.clone())
-            .await
-            .unwrap();
-        assert_eq!(meta.size, body.len() as u64);
-        assert!(meta.etag.is_some());
-        let got = store.get("repos/v1/a/x.tar.zst").await.unwrap().unwrap();
-        assert_eq!(got, body);
-
-        // Head returns matching etag (stable across reads)
-        let h = store.head("repos/v1/a/x.tar.zst").await.unwrap().unwrap();
-        assert_eq!(h.etag, meta.etag);
-
-        // List by prefix
-        store
-            .put("repos/v1/b/y.tar.zst", Bytes::from_static(b"y"))
-            .await
-            .unwrap();
-        let mut keys = store.list("repos/v1/").await.unwrap();
-        keys.sort();
-        assert_eq!(keys, vec!["repos/v1/a/x.tar.zst", "repos/v1/b/y.tar.zst"]);
-
-        // Delete is idempotent
-        store.delete("repos/v1/a/x.tar.zst").await.unwrap();
-        store.delete("repos/v1/a/x.tar.zst").await.unwrap();
-        assert!(store.get("repos/v1/a/x.tar.zst").await.unwrap().is_none());
-    }
-
-    #[tokio::test]
-    async fn etag_is_persisted_content_md5() {
-        let dir = tempfile::tempdir().unwrap();
-        let store = FsBlobStore::new(dir.path()).unwrap();
-        let key = "repos/v1/a/x.tar.zst";
-        let body = Bytes::from_static(b"same bytes");
-
-        // Content-addressed: different same-size content must yield a
-        // different etag even inside one coarse-filesystem timestamp tick
-        // (the size-mtime failure mode); identical content is the same token.
-        let m1 = store.put(key, body.clone()).await.unwrap();
-        let m2 = store
-            .put(key, Bytes::from_static(b"diff bytes"))
-            .await
-            .unwrap();
-        assert_ne!(m1.etag, m2.etag, "different content ⇒ different etag");
-        let m3 = store.put(key, body).await.unwrap();
-        assert_eq!(m1.etag, m3.etag, "identical content ⇒ identical etag");
-        assert_eq!(
-            m3.etag.as_deref(),
-            Some(crate::storage::archive::content_md5_hex(b"same bytes").as_str()),
-            "etag must be the body MD5 (predictable pre-PUT)"
-        );
-
-        // head() reports the persisted sidecar etag, not a recomputed
-        // fingerprint.
-        let h = store.head(key).await.unwrap().unwrap();
-        assert_eq!(h.etag, m3.etag);
-        let sidecar = dir.path().join(format!("{key}.etag"));
-        assert_eq!(
-            std::fs::read_to_string(&sidecar).unwrap(),
-            m3.etag.clone().unwrap(),
-            "etag must come from the persisted sidecar"
-        );
-
-        // delete() removes the sidecar with the object.
-        store.delete(key).await.unwrap();
-        assert!(store.head(key).await.unwrap().is_none());
-        assert!(store.list("repos/v1/").await.unwrap().is_empty());
-    }
-
-    #[tokio::test]
-    async fn rejects_key_traversal() {
-        let dir = tempfile::tempdir().unwrap();
-        let store = FsBlobStore::new(dir.path()).unwrap();
-        assert!(store.get("../escape").await.is_err());
-        assert!(store.put("a/../../etc/passwd", Bytes::new()).await.is_err());
-        // Backslashes are separators on Windows — must be rejected as keys.
-        assert!(store.get("a\\..\\escape").await.is_err());
-        assert!(store.put("repos\\v1\\x", Bytes::new()).await.is_err());
-    }
-
-    #[tokio::test]
-    async fn concurrent_puts_same_key_do_not_corrupt_or_leak_temps() {
-        let dir = tempfile::tempdir().unwrap();
-        let store = FsBlobStore::new(dir.path()).unwrap();
-        let key = "repos/v1/a/x.tar.zst";
-        let body = Bytes::from_static(b"the-one-true-blob");
-
-        // Many concurrent writers of the same key: with a fixed temp name they
-        // would clobber each other's temp file mid-write and corrupt the result.
-        let mut handles = Vec::new();
-        for _ in 0..16 {
-            let store = store.clone();
-            let body = body.clone();
-            handles.push(tokio::spawn(async move { store.put(key, body).await }));
-        }
-        for h in handles {
-            h.await.unwrap().unwrap();
-        }
-
-        // Final content is intact...
-        assert_eq!(store.get(key).await.unwrap().unwrap(), body);
-        // ...and no unique-suffixed temp files were left behind.
-        let leftovers: Vec<String> = store
-            .list("repos/v1/")
-            .await
-            .unwrap()
-            .into_iter()
-            .filter(|k| k.contains("tmp-put"))
-            .collect();
-        assert!(leftovers.is_empty(), "leaked temp files: {leftovers:?}");
-    }
-}

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

@@ -1,173 +0,0 @@
-//! IPFS (content-addressed) blob backend over a Kubo node's Mutable File System.
-//!
-//! Kubo's MFS (`/api/v0/files/*`) provides a path-addressed, mutable namespace
-//! backed by content-addressed IPFS objects — a natural fit for a key→blob store.
-//! Each object's etag is its IPFS CID (from `files/stat`), giving true
-//! content-addressing for the skip-redundant-download optimization.
-//!
-//! Requires a reachable Kubo HTTP API (`GITLAWB_IPFS_API`, e.g.
-//! `http://127.0.0.1:5001`). Objects written here are also retrievable by CID
-//! from the wider IPFS network once pinned/announced by the node.
-//!
-//! DEPLOYMENT RESTRICTION: MFS is the mutable namespace of ONE Kubo daemon —
-//! nothing here publishes a shared CID registry or IPNS mapping. Every
-//! gitlawb node in a network using this backend MUST point at the same Kubo
-//! instance; separate daemons produce disjoint namespaces where a push
-//! accepted on one node is invisible to the others. `storage::build` logs a
-//! startup warning to the same effect.
-
-use anyhow::{Context, Result};
-use async_trait::async_trait;
-use bytes::Bytes;
-
-use super::{validate_key, BlobStore, ObjectMeta};
-
-#[derive(Clone)]
-pub struct IpfsBlobStore {
-    api: String,
-    client: reqwest::Client,
-}
-
-impl IpfsBlobStore {
-    pub fn new(api: &str) -> Result<Self> {
-        // Bound requests so an unresponsive Kubo API can't hang push/write flows
-        // indefinitely. connect_timeout guards the dial; the generous total
-        // timeout still allows large repo-archive transfers.
-        let client = reqwest::Client::builder()
-            .connect_timeout(std::time::Duration::from_secs(5))
-            .timeout(std::time::Duration::from_secs(300))
-            .build()
-            .context("building IPFS HTTP client")?;
-        Ok(Self {
-            api: api.trim_end_matches('/').to_string(),
-            client,
-        })
-    }
-
-    /// MFS path for a key: `/gitlawb/<key>` (namespaced to avoid clobbering
-    /// other MFS users on a shared node).
-    fn mfs_path(key: &str) -> String {
-        format!("/gitlawb/{key}")
-    }
-}
-
-#[async_trait]
-impl BlobStore for IpfsBlobStore {
-    fn backend_name(&self) -> &'static str {
-        "ipfs"
-    }
-
-    async fn get(&self, key: &str) -> Result<Option<Bytes>> {
-        validate_key(key)?;
-        let url = format!("{}/api/v0/files/read", self.api);
-        let resp = self
-            .client
-            .post(&url)
-            .query(&[("arg", Self::mfs_path(key).as_str())])
-            .send()
-            .await
-            .context("IPFS files/read")?;
-        if resp.status().is_success() {
-            Ok(Some(resp.bytes().await.context("reading IPFS body")?))
-        } else {
-            // Kubo returns 500 with a JSON message when the path is absent.
-            let body = resp.text().await.unwrap_or_default();
-            if body.contains("does not exist") || body.contains("no link named") {
-                Ok(None)
-            } else {
-                anyhow::bail!("IPFS files/read {key}: {body}")
-            }
-        }
-    }
-
-    async fn put(&self, key: &str, body: Bytes) -> Result<ObjectMeta> {
-        validate_key(key)?;
-        let size = body.len() as u64;
-        let url = format!("{}/api/v0/files/write", self.api);
-        // Stream the body instead of copying it via to_vec — avoids doubling
-        // peak memory for large archives. Length is known, so set it explicitly.
-        let part = reqwest::multipart::Part::stream_with_length(reqwest::Body::from(body), size)
-            .file_name("blob");
-        let form = reqwest::multipart::Form::new().part("data", part);
-        let resp = self
-            .client
-            .post(&url)
-            .query(&[
-                ("arg", Self::mfs_path(key).as_str()),
-                ("create", "true"),
-                ("parents", "true"),
-                ("truncate", "true"),
-            ])
-            .multipart(form)
-            .send()
-            .await
-            .context("IPFS files/write")?;
-        if !resp.status().is_success() {
-            let status = resp.status();
-            let b = resp.text().await.unwrap_or_default();
-            anyhow::bail!("IPFS files/write {key} returned {status}: {b}");
-        }
-        // etag = CID from stat. The write already succeeded, so a failed stat
-        // must not fail the put — just return without an etag (callers treat a
-        // missing etag as "always re-check", never as a lost write).
-        let etag = match self.head(key).await {
-            Ok(m) => m.and_then(|m| m.etag),
-            Err(e) => {
-                tracing::warn!(key = %key, err = %e, "IPFS stat after write failed — returning no etag");
-                None
-            }
-        };
-        Ok(ObjectMeta { size, etag })
-    }
-
-    async fn head(&self, key: &str) -> Result<Option<ObjectMeta>> {
-        validate_key(key)?;
-        let url = format!("{}/api/v0/files/stat", self.api);
-        let resp = self
-            .client
-            .post(&url)
-            .query(&[("arg", Self::mfs_path(key).as_str())])
-            .send()
-            .await
-            .context("IPFS files/stat")?;
-        if resp.status().is_success() {
-            let v: serde_json::Value = resp.json().await.context("parsing files/stat")?;
-            Ok(Some(ObjectMeta {
-                size: v.get("Size").and_then(|s| s.as_u64()).unwrap_or(0),
-                etag: v
-                    .get("Hash")
-                    .and_then(|h| h.as_str())
-                    .map(|s| s.to_string()),
-            }))
-        } else {
-            let body = resp.text().await.unwrap_or_default();
-            if body.contains("does not exist") || body.contains("no link named") {
-                Ok(None)
-            } else {
-                anyhow::bail!("IPFS files/stat {key}: {body}")
-            }
-        }
-    }
-
-    async fn delete(&self, key: &str) -> Result<()> {
-        validate_key(key)?;
-        let url = format!("{}/api/v0/files/rm", self.api);
-        let resp = self
-            .client
-            .post(&url)
-            .query(&[("arg", Self::mfs_path(key).as_str()), ("force", "true")])
-            .send()
-            .await
-            .context("IPFS files/rm")?;
-        if resp.status().is_success() {
-            Ok(())
-        } else {
-            let body = resp.text().await.unwrap_or_default();
-            if body.contains("does not exist") || body.contains("no link named") {
-                Ok(())
-            } else {
-                anyhow::bail!("IPFS files/rm {key}: {body}")
-            }
-        }
-    }
-}

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

@@ -1,164 +0,0 @@
-//! Storage-agnostic blob layer.
-//!
-//! Repos are persisted to a pluggable object store behind the [`BlobStore`]
-//! trait. Backends:
-//!   - [`s3::S3BlobStore`]  — any S3-compatible service (Tigris, Cloudflare R2,
-//!     AWS S3, MinIO, Backblaze B2). Selected by default when a bucket is set.
-//!   - [`fs::FsBlobStore`]  — a local/mounted directory; for self-hosters & tests.
-//!   - [`ipfs::IpfsBlobStore`] — content-addressed storage over a Kubo (IPFS) node
-//!     using its Mutable File System (MFS) for a key→blob namespace.
-//!
-//! Higher layers ([`archive::RepoArchive`]) compose a bare repo into a single
-//! `repos/v1/{slug}/{repo}.tar.zst` object on top of whichever backend is active,
-//! so the repo-storage semantics are identical regardless of backend.
-
-use std::sync::Arc;
-
-use anyhow::{Context, Result};
-use async_trait::async_trait;
-use bytes::Bytes;
-use tracing::info;
-
-use crate::config::Config;
-
-pub mod archive;
-pub mod fs;
-pub mod ipfs;
-pub mod s3;
-
-/// Metadata about a stored object. `etag` is an opaque change-detection token
-/// (S3 ETag, IPFS CID, or the persisted content MD5 for the filesystem
-/// backend).
-#[derive(Debug, Clone)]
-pub struct ObjectMeta {
-    pub size: u64,
-    pub etag: Option<String>,
-}
-
-/// A backend-agnostic key→bytes object store.
-///
-/// Keys are forward-slash-delimited paths (e.g. `repos/v1/slug/repo.tar.zst`).
-/// Implementations must reject `..` traversal in keys.
-#[async_trait]
-pub trait BlobStore: Send + Sync {
-    /// Short backend name, for logs.
-    fn backend_name(&self) -> &'static str;
-
-    /// Fetch an object. Returns `None` if the key does not exist.
-    async fn get(&self, key: &str) -> Result<Option<Bytes>>;
-
-    /// Store an object, returning its metadata (including the new etag).
-    async fn put(&self, key: &str, body: Bytes) -> Result<ObjectMeta>;
-
-    /// Fetch object metadata without the body. Returns `None` if absent.
-    async fn head(&self, key: &str) -> Result<Option<ObjectMeta>>;
-
-    /// Delete an object. Succeeds (no-op) if the key does not exist.
-    async fn delete(&self, key: &str) -> Result<()>;
-}
-
-/// Build the configured blob store.
-///
-/// Returns `Ok(None)` only when no backend is configured at all (local-only
-/// passthrough mode). A misconfigured backend (missing required setting, or a
-/// client that fails to construct) returns `Err`: we fail closed rather than
-/// silently degrading to local-only, which would accept writes without the
-/// intended durable backend and risk cross-node persistence drift.
-///
-/// Note: this validates configuration and client construction, not live
-/// connectivity — e.g. the S3 client builds successfully against an unreachable
-/// or wrong bucket, and that surfaces as an error on the first real request.
-///
-/// Selection order:
-///   1. Explicit `GITLAWB_STORAGE_BACKEND` (`s3` | `fs` | `ipfs`).
-///   2. Auto: `s3` if a bucket is configured (incl. legacy `GITLAWB_TIGRIS_BUCKET`),
-///      else `fs` if `GITLAWB_STORAGE_FS_DIR` is set,
-///      else local-only.
-///
-/// The `ipfs` backend is never auto-selected: `GITLAWB_IPFS_API` predates this
-/// layer and configures the per-object encrypted pinning path, so treating its
-/// presence as "store repo archives in IPFS" would silently repurpose an
-/// existing pinning config on upgrade. Routing repo archives into IPFS MFS
-/// requires the explicit `GITLAWB_STORAGE_BACKEND=ipfs` opt-in.
-pub async fn build(config: &Config) -> Result<Option<Arc<dyn BlobStore>>> {
-    let bucket = if !config.s3_bucket.is_empty() {
-        config.s3_bucket.clone()
-    } else {
-        config.tigris_bucket.clone()
-    };
-
-    let backend = if !config.storage_backend.is_empty() {
-        config.storage_backend.to_ascii_lowercase()
-    } else if !bucket.is_empty() {
-        "s3".to_string()
-    } else if !config.storage_fs_dir.is_empty() {
-        "fs".to_string()
-    } else {
-        info!("object storage disabled (no backend configured) — local-only mode");
-        return Ok(None);
-    };
-
-    // A backend was selected (explicitly or by auto-detection); fail closed from
-    // here — a missing required setting or an init failure is a hard error.
-    match backend.as_str() {
-        "s3" => {
-            if bucket.is_empty() {
-                anyhow::bail!(
-                    "storage backend=s3 but no bucket configured (set GITLAWB_S3_BUCKET)"
-                );
-            }
-            let endpoint = (!config.s3_endpoint.is_empty()).then(|| config.s3_endpoint.clone());
-            let s = s3::S3BlobStore::new(&bucket, endpoint, config.s3_force_path_style)
-                .await
-                .context("initializing S3 storage")?;
-            info!(bucket = %bucket, backend = "s3", "object storage enabled");
-            Ok(Some(Arc::new(s) as Arc<dyn BlobStore>))
-        }
-        "fs" => {
-            if config.storage_fs_dir.is_empty() {
-                anyhow::bail!("storage backend=fs but GITLAWB_STORAGE_FS_DIR is empty");
-            }
-            let s = fs::FsBlobStore::new(&config.storage_fs_dir)
-                .context("initializing filesystem storage")?;
-            info!(dir = %config.storage_fs_dir, backend = "fs", "object storage enabled");
-            Ok(Some(Arc::new(s) as Arc<dyn BlobStore>))
-        }
-        "ipfs" => {
-            if config.ipfs_api.is_empty() {
-                anyhow::bail!("storage backend=ipfs but GITLAWB_IPFS_API is empty");
-            }
-            let s =
-                ipfs::IpfsBlobStore::new(&config.ipfs_api).context("initializing IPFS storage")?;
-            // MFS is a namespace of ONE Kubo daemon, not a shared network
-            // pointer: no CID registry or IPNS mapping is published, so two
-            // nodes pointed at separate daemons read and write disjoint
-            // storage and every cross-node archive assumption breaks.
-            tracing::warn!(api = %config.ipfs_api, backend = "ipfs",
-                "IPFS backend stores archives in the Kubo daemon's LOCAL MFS namespace — \
-                 ALL nodes sharing this network must point at the SAME Kubo instance; \
-                 separate daemons produce disjoint, silently-diverging storage");
-            info!(api = %config.ipfs_api, backend = "ipfs", "object storage enabled");
-            Ok(Some(Arc::new(s) as Arc<dyn BlobStore>))
-        }
-        other => {
-            anyhow::bail!("unknown GITLAWB_STORAGE_BACKEND: {other}");
-        }
-    }
-}
-
-/// Reject keys that could escape the namespace (`..`) or are absolute.
-pub(crate) fn validate_key(key: &str) -> Result<()> {
-    if key.is_empty() {
-        anyhow::bail!("blob key is empty");
-    }
-    if key.split('/').any(|seg| seg == ".." || seg == ".") {
-        anyhow::bail!("blob key contains traversal segment: {key}");
-    }
-    // Backslash is rejected outright: keys are forward-slash-delimited, and on
-    // Windows `PathBuf::join` treats `\` as a separator, which would let a key
-    // smuggle path components past the segment checks above.
-    if key.starts_with('/') || key.contains('\\') || key.contains('\0') {
-        anyhow::bail!("blob key is absolute, contains '\\', or contains null byte: {key}");
-    }
-    Ok(())
-}

crates/gitlawb-node/src/storage/s3.rs

@@ -1,167 +0,0 @@
-//! S3-compatible blob backend.
-//!
-//! Works with any S3 API implementation: Tigris, Cloudflare R2, AWS S3, MinIO,
-//! Backblaze B2. Credentials and (for Tigris on Fly) the endpoint are read from
-//! the standard AWS env vars (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`,
-//! `AWS_ENDPOINT_URL_S3`, `AWS_REGION`). `endpoint`/`force_path_style` override
-//! those for self-hosted services like MinIO.
-
-use anyhow::{Context, Result};
-use async_trait::async_trait;
-use aws_sdk_s3::Client as S3Client;
-use bytes::Bytes;
-use tracing::debug;
-
-use super::{validate_key, BlobStore, ObjectMeta};
-
-#[derive(Clone)]
-pub struct S3BlobStore {
-    s3: S3Client,
-    bucket: String,
-}
-
-impl S3BlobStore {
-    /// Build a client. `endpoint` overrides `AWS_ENDPOINT_URL_S3` (for R2/MinIO);
-    /// `force_path_style` is required by MinIO and some S3-compatibles.
-    pub async fn new(
-        bucket: &str,
-        endpoint: Option<String>,
-        force_path_style: bool,
-    ) -> Result<Self> {
-        let shared = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
-        let mut builder = aws_sdk_s3::config::Builder::from(&shared);
-        if let Some(ep) = endpoint {
-            builder = builder.endpoint_url(ep);
-        }
-        if force_path_style {
-            builder = builder.force_path_style(true);
-        }
-        // Bound requests so a hung endpoint can't block the calling task forever
-        // (under async_upload it would hold the advisory lock and stall every
-        // later push). attempt_timeout bounds a single try; operation_timeout
-        // bounds the whole call incl. retries — generous enough for large
-        // archive transfers.
-        let timeouts = aws_sdk_s3::config::timeout::TimeoutConfig::builder()
-            .operation_attempt_timeout(std::time::Duration::from_secs(60))
-            .operation_timeout(std::time::Duration::from_secs(300))
-            .build();
-        builder = builder.timeout_config(timeouts);
-        let s3 = S3Client::from_conf(builder.build());
-        Ok(Self {
-            s3,
-            bucket: bucket.to_string(),
-        })
-    }
-
-    /// Test-only constructor with an explicit S3 endpoint and static
-    /// credentials — no env-var reads, so parallel tests cannot race each
-    /// other's `AWS_*` environment the way the env-based `new` would. Lets a
-    /// test point the client at a non-routable endpoint to exercise
-    /// acquire-stall paths; deliberately no operation timeouts, so the stall
-    /// is bounded only by the caller's own deadline.
-    #[cfg(test)]
-    pub(crate) async fn for_testing_with_endpoint(bucket: &str, endpoint_url: &str) -> Self {
-        let creds = aws_sdk_s3::config::Credentials::new("test", "test", None, None, "test");
-        let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
-            .endpoint_url(endpoint_url)
-            .region(aws_config::Region::new("auto"))
-            .credentials_provider(creds)
-            .load()
-            .await;
-        Self {
-            s3: S3Client::new(&config),
-            bucket: bucket.to_string(),
-        }
-    }
-}
-
-#[async_trait]
-impl BlobStore for S3BlobStore {
-    fn backend_name(&self) -> &'static str {
-        "s3"
-    }
-
-    async fn get(&self, key: &str) -> Result<Option<Bytes>> {
-        validate_key(key)?;
-        match self
-            .s3
-            .get_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-        {
-            Ok(resp) => {
-                let data = resp
-                    .body
-                    .collect()
-                    .await
-                    .context("reading S3 response body")?
-                    .into_bytes();
-                Ok(Some(data))
-            }
-            Err(e) => {
-                if e.as_service_error().is_some_and(|e| e.is_no_such_key()) {
-                    Ok(None)
-                } else {
-                    Err(anyhow::anyhow!("S3 GET {key}: {e}"))
-                }
-            }
-        }
-    }
-
-    async fn put(&self, key: &str, body: Bytes) -> Result<ObjectMeta> {
-        validate_key(key)?;
-        let size = body.len() as u64;
-        let resp = self
-            .s3
-            .put_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .body(aws_sdk_s3::primitives::ByteStream::from(body))
-            .send()
-            .await
-            .context(format!("S3 PUT {key}"))?;
-        debug!(key = %key, size, "s3 put");
-        Ok(ObjectMeta {
-            size,
-            etag: resp.e_tag().map(|s| s.to_string()),
-        })
-    }
-
-    async fn head(&self, key: &str) -> Result<Option<ObjectMeta>> {
-        validate_key(key)?;
-        match self
-            .s3
-            .head_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-        {
-            Ok(resp) => Ok(Some(ObjectMeta {
-                size: resp.content_length().unwrap_or(0).max(0) as u64,
-                etag: resp.e_tag().map(|s| s.to_string()),
-            })),
-            Err(e) => {
-                if e.as_service_error().is_some_and(|e| e.is_not_found()) {
-                    Ok(None)
-                } else {
-                    Err(anyhow::anyhow!("S3 HEAD {key}: {e}"))
-                }
-            }
-        }
-    }
-
-    async fn delete(&self, key: &str) -> Result<()> {
-        validate_key(key)?;
-        self.s3
-            .delete_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-            .context(format!("S3 DELETE {key}"))?;
-        Ok(())
-    }
-}

crates/gitlawb-node/src/test_support.rs

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

crates/gitlawb-node/tests/inv22_gates.rs

@@ -210,21 +210,14 @@ fn inv22_concurrency_gates_present_and_not_bypassed() {
     }
 }
 
-/// F4 (repo_store advisory-unlock cancellation safety): the advisory unlock must
-/// be awaited while its owner still holds the pooled connection, and the owner
-/// must not mark itself `released` until that await resolves. Either shape,
+/// F4 (repo_store advisory-unlock cancellation safety): `RepoWriteGuard::release`
+/// must await `pg_advisory_unlock` while `self` still owns the pooled connection,
+/// and must not mark itself `released` until that await resolves. Either shape,
 /// reintroduced, re-opens the mid-unlock cancellation leak: taking the connection
 /// early leaves `Drop` with `conn == None`, and setting `released = true` early
 /// leaves the `Drop` backstop inert — both strand the session lock on cancellation.
 ///
-/// The guarded code MOVED with the storage-abstraction layer: the unlock used to
-/// live in `RepoWriteGuard::release`, and now lives in `LockedConn::unlock`, the
-/// type that owns a lock's whole pinned-connection lifetime (the guard delegates
-/// to it, and the write-back path drives it from a detached task, which is the
-/// cancellation shape this gate exists for). The invariant is unchanged, so this
-/// gate follows the code rather than being deleted with it.
-///
-/// Scoped to the `unlock` fn body: the `Drop` impl legitimately takes the
+/// Scoped to the `release` fn body: the `Drop` impl legitimately takes the
 /// connection and unlocks, so a whole-file scan would match it and read as a false
 /// pass. Reverting either ordering turns this red (proven load-bearing).
 #[test]
@@ -232,12 +225,13 @@ fn f4_release_keeps_conn_owned_until_unlock_resolves() {
     let repo_store = src("git/repo_store.rs");
 
     let rel_start = repo_store
-        .find("async fn unlock(mut self")
-        .expect("F4 gate: repo_store.rs no longer defines LockedConn::unlock");
+        .find("pub async fn release(self")
+        .or_else(|| repo_store.find("pub async fn release(mut self"))
+        .expect("F4 gate: repo_store.rs no longer defines RepoWriteGuard::release");
     let rel_end = repo_store[rel_start..]
-        .find("impl Drop for LockedConn")
+        .find("impl Drop for RepoWriteGuard")
         .map(|off| rel_start + off)
-        .expect("F4 gate: unlock fn / Drop impl markers moved — update this guard");
+        .expect("F4 gate: release fn / Drop impl markers moved — update this guard");
     let release_body = &repo_store[rel_start..rel_end];
 
     let unlock = release_body
@@ -248,7 +242,7 @@ fn f4_release_keeps_conn_owned_until_unlock_resolves() {
     // (a) the connection must still be owned by `self` at the unlock await.
     assert!(
         !before_unlock.contains("self.conn.take()"),
-        "F4 regression: LockedConn::unlock takes self.conn BEFORE awaiting \
+        "F4 regression: RepoWriteGuard::release takes self.conn BEFORE awaiting \
          pg_advisory_unlock. A cancellation during the unlock await then strands the \
          session advisory lock (Drop sees conn == None and skips its backstop). \
          Unlock through the still-owned connection instead."
@@ -257,7 +251,7 @@ fn f4_release_keeps_conn_owned_until_unlock_resolves() {
     // reintroduction shape a single-reorder check on (a) alone is blind to.
     assert!(
         !before_unlock.contains("released = true"),
-        "F4 regression: LockedConn::unlock sets `released = true` BEFORE awaiting \
+        "F4 regression: RepoWriteGuard::release sets `released = true` BEFORE awaiting \
          pg_advisory_unlock. A cancellation during the await then leaves the Drop \
          backstop inert (it early-returns on released). Set released only AFTER the \
          unlock await resolves."