diff --git a/.env.example b/.env.example index bbd9a342..4e5b5adb 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,30 @@ 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. +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. +GITLAWB_ADVISORY_LOCK_POOL_SIZE=16 + # PostgreSQL connection URL. Required. # When using the bundled docker-compose, this is wired automatically. DATABASE_URL=postgresql://gitlawb:changeme@localhost:5432/gitlawb diff --git a/Cargo.lock b/Cargo.lock index 5dd2903d..027368ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3300,7 +3300,7 @@ dependencies = [ [[package]] name = "git-remote-gitlawb" -version = "0.5.1" +version = "0.7.0" dependencies = [ "anyhow", "gitlawb-core", @@ -3312,7 +3312,7 @@ dependencies = [ [[package]] name = "gitlawb-attest" -version = "0.5.1" +version = "0.7.0" dependencies = [ "base64", "ed25519-dalek", @@ -3329,7 +3329,7 @@ dependencies = [ [[package]] name = "gitlawb-core" -version = "0.5.1" +version = "0.7.0" dependencies = [ "anyhow", "base64", @@ -3356,13 +3356,14 @@ dependencies = [ [[package]] name = "gitlawb-node" -version = "0.5.1" +version = "0.7.0" dependencies = [ "alloy", "anyhow", "async-compression", "async-graphql", "async-graphql-axum", + "async-trait", "aws-config", "aws-sdk-s3", "axum", @@ -3412,10 +3413,11 @@ dependencies = [ [[package]] name = "gl" -version = "0.5.1" +version = "0.7.0" dependencies = [ "alloy", "anyhow", + "base64", "chrono", "clap", "dirs", diff --git a/README.md b/README.md index 57ce2885..fab2c773 100644 --- a/README.md +++ b/README.md @@ -344,7 +344,14 @@ 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 may run before it is aborted (504). Default 600. Does not bound `info/refs` or the withheld-blob path. | -| `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. | +| `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_ADVISORY_LOCK_POOL_SIZE` | Dedicated DB pool for per-repo write locks; bounds per-node push concurrency. Default 16. | +| `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. | diff --git a/crates/gitlawb-node/Cargo.toml b/crates/gitlawb-node/Cargo.toml index 72748fae..f580063e 100644 --- a/crates/gitlawb-node/Cargo.toml +++ b/crates/gitlawb-node/Cargo.toml @@ -33,6 +33,7 @@ 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 } @@ -57,7 +58,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. diff --git a/crates/gitlawb-node/src/api/issues.rs b/crates/gitlawb-node/src/api/issues.rs index 17acf9af..6ab192c0 100644 --- a/crates/gitlawb-node/src/api/issues.rs +++ b/crates/gitlawb-node/src/api/issues.rs @@ -70,10 +70,19 @@ 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()))?; + // The git mutation is committed locally and protected by the pending-upload + // marker, so a durable-upload failure here is recoverable (the next + // successful upload re-syncs storage) — NOT a request failure. Failing the + // request would make the caller retry a non-idempotent mutation: the retry + // mints a new issue UUID and both issues eventually publish. + if let Err(e) = release_result { + tracing::error!(repo = %record.name, issue = %issue_id, err = %e, + "issue committed locally but durable upload failed — storage re-syncs on next upload"); + } // 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) @@ -244,11 +253,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())); } }; @@ -257,7 +266,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(), )); @@ -265,12 +274,18 @@ 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")))?; + // Committed locally + marker-protected: a durable-upload failure is + // recoverable, not a request failure (see create_issue). + if let Err(e) = release_result { + tracing::error!(repo = %repo, issue = %issue_id, err = %e, + "issue close committed locally but durable upload failed — storage re-syncs on next upload"); + } let issue: serde_json::Value = serde_json::from_str(&updated) .map_err(|e| AppError::BadRequest(format!("invalid issue data: {e}")))?; diff --git a/crates/gitlawb-node/src/api/pulls.rs b/crates/gitlawb-node/src/api/pulls.rs index 26be6109..e99edc43 100644 --- a/crates/gitlawb-node/src/api/pulls.rs +++ b/crates/gitlawb-node/src/api/pulls.rs @@ -224,10 +224,20 @@ 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()))?; + // The merge commit exists locally and is protected by the pending-upload + // marker, so a durable-upload failure is recoverable — NOT a request + // failure. Failing here would leave the target ref already merged while + // the PR row stays open, and a retry cannot un-merge it; proceed so the + // DB status matches the git state, and let the next successful upload + // re-sync storage. + if let Err(e) = release_result { + tracing::error!(repo = %record.name, pr = %pr.id, err = %e, + "merge committed locally but durable upload failed — storage re-syncs on next upload"); + } state.db.merge_pr(&pr.id, &merger_did).await?; let _ = state.db.touch_repo(&record.id).await; diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b38b177b..333cae14 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -207,17 +207,14 @@ 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. + 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(), @@ -232,9 +229,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 upload the initial archive. On failure, + // compensate by removing our own just-inserted row (keyed by our id) so a + // retry starts clean; init() already removes its local dir on upload + // failure. + 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 { @@ -547,10 +558,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) { @@ -563,7 +574,7 @@ pub async fn git_info_refs( } } - // 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. let disk_path = if service == "git-receive-pack" { state @@ -943,9 +954,67 @@ pub async fn git_receive_pack( let receive_result = smart_http::receive_pack(&disk_path, body, git_timeout).await; // 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. - guard.release(receive_result.is_ok()).await; + let push_ok = receive_result.is_ok(); + // `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(guard); + if push_ok && 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_ok).await { + if push_ok { + // A successful push whose durable upload then failed — the + // client must know the push is not durably stored. + 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"); + } + } let result = receive_result.map_err(|e| { let app = git_service_app_error(&e); @@ -1561,7 +1630,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) @@ -1570,8 +1639,61 @@ pub async fn fork_repo( let disk_path = store::repo_disk_path(&state.config.repos_dir, &forker_did, &fork_name); + // 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 property is that the failure + // compensation below only ever removes state THIS attempt created (its own + // row by id, its own clone dir) and never touches a storage archive that + // may belong to a racing winner's live repo. + let now = Utc::now(); + let record = crate::db::RepoRecord { + id: Uuid::new_v4().to_string(), + name: fork_name.clone(), + owner_did: forker_did.clone(), + description: source.description.clone(), + is_public: source.is_public, + default_branch: source.default_branch.clone(), + created_at: now, + updated_at: now, + disk_path: disk_path.to_string_lossy().to_string(), + forked_from: Some(source.id.clone()), + machine_id: state.machine_id.clone(), + }; + state.db.create_repo(&record).await?; + + // Any failure from here must undo the claim: remove our own row (keyed by + // our generated id) and whatever `git clone` left at `disk_path` — clone + // can create the destination before failing, and an orphaned dir makes a + // retry fail on an existing destination. + // + // KNOWN COMPOUND-FAILURE WINDOW: the row makes the fork pushable before + // the clone+upload complete, and this flow holds no advisory lock across + // that phase. A push acked in the window, whose own upload also failed, + // lives only in this dir — undoing the claim then discards it. Requires + // fork-upload failure + an interleaved push + that push's upload failure; + // the follow-up is to hold the advisory lock across clone+publication. + let undo_claim = |state: AppState, record_id: String, disk_path: std::path::PathBuf| async move { + match tokio::fs::remove_dir_all(&disk_path).await { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => tracing::warn!( + path = %disk_path.display(), err = %e, + "failed to clean up local fork dir after fork error" + ), + } + // Also drop the pending-upload marker a failed upload just wrote: left + // behind, it would read as divergence and wedge a same-name recreate. + crate::git::repo_store::clear_pending_upload(&disk_path); + if let Err(e) = state.db.delete_repo_by_id(&record_id).await { + tracing::warn!(record_id = %record_id, err = %e, + "failed to remove fork row after fork error"); + } + }; + // Clone the source repo as a mirror - let output = std::process::Command::new("git") + let output = match std::process::Command::new("git") .args([ "clone", "--mirror", @@ -1579,37 +1701,34 @@ pub async fn fork_repo( disk_path.to_str().unwrap_or(""), ]) .output() - .map_err(|e| AppError::Git(format!("git clone --mirror failed: {e}")))?; + { + Ok(output) => output, + Err(e) => { + undo_claim(state.clone(), record.id.clone(), disk_path).await; + return Err(AppError::Git(format!("git clone --mirror failed: {e}"))); + } + }; if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + undo_claim(state.clone(), record.id.clone(), disk_path).await; return Err(AppError::Git(format!( "git clone --mirror failed: {stderr}" ))); } - // Upload fork to Tigris - state + // Upload fork to storage — fail closed if the durable upload fails rather + // than reporting a fork that only exists on this node's local disk. + if let Err(e) = state .repo_store .release_after_write(&forker_did, &fork_name) - .await; - - let now = Utc::now(); - let record = crate::db::RepoRecord { - id: Uuid::new_v4().to_string(), - name: fork_name.clone(), - owner_did: forker_did.clone(), - description: source.description.clone(), - is_public: source.is_public, - default_branch: source.default_branch.clone(), - created_at: now, - updated_at: now, - disk_path: disk_path.to_string_lossy().to_string(), - forked_from: Some(source.id.clone()), - machine_id: state.machine_id.clone(), - }; - - state.db.create_repo(&record).await?; + .await + { + undo_claim(state.clone(), record.id.clone(), disk_path).await; + return Err(AppError::Git(format!( + "fork created locally but durable upload failed: {e}" + ))); + } // Persist the proof so the fork carries it when it propagates to peers. if let Some(p) = verified_proof { @@ -2554,7 +2673,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. @@ -2593,7 +2712,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" ); } diff --git a/crates/gitlawb-node/src/config.rs b/crates/gitlawb-node/src/config.rs index fc2247d9..601769fd 100644 --- a/crates/gitlawb-node/src/config.rs +++ b/crates/gitlawb-node/src/config.rs @@ -129,11 +129,61 @@ 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. + #[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, + + /// Max connections in the dedicated advisory-lock pool. A push pins one + /// connection for its whole lifetime (lock held across receive-pack and + /// the storage upload), so this bounds per-node push concurrency. Counts + /// against Postgres `max_connections` on top of the handler pool. Must be + /// at least 1 — zero would let the node start but time out every write. + #[arg( + long, + env = "GITLAWB_ADVISORY_LOCK_POOL_SIZE", + default_value_t = 16, + value_parser = clap::value_parser!(u32).range(1..) + )] + pub advisory_lock_pool_size: u32, + /// 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. @@ -272,4 +322,21 @@ mod tests { Config::try_parse_from(["gitlawb-node", "--git-service-timeout-secs", "0"]).is_err() ); } + + #[test] + fn advisory_lock_pool_size_defaults_to_16_and_rejects_zero() { + assert_eq!( + Config::parse_from(["gitlawb-node"]).advisory_lock_pool_size, + 16 + ); + assert_eq!( + Config::parse_from(["gitlawb-node", "--advisory-lock-pool-size", "1"]) + .advisory_lock_pool_size, + 1 + ); + // 0 would let the node start but time out every write on pool acquire. + assert!( + Config::try_parse_from(["gitlawb-node", "--advisory-lock-pool-size", "0"]).is_err() + ); + } } diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 33f5bd67..0b947ed8 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -250,11 +250,6 @@ pub struct Db { } impl Db { - /// Access the underlying Postgres connection pool. - pub fn pool(&self) -> &PgPool { - &self.pool - } - #[cfg(test)] pub fn for_testing(pool: PgPool) -> Self { Self { pool } @@ -989,6 +984,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( diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index 59e34c84..eec7f23d 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/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; diff --git a/crates/gitlawb-node/src/git/repo_store.rs b/crates/gitlawb-node/src/git/repo_store.rs index a5c367e9..50fcc3fa 100644 --- a/crates/gitlawb-node/src/git/repo_store.rs +++ b/crates/gitlawb-node/src/git/repo_store.rs @@ -1,35 +1,49 @@ -//! Centralized repo storage layer — local disk cache backed by Tigris (S3). +//! Centralized repo storage layer — local disk cache backed by a pluggable +//! object store (S3-compatible / filesystem / IPFS) via [`RepoArchive`]. //! //! Every handler that needs access to a git repo on disk goes through `RepoStore`: //! -//! - `acquire()` — ensures the repo is on local disk (downloads from Tigris on cache miss). -//! - `release_after_write()` — uploads the updated repo to Tigris after a write operation. -//! - `init()` — creates a new bare repo locally and uploads to Tigris. +//! - `acquire()` — ensures the repo is on local disk (downloads on cache miss). +//! - `acquire_write()` — write lock + ensures local matches storage (skips the +//! download when the cached etag already matches — the push-latency win). +//! - `release()` / `release_after_write()` — upload the updated repo to storage. +//! - `init()` — creates a new bare repo locally and uploads to storage. //! -//! When Tigris is disabled (bucket empty), this is a simple passthrough to local disk. +//! When no backend is configured, this is a simple passthrough to local disk. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::{Context, Result}; -use sqlx::PgPool; +use sqlx::pool::PoolConnection; +use sqlx::{PgPool, Postgres}; use tokio::sync::Mutex; -use tracing::{debug, info, warn}; +use tracing::{debug, warn}; use super::store; -use super::tigris::TigrisClient; +use crate::storage::archive::RepoArchive; -/// Centralized repo storage: local disk cache + optional Tigris backend. +/// Centralized repo storage: local disk cache + optional object-storage backend +/// (S3-compatible / filesystem / IPFS) behind the [`RepoArchive`] layer. #[derive(Clone)] pub struct RepoStore { repos_dir: PathBuf, - tigris: Option, - /// Shared Postgres pool for advisory locks. - pool: PgPool, - /// Tracks repos already confirmed to exist in Tigris — avoids redundant + archive: Option, + /// Bounded pool dedicated to advisory-lock connections — the only DB pool + /// RepoStore needs, as it touches Postgres solely for advisory locks. A push + /// pins one connection for its whole lifetime (the lock is held across both + /// receive-pack and the upload), so a separate budget keeps a burst of + /// concurrent/slow pushes from draining the handler pool and starving every + /// other DB handler. + lock_pool: PgPool, + /// Tracks repos already confirmed to exist in storage — avoids redundant /// HEAD checks and background uploads for repos we've already migrated. migrated: Arc>>, + /// Last-known archive etag per `owner_slug/repo` key. Lets a write skip the + /// pre-write download when our local copy already matches storage (the + /// common case under sticky routing) — the main push-latency win. + versions: Arc>>, } impl RepoStore { @@ -37,80 +51,233 @@ impl RepoStore { pub fn for_testing(repos_dir: PathBuf, pool: PgPool) -> Self { Self { repos_dir, - tigris: None, - pool, - migrated: Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())), + archive: None, + lock_pool: pool, + migrated: Arc::new(Mutex::new(HashSet::new())), + versions: Arc::new(Mutex::new(HashMap::new())), } } - pub fn new(repos_dir: PathBuf, tigris: Option, pool: PgPool) -> Self { + /// `lock_pool` is a bounded pool that advisory-lock connections are pinned + /// from, kept separate from the handler pool so push concurrency can't + /// drain it. + pub fn new(repos_dir: PathBuf, archive: Option, lock_pool: PgPool) -> Self { Self { repos_dir, - tigris, - pool, + archive, + lock_pool, migrated: Arc::new(Mutex::new(HashSet::new())), + versions: Arc::new(Mutex::new(HashMap::new())), + } + } + + /// Ensure the local copy matches storage, skipping the download when our + /// cached etag already equals the current archive etag. + /// + /// `require_fresh` selects the failure policy: + /// - `false` (read path, `acquire_fresh`): self-heal — if a storage HEAD or + /// download fails but a valid local copy exists, use it; a later upload + /// re-syncs storage. + /// - `true` (write path, `acquire_write`): fail closed — never fall back to + /// a possibly-stale local copy. The remote etag differs (remote is newer), + /// so uploading our stale copy after the write would clobber it (lost + /// update). Propagate the error so the write is rejected instead. + async fn sync_down_if_stale( + &self, + owner_slug: &str, + repo_name: &str, + local_path: &Path, + require_fresh: bool, + ) -> Result<()> { + let Some(ref archive) = self.archive else { + return Ok(()); + }; + + let marker = pending_upload_marker(local_path); + // try_exists, not exists(): a transient EACCES/EIO must not read as + // "no marker" — that path downloads and can roll back a pending local + // write. Fail the write path closed; treat as present on the read path. + let marker_present = match marker.try_exists() { + Ok(present) => present, + Err(e) => { + if require_fresh { + return Err(e).context("probing pending-upload marker"); + } + warn!(repo = %repo_name, err = %e, + "pending-upload marker probe failed — treating as present"); + true + } + }; + if marker_present { + if local_path.exists() { + // The local copy has a write that storage never received (its + // upload failed, or the node stopped first). The marker records + // the storage etag that write was BASED on, so we can tell + // "storage unchanged — local strictly ahead" apart from + // "another node advanced storage — genuine divergence" rather + // than treating local as authoritative forever. + let base = std::fs::read_to_string(&marker).unwrap_or_default(); + let base = base.trim(); + let remote = match archive.head_etag(owner_slug, repo_name).await { + Ok(r) => r, + Err(e) => { + if require_fresh { + return Err(e).context("storage head while local pending upload"); + } + warn!(repo = %repo_name, err = %e, + "storage head failed while pending upload — using local copy"); + return Ok(()); + } + }; + match remote.as_deref() { + // Storage empty, or exactly the version our write built on: + // local is strictly ahead. Serve it; the next successful + // post-write upload re-syncs storage and clears the marker. + None => return Ok(()), + Some(r) if r == base => { + warn!(repo = %repo_name, + "local copy ahead of storage (pending upload) — skipping download"); + return Ok(()); + } + // Storage advanced past our base while this node held + // un-uploaded local changes: both sides have writes the + // other lacks. Overwriting either silently loses a push. + Some(_) => { + if require_fresh { + anyhow::bail!( + "storage for {owner_slug}/{repo_name} advanced while local \ + changes were pending upload — refusing to overwrite either \ + side; reconcile manually (fetch both, merge, remove the \ + pending-upload marker)" + ); + } + warn!(repo = %repo_name, + "storage diverged from pending local copy — serving local for read"); + return Ok(()); + } + } + } + // Marker without a local copy: the repo dir was removed out from + // under us, so the storage copy is the best remaining state. Drop + // the stale marker and fall through to the normal download. + let _ = std::fs::remove_file(&marker); + } + let key = format!("{owner_slug}/{repo_name}"); + + let remote_etag = match archive.head_etag(owner_slug, repo_name).await { + Ok(Some(etag)) => etag, + Ok(None) => return Ok(()), // not in storage yet — local is authoritative + Err(e) => { + // HEAD failed. Read path: fall back to a valid local copy if we + // have one. Write path: fail closed (see `require_fresh`). + if !require_fresh && local_path.exists() { + warn!(repo = %repo_name, err = %e, "storage head failed — using local copy"); + return Ok(()); + } + return Err(e).context("storage head before access"); + } + }; + + if local_path.exists() { + let known = self.versions.lock().await.get(&key).cloned(); + if known.as_deref() == Some(remote_etag.as_str()) { + debug!(repo = %repo_name, "local copy current (etag match) — skipping download"); + return Ok(()); + } + } + + // KNOWN LIMITATION (pre-dates this layer): read-path downloads and + // their swap-into-place are not serialized against the advisory write + // lock, so a slow in-flight download decided before a push began can + // swap a stale tree under a running receive-pack on the same node. + // Requires a cache-miss/stale read racing a same-repo write; the + // follow-up is to serialize download+swap with writers. + match archive.download(owner_slug, repo_name, local_path).await { + Ok(()) => { + self.versions.lock().await.insert(key, remote_etag); + Ok(()) + } + Err(e) => { + // Read path self-heal only: a corrupt/unreadable archive must not + // block access when a valid local copy exists. On the write path + // the remote etag differs (remote is newer), so falling back and + // later uploading our stale copy would clobber it — fail closed. + if !require_fresh && local_path.exists() { + warn!(repo = %repo_name, err = %e, + "archive download failed — falling back to local copy"); + Ok(()) + } else { + Err(e).context("downloading repo archive") + } + } } } - /// Ensure a repo is available on local disk, downloading from Tigris if needed. - /// If the repo exists locally but not yet in Tigris, a background upload is - /// spawned to lazily migrate it (on-demand migration for pre-Tigris repos). + /// Ensure a repo is available on local disk, downloading from storage if needed. + /// If the repo exists locally but not yet in storage, a background upload is + /// spawned to lazily migrate it (on-demand migration for pre-storage repos). /// Returns the local path to the bare repo. pub async fn acquire(&self, owner_did: &str, repo_name: &str) -> Result { let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?; // Fast path: repo exists locally if local_path.exists() { - // Lazy migration: if Tigris is enabled and we haven't confirmed this - // repo is in Tigris yet, check and upload in the background. - if let Some(ref tigris) = self.tigris { + // Lazy migration: if storage is enabled and we haven't confirmed this + // repo is in storage yet, check and upload in the background. + if self.archive.is_some() { let key = format!("{owner_slug}/{repo_name}"); let already_migrated = self.migrated.lock().await.contains(&key); - if !already_migrated { - let tigris = tigris.clone(); + // A pending-upload marker means the marker machinery already + // owns this repo's next upload (next write, or the startup + // retry). Migration must not steal it: `upload_under_lock` + // knows nothing about markers, so its upload would strand the + // marker with a base that no longer matches storage, wedging + // the repo's writes on a spurious divergence. + let marker_pending = pending_upload_marker(&local_path).exists(); + if !already_migrated && !marker_pending { + let this = self.clone(); let slug = owner_slug.clone(); let name = repo_name.to_string(); let path = local_path.clone(); - let migrated = Arc::clone(&self.migrated); + let key = key.clone(); tokio::spawn(async move { - // Check if already in Tigris before uploading - match tigris.exists(&slug, &name).await { - Ok(true) => { - debug!(repo = %name, "repo already in tigris — skipping migration"); - } - Ok(false) => { - info!(repo = %name, "migrating local repo to tigris"); - if let Err(e) = tigris.upload(&slug, &name, &path).await { - warn!(repo = %name, err = %e, "lazy migration to tigris failed"); - return; - } - info!(repo = %name, "lazy migration to tigris complete"); + // Upload under the advisory lock (skip if already present) + // so this opportunistic migration can't clobber a + // concurrent locked push by landing a stale snapshot. + match this.upload_under_lock(&slug, &name, &path, true).await { + Ok(()) => { + this.migrated.lock().await.insert(key); + debug!(repo = %name, "lazy migration to storage complete (or already present)"); } Err(e) => { - warn!(repo = %name, err = %e, "tigris existence check failed"); - return; + warn!(repo = %name, err = %e, "lazy migration to storage failed"); } } - migrated.lock().await.insert(format!("{slug}/{name}")); }); } } return Ok(local_path); } - // Try downloading from Tigris - if let Some(ref tigris) = self.tigris { - if tigris.exists(&owner_slug, repo_name).await.unwrap_or(false) { - debug!(repo = %repo_name, "cache miss — downloading from tigris"); - tigris + // Try downloading from storage + if let Some(ref archive) = self.archive { + if let Some(remote_etag) = archive + .head_etag(&owner_slug, repo_name) + .await + .context("checking storage for repo")? + { + debug!(repo = %repo_name, "cache miss — downloading from storage"); + archive .download(&owner_slug, repo_name, &local_path) .await - .context("downloading repo from tigris")?; - // Mark as migrated since we just downloaded it - self.migrated - .lock() - .await - .insert(format!("{owner_slug}/{repo_name}")); + .context("downloading repo from storage")?; + // The local copy didn't exist, so any pending-upload marker + // here is stale litter — clear it or it would wrongly pin the + // just-downloaded copy as "ahead of storage". + clear_pending_upload(&local_path); + let key = format!("{owner_slug}/{repo_name}"); + self.migrated.lock().await.insert(key.clone()); + self.versions.lock().await.insert(key, remote_etag); return Ok(local_path); } } @@ -120,34 +287,14 @@ impl RepoStore { Ok(local_path) } - /// Ensure a repo is available on local disk with the **latest** Tigris state. + /// Ensure a repo is available on local disk with the **latest** storage state. /// Use this for operations that precede a write (e.g. `info/refs` for /// `git-receive-pack`) so the client sees the same refs that `acquire_write()` /// will operate on. pub async fn acquire_fresh(&self, owner_did: &str, repo_name: &str) -> Result { let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?; - - if let Some(ref tigris) = self.tigris { - if tigris.exists(&owner_slug, repo_name).await.unwrap_or(false) { - debug!(repo = %repo_name, "acquire_fresh: downloading latest from tigris"); - if let Err(e) = tigris.download(&owner_slug, repo_name, &local_path).await { - // The Tigris archive is present (HEAD ok) but unreadable — a - // corrupt/partial upload, or a transient GET failure. If we have a - // valid local copy, proceed with it rather than blocking the write; - // the post-write upload re-syncs (self-heals) Tigris. Only hard-fail - // when there is no local copy to fall back to. - if local_path.exists() { - warn!(repo = %repo_name, err = %e, - "acquire_fresh: tigris download failed — falling back to local copy"); - return Ok(local_path); - } - return Err(e).context("downloading repo from tigris (fresh)"); - } - return Ok(local_path); - } - } - - // Tigris disabled or repo not in Tigris — fall back to local + self.sync_down_if_stale(&owner_slug, repo_name, &local_path, false) + .await?; Ok(local_path) } @@ -156,93 +303,317 @@ impl RepoStore { pub async fn acquire_write(&self, owner_did: &str, repo_name: &str) -> Result { let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?; let lock_key = advisory_lock_key(&owner_slug, repo_name); + let label = format!("{owner_slug}/{repo_name}"); + let lock = LockedConn::acquire(&self.lock_pool, lock_key, &label).await?; - // Acquire Postgres advisory lock with retry using pg_try_advisory_lock - // to avoid blocking indefinitely on stale locks from crashed connections. - let mut acquired = false; - for attempt in 0..60 { - let row: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)") - .bind(lock_key) - .fetch_one(&self.pool) - .await - .context("trying advisory lock")?; - if row.0 { - acquired = true; - break; - } - if attempt < 59 { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - } - } - if !acquired { - anyhow::bail!("could not acquire advisory lock after 60s — possible stale lock for {owner_slug}/{repo_name}"); - } - - // Always download the latest from Tigris before writing. - // Local disk may be stale if another machine pushed since our last access. - if let Some(ref tigris) = self.tigris { - if tigris.exists(&owner_slug, repo_name).await.unwrap_or(false) { - debug!(repo = %repo_name, "write acquire: downloading latest from tigris"); - if let Err(e) = tigris.download(&owner_slug, repo_name, &local_path).await { - // Same self-healing fallback as acquire_fresh: a corrupt/unreadable - // Tigris archive must not block a write when a valid local copy - // exists — release(success) will re-upload a good archive. - if local_path.exists() { - warn!(repo = %repo_name, err = %e, - "write acquire: tigris download failed — falling back to local copy"); - } else { - return Err(e).context("downloading repo from tigris for write"); - } - } - } + // Ensure local matches the latest in storage before writing. The etag + // cache skips the full download when our copy is already current (the + // common single-machine case under sticky routing); a stale copy (another + // machine pushed since) still triggers a download. The advisory lock above + // serializes this so the post-write upload can't race a concurrent writer. + if let Err(e) = self + .sync_down_if_stale(&owner_slug, repo_name, &local_path, true) + .await + { + lock.unlock().await; + return Err(e); } Ok(RepoWriteGuard { owner_slug, repo_name: repo_name.to_string(), local_path, - lock_key, - pool: self.pool.clone(), - tigris: self.tigris.clone(), + lock, + archive: self.archive.clone(), + versions: Arc::clone(&self.versions), }) } - /// Initialize a new bare repo on local disk and upload to Tigris. + /// Initialize a new bare repo on local disk and upload to storage. pub async fn init(&self, owner_did: &str, repo_name: &str) -> Result { let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?; - store::init_bare(&local_path).context("initializing bare repo")?; - - // Upload to Tigris in background - if let Some(ref tigris) = self.tigris { - let tigris = tigris.clone(); - let owner_slug = owner_slug.clone(); - let repo_name = repo_name.to_string(); - let path = local_path.clone(); - tokio::spawn(async move { - if let Err(e) = tigris.upload(&owner_slug, &repo_name, &path).await { - warn!(repo = %repo_name, err = %e, "failed to upload new repo to tigris"); - } - }); + if let Err(e) = store::init_bare(&local_path) { + // A half-initialized dir would block a retry on "already exists". + let _ = std::fs::remove_dir_all(&local_path); + return Err(e).context("initializing bare repo"); + } + // A marker left by a previous same-name repo (failed creation, deleted + // repo) describes THAT repo's history, not this fresh one — once this + // repo's archive exists, a stale empty-base marker would read as + // divergence and wedge its writes. + clear_pending_upload(&local_path); + + // Upload the new repo synchronously under the advisory lock: a background + // upload could land the empty repo *after* a racing first push and clobber + // it, and a silent failure would leave the repo absent from storage. Fail + // closed instead — surface upload errors to the caller, and remove the + // just-created local dir so a retry of the same name doesn't hit an + // existing destination. + if let Err(e) = self + .upload_under_lock(&owner_slug, repo_name, &local_path, false) + .await + { + if let Err(cleanup_err) = std::fs::remove_dir_all(&local_path) { + warn!(repo = %repo_name, err = %cleanup_err, + "failed to remove local repo dir after init upload failure"); + } + return Err(e).context("uploading new repo to storage"); } Ok(local_path) } - /// Upload a repo to Tigris after a write operation (push, merge, fork, etc.). - /// Call this after any operation that modifies the git repo on disk. - pub async fn release_after_write(&self, owner_did: &str, repo_name: &str) { - if let Some(ref tigris) = self.tigris { - let (owner_slug, local_path) = match self.local_path(owner_did, repo_name) { - Ok(p) => p, + /// Upload a repo to storage after a write operation (merge, fork, etc.). + /// Call this after any operation that modifies the git repo on disk. Returns + /// `Err` if the durable upload fails so the caller can surface it rather than + /// acking a write that never reached storage. + /// + /// The upload runs under the per-repo advisory lock: claim-first creation + /// makes a fork addressable (and pushable) before this upload finishes, so + /// an unlocked PUT could compress a pre-push snapshot and land it AFTER a + /// concurrent locked push's upload — which that push's cleared marker no + /// longer protects against. + pub async fn release_after_write(&self, owner_did: &str, repo_name: &str) -> Result<()> { + if self.archive.is_none() { + return Ok(()); + } + let (owner_slug, local_path) = self + .local_path(owner_did, repo_name) + .context("rejected unsafe path in release_after_write")?; + let key = format!("{owner_slug}/{repo_name}"); + let base = self.versions.lock().await.get(&key).cloned(); + mark_pending_upload(&local_path, base.as_deref()) + .context("persisting pending-upload marker")?; + // The marker's recorded base is authoritative, not the cache: if a + // marker already existed (earlier failed upload), mark_pending_upload + // preserved its ORIGINAL base while the versions cache was invalidated + // by that failure — or emptied entirely by a restart. Comparing + // against the cache value would misread that state as divergence. + let base = std::fs::read_to_string(pending_upload_marker(&local_path)) + .map(|s| s.trim().to_string()) + .unwrap_or_else(|_| base.unwrap_or_default()); + match self + .upload_locked_with_marker(&owner_slug, repo_name, &local_path, &base) + .await + { + Ok(PendingUploadOutcome::Uploaded) => Ok(()), + Ok(PendingUploadOutcome::Diverged) => { + // Another writer advanced storage between our sync and this + // upload. Refusing (rather than uploading) protects their + // acked write; ours stays local behind the marker. + self.versions.lock().await.remove(&key); + anyhow::bail!( + "storage for {key} advanced during the write — upload aborted to \ + avoid clobbering the concurrent writer; local copy left marked" + ) + } + Err(e) => { + // Storage is now behind local. Drop the cached etag, and leave + // the pending-upload marker in place so sync_down_if_stale + // serves the local copy instead of rolling it back to the + // stale archive. + self.versions.lock().await.remove(&key); + Err(e).context("uploading repo to storage after write") + } + } + } + + /// Startup sweep re-attempting the durable upload for every repo whose + /// pending-upload marker survived a crash or a failed upload. Without this, + /// a repo that receives no further writes stays divergent from storage + /// indefinitely, visible only as one log line at failure time. + /// + /// Applies the same base-etag rule as `sync_down_if_stale`: a repo whose + /// storage advanced past the marker's base is left marked (its writes stay + /// wedged pending manual reconciliation) and only logged. Returns + /// `(reuploaded, still_pending)`. + pub async fn retry_pending_uploads(&self) -> (usize, usize) { + if self.archive.is_none() { + return (0, 0); + } + let mut reuploaded = 0usize; + let mut still_pending = 0usize; + + let mut markers: Vec<(String, String, PathBuf)> = Vec::new(); // (slug, repo, local) + let Ok(owners) = std::fs::read_dir(&self.repos_dir) else { + return (0, 0); + }; + for owner in owners.flatten() { + if !owner.path().is_dir() { + continue; + } + let slug = owner.file_name().to_string_lossy().into_owned(); + let Ok(entries) = std::fs::read_dir(owner.path()) else { + continue; + }; + for entry in entries.flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + // Marker layout: `.{repo}.git.pending-upload` + let Some(repo_dir) = name + .strip_prefix('.') + .and_then(|n| n.strip_suffix(".pending-upload")) + else { + continue; + }; + let Some(repo_name) = repo_dir.strip_suffix(".git") else { + continue; + }; + markers.push(( + slug.clone(), + repo_name.to_string(), + owner.path().join(repo_dir), + )); + } + } + + // Seed the gauge with the surviving-marker count before processing; + // the clears below (and all runtime marker churn) then keep it + // current via deltas. + crate::metrics::set_pending_upload_markers(markers.len() as i64); + + for (slug, repo_name, local_path) in markers { + if !local_path.exists() { + // Stale litter (repo dir gone) — storage is the best remaining + // state; drop the marker. + clear_pending_upload(&local_path); + continue; + } + let base = + std::fs::read_to_string(pending_upload_marker(&local_path)).unwrap_or_default(); + let base = base.trim().to_string(); + // The base-vs-remote decision and the upload both happen inside + // `upload_locked_with_marker`, UNDER the advisory lock: an + // unlocked pre-check here could pass, then block on a concurrent + // push's lock for that push's whole duration, and the stale + // decision would clobber the push's freshly-uploaded archive. + match self + .upload_locked_with_marker(&slug, &repo_name, &local_path, &base) + .await + { + Ok(PendingUploadOutcome::Uploaded) => { + debug!(repo = %repo_name, "pending-upload retry: re-synced storage"); + reuploaded += 1; + } + Ok(PendingUploadOutcome::Diverged) => { + warn!(repo = %repo_name, + "pending-upload retry: storage diverged from marker base — \ + leaving marked; writes stay blocked pending manual reconciliation"); + still_pending += 1; + } Err(e) => { - warn!(repo = %repo_name, err = %e, "rejected unsafe path in release_after_write"); - return; + warn!(repo = %repo_name, err = %e, + "pending-upload retry: upload failed — will retry on next write"); + still_pending += 1; + } + } + } + (reuploaded, still_pending) + } + + /// Marker-protected upload: takes the per-repo advisory lock, re-checks + /// that storage still matches `base` UNDER the lock, and only then uploads, + /// updates the versions cache, and clears the marker — all before the lock + /// is released. + /// + /// Both halves of that ordering are load-bearing: + /// - The divergence check must run under the lock. An unlocked check can + /// pass just before a concurrent locked push advances storage (the check + /// then blocks on that push's lock), and blindly uploading afterwards + /// would clobber the acked push it lost the race to. + /// - The marker must be cleared before the lock is released, or a writer + /// queued on the lock could observe marker + fresh etag and fail with a + /// spurious "diverged — reconcile manually" on a consistent repo. + async fn upload_locked_with_marker( + &self, + owner_slug: &str, + repo_name: &str, + local_path: &Path, + base: &str, + ) -> Result { + let Some(ref archive) = self.archive else { + anyhow::bail!("upload_locked_with_marker called without a storage backend"); + }; + let lock_key = advisory_lock_key(owner_slug, repo_name); + let label = format!("{owner_slug}/{repo_name}"); + let lock = LockedConn::acquire(&self.lock_pool, lock_key, &label).await?; + + let outcome: Result = async { + let remote = archive + .head_etag(owner_slug, repo_name) + .await + .context("storage head under lock before pending upload")?; + if remote.as_deref().unwrap_or("") != base { + return Ok(PendingUploadOutcome::Diverged); + } + let etag = archive + .upload(owner_slug, repo_name, local_path) + .await + .context("uploading repo to storage under lock")?; + if let Some(ref etag) = etag { + self.versions + .lock() + .await + .insert(label.clone(), etag.clone()); + } + clear_pending_upload_after_success(local_path, etag.as_deref()); + Ok(PendingUploadOutcome::Uploaded) + } + .await; + + lock.unlock().await; + outcome + } + + /// Upload `local_path` to storage while holding the per-repo advisory lock, + /// so a background or init-time upload can't clobber a concurrent locked + /// write by landing an older snapshot after it. With `skip_if_exists`, skips + /// the upload when the archive is already present (used by lazy migration). + async fn upload_under_lock( + &self, + owner_slug: &str, + repo_name: &str, + local_path: &Path, + skip_if_exists: bool, + ) -> Result<()> { + let Some(ref archive) = self.archive else { + return Ok(()); + }; + let lock_key = advisory_lock_key(owner_slug, repo_name); + let label = format!("{owner_slug}/{repo_name}"); + let lock = LockedConn::acquire(&self.lock_pool, lock_key, &label).await?; + + let outcome: Result> = async { + if skip_if_exists { + // Propagate a failed existence check instead of treating it as + // "absent": HEAD failing transiently while PUT would succeed + // must not let this node's cache overwrite a newer shared + // archive. The lazy-migration caller just retries later. + let exists = archive + .exists(owner_slug, repo_name) + .await + .context("checking storage before migration upload")?; + if exists { + return Ok(None); // already present — nothing to upload } - }; - if let Err(e) = tigris.upload(&owner_slug, repo_name, &local_path).await { - warn!(repo = %repo_name, err = %e, "failed to upload repo to tigris after write"); } + archive.upload(owner_slug, repo_name, local_path).await + } + .await; + + // Release the lock on the same connection regardless of outcome. + lock.unlock().await; + + match outcome { + Ok(Some(etag)) => { + self.versions + .lock() + .await + .insert(format!("{owner_slug}/{repo_name}"), etag); + Ok(()) + } + Ok(None) => Ok(()), + Err(e) => Err(e).context("uploading repo to storage under lock"), } } @@ -347,15 +718,142 @@ fn validate_repo_name(repo_name: &str) -> Result<()> { Ok(()) } +/// How long to retry acquiring the per-repo advisory lock before giving up. +/// Matches the storage backends' total operation timeout (300s in `s3.rs` and +/// `ipfs.rs`): the writer holding the lock may legitimately be mid-upload of a +/// large archive, so a concurrent push must be willing to outwait the longest +/// possible upload rather than failing while the lock holder is still healthy. +pub(crate) const LOCK_ACQUIRE_TIMEOUT_SECS: u64 = 300; + +/// A pool connection pinned for the lifetime of a session-scoped advisory lock. +/// +/// Postgres advisory locks bind to one backend connection and only release on +/// that same connection; with a pool, acquiring and releasing on different +/// checked-out connections means the unlock silently no-ops while the lock +/// lingers on the original. So the lock's whole lifetime — acquire, use, +/// unlock — runs on this single pinned connection. +/// +/// `unlock()` is the graceful path: it releases the lock and returns the +/// connection to the pool. If the holder is instead *dropped* while the lock is +/// held — e.g. a detached write-back task cancelled by runtime shutdown mid +/// upload — `Drop` detaches the connection from the pool and closes it, which +/// ends the Postgres session and frees the lock server-side. The one thing this +/// type never does is return a still-locked connection to the pool. +struct LockedConn { + /// `None` once `unlock()` has run (or after a timed-out acquire hands the + /// never-locked connection back to the pool). + conn: Option>, + lock_key: i64, + repo_label: String, +} + +impl LockedConn { + /// Acquire `lock_key` on a freshly pinned connection, polling + /// `pg_try_advisory_lock` once per second up to [`LOCK_ACQUIRE_TIMEOUT_SECS`]. + /// Polling (rather than the blocking `pg_advisory_lock`) keeps a stale lock + /// from a crashed session from wedging writers indefinitely. + async fn acquire(pool: &PgPool, lock_key: i64, repo_label: &str) -> Result { + let conn = pool + .acquire() + .await + .context("acquiring db connection for advisory lock")?; + // Wrap the connection before the first lock attempt so cancellation at + // any point in the retry loop hits `Drop` and closes the connection — + // a lock granted just as the caller was cancelled can't strand. + let mut this = Self { + conn: Some(conn), + lock_key, + repo_label: repo_label.to_string(), + }; + for attempt in 0..LOCK_ACQUIRE_TIMEOUT_SECS { + let conn = this.conn.as_mut().expect("connection present until unlock"); + let row: (bool,) = match sqlx::query_as("SELECT pg_try_advisory_lock($1)") + .bind(lock_key) + .fetch_one(&mut **conn) + .await + { + Ok(row) => row, + Err(e) => { + // The poll itself failed, so the lock's server-side state + // is unknown: if the query executed but the response was + // lost, this session HOLDS the lock, and repooling the + // connection would strand it. Close the connection + // deliberately (freeing any lock it may hold) instead of + // going through Drop's generic dropped-holder warning. + if let Some(conn) = this.conn.take() { + drop(conn.detach()); + } + return Err(e).context("trying advisory lock"); + } + }; + if row.0 { + return Ok(this); + } + if attempt < LOCK_ACQUIRE_TIMEOUT_SECS - 1 { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } + // Timed out without ever holding the lock — return the connection to + // the pool normally rather than letting `Drop` close it. + drop(this.conn.take()); + anyhow::bail!( + "could not acquire advisory lock for {} after {LOCK_ACQUIRE_TIMEOUT_SECS}s — \ + possible stale lock or a long-running upload", + this.repo_label + ); + } + + /// Release the lock on the pinned connection and return it to the pool. + async fn unlock(mut self) { + if let Some(mut conn) = self.conn.take() { + if let Err(e) = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(self.lock_key) + .execute(&mut *conn) + .await + { + // The unlock failed, so the session may still hold the lock — + // close the connection (like `Drop`) instead of repooling it, + // or the lock would wedge this repo's writes until the pool + // happened to recycle that connection. + warn!(repo = %self.repo_label, err = %e, + "failed to release advisory lock — closing its connection"); + drop(conn.detach()); + } + } + } +} + +impl Drop for LockedConn { + fn drop(&mut self) { + if let Some(conn) = self.conn.take() { + // Dropped while the lock is (or may be) held: closing the socket + // ends the Postgres session, which releases every advisory lock it + // held. Detach first so the closed connection is never handed back + // to the pool; the pool replaces it on demand. + warn!(repo = %self.repo_label, + "advisory-lock holder dropped without unlock — closing its connection to free the lock"); + drop(conn.detach()); + } + } +} + /// Guard returned by `acquire_write()`. Holds the Postgres advisory lock and -/// uploads to Tigris + releases the lock on `release()`. +/// uploads to storage + releases the lock on `release()`. +/// +/// `#[must_use]`: dropping the guard without calling `release()` skips the +/// storage upload and force-closes the pinned lock connection to free the +/// advisory lock (see [`LockedConn`]) — safe, but never what a caller wants. +#[must_use = "call release() — dropping the guard skips the upload and force-closes the lock connection"] pub struct RepoWriteGuard { owner_slug: String, repo_name: String, pub local_path: PathBuf, - lock_key: i64, - pool: PgPool, - tigris: Option, + /// The pinned advisory-lock connection; freed on `release()`, or by + /// `LockedConn::drop` if the guard (or a write-back task driving it) is + /// dropped or cancelled mid-flight. + lock: LockedConn, + archive: Option, + versions: Arc>>, } impl RepoWriteGuard { @@ -364,41 +862,197 @@ impl RepoWriteGuard { &self.local_path } - /// Upload to Tigris (only when the write succeeded) and release the advisory + /// Durably record intent-to-upload NOW, before the caller acks the client. + /// Write-back callers must call this before spawning `release()` — the + /// spawned task may never be polled if the process stops right after the + /// ack, and without the marker already on disk a restart would treat the + /// stale storage archive as newer and roll the acked write back. On `Err` + /// the caller must NOT ack early; fall back to strict upload-before-ack. + /// Idempotent with the marker `release()` writes itself. No-op without a + /// storage backend (markers would be inert until a backend appears, then + /// wedge repos whose archives predate them). + pub async fn mark_pending(&self) -> Result<()> { + if self.archive.is_none() { + return Ok(()); + } + let key = format!("{}/{}", self.owner_slug, self.repo_name); + let base = self.versions.lock().await.get(&key).cloned(); + mark_pending_upload(&self.local_path, base.as_deref()) + } + + /// Upload to storage (only when the write succeeded) and release the advisory /// lock. Pass `success = false` when the write operation failed — uploading a /// half-applied or otherwise inconsistent repo would propagate corruption to - /// Tigris (and to every node that later downloads it). The lock is always + /// storage (and to every node that later downloads it). The lock is always /// released regardless, to avoid stale locks blocking future writes. - pub async fn release(self, success: bool) { - // Upload to Tigris only on success. - if success { - if let Some(ref tigris) = self.tigris { - if let Err(e) = tigris + /// + /// IMPORTANT: the advisory lock is held until the upload finishes, so a + /// concurrent writer on another machine cannot read a stale archive. When + /// callers want a fast client ack, they spawn this future as a background + /// task (write-back) — the lock + etag-cache update still complete in order. + pub async fn release(self, success: bool) -> Result<()> { + let key = format!("{}/{}", self.owner_slug, self.repo_name); + + // Upload to storage only on success. Capture the outcome so we can both + // release the lock unconditionally and propagate a durable-upload + // failure to the caller (a synchronous caller turns it into a client + // error; a write-back caller logs it). + let upload_result: Result<()> = if success { + if let Some(ref archive) = self.archive { + let base = self.versions.lock().await.get(&key).cloned(); + if let Err(e) = mark_pending_upload(&self.local_path, base.as_deref()) { + // Proceed with the upload anyway: if it succeeds, no marker + // is needed; if both fail, the error below reaches the + // caller (double-failure corner, same exposure as + // pre-marker behavior). + warn!(repo = %self.repo_name, err = %e, "failed to write pending-upload marker"); + } + match archive .upload(&self.owner_slug, &self.repo_name, &self.local_path) .await { - warn!(repo = %self.repo_name, err = %e, "failed to upload repo to tigris after write"); + Ok(Some(etag)) => { + self.versions.lock().await.insert(key.clone(), etag.clone()); + clear_pending_upload_after_success(&self.local_path, Some(&etag)); + Ok(()) + } + Ok(None) => { + clear_pending_upload(&self.local_path); + Ok(()) + } + Err(e) => { + // Storage is now behind local (this holds even for an + // already-acked write-back push). Drop the cached etag, + // and leave the pending-upload marker so the next + // access serves the local copy instead of rolling it + // back to the stale archive; the next successful + // upload re-syncs storage and clears the marker. + self.versions.lock().await.remove(&key); + Err(e).context("uploading repo to storage after write") + } } + } else { + Ok(()) } } else { - warn!(repo = %self.repo_name, "write failed — skipping tigris upload to avoid propagating an inconsistent repo"); - } + // Write failed: skip the upload (a half-applied repo must not reach + // storage) and invalidate the cached etag — the local copy may be + // dirty, so the next write must re-download instead of skipping on a + // now-misleading etag match. + warn!(repo = %self.repo_name, "write failed — skipping storage upload and invalidating etag cache"); + self.versions.lock().await.remove(&key); + Ok(()) + }; - // Release advisory lock - let _ = sqlx::query("SELECT pg_advisory_unlock($1)") - .bind(self.lock_key) - .execute(&self.pool) - .await; + // Release the advisory lock on the same connection it was taken on + // regardless of the upload outcome, then return it to the pool. + self.lock.unlock().await; + + upload_result + } +} + +/// Sibling marker file recording that `local_path` holds writes storage has +/// not received yet ("local is ahead"). Written before every post-write upload +/// and removed only when the upload succeeds, so it survives process death and +/// lets `sync_down_if_stale` distinguish "storage is ahead of local" (download) +/// from "local is ahead of storage" (never download — that would roll back an +/// acked write). Lives next to the repo dir, not inside it, so it is never +/// packed into the archive. +fn pending_upload_marker(local_path: &Path) -> PathBuf { + let name = local_path + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + local_path.with_file_name(format!(".{name}.pending-upload")) +} + +/// Persist the intent-to-upload marker. Fallible (write-back callers must NOT +/// ack the client if this fails) and atomic (tmp + rename, so a crash cannot +/// leave a torn marker). +/// +/// `base_etag` is the storage etag the local write was built on (empty when +/// storage held nothing). `sync_down_if_stale` compares it against the current +/// remote etag to distinguish "local strictly ahead" from cross-node +/// divergence. +/// +/// An existing marker is preserved untouched: its base is the last storage +/// etag this node confirmed, which stays correct for every further write +/// stacked on the same undiverged local copy. Re-marking would record the +/// CURRENT cache — emptied by the preceding upload failure — and a corrupted +/// (empty) base makes the next sync read unchanged storage as divergence, +/// wedging the repo's whole write surface after two consecutive upload +/// failures. +fn mark_pending_upload(local_path: &Path, base_etag: Option<&str>) -> Result<()> { + let marker = pending_upload_marker(local_path); + match marker.try_exists() { + Ok(true) => return Ok(()), // keep the original base + Ok(false) => {} + Err(e) => return Err(e).context("probing pending-upload marker"), + } + let tmp = marker.with_file_name(format!(".pending-upload.tmp-{}", uuid::Uuid::new_v4())); + std::fs::write(&tmp, base_etag.unwrap_or_default()).context("writing pending-upload marker")?; + std::fs::rename(&tmp, &marker) + .inspect_err(|_| { + let _ = std::fs::remove_file(&tmp); + }) + .context("publishing pending-upload marker")?; + crate::metrics::add_pending_upload_markers(1); + Ok(()) +} + +pub(crate) fn clear_pending_upload(local_path: &Path) { + if std::fs::remove_file(pending_upload_marker(local_path)).is_ok() { + crate::metrics::add_pending_upload_markers(-1); + } +} + +/// Remove the marker after a *successful* upload, first atomically rewriting +/// its base to the just-uploaded etag. A crash between the rewrite and the +/// unlink then reads as "local ahead, base matches" — which self-heals on the +/// next write or startup retry — instead of "base predates storage", which +/// would wedge the repo behind a spurious permanent divergence even though +/// local and storage are identical. +fn clear_pending_upload_after_success(local_path: &Path, new_etag: Option<&str>) { + let marker = pending_upload_marker(local_path); + if let Some(etag) = new_etag { + if marker.exists() { + let tmp = + marker.with_file_name(format!(".pending-upload.tmp-{}", uuid::Uuid::new_v4())); + if std::fs::write(&tmp, etag).is_ok() { + let _ = std::fs::rename(&tmp, &marker); + } else { + let _ = std::fs::remove_file(&tmp); + } + } + } + if std::fs::remove_file(&marker).is_ok() { + crate::metrics::add_pending_upload_markers(-1); } } +/// Outcome of a marker-protected upload attempt. +enum PendingUploadOutcome { + /// Uploaded, versions cache updated, marker cleared — all under the lock. + Uploaded, + /// Storage no longer matches the marker's base: another writer advanced it. + /// Nothing was uploaded and the marker was left in place. + Diverged, +} + /// Compute a stable i64 hash for a Postgres advisory lock key. +/// +/// SHA-256 prefix, NOT `DefaultHasher`: the default hasher's algorithm is +/// explicitly unspecified across Rust releases, and this lock is the sole +/// cross-machine write serializer — two machines built with different +/// toolchains hashing the same repo to different keys would silently stop +/// excluding each other mid rolling deploy. (Changing the scheme is itself a +/// one-deploy exclusion gap between old and new binaries; accepted once, +/// here, to get onto a stable function.) fn advisory_lock_key(owner_slug: &str, repo_name: &str) -> i64 { - use std::hash::{Hash, Hasher}; - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - owner_slug.hash(&mut hasher); - repo_name.hash(&mut hasher); - hasher.finish() as i64 + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(format!("{owner_slug}/{repo_name}").as_bytes()); + i64::from_be_bytes(digest[..8].try_into().expect("digest has at least 8 bytes")) } #[cfg(test)] @@ -562,4 +1216,548 @@ mod tests { ); } } + + // ── sync_down_if_stale (fs-backed archive, lazy pool) ────────────────── + + /// A RepoStore over an fs-backed archive. `sync_down_if_stale` never touches + /// the pool, so a lazy (never-connected) pool is fine. + fn store_with_fs_archive(repos_dir: PathBuf, store_root: &Path) -> RepoStore { + let blob: Arc = + Arc::new(crate::storage::fs::FsBlobStore::new(store_root).unwrap()); + let archive = crate::storage::archive::RepoArchive::new(blob); + let pool = sqlx::PgPool::connect_lazy("postgres://invalid").unwrap(); + RepoStore::new(repos_dir, Some(archive), pool) + } + + #[tokio::test] + async fn sync_down_if_stale_downloads_then_skips_on_etag_match() { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let store = store_with_fs_archive(repos_dir.path().to_path_buf(), store_root.path()); + + // Seed the archive with a repo. + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + + let local = repos_dir.path().join("owner").join("repo.git"); + + // First call downloads. + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .unwrap(); + assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"v1\n"); + + // Locally mutate, then sync again: the cached etag still matches the + // remote, so the download is skipped and our local edit survives. + std::fs::write(local.join("HEAD"), b"LOCAL-EDIT\n").unwrap(); + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .unwrap(); + assert_eq!( + std::fs::read(local.join("HEAD")).unwrap(), + b"LOCAL-EDIT\n", + "etag match must skip the download (local copy preserved)" + ); + } + + // Needs a real pool: `release_after_write` uploads under the advisory lock. + #[sqlx::test] + async fn pending_marker_prevents_rollback_and_clears_on_next_upload(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let blob: Arc = + Arc::new(crate::storage::fs::FsBlobStore::new(store_root.path()).unwrap()); + let store = RepoStore::new( + repos_dir.path().to_path_buf(), + Some(crate::storage::archive::RepoArchive::new(blob)), + pool, + ); + + // Storage holds v1; local downloads it. + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + let local = repos_dir.path().join("owner").join("repo.git"); + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .unwrap(); + + // Simulate an acked write whose upload failed: local advances, the + // pending marker (recording the storage etag the write was based on) + // persists, and the in-memory cache was invalidated on failure. + let base = store + .archive + .as_ref() + .unwrap() + .head_etag("owner", "repo") + .await + .unwrap() + .unwrap(); + std::fs::write(local.join("HEAD"), b"ACKED-WRITE\n").unwrap(); + mark_pending_upload(&local, Some(&base)).unwrap(); + store.versions.lock().await.clear(); + + // Both the read and the write path must serve local, not roll it back. + for require_fresh in [false, true] { + store + .sync_down_if_stale("owner", "repo", &local, require_fresh) + .await + .unwrap(); + assert_eq!( + std::fs::read(local.join("HEAD")).unwrap(), + b"ACKED-WRITE\n", + "pending marker must prevent rollback (require_fresh={require_fresh})" + ); + } + + // The next successful upload re-syncs storage and clears the marker. + store.release_after_write("owner", "repo").await.unwrap(); + assert!( + !pending_upload_marker(&local).exists(), + "marker must be cleared by a successful upload" + ); + let out = tempfile::tempdir().unwrap(); + let restored = out.path().join("restored.git"); + store + .archive + .as_ref() + .unwrap() + .download("owner", "repo", &restored) + .await + .unwrap(); + assert_eq!( + std::fs::read(restored.join("HEAD")).unwrap(), + b"ACKED-WRITE\n" + ); + } + + #[tokio::test] + async fn pending_marker_detects_cross_node_divergence() { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let store = store_with_fs_archive(repos_dir.path().to_path_buf(), store_root.path()); + + // Storage v1; local synced, then advanced with a failed upload (marker + // records v1's etag as its base). + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + let archive = store.archive.as_ref().unwrap(); + archive.upload("owner", "repo", seed.path()).await.unwrap(); + let local = repos_dir.path().join("owner").join("repo.git"); + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .unwrap(); + let base = archive.head_etag("owner", "repo").await.unwrap().unwrap(); + std::fs::write(local.join("HEAD"), b"LOCAL-AHEAD\n").unwrap(); + mark_pending_upload(&local, Some(&base)).unwrap(); + store.versions.lock().await.clear(); + + // Another node advances storage past our base. + let seed2 = tempfile::tempdir().unwrap(); + std::fs::write(seed2.path().join("HEAD"), b"OTHER-NODE\n").unwrap(); + archive.upload("owner", "repo", seed2.path()).await.unwrap(); + + // Write path: refuse — proceeding would clobber one side or the other. + assert!( + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .is_err(), + "diverged marker must fail the write path closed" + ); + // Read path: serve local (read-only cannot propagate damage), and the + // local copy must be untouched either way. + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .unwrap(); + assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"LOCAL-AHEAD\n"); + } + + #[tokio::test] + async fn stale_pending_marker_without_local_copy_is_dropped() { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let store = store_with_fs_archive(repos_dir.path().to_path_buf(), store_root.path()); + + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + + // Marker exists but the repo dir does not (removed out from under us): + // the marker is stale — drop it and download normally. + let local = repos_dir.path().join("owner").join("repo.git"); + std::fs::create_dir_all(local.parent().unwrap()).unwrap(); + mark_pending_upload(&local, Some("whatever")).unwrap(); + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .unwrap(); + assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"v1\n"); + assert!(!pending_upload_marker(&local).exists()); + } + + #[tokio::test] + async fn sync_down_if_stale_require_fresh_fails_closed_on_bad_remote() { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let store = store_with_fs_archive(repos_dir.path().to_path_buf(), store_root.path()); + + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + + let local = repos_dir.path().join("owner").join("repo.git"); + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .unwrap(); + + // Corrupt the stored archive: HEAD now succeeds with a *new* etag (so the + // cache no longer matches and a download is forced), but the download + // decompresses garbage and fails. + let blob_path = store_root.path().join("repos/v1/owner/repo.tar.zst"); + std::fs::write(&blob_path, b"corrupted not-a-tar-zst").unwrap(); + // The fs backend's etag lives in a sidecar, so a direct file overwrite + // must also bump it for the change to be visible (as any real writer's + // put() would). + std::fs::write( + store_root.path().join("repos/v1/owner/repo.tar.zst.etag"), + "corrupted-generation", + ) + .unwrap(); + + // Write path: must fail closed rather than fall back to the stale local + // copy (which a later upload would use to clobber the newer remote). + assert!( + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .is_err(), + "require_fresh=true must propagate the download error" + ); + + // Read path: self-heals — falls back to the valid local copy. + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .expect("require_fresh=false must fall back to the local copy"); + assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"v1\n"); + } + + // ── failing-store double: exercises error branches no real backend can ── + + /// BlobStore wrapper whose `put`/`head` can be flipped to fail, unlocking + /// deterministic coverage of the upload-failure and head-failure branches. + struct FlakyStore { + inner: crate::storage::fs::FsBlobStore, + fail_put: std::sync::atomic::AtomicBool, + fail_head: std::sync::atomic::AtomicBool, + } + + impl FlakyStore { + fn new(root: &Path) -> Arc { + Arc::new(Self { + inner: crate::storage::fs::FsBlobStore::new(root).unwrap(), + fail_put: std::sync::atomic::AtomicBool::new(false), + fail_head: std::sync::atomic::AtomicBool::new(false), + }) + } + } + + #[async_trait::async_trait] + impl crate::storage::BlobStore for FlakyStore { + fn backend_name(&self) -> &'static str { + "flaky" + } + async fn get(&self, key: &str) -> Result> { + self.inner.get(key).await + } + async fn put(&self, key: &str, body: bytes::Bytes) -> Result { + if self.fail_put.load(std::sync::atomic::Ordering::Relaxed) { + anyhow::bail!("injected put failure"); + } + self.inner.put(key, body).await + } + async fn head(&self, key: &str) -> Result> { + if self.fail_head.load(std::sync::atomic::Ordering::Relaxed) { + anyhow::bail!("injected head failure"); + } + self.inner.head(key).await + } + async fn delete(&self, key: &str) -> Result<()> { + self.inner.delete(key).await + } + } + + fn store_with_flaky(repos_dir: PathBuf, flaky: Arc, pool: PgPool) -> RepoStore { + let blob: Arc = flaky; + let archive = crate::storage::archive::RepoArchive::new(blob); + RepoStore::new(repos_dir, Some(archive), pool) + } + + /// beardthelion P1 regression: two consecutive failed uploads must not + /// corrupt the marker's base — the second `release` re-marks while the + /// versions cache is empty, and overwriting the base with "" would make + /// the third write read unchanged storage as divergence and wedge the + /// repo's entire write surface. + #[sqlx::test] + async fn two_consecutive_failed_uploads_preserve_marker_base(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + + // Storage v1, synced down. + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + let local = repos_dir.path().join("owner").join("repo.git"); + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .unwrap(); + + // Write 1: mutate, upload fails. + flaky + .fail_put + .store(true, std::sync::atomic::Ordering::Relaxed); + std::fs::write(local.join("HEAD"), b"write-1\n").unwrap(); + let guard = store.acquire_write("owner", "repo").await.unwrap(); + assert!(guard.release(true).await.is_err(), "injected put must fail"); + let base_after_first = std::fs::read_to_string(pending_upload_marker(&local)).unwrap(); + assert!(!base_after_first.trim().is_empty(), "base must be recorded"); + + // Write 2: acquire must succeed (storage unchanged == local-ahead), + // and the second failed release must NOT re-mark with an empty base. + let guard = store.acquire_write("owner", "repo").await.unwrap(); + std::fs::write(local.join("HEAD"), b"write-2\n").unwrap(); + assert!(guard.release(true).await.is_err()); + assert_eq!( + std::fs::read_to_string(pending_upload_marker(&local)).unwrap(), + base_after_first, + "an existing marker's base must be preserved on re-mark" + ); + + // Write 3: still not wedged — and once the store heals, everything + // re-syncs and the marker clears. + let guard = store + .acquire_write("owner", "repo") + .await + .expect("repeated upload failures must not wedge the write surface"); + flaky + .fail_put + .store(false, std::sync::atomic::Ordering::Relaxed); + guard.release(true).await.unwrap(); + assert!(!pending_upload_marker(&local).exists()); + } + + /// jatmn P1 regression: the write-back ack window. `mark_pending` runs + /// before the ack; if the process dies before the spawned release is ever + /// polled (simulated by dropping the guard), the marker alone must keep + /// the next sync from rolling the acked write back. + #[sqlx::test] + async fn mark_pending_alone_protects_the_ack_window(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + store + .archive + .as_ref() + .unwrap() + .upload("owner", "repo", seed.path()) + .await + .unwrap(); + let local = repos_dir.path().join("owner").join("repo.git"); + + let guard = store.acquire_write("owner", "repo").await.unwrap(); + std::fs::write(local.join("HEAD"), b"ACKED\n").unwrap(); + guard.mark_pending().await.unwrap(); + drop(guard); // crash before release() is ever polled + + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .unwrap(); + assert_eq!( + std::fs::read(local.join("HEAD")).unwrap(), + b"ACKED\n", + "the pre-ack marker alone must prevent rollback" + ); + } + + /// The lazy-migration existence check must propagate failure instead of + /// reading it as "absent" and uploading over a possibly-newer archive. + #[sqlx::test] + async fn upload_under_lock_propagates_failed_existence_check(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + + let local = repos_dir.path().join("owner").join("repo.git"); + std::fs::create_dir_all(&local).unwrap(); + std::fs::write(local.join("HEAD"), b"local\n").unwrap(); + + flaky + .fail_head + .store(true, std::sync::atomic::Ordering::Relaxed); + assert!( + store + .upload_under_lock("owner", "repo", &local, true) + .await + .is_err(), + "a failed existence check must not read as absent" + ); + assert!( + store + .archive + .as_ref() + .unwrap() + .head_etag("owner", "repo") + .await + .is_err(), + "sanity: head still failing" + ); + } + + /// init() must remove its local dir when the initial upload fails, so a + /// retry of the same name doesn't hit an existing destination. + #[sqlx::test] + async fn init_removes_local_dir_when_upload_fails(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + + flaky + .fail_put + .store(true, std::sync::atomic::Ordering::Relaxed); + assert!(store.init("did:key:z6MkOwner", "newrepo").await.is_err()); + let local = repos_dir + .path() + .join("did_key_z6MkOwner") + .join("newrepo.git"); + assert!( + !local.exists(), + "failed init must not leave a local dir behind" + ); + } + + /// Marker + head failure: the write path fails closed, the read path + /// serves the local copy. + #[tokio::test] + async fn marker_with_failing_head_fails_write_closed_serves_read() { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + // sync_down never touches the pool — lazy is fine here. + let pool = sqlx::PgPool::connect_lazy("postgres://invalid").unwrap(); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + + let local = repos_dir.path().join("owner").join("repo.git"); + std::fs::create_dir_all(&local).unwrap(); + std::fs::write(local.join("HEAD"), b"pending\n").unwrap(); + mark_pending_upload(&local, Some("base-etag")).unwrap(); + + flaky + .fail_head + .store(true, std::sync::atomic::Ordering::Relaxed); + assert!( + store + .sync_down_if_stale("owner", "repo", &local, true) + .await + .is_err(), + "write path must fail closed when freshness is unknowable" + ); + store + .sync_down_if_stale("owner", "repo", &local, false) + .await + .expect("read path serves the local copy"); + assert_eq!(std::fs::read(local.join("HEAD")).unwrap(), b"pending\n"); + } + + /// Startup sweep: re-uploads marked repos whose storage didn't move, and + /// leaves diverged ones marked. + #[sqlx::test] + async fn retry_pending_uploads_heals_and_respects_divergence(pool: PgPool) { + let store_root = tempfile::tempdir().unwrap(); + let repos_dir = tempfile::tempdir().unwrap(); + let flaky = FlakyStore::new(store_root.path()); + let store = store_with_flaky(repos_dir.path().to_path_buf(), Arc::clone(&flaky), pool); + let archive = store.archive.as_ref().unwrap(); + + // Repo A: storage v1, local ahead with matching base — heals. + let seed = tempfile::tempdir().unwrap(); + std::fs::write(seed.path().join("HEAD"), b"v1\n").unwrap(); + archive.upload("owner", "heals", seed.path()).await.unwrap(); + let base_a = archive.head_etag("owner", "heals").await.unwrap().unwrap(); + let local_a = repos_dir.path().join("owner").join("heals.git"); + std::fs::create_dir_all(&local_a).unwrap(); + std::fs::write(local_a.join("HEAD"), b"local-ahead\n").unwrap(); + mark_pending_upload(&local_a, Some(&base_a)).unwrap(); + + // Repo B: marker base predates current storage — stays marked. + archive + .upload("owner", "diverged", seed.path()) + .await + .unwrap(); + let local_b = repos_dir.path().join("owner").join("diverged.git"); + std::fs::create_dir_all(&local_b).unwrap(); + std::fs::write(local_b.join("HEAD"), b"local-b\n").unwrap(); + mark_pending_upload(&local_b, Some("stale-base")).unwrap(); + + let (reuploaded, still_pending) = store.retry_pending_uploads().await; + assert_eq!((reuploaded, still_pending), (1, 1)); + assert!(!pending_upload_marker(&local_a).exists()); + assert!(pending_upload_marker(&local_b).exists()); + + // A's local content is now durably in storage. + let out = tempfile::tempdir().unwrap(); + let restored = out.path().join("restored.git"); + archive.download("owner", "heals", &restored).await.unwrap(); + assert_eq!( + std::fs::read(restored.join("HEAD")).unwrap(), + b"local-ahead\n" + ); + } } diff --git a/crates/gitlawb-node/src/git/tigris.rs b/crates/gitlawb-node/src/git/tigris.rs deleted file mode 100644 index ad26ddc5..00000000 --- a/crates/gitlawb-node/src/git/tigris.rs +++ /dev/null @@ -1,226 +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 { - 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(), - }) - } - - /// 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 { - 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> { - 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> { - // KNOWN LIMITATION: this map is never evicted — one (PathBuf, Arc) - // 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>>>> = 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(()) -} diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index aa0483db..4484f8f5 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/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; @@ -262,25 +263,59 @@ 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 - } + // 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); + + // Dedicated, bounded pool for advisory-lock connections. A push pins one + // connection for its whole lifetime (lock held across receive-pack + + // upload), so giving locks their own budget keeps a burst of concurrent or + // slow pushes from draining the main pool and stalling every other DB + // handler node-wide. Sized independently of the handler pool + // (GITLAWB_ADVISORY_LOCK_POOL_SIZE). acquire_timeout matches the advisory + // lock's own wait budget: with the default 30s, a burst that pins every + // connection would fail waiters at 30s even though the lock path is + // prepared to outwait a full 300s storage upload. + let lock_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(config.advisory_lock_pool_size) + .acquire_timeout(std::time::Duration::from_secs( + git::repo_store::LOCK_ACQUIRE_TIMEOUT_SECS, + )) + .connect(&config.database_url) + .await + .context("creating advisory-lock connection 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"); } - } else { - info!("tigris storage disabled (no bucket configured)"); - None - }; + } - let repo_store = - git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, db.pool().clone()); + 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. diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index a21c6d8b..0caf74c3 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/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 = OnceLock::new(); static WEBHOOK_DELIVERIES: OnceLock = OnceLock::new(); static PACK_SIZE: OnceLock = OnceLock::new(); static PEERS_CONNECTED: OnceLock = OnceLock::new(); +static PENDING_UPLOAD_MARKERS: OnceLock = OnceLock::new(); /// One-time initializer. Builds the registry, registers every metric, /// and sets the constant `gitlawb_info` gauge. Idempotent — calling @@ -197,6 +200,18 @@ pub fn init(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)"); @@ -264,6 +279,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 { diff --git a/crates/gitlawb-node/src/state.rs b/crates/gitlawb-node/src/state.rs index d3e53f3a..40f509b4 100644 --- a/crates/gitlawb-node/src/state.rs +++ b/crates/gitlawb-node/src/state.rs @@ -48,7 +48,7 @@ pub struct AppState { pub graphql_schema: Arc, /// Fly.io machine ID — used for fly-replay routing in multi-machine deployments pub machine_id: Option, - /// 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, diff --git a/crates/gitlawb-node/src/storage/archive.rs b/crates/gitlawb-node/src/storage/archive.rs new file mode 100644 index 00000000..5d319657 --- /dev/null +++ b/crates/gitlawb-node/src/storage/archive.rs @@ -0,0 +1,389 @@ +//! 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, +} + +impl RepoArchive { + pub fn new(store: Arc) -> 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> { + 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 { + 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> { + 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")?; + + 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(()) + } + + /// 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 + } +} + +/// 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> { + 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> { + // KNOWN LIMITATION: this map is never evicted — one (PathBuf, Arc) + // 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>>>> = 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")?; + + 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() { + let b = parent.join(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 = + 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"); + } +} diff --git a/crates/gitlawb-node/src/storage/fs.rs b/crates/gitlawb-node/src/storage/fs.rs new file mode 100644 index 00000000..a7504636 --- /dev/null +++ b/crates/gitlawb-node/src/storage/fs.rs @@ -0,0 +1,346 @@ +//! 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 a fresh UUID persisted in a `.etag` +//! sidecar on every write, so the skip-redundant-download optimization can rely +//! on "etag unchanged ⇒ content unchanged" even on filesystems with coarse +//! timestamps (objects predating the sidecar fall back to size-mtime). + +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) -> Result { + 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 { + 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: a fresh UUID per `put`. + /// RepoStore treats etag equality as proof the local copy is current, so + /// the token must change on EVERY write. A `size-mtime` fingerprint cannot + /// guarantee that on mounted filesystems with coarse timestamp precision + /// (two same-size writes in one tick collide); a per-write UUID can. + 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>> = 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> { + 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 { + 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 { + std::fs::create_dir_all(&parent).context("creating blob parent dir")?; + let etag = uuid::Uuid::new_v4().to_string(); + let sidecar = Self::sidecar_of(&path2); + let sidecar_tmp = sidecar.with_extension(format!("{}.tmp-put", uuid::Uuid::new_v4())); + 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); + 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> { + 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> { + let root = self.root.clone(); + let prefix = prefix.to_string(); + tokio::task::spawn_blocking(move || -> Result> { + 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 rewrite_always_changes_etag_even_for_identical_content() { + 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"); + + // Two writes of identical content back-to-back: a size-mtime + // fingerprint can collide inside one timestamp tick on coarse + // filesystems; the persisted per-write etag must always differ. + let m1 = store.put(key, body.clone()).await.unwrap(); + let m2 = store.put(key, body).await.unwrap(); + assert_ne!(m1.etag, m2.etag, "every put must produce a new etag"); + + // head() reports the latest persisted etag, and it really is the + // sidecar's content — not a size-mtime fingerprint that would also + // pass the assert_ne above when consecutive writes get distinct + // mtimes. + let h = store.head(key).await.unwrap().unwrap(); + assert_eq!(h.etag, m2.etag); + let sidecar = dir.path().join(format!("{key}.etag")); + assert_eq!( + std::fs::read_to_string(&sidecar).unwrap(), + m2.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 = store + .list("repos/v1/") + .await + .unwrap() + .into_iter() + .filter(|k| k.contains("tmp-put")) + .collect(); + assert!(leftovers.is_empty(), "leaked temp files: {leftovers:?}"); + } +} diff --git a/crates/gitlawb-node/src/storage/ipfs.rs b/crates/gitlawb-node/src/storage/ipfs.rs new file mode 100644 index 00000000..c99af1fc --- /dev/null +++ b/crates/gitlawb-node/src/storage/ipfs.rs @@ -0,0 +1,166 @@ +//! 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. + +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 { + // 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/` (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> { + 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 { + 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> { + 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}") + } + } + } +} diff --git a/crates/gitlawb-node/src/storage/mod.rs b/crates/gitlawb-node/src/storage/mod.rs new file mode 100644 index 00000000..d9c47e27 --- /dev/null +++ b/crates/gitlawb-node/src/storage/mod.rs @@ -0,0 +1,156 @@ +//! 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 a persisted per-write UUID for the filesystem +/// backend). +#[derive(Debug, Clone)] +pub struct ObjectMeta { + pub size: u64, + pub etag: Option, +} + +/// 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>; + + /// Store an object, returning its metadata (including the new etag). + async fn put(&self, key: &str, body: Bytes) -> Result; + + /// Fetch object metadata without the body. Returns `None` if absent. + async fn head(&self, key: &str) -> Result>; + + /// 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>> { + 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)) + } + "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)) + } + "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")?; + info!(api = %config.ipfs_api, backend = "ipfs", "object storage enabled"); + Ok(Some(Arc::new(s) as Arc)) + } + 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(()) +} diff --git a/crates/gitlawb-node/src/storage/s3.rs b/crates/gitlawb-node/src/storage/s3.rs new file mode 100644 index 00000000..d3310f73 --- /dev/null +++ b/crates/gitlawb-node/src/storage/s3.rs @@ -0,0 +1,146 @@ +//! 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, + force_path_style: bool, + ) -> Result { + 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(), + }) + } +} + +#[async_trait] +impl BlobStore for S3BlobStore { + fn backend_name(&self) -> &'static str { + "s3" + } + + async fn get(&self, key: &str) -> Result> { + 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 { + 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> { + 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(()) + } +}