fix(storage): path-traversal barrier for sibling paths; restore the F4 unlock shape

Two CI failures from the storage-abstraction merge, both in this layer.

CodeQL (rust/path-injection). Every path derived from a repo path — the
pending-upload marker, its rename-temp, and the swap phase's .bak- and
.tmp-extract. work dirs — is rebuilt with join/with_file_name, and a rebuilt
path is a new question no earlier validation answers. Add
validated_sibling_path (single-segment name, containment under the repo
parent, explicit Component::Normal walk) and validated_repo_path_in (the same
walk for a repo path arriving as a parameter, applied where the filesystem
work happens rather than trusted across a call boundary), then route every
marker/backup/temp construction and the sweep's entry-derived repo paths
through them before any fs call. pending_upload_marker is fallible as a
result; the cleanup paths that cannot propagate treat a rejected path as
"no marker", the same conservative answer they already give for an absent one.

INV-22 F4 gate. The merge moved the advisory unlock from
RepoWriteGuard::release into LockedConn::unlock and, in doing so, took the
connection out of self BEFORE awaiting pg_advisory_unlock — the exact shape
#174 F4 forbids, since a cancellation mid-await then leaves Drop with
conn == None and strands the session lock. Restore the reviewed shape: unlock
through the still-owned connection, dispose of it only when the unlock errors
(#174 F3b), and set the new released flag only once the await resolves so the
Drop backstop stays live until then. tests/inv22_gates.rs follows the code to
its new home rather than being deleted with it; the invariant is unchanged.

.env.example

@@ -17,6 +17,31 @@ 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

Cargo.lock

@@ -3441,6 +3441,7 @@ dependencies = [
  "async-compression",
  "async-graphql",
  "async-graphql-axum",
+ "async-trait",
  "aws-config",
  "aws-sdk-s3",
  "axum",
@@ -3465,6 +3466,7 @@ 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 (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_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_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,7 +412,13 @@ 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_TIGRIS_BUCKET` | Optional S3/Tigris shared repo storage bucket. |
+| `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_PINATA_JWT` | Optional Pinata/IPFS warm-storage pinning. |
 | `GITLAWB_IRYS_URL` | Optional Irys/Arweave permanent anchoring. |
 

crates/gitlawb-node/Cargo.toml

@@ -33,9 +33,15 @@ 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"] }
@@ -57,7 +63,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 = "0.13"
+zstd = { version = "0.13", features = ["zstdmt"] }
 # 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::tigris_stall`: hold one specific await open so the clamp around it is
+// `RepoStore::storage_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,10 +3757,12 @@ mod tests {
         // parallel test run); the silent local endpoint stalls the HEAD
         // deterministically.
         let endpoint = crate::test_support::silent_http_endpoint().await;
-        let tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+        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 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;
@@ -3830,10 +3832,12 @@ 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 tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+        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 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);
@@ -7955,10 +7959,12 @@ 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 tigris =
-            crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint)
-                .await;
-        state.repo_store = crate::git::repo_store::RepoStore::new(repos_dir, Some(tigris), pool);
+        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 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.ipfs_request_budget_secs = 1;
         cfg.git_acquire_timeout_secs = 2;

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

@@ -73,10 +73,32 @@ pub async fn create_issue(
 
     let create_result = git_issues::create_issue(&disk_path, &issue_id, &json_str);
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(create_result.is_ok()).await;
+    // Always release the advisory lock — even on error; upload to storage only on success.
+    let release_result = guard.release(create_result.is_ok()).await;
 
     create_result.map_err(|e| AppError::Git(e.to_string()))?;
+    // A durable-upload failure is recoverable ONLY while the pending-upload
+    // marker protects the committed mutation (the next successful upload
+    // re-syncs storage). Verify the marker actually exists before choosing to
+    // succeed: if the marker write itself also failed, nothing protects the
+    // mutation from a stale-archive rollback, and the request must fail.
+    // (Succeeding here avoids non-idempotent retries: a retried create mints
+    // a second issue UUID and both eventually publish.)
+    if let Err(e) = release_result {
+        if state
+            .repo_store
+            .pending_marker_exists(&record.owner_did, &record.name)
+        {
+            tracing::error!(repo = %record.name, issue = %issue_id, err = %e,
+                "issue committed locally but durable upload failed — storage re-syncs on next upload");
+        } else {
+            tracing::error!(repo = %record.name, issue = %issue_id, err = %e,
+                "issue committed locally with NO durable protection — failing the request");
+            return Err(AppError::Git(format!(
+                "issue stored locally but durability could not be guaranteed: {e}"
+            )));
+        }
+    }
 
     // Bump trust score for the issue author — increment current score by 0.05
     // (avoids the push_count=0 stuck-at-0.05 bug for agents who only file issues)
@@ -249,11 +271,11 @@ pub async fn close_issue(
             .ok()
             .and_then(|i| i.author),
         Ok(None) => {
-            guard.release(false).await;
+            let _ = guard.release(false).await;
             return Err(AppError::NotFound(format!("issue {issue_id} not found")));
         }
         Err(e) => {
-            guard.release(false).await;
+            let _ = guard.release(false).await;
             return Err(AppError::Git(e.to_string()));
         }
     };
@@ -262,7 +284,7 @@ pub async fn close_issue(
         .as_deref()
         .is_some_and(|a| crate::api::did_matches(&auth.0, a));
     if !is_owner && !is_author {
-        guard.release(false).await;
+        let _ = guard.release(false).await;
         return Err(AppError::Forbidden(
             "only the repo owner or the issue author can close this issue".into(),
         ));
@@ -270,12 +292,29 @@ pub async fn close_issue(
 
     let close_result = git_issues::close_issue(&disk_path, &issue_id);
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(close_result.is_ok()).await;
+    // Always release the advisory lock — even on error; upload to storage only on success.
+    let release_result = guard.release(close_result.is_ok()).await;
 
     let updated = close_result
         .map_err(|e| AppError::Git(e.to_string()))?
         .ok_or_else(|| AppError::RepoNotFound(format!("issue {issue_id} not found")))?;
+    // Recoverable only while the marker protects the committed mutation
+    // (see create_issue).
+    if let Err(e) = release_result {
+        if state
+            .repo_store
+            .pending_marker_exists(&record.owner_did, &record.name)
+        {
+            tracing::error!(repo = %repo, issue = %issue_id, err = %e,
+                "issue close committed locally but durable upload failed — storage re-syncs on next upload");
+        } else {
+            tracing::error!(repo = %repo, issue = %issue_id, err = %e,
+                "issue close committed locally with NO durable protection — failing the request");
+            return Err(AppError::Git(format!(
+                "issue close stored locally but durability could not be guaranteed: {e}"
+            )));
+        }
+    }
 
     let issue: serde_json::Value = serde_json::from_str(&updated)
         .map_err(|e| AppError::BadRequest(format!("invalid issue data: {e}")))?;
@@ -382,7 +421,7 @@ mod lock_pool_shed_tests {
 
         // MUST-NOT: with the pool free again the call is not shed as capacity (it
         // fails later on the nonexistent on-disk repo, which is a git 500).
-        held.release(false).await;
+        held.release(false).await.ok();
         let admitted = create_issue(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),
@@ -431,7 +470,7 @@ mod lock_pool_shed_tests {
         let err = shed.expect_err("an exhausted lock pool must fail the call");
         assert_sheds_503_with_retry_after(err, "close_issue");
 
-        held.release(false).await;
+        held.release(false).await.ok();
         let admitted = close_issue(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),
@@ -449,198 +488,3 @@ mod lock_pool_shed_tests {
         );
     }
 }
-
-/// #426: end-to-end at the handler seam. A corrupt object store behind a
-/// resolving issue ref used to look identical to an absent issue: get returned
-/// a fake 404, close returned a fake 404 after releasing the write guard, and
-/// list silently under-reported. All three must now surface a git error.
-#[cfg(test)]
-mod corrupt_store_tests {
-    use super::*;
-    use axum::response::IntoResponse;
-    use sqlx::PgPool;
-    use tempfile::TempDir;
-
-    fn seed_repo(owner_did: &str, name: &str) -> crate::db::RepoRecord {
-        let now = Utc::now();
-        crate::db::RepoRecord {
-            id: Uuid::new_v4().to_string(),
-            name: name.to_string(),
-            owner_did: owner_did.to_string(),
-            description: None,
-            is_public: true,
-            default_branch: "main".to_string(),
-            created_at: now,
-            updated_at: now,
-            disk_path: format!("/tmp/{name}"),
-            forked_from: None,
-            machine_id: None,
-        }
-    }
-
-    /// A state whose repo store roots in `repos_dir`, so the test can lay down
-    /// a real repo and then corrupt it without touching /tmp.
-    async fn state_with_repos_dir(pool: &PgPool, repos_dir: &std::path::Path) -> AppState {
-        let mut state = crate::test_support::test_state(pool.clone()).await;
-        state.repo_store =
-            crate::git::repo_store::RepoStore::for_testing(repos_dir.to_path_buf(), pool.clone());
-        state
-    }
-
-    /// Bare repo at the store's expected path, holding one issue ref.
-    fn bare_repo_with_issue(
-        repos_dir: &std::path::Path,
-        owner: &str,
-        name: &str,
-    ) -> (std::path::PathBuf, String) {
-        let repo_path = repos_dir
-            .join(owner.replace([':', '/'], "_"))
-            .join(format!("{name}.git"));
-        std::fs::create_dir_all(&repo_path).unwrap();
-        let out = std::process::Command::new("git")
-            .args(["init", "--bare"])
-            .arg(&repo_path)
-            .output()
-            .unwrap();
-        assert!(out.status.success());
-        let issue_id = "aaaabbbb-0000-0000-0000-000000000000".to_string();
-        git_issues::create_issue(
-            &repo_path,
-            &issue_id,
-            &serde_json::json!({
-                "id": issue_id,
-                "title": "t",
-                "status": "open",
-            })
-            .to_string(),
-        )
-        .unwrap();
-        (repo_path, issue_id)
-    }
-
-    /// Delete the blob object behind the issue ref and put garbage in its
-    /// place: `cat-file` fails while `rev-parse` still resolves the ref.
-    fn corrupt_issue_object(repo_path: &std::path::Path, issue_id: &str) {
-        let out = std::process::Command::new("git")
-            .args(["rev-parse", &format!("refs/gitlawb/issues/{issue_id}")])
-            .current_dir(repo_path)
-            .output()
-            .unwrap();
-        let oid = String::from_utf8(out.stdout).unwrap().trim().to_string();
-        let obj = repo_path.join("objects").join(&oid[..2]).join(&oid[2..]);
-        std::fs::remove_file(&obj).unwrap();
-        std::fs::write(&obj, b"garbage").unwrap();
-    }
-
-    /// The regression: closing over a corrupt store used to return a fake 404
-    /// (or panic deeper in). It must reach the caller as a git 500, and the
-    /// handler must return rather than unwind.
-    #[sqlx::test]
-    async fn close_issue_on_corrupt_store_returns_git_error_not_panic(pool: PgPool) {
-        let owner = "did:key:zISSUECORRUPTCCCCCCCCCCCCCCCCCCCCCCCCC";
-        let dir = TempDir::new().unwrap();
-        let state = state_with_repos_dir(&pool, dir.path()).await;
-        state
-            .db
-            .create_repo(&seed_repo(owner, "corrupt-close"))
-            .await
-            .expect("seed repo");
-        let (repo_path, issue_id) = bare_repo_with_issue(dir.path(), owner, "corrupt-close");
-        corrupt_issue_object(&repo_path, &issue_id);
-
-        let err = close_issue(
-            State(state),
-            Extension(AuthenticatedDid(owner.to_string())),
-            Path((owner.to_string(), "corrupt-close".to_string(), issue_id)),
-        )
-        .await
-        .expect_err("a corrupt store must error, not panic or 404");
-        assert_eq!(
-            err.into_response().status(),
-            StatusCode::INTERNAL_SERVER_ERROR,
-            "a read failure is a git 500, not a fake not-found"
-        );
-    }
-
-    /// get on a corrupt store: 500, not the 404 the fold used to produce.
-    #[sqlx::test]
-    async fn get_issue_on_corrupt_store_returns_500_not_404(pool: PgPool) {
-        let owner = "did:key:zISSUECORRUPTGETDDDDDDDDDDDDDDDDDDDDDDD";
-        let dir = TempDir::new().unwrap();
-        let state = state_with_repos_dir(&pool, dir.path()).await;
-        state
-            .db
-            .create_repo(&seed_repo(owner, "corrupt-get"))
-            .await
-            .expect("seed repo");
-        let (repo_path, issue_id) = bare_repo_with_issue(dir.path(), owner, "corrupt-get");
-        corrupt_issue_object(&repo_path, &issue_id);
-
-        let err = get_issue(
-            State(state),
-            Path((owner.to_string(), "corrupt-get".to_string(), issue_id)),
-            None,
-        )
-        .await
-        .expect_err("a corrupt store must error, not 404");
-        assert_eq!(
-            err.into_response().status(),
-            StatusCode::INTERNAL_SERVER_ERROR,
-        );
-    }
-
-    /// list on a corrupt store must fail outright; a 200 that silently drops
-    /// the unreadable issue is the other half of the fold.
-    #[sqlx::test]
-    async fn list_issues_on_corrupt_store_returns_git_error(pool: PgPool) {
-        let owner = "did:key:zISSUECORRUPTLISTEEEEEEEEEEEEEEEEEEEEEE";
-        let dir = TempDir::new().unwrap();
-        let state = state_with_repos_dir(&pool, dir.path()).await;
-        state
-            .db
-            .create_repo(&seed_repo(owner, "corrupt-list"))
-            .await
-            .expect("seed repo");
-        let (repo_path, issue_id) = bare_repo_with_issue(dir.path(), owner, "corrupt-list");
-        corrupt_issue_object(&repo_path, &issue_id);
-
-        let err = list_issues(
-            State(state),
-            Path((owner.to_string(), "corrupt-list".to_string())),
-            None,
-        )
-        .await
-        .expect_err("a corrupt store must error, not under-report");
-        assert_eq!(
-            err.into_response().status(),
-            StatusCode::INTERNAL_SERVER_ERROR,
-        );
-    }
-
-    /// Must-not direction: a genuinely absent issue still gets the 404.
-    #[sqlx::test]
-    async fn get_issue_on_absent_ref_still_returns_404(pool: PgPool) {
-        let owner = "did:key:zISSUEABSENTFFFFFFFFFFFFFFFFFFFFFFFFFFF";
-        let dir = TempDir::new().unwrap();
-        let state = state_with_repos_dir(&pool, dir.path()).await;
-        state
-            .db
-            .create_repo(&seed_repo(owner, "absent-get"))
-            .await
-            .expect("seed repo");
-        bare_repo_with_issue(dir.path(), owner, "absent-get");
-
-        let err = get_issue(
-            State(state),
-            Path((
-                owner.to_string(),
-                "absent-get".to_string(),
-                "ffffffff-0000-0000-0000-000000000000".to_string(),
-            )),
-            None,
-        )
-        .await
-        .expect_err("a missing issue is still not-found");
-        assert_eq!(err.into_response().status(), StatusCode::NOT_FOUND);
-    }
-}

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

@@ -227,10 +227,31 @@ pub async fn merge_pr(
         &pr.title,
     );
 
-    // Always release the advisory lock — even on error; upload to Tigris only on success.
-    guard.release(merge_result.is_ok()).await;
+    // Always release the advisory lock — even on error; upload to storage only on success.
+    let release_result = guard.release(merge_result.is_ok()).await;
 
     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;
@@ -526,7 +547,7 @@ mod lock_pool_shed_tests {
 
         // 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;
+        held.release(false).await.ok();
         let admitted = merge_pr(
             State(state.clone()),
             Extension(AuthenticatedDid(owner.to_string())),

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

@@ -253,17 +253,17 @@ pub async fn create_repo(
     // Request is admissible — spend the proof now, immediately before the write.
     let verified_proof = proof.consume(&state.db).await?;
 
-    let disk_path = state
-        .repo_store
-        .init(&owner_did, &req.name)
-        .await
-        .map_err(|e| {
-            // `{:#}` walks the anyhow chain to the leaf cause; the other git
-            // handlers log their failures, this one didn't.
-            tracing::error!(owner = %owner_did, repo = %req.name, err = %format!("{e:#}"), "repo create failed");
-            AppError::Git(e.to_string())
-        })?;
-
+    // Claim-first ordering: insert the DB row before creating anything durable.
+    // Within one node's database the row is the claim on (owner, name) — a
+    // concurrent same-name create loses at the insert with nothing on disk or
+    // in storage yet. Across nodes (each fly app has its own Postgres) the
+    // insert arbitrates nothing; the cross-node safety property is that
+    // failure compensation below only ever touches state THIS attempt created
+    // (its own row by id, its own local dir) and never deletes a storage key.
+    // init() publishes under the per-repo advisory lock, so a push arriving
+    // through the just-visible row serializes behind publication and can
+    // never be destroyed by this compensation.
+    let disk_path = store::repo_disk_path(&state.config.repos_dir, &owner_did, &req.name);
     let now = Utc::now();
     let record = crate::db::RepoRecord {
         id: Uuid::new_v4().to_string(),
@@ -278,9 +278,23 @@ pub async fn create_repo(
         forked_from: None,
         machine_id: state.machine_id.clone(),
     };
-
     state.db.create_repo(&record).await?;
 
+    // Create the bare repo locally and publish the initial archive (under the
+    // advisory lock). On failure, compensate by removing our own just-inserted
+    // row (keyed by our id) so a retry starts clean; create_published removes
+    // its local dir itself.
+    if let Err(e) = state.repo_store.init(&owner_did, &req.name).await {
+        // `{:#}` walks the anyhow chain to the leaf cause; the other git
+        // handlers log their failures, this one didn't.
+        tracing::error!(owner = %owner_did, repo = %req.name, err = %format!("{e:#}"), "repo create failed");
+        if let Err(db_err) = state.db.delete_repo_by_id(&record.id).await {
+            tracing::warn!(repo = %req.name, err = %db_err,
+                "failed to remove repo row after init failure");
+        }
+        return Err(AppError::Git(e.to_string()));
+    }
+
     // Persist the proof so it can travel with the repo and a mirroring peer can
     // re-verify it (enforce-mode origins only; off/shadow yield no proof here).
     if let Some(p) = verified_proof {
@@ -631,10 +645,10 @@ pub async fn git_info_refs(
     }
 
     // Push flood brake on the advertisement phase. A push always hits this
-    // GET first, and for receive-pack it forces a fresh Tigris download below;
+    // GET first, and for receive-pack it forces a fresh storage download below;
     // throttling only the receive-pack POST would leave the expensive
     // fresh-acquire reachable unauthenticated and unlimited. Applied before the
-    // acquire so a rejected request does no Tigris work. Same per-IP limiter and
+    // acquire so a rejected request does no storage work. Same per-IP limiter and
     // trusted-proxy policy as the POST middleware (shared buckets).
     if service == "git-receive-pack" {
         if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
@@ -692,7 +706,7 @@ pub async fn git_info_refs(
         git_permit(&state.git_read_semaphore)?
     };
 
-    // For receive-pack (push), download the latest from Tigris so the client
+    // For receive-pack (push), download the latest from storage so the client
     // sees the same refs that acquire_write() will operate on.
     //
     // Bound the acquire under `git_acquire_timeout_secs`: the concurrency permit is
@@ -2308,7 +2322,7 @@ pub async fn git_receive_pack(
     }
 
     // Always release the advisory lock — even on error — to prevent stale locks
-    // from blocking subsequent pushes. Only upload to Tigris when the push
+    // from blocking subsequent pushes. Only upload to storage when the push
     // succeeded; uploading a half-applied repo would propagate corruption.
     // Reclaim the write lock from the shared cell (#173 F2). This is only reachable
     // once `receive_pack` has returned, so the admission guard's copy can only ever
@@ -2319,12 +2333,74 @@ pub async fn git_receive_pack(
         .expect("repo write-lock mutex poisoned")
         .take()
         .expect("the write lock is only taken here, and only once");
-    reclaimed.release(push_succeeded).await;
+    // `Some` = the guard still needs a synchronous (strict) release; taken by
+    // the write-back path only once its intent marker is durably on disk.
+    let mut strict_guard = Some(reclaimed);
+    if push_succeeded && state.config.async_upload {
+        let guard = strict_guard.take().expect("guard present before release");
+        // Write-back: ack the client now; the durable upload to object storage
+        // and the advisory-lock release run in the background. The lock is held
+        // until the upload finishes, so a concurrent writer on another machine
+        // can't observe a stale archive. If this detached task is cancelled by
+        // runtime shutdown mid-upload, the guard's lock connection is closed
+        // rather than repooled, so Postgres frees the advisory lock (see
+        // `LockedConn`). Durability tradeoff: if the upload fails (or the node
+        // stops first), storage stays stale until this repo's next successful
+        // upload. The persisted pending-upload marker keeps the local copy
+        // authoritative on THIS node in that window — no access rolls it back
+        // to the stale archive — but other nodes still serve the stale archive
+        // until the re-upload lands. Hence async_upload is opt-in.
+        //
+        // The intent marker must be on disk BEFORE the ack: the spawned task
+        // may never be polled if the process stops right after the response,
+        // and without the marker a restart would treat the stale storage
+        // archive as newer and roll the acked push back. If the marker itself
+        // cannot be persisted, do NOT ack early — fall back to the strict
+        // upload-before-ack path below.
+        match guard.mark_pending().await {
+            Ok(()) => {
+                let repo_label = name.to_string();
+                tokio::spawn(async move {
+                    if let Err(e) = guard.release(true).await {
+                        tracing::error!(repo = %repo_label, err = %e,
+                            "write-back durable upload failed after push was acked");
+                    }
+                });
+            }
+            Err(e) => {
+                tracing::warn!(repo = %name, err = %e,
+                    "pending-upload marker write failed — falling back to strict upload-before-ack");
+                strict_guard = Some(guard);
+            }
+        }
+    }
+    if let Some(guard) = strict_guard {
+        // Strict path (failed push, async_upload off, or marker write failure):
+        // upload-before-ack.
+        if let Err(e) = guard.release(push_succeeded).await {
+            if push_succeeded {
+                // A successful push whose durable upload then failed — the
+                // client must know the push is not durably stored. The lease
+                // drops on this return, and the advisory lock was already
+                // released inside `release`.
+                tracing::error!(repo = %name, err = %e, "durable upload failed after push");
+                return Err(AppError::Git(format!(
+                    "push applied locally but durable upload to storage failed: {e}"
+                )));
+            }
+            // The push itself failed; log the release error but fall through
+            // so the real git failure (below) is what the client sees.
+            tracing::error!(repo = %name, err = %e, "lock release failed after failed push");
+        }
+    }
     // Clean path: clone (a) already dropped inside run_git_service when the receive-pack
-    // group was reaped; clone (b) held here spanned the success-only Tigris upload that
-    // ran inside release() above. Drop it now so a second same-repo push proceeds the
-    // moment this write is durable, rather than at end of the (longer) handler tail. On
-    // the disconnect path this line is never reached: clone (a) rides the reaper (F3).
+    // group was reaped; clone (b) held here spanned the success-only storage upload that
+    // ran inside release() above (or, under async_upload, only the marker write — the
+    // upload continues under the advisory lock, which is what keeps a second push from
+    // reading a stale archive). Drop it now so a second same-repo push proceeds the
+    // moment this write is durable (or acked), rather than at end of the (longer)
+    // handler tail. On the disconnect path this line is never reached: clone (a) rides
+    // the reaper (F3).
     drop(lease);
 
     let result = receive_result.map_err(|e| {
@@ -3057,7 +3133,7 @@ pub async fn fork_repo(
     // Request is admissible — spend the proof now, immediately before the write.
     let verified_proof = proof.consume(&state.db).await?;
 
-    // Ensure source repo is on local disk (downloads from Tigris on cache miss)
+    // Ensure source repo is on local disk (downloads from storage on cache miss)
     let source_path = state
         .repo_store
         .acquire(&source.owner_did, &source.name)
@@ -3066,30 +3142,14 @@ pub async fn fork_repo(
 
     let disk_path = store::repo_disk_path(&state.config.repos_dir, &forker_did, &fork_name);
 
-    // Clone the source repo as a mirror
-    let output = std::process::Command::new("git")
-        .args([
-            "clone",
-            "--mirror",
-            source_path.to_str().unwrap_or(""),
-            disk_path.to_str().unwrap_or(""),
-        ])
-        .output()
-        .map_err(|e| AppError::Git(format!("git clone --mirror failed: {e}")))?;
-
-    if !output.status.success() {
-        let stderr = String::from_utf8_lossy(&output.stderr);
-        return Err(AppError::Git(format!(
-            "git clone --mirror failed: {stderr}"
-        )));
-    }
-
-    // Upload fork to Tigris
-    state
-        .repo_store
-        .release_after_write(&forker_did, &fork_name)
-        .await;
-
+    // Claim-first ordering: insert the DB row before cloning or uploading
+    // anything. Within one node's database the row is the claim on (owner,
+    // name) — a concurrent same-name fork loses at the insert with nothing on
+    // disk or in storage. Across nodes (per-node Postgres) the insert
+    // arbitrates nothing; the cross-node safety properties are that
+    // publication runs under the per-repo advisory lock (below) and that the
+    // failure compensation only ever removes state THIS attempt created (its
+    // own row by id, its own clone dir) — never a storage archive.
     let now = Utc::now();
     let record = crate::db::RepoRecord {
         id: Uuid::new_v4().to_string(),
@@ -3104,9 +3164,42 @@ pub async fn fork_repo(
         forked_from: Some(source.id.clone()),
         machine_id: state.machine_id.clone(),
     };
-
     state.db.create_repo(&record).await?;
 
+    // The whole clone-and-publish lifecycle runs under the per-repo advisory
+    // lock inside `create_published`: pushes to the just-visible row serialize
+    // on the same lock, so none can execute (let alone be destroyed by the
+    // compensation below) until publication has succeeded or been unwound.
+    // On failure, compensation removes only what THIS attempt created: its
+    // clone dir (inside create_published) and its own row, keyed by our
+    // generated id.
+    let clone_result = state
+        .repo_store
+        .create_published(&forker_did, &fork_name, |dest| {
+            let output = std::process::Command::new("git")
+                .args([
+                    "clone",
+                    "--mirror",
+                    source_path.to_str().unwrap_or(""),
+                    dest.to_str().unwrap_or(""),
+                ])
+                .output()
+                .map_err(|e| anyhow::anyhow!("git clone --mirror failed: {e}"))?;
+            if !output.status.success() {
+                let stderr = String::from_utf8_lossy(&output.stderr);
+                anyhow::bail!("git clone --mirror failed: {stderr}");
+            }
+            Ok(())
+        })
+        .await;
+    if let Err(e) = clone_result {
+        if let Err(db_err) = state.db.delete_repo_by_id(&record.id).await {
+            tracing::warn!(record_id = %record.id, err = %db_err,
+                "failed to remove fork row after fork error");
+        }
+        return Err(AppError::Git(format!("fork failed: {e}")));
+    }
+
     // Persist the proof so the fork carries it when it propagates to peers.
     if let Some(p) = verified_proof {
         if let Err(e) = p.record_for_repo(&state.db, &record.id).await {
@@ -4693,7 +4786,7 @@ mod tests {
 
     /// The receive-pack *advertisement* (`GET info/refs?service=git-receive-pack`)
     /// must be throttled by the per-IP push limiter BEFORE it does the fresh
-    /// Tigris acquire — otherwise the flood brake on the POST is bypassable via
+    /// storage acquire — otherwise the flood brake on the POST is bypassable via
     /// the cheaper unauthenticated GET (PR #152 review P1). Pre-filling the
     /// bucket makes the assertion deterministic and keeps the test off the
     /// acquire path entirely.
@@ -4732,7 +4825,7 @@ mod tests {
         assert_eq!(
             status,
             StatusCode::TOO_MANY_REQUESTS,
-            "receive-pack advertisement must be throttled before the Tigris acquire"
+            "receive-pack advertisement must be throttled before the storage acquire"
         );
     }
 
@@ -6395,7 +6488,7 @@ mod tests {
         // observes the same counter at 1: without it, a zero here would pass on any build
         // where an upload is simply impossible.
         assert_eq!(
-            state.repo_store.tigris_upload_site_reached(),
+            state.repo_store.storage_upload_site_reached(),
             0,
             "a push interrupted by a client disconnect must not reach the Tigris upload \
              site: publishing a half-applied repo propagates it to every node that later \
@@ -6500,7 +6593,7 @@ mod tests {
         // not at least once: a retried exec race releases with success = false and must
         // not count.
         assert_eq!(
-            state.repo_store.tigris_upload_site_reached(),
+            state.repo_store.storage_upload_site_reached(),
             1,
             "a completed push must reach the Tigris upload site once"
         );
@@ -6565,7 +6658,7 @@ mod tests {
 
         // MUST-NOT: with the pool free again, the push is not shed as capacity (it fails
         // later on the nonexistent on-disk repo, which is a git error, not Overloaded).
-        held.release(false).await;
+        held.release(false).await.ok();
         let admitted = git_receive_pack(
             State(state.clone()),
             Path((owner.to_string(), name.to_string())),

crates/gitlawb-node/src/config.rs

@@ -172,11 +172,50 @@ 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.
+    /// Tigris (S3-compatible) bucket for repo storage. Legacy alias for
+    /// `s3_bucket` — still honoured so existing deployments keep working.
     /// 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.

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

@@ -1239,6 +1239,17 @@ 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/git/issues.rs

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

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

@@ -3,5 +3,4 @@ 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/tigris.rs

@@ -1,245 +0,0 @@
-//! 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::{Arc, Mutex, OnceLock};
-
-use anyhow::{Context, Result};
-use aws_sdk_s3::Client as S3Client;
-use tracing::{debug, info};
-
-/// 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(),
-        })
-    }
-
-    /// Test-only constructor with an explicit S3 endpoint, region, and static
-    /// credentials — no env-var reads, so parallel tests cannot race each other's
-    /// `AWS_*` environment the way the env-based `new` would. Lets a test point
-    /// the client at a non-routable endpoint to exercise acquire-stall paths.
-    #[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(),
-        }
-    }
-
-    /// 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}"))
-                }
-            }
-        }
-    }
-
-    /// Upload a local bare repo directory to Tigris as a tar.zst archive.
-    pub async fn upload(&self, owner_slug: &str, repo_name: &str, local_path: &Path) -> Result<()> {
-        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);
-
-        self.s3
-            .put_object()
-            .bucket(&self.bucket)
-            .key(&key)
-            .body(body)
-            .content_type("application/zstd")
-            .send()
-            .await
-            .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: &Path,
-    ) -> Result<()> {
-        let key = Self::repo_key(owner_slug, repo_name);
-        debug!(key = %key, path = %local_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();
-
-        // Extract tar.zst to local path
-        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 from tigris");
-        Ok(())
-    }
-
-    /// Delete a repo archive from Tigris.
-    #[allow(dead_code)]
-    pub async fn delete(&self, owner_slug: &str, repo_name: &str) -> Result<()> {
-        let key = Self::repo_key(owner_slug, repo_name);
-        self.s3
-            .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.
-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.
-    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 the freshly-extracted repo into place. rename within the same parent
-    // is effectively atomic, but most platforms refuse to rename onto a
-    // non-empty dir, so remove the old copy first. Serialize this per repo path:
-    // concurrent extractions unpack into isolated temp dirs, but their swaps
-    // must not interleave or they race to a nondeterministic overwrite/failure.
-    let lock = publish_lock(local_path);
-    let _publish = lock.lock().expect("publish lock poisoned");
-    if local_path.exists() {
-        std::fs::remove_dir_all(local_path).context("removing stale repo dir")?;
-    }
-    std::fs::rename(&tmp_dir, local_path).context("swapping extracted repo into place")?;
-
-    Ok(())
-}

crates/gitlawb-node/src/main.rs

@@ -18,6 +18,7 @@ mod pinata;
 mod rate_limit;
 mod server;
 mod state;
+mod storage;
 mod sync;
 #[cfg(test)]
 mod test_support;
@@ -299,22 +300,14 @@ async fn main() -> Result<()> {
         info!("  fly machine: {mid}");
     }
 
-    // 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
-    };
+    // 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
@@ -323,13 +316,42 @@ async fn main() -> Result<()> {
     // 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).
+    // 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),
     );
-    let repo_store = git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, lock_pool);
+
+    // 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");
+            }
+        });
+    }
 
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
     // DID (attacker-varied), so bound its key set to cap memory.

crates/gitlawb-node/src/metrics.rs

@@ -15,6 +15,8 @@
 //!     `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`, ...)
@@ -51,6 +53,7 @@ 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
@@ -202,6 +205,18 @@ 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)");
@@ -284,6 +299,24 @@ 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 Tigris backend
+    /// Centralized repo storage: local disk cache + optional object-store backend
     pub repo_store: RepoStore,
     /// Per-DID rate limiter for creation endpoints (repos, issues, PRs)
     pub rate_limiter: RateLimiter,

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

@@ -0,0 +1,446 @@
+//! 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

@@ -0,0 +1,374 @@
+//! 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

@@ -0,0 +1,173 @@
+//! 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

@@ -0,0 +1,164 @@
+//! 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

@@ -0,0 +1,167 @@
+//! 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

@@ -10687,6 +10687,9 @@ mod tests {
             "TigrisClient",
             ".download(",
             "tigris",
+            "RepoArchive",
+            "BlobStore",
+            "fetch_raw",
         ] {
             assert!(
                 !code.contains(shape),

crates/gitlawb-node/tests/inv22_gates.rs

@@ -210,14 +210,21 @@ fn inv22_concurrency_gates_present_and_not_bypassed() {
     }
 }
 
-/// 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,
+/// 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,
 /// 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.
 ///
-/// Scoped to the `release` fn body: the `Drop` impl legitimately takes the
+/// 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
 /// 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]
@@ -225,12 +232,12 @@ fn f4_release_keeps_conn_owned_until_unlock_resolves() {
     let repo_store = src("git/repo_store.rs");
 
     let rel_start = repo_store
-        .find("pub async fn release(mut self")
-        .expect("F4 gate: repo_store.rs no longer defines RepoWriteGuard::release");
+        .find("async fn unlock(mut self")
+        .expect("F4 gate: repo_store.rs no longer defines LockedConn::unlock");
     let rel_end = repo_store[rel_start..]
-        .find("impl Drop for RepoWriteGuard")
+        .find("impl Drop for LockedConn")
         .map(|off| rel_start + off)
-        .expect("F4 gate: release fn / Drop impl markers moved — update this guard");
+        .expect("F4 gate: unlock fn / Drop impl markers moved — update this guard");
     let release_body = &repo_store[rel_start..rel_end];
 
     let unlock = release_body
@@ -241,7 +248,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: RepoWriteGuard::release takes self.conn BEFORE awaiting \
+        "F4 regression: LockedConn::unlock 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."
@@ -250,7 +257,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: RepoWriteGuard::release sets `released = true` BEFORE awaiting \
+        "F4 regression: LockedConn::unlock 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."