feat(storage): storage-agnostic BlobStore layer + push-latency quick wins

.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

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,20 +284,37 @@ 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(),
         ));
     }
 
     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())),

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())),