diff --git a/.env.example b/.env.example index bbd9a34..cbf3db3 100644 --- a/.env.example +++ b/.env.example @@ -109,10 +109,32 @@ GITLAWB_MAX_PACK_BYTES=2147483648 # Max seconds a served git upload-pack / receive-pack (clone / push) may run # before it is aborted with a 504. Bounds a hung git that would otherwise pin a -# worker and, on push, the repo write lock. Does NOT cover the info/refs -# advertisement or the withheld-blob path, which remain unbounded. Default 600. +# worker and, on push, the repo write lock. Also bounds the info/refs +# advertisement and the withheld-blob pack build (#174). Default 600. GITLAWB_GIT_SERVICE_TIMEOUT_SECS=600 +# Max concurrent git read ops (upload-pack + info/refs advertisements) served at +# once, a global pool separate from the push pool below. Over-cap sheds a clean +# 503 + Retry-After. Anonymous reads draw from here, so pair it with +# GITLAWB_MAX_CONCURRENT_READS_PER_CALLER (below) so one caller cannot monopolize +# the pool. Default 128. +GITLAWB_MAX_CONCURRENT_GIT_OPS=128 + +# Max concurrent git-receive-pack (push) operations, in a pool separate from the +# read pool (GITLAWB_MAX_CONCURRENT_GIT_OPS) so anonymous reads cannot shed an +# authenticated push at admission. Over-cap sheds a 503 + Retry-After. Default 32. +GITLAWB_MAX_CONCURRENT_GIT_PUSHES=32 + +# Max concurrent read ops (upload-pack + info/refs advertisements) a single caller +# may hold, so one caller cannot monopolize the read pool. Keyed per-DID when +# signed, else per-source-IP. The per-IP key is only as granular as +# GITLAWB_TRUSTED_PROXY below: left unset, a node behind an edge/NAT keys all +# anonymous callers on the edge IP and this collapses to one global anonymous cap. +# Set GITLAWB_TRUSTED_PROXY for per-client keying; a high-fanout caller (CI behind +# one NAT) should authenticate for a per-DID budget or the operator raises this. +# Default 16. +GITLAWB_MAX_CONCURRENT_READS_PER_CALLER=16 + # ── Push rate limiting (git-receive-pack flood brake) ───────────────────── # Max receive-pack requests (info/refs advertisement + push POST) per client # IP per hour. 0 disables. Default 600. diff --git a/README.md b/README.md index 57ce288..0f76044 100644 --- a/README.md +++ b/README.md @@ -343,7 +343,7 @@ Important node settings: | `GITLAWB_REQUIRE_SIGNED_PEER_WRITES` | Require signed peer announce/sync writes. | | `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_GIT_SERVICE_TIMEOUT_SECS` | Max seconds a served git upload-pack, receive-pack, or `info/refs` advertisement may run before it is aborted (504). Default 600. Also bounds the withheld-blob classification walk (on both the upload-pack serve and receive-pack replication paths), which is reaped at the deadline. | | `GITLAWB_TIGRIS_BUCKET` | Optional S3/Tigris shared repo storage bucket. | | `GITLAWB_PINATA_JWT` | Optional Pinata/IPFS warm-storage pinning. | | `GITLAWB_IRYS_URL` | Optional Irys/Arweave permanent anchoring. | diff --git a/crates/gitlawb-node/src/api/ipfs.rs b/crates/gitlawb-node/src/api/ipfs.rs index f3de757..e921c39 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -27,7 +27,7 @@ use std::str::FromStr; use crate::auth::AuthenticatedDid; use crate::error::{AppError, Result}; use crate::git::store; -use crate::git::visibility_pack::{allowed_blob_set_for_caller, has_path_scoped_rule}; +use crate::git::visibility_pack::{allowed_blob_set_for_caller_bounded, has_path_scoped_rule}; use crate::state::AppState; use crate::visibility::{visibility_check, Decision}; @@ -142,10 +142,16 @@ pub async fn get_by_cid( let is_public = repo.is_public; let owner = repo.owner_did.clone(); let caller_for_walk = caller_owned.clone(); - // Full-history walk shells out to git — keep it off the async runtime. + let git_bin = state.git_bin.clone(); + let walk_timeout = + std::time::Duration::from_secs(state.config.git_service_timeout_secs); + // Full-history walk shells out to git — keep it off the async runtime, + // bounded and reaped like the served-git ops (#174). let walk = tokio::task::spawn_blocking(move || { - allowed_blob_set_for_caller( + allowed_blob_set_for_caller_bounded( &rp, + &git_bin, + walk_timeout, &r, is_public, &owner, diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b9cbc35..85eaa22 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -44,6 +44,8 @@ async fn replication_withheld_set( owner_did: &str, is_public: bool, disk_path: std::path::PathBuf, + git_bin: String, + timeout: std::time::Duration, ) -> (bool, Option>) { let announce = match &rules { Some(rules) => crate::visibility::listable_at_root(rules, is_public, owner_did, None), @@ -65,8 +67,8 @@ async fn replication_withheld_set( Some(rules) => { let owner_did = owner_did.to_string(); tokio::task::spawn_blocking(move || { - crate::git::visibility_pack::withheld_blob_oids( - &disk_path, &rules, is_public, &owner_did, None, + crate::git::visibility_pack::withheld_blob_oids_bounded( + &disk_path, &git_bin, timeout, &rules, is_public, &owner_did, None, ) }) .await @@ -106,10 +108,12 @@ async fn fail_closed_full_scan_objects( is_public: bool, owner_did: String, candidates: Vec, + git_bin: String, + timeout: std::time::Duration, ) -> Vec { tokio::task::spawn_blocking(move || -> anyhow::Result> { - let allowed = crate::git::visibility_pack::replicable_blob_set( - &disk_path, &rules, is_public, &owner_did, + let allowed = crate::git::visibility_pack::replicable_blob_set_bounded( + &disk_path, &git_bin, timeout, &rules, is_public, &owner_did, )?; let all_blobs = crate::git::push_delta::all_blob_oids(&disk_path)?; Ok(crate::git::visibility_pack::replicable_objects_fail_closed( @@ -508,6 +512,30 @@ pub async fn git_info_refs( auth: Option>, ) -> Result { let name = smart_http_repo_name(&repo)?; + let service = query + .service + .ok_or_else(|| AppError::BadRequest("missing ?service= parameter".into()))?; + // #62 cheap pre-DB load shed: if the pool this service draws from is already + // saturated, shed with 503 before any DB/disk work. Best-effort (holds no + // permit); the authoritative hold is `git_permit` below, after the per-source + // cap. Restores the shed-before-DB property the reordered held acquire alone + // would drop, while the reorder still prevents one source from occupying global + // slots during the DB/visibility window. + { + let pool = if service == "git-receive-pack" { + &state.git_write_semaphore + } else { + &state.git_read_semaphore + }; + if pool.available_permits() == 0 { + tracing::warn!( + "served-git concurrency cap reached; shedding request with 503 (pre-DB)" + ); + return Err(AppError::Overloaded( + "git service at capacity, retry shortly".into(), + )); + } + } tracing::info!(owner = %owner, repo = %name, "info/refs request"); let record = state .db @@ -521,9 +549,6 @@ pub async fn git_info_refs( return Err(AppError::RepoNotFound(format!("{owner}/{name}"))); } - let service = query - .service - .ok_or_else(|| AppError::BadRequest("missing ?service= parameter".into()))?; tracing::debug!(service = %service, repo = %name, "info/refs service"); // Enforce read visibility on the ref advertisement, for BOTH services. The @@ -563,6 +588,48 @@ pub async fn git_info_refs( } } + // Per-source concurrency sub-cap (#174), keyed on the resolved source IP and + // acquired AFTER the visibility + push-rate gates (KTD7) so a denied or + // rate-limited request never consumes a slot; held for the whole op. The + // upload-pack advertisement is bounded on the read pool (git_read_per_caller). + // The receive-pack advertisement draws from the write pool, so it is bounded per + // source (git_push_advert_per_caller) instead: without this, an anonymous + // multi-source flood of push-handshake advertisements could hold the write pool's + // slots across acquire_fresh and shed authenticated pushes, since the per-IP push + // rate limiter caps rate, not concurrency (#174 review fix). + let caller_key = read_caller_key(&headers, peer, state.push_limiter_trust); + let _caller_permit = if service == "git-receive-pack" { + acquire_read_caller_permit( + &state.git_push_advert_per_caller, + caller_key.as_deref(), + name, + "receive-pack advert", + )? + } else { + acquire_read_caller_permit( + &state.git_read_per_caller, + caller_key.as_deref(), + name, + "info/refs", + )? + }; + + // Shed with a 503 before spawning git when the concurrency cap is saturated; + // held for the whole op (incl. the smart_http call), released on return. Taken + // AFTER the per-source cap above so one source cannot occupy global slots it + // would be sub-cap-denied for during the DB/visibility window and starve other + // sources; still before acquire_fresh/git so it bounds the fresh Tigris acquire + // and git exec (INV-10). The receive-pack advertisement is phase one of a push, + // so it draws from the WRITE pool (like the git-receive-pack POST), not the + // global read pool an anonymous clone flood can exhaust and thereby starve the + // push handshake (#174 U2). The upload-pack advertisement stays on the read + // pool with its per-caller sub-cap. + let _permit = if service == "git-receive-pack" { + git_permit(&state.git_write_semaphore)? + } else { + git_permit(&state.git_read_semaphore)? + }; + // For receive-pack (push), download the latest from Tigris so the client // sees the same refs that acquire_write() will operate on. let disk_path = if service == "git-receive-pack" { @@ -581,14 +648,80 @@ pub async fn git_info_refs( AppError::Git(e.to_string()) })?; - smart_http::info_refs(&disk_path, &service) + let git_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs); + smart_http::info_refs("git", &service, &disk_path, git_timeout) .await .map_err(|e| { - tracing::error!(repo = %name, service = %service, err = %e, "info_refs git failed"); - AppError::Git(e.to_string()) + let app = git_service_app_error(&e); + match &app { + AppError::Timeout(_) => { + tracing::warn!(repo = %name, service = %service, "info/refs advertisement timed out") + } + _ => { + tracing::error!(repo = %name, service = %service, err = %e, "info_refs git failed") + } + } + app }) } +/// Acquire a permit from the served-git concurrency semaphore, or shed the +/// request with a 503 + Retry-After when every slot is in use. Bind the returned +/// permit to a named local so it is held for the whole git op (it releases on +/// drop); a bare `_` would release it immediately. +fn git_permit( + sem: &std::sync::Arc, +) -> Result { + sem.clone().try_acquire_owned().map_err(|_| { + // Surface the shed so operators can see the cap engaging, mirroring the + // receive-pack rate-limit warn above. A silent 503 makes a saturated or + // misconfigured cap look like a client problem instead of a capacity one. + tracing::warn!("served-git concurrency cap reached; shedding request with 503"); + AppError::Overloaded("git service at capacity, retry shortly".into()) + }) +} + +/// Resolve the per-caller key for the read sub-cap (#174): always the resolved +/// source IP (`client_key`), never the signed DID. Public read routes accept any +/// valid `did:key` via `optional_signature` with no admission step, so keying on +/// the DID would let one host mint disposable DIDs to multiply its per-source +/// budget; the push path already throttles on the resolved source IP for exactly +/// this DID-farm reason (`rate_limit.rs`, `IpRateLimiter`). `None` when no key +/// resolves (no trusted header and no peer): such a request is bounded by the +/// global read pool only, never a 500. The per-source-IP key is only as granular +/// as `trust`; see the `max_concurrent_reads_per_caller` config doc. +fn read_caller_key( + headers: &axum::http::HeaderMap, + peer: Option, + trust: crate::rate_limit::TrustedProxy, +) -> Option { + crate::rate_limit::client_key(headers, peer, trust) +} + +/// Acquire the per-caller read sub-cap permit (#174), or shed with a 503. `key` is +/// `None` when no caller key resolves — that request is bounded by the global read +/// pool only and is never shed here (returns `Ok(None)`). `handler` labels the shed +/// log line. Shared by both read handlers so the two acquire sites cannot drift. +fn acquire_read_caller_permit( + limiter: &crate::rate_limit::PerCallerConcurrency, + key: Option<&str>, + repo: &str, + handler: &str, +) -> Result> { + match key { + Some(k) => match limiter.try_acquire(k) { + Some(p) => Ok(Some(p)), + None => { + tracing::warn!(repo = %repo, caller = %k, handler, "per-caller read cap reached; shedding with 503"); + Err(AppError::Overloaded( + "git service at capacity for this caller, retry shortly".into(), + )) + } + }, + None => Ok(None), + } +} + /// Map an error from a `smart_http` git service call to the right `AppError`: /// [`smart_http::GitServiceTimeout`] to 504, a malformed client request to 400, /// anything else to a 500 git error. Pure (no logging) so it is unit-testable; @@ -614,8 +747,19 @@ pub async fn git_upload_pack( State(state): State, Path((owner, repo)): Path<(String, String)>, auth: Option>, + crate::rate_limit::PeerAddr(peer): crate::rate_limit::PeerAddr, + headers: axum::http::HeaderMap, body: Bytes, ) -> Result { + // #62 cheap pre-DB load shed (see git_info_refs): shed before DB when the read + // pool is saturated; the authoritative hold is `git_permit` below, after the + // per-source cap. + if state.git_read_semaphore.available_permits() == 0 { + tracing::warn!("served-git concurrency cap reached; shedding request with 503 (pre-DB)"); + return Err(AppError::Overloaded( + "git service at capacity, retry shortly".into(), + )); + } let name = smart_http_repo_name(&repo)?; let record = state .db @@ -636,6 +780,26 @@ pub async fn git_upload_pack( return Err(AppError::RepoNotFound(format!("{owner}/{name}"))); } + // Per-caller read sub-cap (#174): after the visibility gate (KTD7) so a + // visibility-denied caller never consumes a scarce read slot. Keyed on the + // resolved source IP (never the signed DID, #174 U1); no resolvable key -> + // global read pool only. + let caller_key = read_caller_key(&headers, peer, state.push_limiter_trust); + let _caller_permit = acquire_read_caller_permit( + &state.git_read_per_caller, + caller_key.as_deref(), + name, + "upload-pack", + )?; + + // Shed with a 503 before spawning git when the concurrency cap is saturated; + // held for the whole op (incl. the smart_http call), released on return. Taken + // AFTER the per-source cap above so one source cannot occupy global slots it + // would be sub-cap-denied for during the DB/visibility window and starve other + // sources; still before acquire/git so it bounds the Tigris acquire and git + // exec (INV-10). + let _permit = git_permit(&state.git_read_semaphore)?; + let disk_path = state .repo_store .acquire(&record.owner_did, &record.name) @@ -658,9 +822,12 @@ pub async fn git_upload_pack( let owner_did = record.owner_did.clone(); let caller_owned = caller.map(str::to_string); let is_public = record.is_public; + let git_bin = state.git_bin.clone(); tokio::task::spawn_blocking(move || { - visibility_pack::withheld_blob_oids( + visibility_pack::withheld_blob_oids_bounded( &path, + &git_bin, + git_timeout, &rules, is_public, &owner_did, @@ -669,14 +836,16 @@ pub async fn git_upload_pack( }) .await .map_err(|e| AppError::Git(e.to_string()))? - .map_err(|e| AppError::Git(e.to_string()))? + // A walk that hit its deadline carries GitServiceTimeout; map it to 504 + // like the smart_http paths, not a generic 500 (#174 U3). + .map_err(|e| git_service_app_error(&e))? }; if withheld.is_empty() { smart_http::upload_pack(&disk_path, body, git_timeout).await } else { tracing::info!(repo = %name, caller = ?caller, withheld = withheld.len(), "serving filtered pack"); - smart_http::upload_pack_excluding(&disk_path, body, &withheld).await + smart_http::upload_pack_excluding(&disk_path, body, &withheld, git_timeout).await } } .map_err(|e| { @@ -853,6 +1022,12 @@ pub async fn git_receive_pack( Extension(auth): Extension, body: Bytes, ) -> Result { + // Shed with a 503 before spawning git when the concurrency cap is saturated. + // Pushes draw from the dedicated WRITE pool, separate from reads, so a flood of + // anonymous reads cannot shed an authenticated push (#174). Acquired at the very + // top so it wraps the write-guard below (and precedes the Tigris acquire_write, + // bounding concurrent fresh acquires — INV-10); held for the whole op. + let _permit = git_permit(&state.git_write_semaphore)?; let name = smart_http_repo_name(&repo)?; tracing::info!(owner = %owner, repo = %name, "receive-pack request"); let record = state @@ -1057,6 +1232,8 @@ pub async fn git_receive_pack( &record.owner_did, record.is_public, disk_path.clone(), + state.git_bin.clone(), + std::time::Duration::from_secs(state.config.git_service_timeout_secs), ) .await; @@ -1091,6 +1268,8 @@ pub async fn git_receive_pack( record.is_public, record.owner_did.clone(), pin_set.candidates, + state.git_bin.clone(), + std::time::Duration::from_secs(state.config.git_service_timeout_secs), ) .await } else { @@ -1116,6 +1295,8 @@ pub async fn git_receive_pack( let node_did_str = state.node_did.to_string(); let node_seed = state.node_keypair.to_seed(); let repo_name = record.name.clone(); + let enc_git_bin = state.git_bin.clone(); + let enc_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs); tokio::spawn(async move { let pinned = crate::ipfs_pin::pin_new_objects( &ipfs_api, @@ -1140,9 +1321,15 @@ pub async fn git_receive_pack( { let p = repo_path_clone.clone(); let owner = owner_did.clone(); + let git_bin = enc_git_bin.clone(); let recip = tokio::task::spawn_blocking(move || { - crate::git::visibility_pack::withheld_blob_recipients( - &p, &rules, is_public, &owner, + crate::git::visibility_pack::withheld_blob_recipients_bounded( + &p, + &git_bin, + enc_timeout, + &rules, + is_public, + &owner, ) }) .await; @@ -1862,6 +2049,17 @@ mod tests { assert!(matches!(git_service_app_error(&other), AppError::Git(_))); } + #[test] + fn git_permit_sheds_at_capacity_and_releases() { + let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(1)); + let p1 = git_permit(&sem).expect("first acquire succeeds"); + // At capacity the next request is shed with Overloaded (-> 503), not queued. + assert!(matches!(git_permit(&sem), Err(AppError::Overloaded(_)))); + // Releasing the permit frees the slot for the next request. + drop(p1); + assert!(git_permit(&sem).is_ok()); + } + fn repo_owned_by(owner_did: &str) -> crate::db::RepoRecord { let now = chrono::Utc::now(); crate::db::RepoRecord { @@ -1892,16 +2090,39 @@ mod tests { let dummy = std::path::PathBuf::from("/nonexistent"); // Private: no rules at all. - let (announce, _) = replication_withheld_set(None, OWNER_DID, false, dummy.clone()).await; + let (announce, _) = replication_withheld_set( + None, + OWNER_DID, + false, + dummy.clone(), + "git".into(), + std::time::Duration::from_secs(600), + ) + .await; assert!(!announce, "private repo (no rules) must not announce"); // Private: empty rule set, is_public=false → still not listable at root. - let (announce, _) = - replication_withheld_set(Some(vec![]), OWNER_DID, false, dummy.clone()).await; + let (announce, _) = replication_withheld_set( + Some(vec![]), + OWNER_DID, + false, + dummy.clone(), + "git".into(), + std::time::Duration::from_secs(600), + ) + .await; assert!(!announce, "private repo (empty rules) must not announce"); // Public: empty rule set, is_public=true → listable at root, announces. - let (announce, _) = replication_withheld_set(Some(vec![]), OWNER_DID, true, dummy).await; + let (announce, _) = replication_withheld_set( + Some(vec![]), + OWNER_DID, + true, + dummy, + "git".into(), + std::time::Duration::from_secs(600), + ) + .await; assert!(announce, "public repo must announce"); } @@ -1993,6 +2214,150 @@ mod tests { } } + #[cfg(unix)] + fn write_fake_git(dir: &std::path::Path, body: &str) -> String { + use std::os::unix::fs::PermissionsExt; + let p = dir.join("fakegit"); + std::fs::write(&p, body).unwrap(); + let mut perm = std::fs::metadata(&p).unwrap().permissions(); + perm.set_mode(0o755); + std::fs::set_permissions(&p, perm).unwrap(); + p.to_str().unwrap().to_string() + } + + /// #174 (write-pool twin, vetted by execution not reasoning): the receive-pack + /// post-push replication walk is bounded. Drive `replication_withheld_set` with an + /// injected fake git that hangs on `rev-list` and a short budget: it must RETURN + /// within the budget (so `git_receive_pack` releases the write permit it holds + /// across this await, rather than pinning it for the hang) AND fail closed + /// (announce suppressed) because the walk could not be vetted. Proves this path + /// funnels through the bounded `blob_paths`, on the write-permit-holding side. + #[cfg(unix)] + #[tokio::test] + async fn replication_walk_is_bounded_and_fails_closed_on_a_hung_git() { + use std::time::Duration; + let tmp = tempfile::TempDir::new().unwrap(); + let body = "#!/bin/sh\ncase \"$1\" in\n rev-list) sleep 30 ;;\n rev-parse) echo deadbeef ;;\n *) : ;;\nesac\nexit 0\n"; + let git_bin = write_fake_git(tmp.path(), body); + // Public root (announceable) + a path-scoped rule, so the walk actually runs + // rather than taking the has_path_scoped_rule short-circuit. + let rules = Some(vec![vis_rule("/secret/**", &[])]); + + let result = tokio::time::timeout( + Duration::from_secs(10), + replication_withheld_set( + rules, + OWNER_DID, + true, + tmp.path().to_path_buf(), + git_bin, + Duration::from_millis(200), + ), + ) + .await + .expect( + "replication_withheld_set must return within the budget; a hung walk must \ + not pin the write permit git_receive_pack holds across it", + ); + assert_eq!( + result, + (false, None), + "a walk that could not be vetted must suppress the announce (fail closed)" + ); + } + + /// #174 (serve-path 504, vetted by execution): a hung withheld-blob walk on the + /// upload-pack POST maps to 504, not a generic 500. Real repo dir on disk (so + /// acquire's fast path returns it) + a path-scoped rule (so the walk runs) + + /// an injected fake git that hangs on rev-list. The handler must return 504, + /// proving git_upload_pack routes the walk's GitServiceTimeout through + /// git_service_app_error end to end. + #[cfg(unix)] + #[sqlx::test] + async fn upload_pack_hung_withheld_walk_returns_504(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let tmp = tempfile::TempDir::new().unwrap(); + let body = "#!/bin/sh\ncase \"$1\" in\n rev-list) sleep 30 ;;\n rev-parse) echo deadbeef ;;\n *) : ;;\nesac\nexit 0\n"; + let fake = write_fake_git(tmp.path(), body); + + let mut state = crate::test_support::test_state(pool).await; + state.git_bin = fake; + let mut cfg = (*state.config).clone(); + cfg.git_service_timeout_secs = 1; + state.config = std::sync::Arc::new(cfg); + state + .db + .upsert_mirror_repo("z6srv504", "sv", "/tmp/z6srv504-sv", None, false) + .await + .unwrap(); + let rec = state.db.get_repo("z6srv504", "sv").await.unwrap().unwrap(); + // Path-scoped rule so has_path_scoped_rule() is true and the walk runs; the + // public root still lets an anonymous caller past the "/" gate. + state + .db + .set_visibility_rule( + &rec.id, + "/secret/**", + crate::db::VisibilityMode::B, + &[], + OWNER_DID, + ) + .await + .unwrap(); + // acquire()'s fast path returns the local path when it exists on disk. + let disk = std::path::Path::new("/tmp/z6srv504/sv.git"); + std::fs::create_dir_all(disk).unwrap(); + + let peer: SocketAddr = "203.0.113.91:7000".parse().unwrap(); + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::POST) + .uri("/z6srv504/sv/git-upload-pack") + .body(Body::from(&b"0000"[..])) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + let status = router.oneshot(req).await.unwrap().status(); + let _ = std::fs::remove_dir_all("/tmp/z6srv504"); + assert_eq!( + status, + StatusCode::GATEWAY_TIMEOUT, + "a hung withheld-blob walk must surface as 504, not a generic 500" + ); + } + + /// #174 (F2 sizing edge, vetted by execution): the receive-pack advertisement + /// per-source cap is derived in main.rs as `(max_concurrent_git_pushes / 8).max(1)`, + /// so it is never 0 even at the minimum write-pool size (1). A 0 cap would make + /// PerCallerConcurrency shed EVERY receive-pack advertisement and break all pushes. + #[test] + fn advert_per_caller_cap_sizing_is_never_zero() { + let cap = |pushes: usize| (pushes / 8).max(1); + for pushes in [1usize, 4, 8, 32, 256] { + assert!( + cap(pushes) >= 1, + "advert cap must be >= 1 for pushes={pushes}" + ); + } + assert_eq!(cap(1), 1, "minimum write pool must derive cap 1, not 0"); + assert_eq!( + cap(32), + 4, + "default write pool 32 derives cap 4 (~8 source IPs to fill)" + ); + // A cap of 1 admits one and sheds the second from the same source. + let lim = crate::rate_limit::PerCallerConcurrency::new(cap(1), 100); + let _held = lim.try_acquire("src").expect("first advert admitted"); + assert!( + lim.try_acquire("src").is_none(), + "second advert from the same source is shed" + ); + } + #[test] fn fork_owner_full_did_with_path_rule_allowed() { // Owner reads everything (implicit reader), so nothing is withheld. @@ -2579,6 +2944,621 @@ mod tests { ); } + /// #174 U2: the receive-pack advertisement (`GET info/refs?service=git-receive-pack`) + /// is phase one of a push, so it draws from the WRITE pool, not the global read + /// pool an anonymous clone flood can exhaust. Proven at the handler by saturating + /// each pool to zero and checking who shares it (INV-10). Revert the branch to the + /// read pool and the read-saturated receive-pack assertion goes 503 (the exact + /// push-handshake starvation jatmn flagged). + #[sqlx::test] + async fn receive_pack_advertisement_draws_from_write_pool(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use std::sync::Arc; + use tokio::sync::Semaphore; + use tower::ServiceExt; + + // Build a fresh state with the two pools sized independently, then drive one + // info/refs advertisement for `service` and return its handler status. + async fn advert_status( + pool: &sqlx::PgPool, + read_permits: usize, + write_permits: usize, + service: &str, + ) -> StatusCode { + let mut state = crate::test_support::test_state(pool.clone()).await; + state.git_read_semaphore = Arc::new(Semaphore::new(read_permits)); + state.git_write_semaphore = Arc::new(Semaphore::new(write_permits)); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6wpadv", "wp", "/tmp/wp-nonexistent", None, false) + .await + .unwrap(); + let peer: SocketAddr = "203.0.113.61:6000".parse().unwrap(); + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::GET) + .uri(format!("/z6wpadv/wp/info/refs?service={service}")) + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + router.oneshot(req).await.unwrap().status() + } + + // Read pool saturated, write pool free: the push handshake SURVIVES. + assert_ne!( + advert_status(&pool, 0, 8, "git-receive-pack").await, + StatusCode::SERVICE_UNAVAILABLE, + "receive-pack advertisement must draw from the write pool: a saturated read pool must not shed it" + ); + // Write pool saturated, read pool free: the receive-pack advertisement SHEDS, + // proving it now consumes the write pool (not the read pool). + assert_eq!( + advert_status(&pool, 8, 0, "git-receive-pack").await, + StatusCode::SERVICE_UNAVAILABLE, + "receive-pack advertisement draws from the write pool, so a saturated write pool sheds it 503" + ); + // Read pool saturated: the upload-pack advertisement still SHEDS (unchanged). + assert_eq!( + advert_status(&pool, 0, 8, "git-upload-pack").await, + StatusCode::SERVICE_UNAVAILABLE, + "upload-pack advertisement stays on the read pool: a saturated read pool sheds it 503" + ); + // Write pool saturated, read pool free: the upload-pack advertisement is + // UNAFFECTED, proving reads never touch the write pool in either direction. + assert_ne!( + advert_status(&pool, 8, 0, "git-upload-pack").await, + StatusCode::SERVICE_UNAVAILABLE, + "upload-pack advertisement never touches the write pool" + ); + } + + /// #174 U2: the receive-pack advertisement is a write-path op, so it must not be + /// shed by the READ per-caller sub-cap even when the caller's source IP has + /// exhausted its read budget (e.g. concurrent clones from the same host). Fill + /// the IP's read per-caller slot, then the receive-pack advertisement from that + /// same IP must still get through. Restore the unconditional read-cap acquire on + /// the receive-pack branch and this goes 503. + #[sqlx::test] + async fn receive_pack_advertisement_ignores_read_per_caller_cap(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6wpc", "wp", "/tmp/wp-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.71:6000".parse().unwrap(); + // Exhaust the source IP's single READ per-caller slot, as concurrent clones + // from the same host would. + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("fill the IP's read per-caller slot"); + + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6wpc/wp/info/refs?service=git-receive-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + assert_ne!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "receive-pack advertisement must not be shed by the read per-caller cap: it is a write-path op" + ); + } + + /// #174 (review fix): the anon-reachable receive-pack advertisement draws from + /// the write pool, so it is bounded per source by `git_push_advert_per_caller` to + /// stop one source from monopolizing the write pool and shedding authenticated + /// pushes. Fill one source IP's advert slot; its next receive-pack advertisement + /// sheds 503, while a different source and the upload-pack advertisement are + /// unaffected. Remove the advert-cap acquisition and the same-source assertion + /// goes green-not-503. + #[sqlx::test] + async fn receive_pack_advertisement_capped_per_source(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_push_advert_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6advcap", "ac", "/tmp/ac-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.81:6000".parse().unwrap(); + // Fill this source IP's single receive-pack-advertisement slot. + let _slot = state + .git_push_advert_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first advert slot for this source IP"); + + // Same source: the receive-pack advertisement sheds 503 (advert cap full). + let router = crate::server::build_router(state.clone()); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6advcap/ac/info/refs?service=git-receive-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + assert_eq!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a source at its receive-pack advertisement cap must shed 503, so it cannot monopolize the write pool" + ); + + // A DIFFERENT source keeps its own advert budget -> not shed. + let other: SocketAddr = "203.0.113.82:6000".parse().unwrap(); + let router2 = crate::server::build_router(state.clone()); + let mut req2 = Request::builder() + .method(Method::GET) + .uri("/z6advcap/ac/info/refs?service=git-receive-pack") + .body(Body::empty()) + .unwrap(); + req2.extensions_mut().insert(ConnectInfo(other)); + assert_ne!( + router2.oneshot(req2).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a different source must keep its own receive-pack advertisement budget" + ); + + // The upload-pack advertisement is NOT bounded by the receive-pack advert cap. + let router3 = crate::server::build_router(state); + let mut req3 = Request::builder() + .method(Method::GET) + .uri("/z6advcap/ac/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req3.extensions_mut().insert(ConnectInfo(peer)); + assert_ne!( + router3.oneshot(req3).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "the upload-pack advertisement must not be shed by the receive-pack advert cap" + ); + } + + /// #174 SC2 (info_refs probe): the per-caller read sub-cap sheds a caller that + /// is already at its concurrency budget on the upload-pack advertisement, while + /// a DIFFERENT caller still enters. Remove the sub-cap from `git_info_refs` and + /// the same-caller assertion goes green-not-503 — this is the info_refs half of + /// the two-handler mutation probe. + #[sqlx::test] + async fn info_refs_per_caller_cap_sheds_one_caller_not_others(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6pcadv", "pc", "/tmp/pc-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.31:5000".parse().unwrap(); + // Fill this caller's single read slot (a clone shares the Arc-backed map). + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first slot for this caller"); + + // Same caller (IP) at its cap -> shed 503 before the git/Tigris work. + let router = crate::server::build_router(state.clone()); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6pcadv/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + assert_eq!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a caller already at its per-caller read cap must shed the advertisement with 503" + ); + + // A DIFFERENT caller (IP) has its own budget -> not shed by the per-caller cap. + let other: SocketAddr = "203.0.113.32:5000".parse().unwrap(); + let router2 = crate::server::build_router(state.clone()); + let mut req2 = Request::builder() + .method(Method::GET) + .uri("/z6pcadv/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req2.extensions_mut().insert(ConnectInfo(other)); + assert_ne!( + router2.oneshot(req2).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a different caller must not be shed by another caller's saturated budget" + ); + } + + /// #174 SC2 (upload_pack probe): the same per-caller shed on the POST + /// upload-pack path. Remove the sub-cap from `git_upload_pack` and this goes + /// green-not-503 — the upload_pack half of the two-handler mutation probe. + #[sqlx::test] + async fn upload_pack_per_caller_cap_sheds_one_caller_not_others(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6pcupl", "pc", "/tmp/pc-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.41:5000".parse().unwrap(); + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first slot for this caller"); + + let router = crate::server::build_router(state.clone()); + let mut req = Request::builder() + .method(Method::POST) + .uri("/z6pcupl/pc/git-upload-pack") + .body(Body::from(&b"0000"[..])) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + assert_eq!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a caller already at its per-caller read cap must shed upload-pack with 503" + ); + + let other: SocketAddr = "203.0.113.42:5000".parse().unwrap(); + let router2 = crate::server::build_router(state.clone()); + let mut req2 = Request::builder() + .method(Method::POST) + .uri("/z6pcupl/pc/git-upload-pack") + .body(Body::from(&b"0000"[..])) + .unwrap(); + req2.extensions_mut().insert(ConnectInfo(other)); + assert_ne!( + router2.oneshot(req2).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a different caller must not be shed by another caller's saturated budget" + ); + } + + /// #174 (review fix): the per-source caller cap is an independent brake that + /// sheds a capped source even when the global pool has free capacity — the + /// sub-cap is not a mere pre-filter for pool exhaustion. Proven by leaving the + /// global read pool with capacity (so the pre-DB early shed passes) AND + /// pre-holding the source's upload-pack read sub-cap: the request reaches the + /// caller cap and sheds there, so its 503 body reads "for this caller". Remove + /// the `acquire_read_caller_permit` call and the capped source falls through to + /// the git op instead of shedding with "for this caller" — this is the + /// caller-cap acquire probe for the info/refs upload-pack branch. + #[sqlx::test] + async fn info_refs_upload_pack_per_source_cap_sheds_with_global_capacity(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use std::sync::Arc; + use tokio::sync::Semaphore; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + // Global read pool has free capacity (early shed passes); source pre-held at + // its per-caller cap so it sheds on the caller cap, not the global pool. + state.git_read_semaphore = Arc::new(Semaphore::new(4)); + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6ordir", "oi", "/tmp/oi-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.91:5000".parse().unwrap(); + // Pin this source at its single upload-pack read slot. + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first read slot for this source IP"); + + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6ordir/oi/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + let resp = router.oneshot(req).await.unwrap(); + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "a source at its read sub-cap must shed 503 even with global pool capacity" + ); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .expect("body bytes"); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("for this caller"), + "the per-source cap is an independent brake: with global capacity free, the capped source must still shed with the caller-cap body, got {body}" + ); + } + + /// #174 (review fix): same independent-brake guarantee for the receive-pack + /// advertisement branch of info/refs — its per-source cap + /// (`git_push_advert_per_caller`) sheds a capped source even when the global + /// write pool has capacity. Leave the global write pool with capacity (so the + /// pre-DB early shed passes) and pre-hold the source's advert slot: the request + /// reaches the caller cap, so the 503 body reads "for this caller". Remove the + /// caller-cap acquire and the capped source falls through instead of shedding + /// with "for this caller". The push rate limiter is left permissive so the + /// request reaches the caller cap. + #[sqlx::test] + async fn info_refs_receive_pack_per_source_cap_sheds_with_global_capacity(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use std::sync::Arc; + use std::time::Duration; + use tokio::sync::Semaphore; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + // Global write pool has free capacity (early shed passes); source pre-held at + // its advert sub-cap so it sheds on the caller cap, not the global pool. + state.git_write_semaphore = Arc::new(Semaphore::new(4)); + state.git_push_advert_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + // Permissive push rate limiter so the advertisement passes the rate gate and + // reaches the per-source concurrency cap. + state.push_rate_limiter = crate::rate_limit::RateLimiter::new(100, Duration::from_secs(60)); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6ordrp", "or", "/tmp/or-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.92:5000".parse().unwrap(); + // Pin this source at its single receive-pack advertisement slot. + let _slot = state + .git_push_advert_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first advert slot for this source IP"); + + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6ordrp/or/info/refs?service=git-receive-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + let resp = router.oneshot(req).await.unwrap(); + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "a source at its advert sub-cap must shed 503 even with global write pool capacity" + ); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .expect("body bytes"); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("for this caller"), + "the per-source advert cap is an independent brake: with global write capacity free, the capped source must still shed with the caller-cap body, got {body}" + ); + } + + /// #174 (review fix): same independent-brake guarantee for the POST upload-pack + /// handler — its per-source read cap sheds a capped source even when the global + /// read pool has capacity. Leave the global read pool with capacity (so the + /// pre-DB early shed passes) and pre-hold the source's read slot: the request + /// reaches the caller cap, so the 503 body reads "for this caller". Remove the + /// caller-cap acquire and the capped source falls through instead of shedding + /// with "for this caller". + #[sqlx::test] + async fn upload_pack_per_source_cap_sheds_with_global_capacity(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use std::sync::Arc; + use tokio::sync::Semaphore; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + // Global read pool has free capacity (early shed passes); source pre-held at + // its per-caller cap so it sheds on the caller cap, not the global pool. + state.git_read_semaphore = Arc::new(Semaphore::new(4)); + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6ordup", "ou", "/tmp/ou-nonexistent", None, false) + .await + .unwrap(); + + let peer: SocketAddr = "203.0.113.93:5000".parse().unwrap(); + // Pin this source at its single read slot. + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first read slot for this source IP"); + + let router = crate::server::build_router(state); + let mut req = Request::builder() + .method(Method::POST) + .uri("/z6ordup/ou/git-upload-pack") + .body(Body::from(&b"0000"[..])) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + let resp = router.oneshot(req).await.unwrap(); + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "a source at its read sub-cap must shed 503 even with global pool capacity" + ); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .expect("body bytes"); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("for this caller"), + "the per-source cap is an independent brake: with global capacity free, the capped source must still shed with the caller-cap body, got {body}" + ); + } + + /// #174 SC2 (per-source key, U1): the per-caller read sub-cap keys on the + /// resolved source IP, NOT the signed DID, so a disposable-DID farm cannot + /// multiply its budget. Fill the source IP's single read slot, then drive two + /// requests signed under DIFFERENT DIDs from that SAME IP: both must shed 503 + /// (keyed by the saturated IP, not their own free DID slots). A signed request + /// from a DIFFERENT source IP keeps its own budget. Revert `read_caller_key` to + /// prefer the DID and the same-IP assertions go green-not-503 (each fresh DID + /// gets a free slot) -- the farm-defeat mutation probe. + #[sqlx::test] + async fn info_refs_per_caller_cap_keys_on_ip_not_did(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::extract::ConnectInfo; + use axum::http::{Method, Request, StatusCode}; + use std::net::SocketAddr; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6pcip", "pc", "/tmp/pc-nonexistent", None, false) + .await + .unwrap(); + + let did_a = "did:key:z6MkPerCallerKeyingProofDidAAAAAAAAAAAAAAAA"; + let did_b = "did:key:z6MkPerCallerKeyingProofDidBBBBBBBBBBBBBBBB"; + let peer: SocketAddr = "203.0.113.51:5000".parse().unwrap(); + + // Fill the SOURCE IP's single read slot; both DIDs' own slots stay free. + let _slot = state + .git_read_per_caller + .try_acquire(&peer.ip().to_string()) + .expect("first slot for this source IP"); + + // Signed as DID_A from `peer`: keyed by the saturated source IP -> shed 503. + let router = crate::server::build_router(state.clone()); + let mut req = Request::builder() + .method(Method::GET) + .uri("/z6pcip/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req.extensions_mut().insert(ConnectInfo(peer)); + req.extensions_mut() + .insert(crate::auth::AuthenticatedDid(did_a.to_string())); + assert_eq!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a signed caller must be keyed by its source IP, not its DID: the saturated IP must shed it 503" + ); + + // Same IP, a DIFFERENT DID: still keyed by the same saturated IP -> also shed. + // The farm defeat: minting a fresh DID buys no fresh per-source budget. + let router2 = crate::server::build_router(state.clone()); + let mut req2 = Request::builder() + .method(Method::GET) + .uri("/z6pcip/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req2.extensions_mut().insert(ConnectInfo(peer)); + req2.extensions_mut() + .insert(crate::auth::AuthenticatedDid(did_b.to_string())); + assert_eq!( + router2.oneshot(req2).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a second DID from the same source IP must also shed 503: a DID farm cannot multiply the per-source budget" + ); + + // A signed caller from a DIFFERENT source IP keeps its own budget -> not shed. + let other: SocketAddr = "203.0.113.52:5000".parse().unwrap(); + let router3 = crate::server::build_router(state.clone()); + let mut req3 = Request::builder() + .method(Method::GET) + .uri("/z6pcip/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + req3.extensions_mut().insert(ConnectInfo(other)); + req3.extensions_mut() + .insert(crate::auth::AuthenticatedDid(did_a.to_string())); + assert_ne!( + router3.oneshot(req3).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a signed caller from a different source IP must keep its own per-source budget" + ); + } + + /// #174 SC2 (None-key): a request with no resolvable caller key (no ConnectInfo, + /// no trusted header) must NOT be shed by the per-caller cap even when another + /// caller's budget is full — it is bounded by the global read pool only. A None + /// key never keys into the map, so it never 503s from the per-caller sub-cap. + #[sqlx::test] + async fn info_refs_none_key_bypasses_per_caller_cap(pool: sqlx::PgPool) { + use axum::body::Body; + use axum::http::{Method, Request, StatusCode}; + use tower::ServiceExt; + + let mut state = crate::test_support::test_state(pool).await; + state.git_read_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100); + state.push_limiter_trust = crate::rate_limit::TrustedProxy::None; + state + .db + .upsert_mirror_repo("z6pcnone", "pc", "/tmp/pc-nonexistent", None, false) + .await + .unwrap(); + // Saturate an unrelated caller's budget; the None-key request must be + // unaffected because it never keys into the per-caller map. + let _slot = state + .git_read_per_caller + .try_acquire("203.0.113.99") + .expect("hold an unrelated caller's slot"); + + // No ConnectInfo inserted -> PeerAddr is None -> no per-caller key. + let router = crate::server::build_router(state.clone()); + let req = Request::builder() + .method(Method::GET) + .uri("/z6pcnone/pc/info/refs?service=git-upload-pack") + .body(Body::empty()) + .unwrap(); + assert_ne!( + router.oneshot(req).await.unwrap().status(), + StatusCode::SERVICE_UNAVAILABLE, + "a request with no resolvable caller key must not be shed by the per-caller cap" + ); + } + /// Repo creation must be throttled by the per-IP creation limiter BEFORE /// signature verification — otherwise a DID farm (one throwaway did:key per /// repo, each carrying a valid but machine-solved iCaptcha proof) walks past diff --git a/crates/gitlawb-node/src/auth/mod.rs b/crates/gitlawb-node/src/auth/mod.rs index 720fb3a..5befa7f 100644 --- a/crates/gitlawb-node/src/auth/mod.rs +++ b/crates/gitlawb-node/src/auth/mod.rs @@ -520,6 +520,12 @@ mod tests { sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)), peer_write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)), shutdown_tx: tokio::sync::watch::channel(false).0, + git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(64)), + git_write_semaphore: Arc::new(tokio::sync::Semaphore::new(64)), + git_read_per_caller: crate::rate_limit::PerCallerConcurrency::with_default_max_keys(16), + git_push_advert_per_caller: + crate::rate_limit::PerCallerConcurrency::with_default_max_keys(8), + git_bin: "git".to_string(), } } diff --git a/crates/gitlawb-node/src/config.rs b/crates/gitlawb-node/src/config.rs index fc2247d..90a061e 100644 --- a/crates/gitlawb-node/src/config.rs +++ b/crates/gitlawb-node/src/config.rs @@ -172,9 +172,9 @@ pub struct Config { /// its process group torn down, in seconds. Bounds a git that neither /// finishes nor disconnects. Must be positive; set it very large to /// effectively disable the bound. Default: 600s (10 min), generous for large - /// clones. Does not cover the ref advertisement (`info/refs`) or the - /// withheld-blob fetch path (`upload_pack_excluding`, a blocking - /// `spawn_blocking` a tokio timeout cannot cancel); both remain unbounded. + /// clones. Also bounds the ref advertisement (`info/refs`) and the withheld-blob + /// pack build (`upload_pack_excluding`'s pack-objects stage), which now share the + /// same timeout + process-group teardown (#174). #[arg( long, env = "GITLAWB_GIT_SERVICE_TIMEOUT_SECS", @@ -234,6 +234,80 @@ pub struct Config { value_parser = clap::value_parser!(u64).range(1..) )] pub db_retry_max_secs: u64, + + /// Maximum number of served git operations (upload-pack / receive-pack / + /// info-refs) allowed to run concurrently. Beyond this the node sheds the + /// request with a clean 503 + Retry-After instead of spawning another git + /// subprocess and risking PID/thread exhaustion. Portable backstop: the + /// compose `pids_limit` is not present on Fly, whose connection-concurrency + /// cap is a different axis (500 connections each fan out to git + + /// pack-objects + threads). Size below the process budget with headroom. + /// + /// This is the READ pool (`git_read_semaphore`): upload-pack and both info/refs + /// advertisements. The authenticated push POST draws from a separate write pool + /// (`max_concurrent_git_pushes`) that anonymous reads can never reach, and each + /// caller is additionally bounded by `max_concurrent_reads_per_caller`, so an + /// anonymous flood cannot shed the actual push nor monopolize reads (#174). (The + /// receive-pack advertisement itself shares the read pool; a shed advertisement + /// is a cheap retryable GET, and the write POST it precedes always has capacity.) + /// + /// A permit is held for the whole op. Every git subprocess that STREAMS is + /// duration-bounded and reaps its process group on disconnect: upload-pack, + /// receive-pack, and both info/refs advertisements run under + /// `git_service_timeout_secs` with `process_group(0)` teardown, and the + /// withheld-blob (`upload_pack_excluding`) pack-objects stage runs on the async + /// side under the same teardown (#174 closed the two duration/cancellation gaps + /// this comment previously tracked). The one remaining blocking stage is the + /// `rev-list` object enumeration in the withheld-blob path — a bounded walk that + /// terminates, run off the async runtime; it is not process-group-reaped on + /// disconnect, so a stuck rev-list can still hold its slot until git exits. + /// + /// Default: 128. Must be between 1 and 1_048_576; the ceiling keeps the value + /// well under tokio's `Semaphore` permit limit so an oversized value is a + /// clean CLI error rather than a boot-time panic. + #[arg( + long, + env = "GITLAWB_MAX_CONCURRENT_GIT_OPS", + default_value_t = 128, + value_parser = clap::builder::RangedU64ValueParser::::new().range(1..=1_048_576) + )] + pub max_concurrent_git_ops: usize, + + /// Maximum number of concurrent `git-receive-pack` (push) operations. Pushes + /// draw from this dedicated pool, separate from `max_concurrent_git_ops` + /// (reads), so a flood of anonymous reads cannot shed an authenticated push at + /// admission (#174). Beyond this a push sheds a clean 503 + Retry-After. + /// + /// Default: 32. Must be between 1 and 1_048_576 (the ceiling keeps the value + /// under tokio's `Semaphore` permit limit so an oversized value is a clean CLI + /// error rather than a boot-time panic). + #[arg( + long, + env = "GITLAWB_MAX_CONCURRENT_GIT_PUSHES", + default_value_t = 32, + value_parser = clap::builder::RangedU64ValueParser::::new().range(1..=1_048_576) + )] + pub max_concurrent_git_pushes: usize, + + /// Maximum concurrent read operations (`upload-pack` and the `info/refs` + /// advertisements) a single caller may hold at once, so one caller cannot + /// monopolize the `max_concurrent_git_ops` read pool (#174). Callers are keyed + /// per-DID when signed, else per-source-IP. IMPORTANT: the per-source-IP key is + /// only as granular as `GITLAWB_TRUSTED_PROXY`. Left unset (the default), a node + /// behind an edge/NAT keys all anonymous callers on the edge IP, so this cap + /// collapses to a single global anonymous cap rather than per-client. Set + /// `GITLAWB_TRUSTED_PROXY` to key on the real client; a known high-fanout caller + /// (a CI fleet behind one NAT) should authenticate for a per-DID budget or the + /// operator raises this. Over-cap for a caller sheds a clean 503 + Retry-After. + /// + /// Default: 16. Must be between 1 and 1_048_576. + #[arg( + long, + env = "GITLAWB_MAX_CONCURRENT_READS_PER_CALLER", + default_value_t = 16, + value_parser = clap::builder::RangedU64ValueParser::::new().range(1..=1_048_576) + )] + pub max_concurrent_reads_per_caller: usize, } impl Config { @@ -272,4 +346,91 @@ mod tests { Config::try_parse_from(["gitlawb-node", "--git-service-timeout-secs", "0"]).is_err() ); } + + #[test] + fn max_concurrent_git_ops_defaults_and_rejects_out_of_range() { + assert_eq!( + Config::parse_from(["gitlawb-node"]).max_concurrent_git_ops, + 128 + ); + assert_eq!( + Config::parse_from(["gitlawb-node", "--max-concurrent-git-ops", "8"]) + .max_concurrent_git_ops, + 8 + ); + // 0 permits would shed every served-git request with a 503; clap must reject it. + assert!(Config::try_parse_from(["gitlawb-node", "--max-concurrent-git-ops", "0"]).is_err()); + // Above the ceiling would panic tokio's Semaphore::new at boot (permits > + // usize::MAX >> 3); clap must reject it as a clean CLI error instead. + assert!( + Config::try_parse_from(["gitlawb-node", "--max-concurrent-git-ops", "1048577"]) + .is_err() + ); + // The ceiling itself is accepted. + assert_eq!( + Config::parse_from(["gitlawb-node", "--max-concurrent-git-ops", "1048576"]) + .max_concurrent_git_ops, + 1_048_576 + ); + } + + #[test] + fn max_concurrent_git_pushes_defaults_and_rejects_out_of_range() { + assert_eq!( + Config::parse_from(["gitlawb-node"]).max_concurrent_git_pushes, + 32 + ); + assert_eq!( + Config::parse_from(["gitlawb-node", "--max-concurrent-git-pushes", "8"]) + .max_concurrent_git_pushes, + 8 + ); + // 0 permits would shed every push with a 503; clap must reject it. + assert!( + Config::try_parse_from(["gitlawb-node", "--max-concurrent-git-pushes", "0"]).is_err() + ); + // Above the ceiling would panic tokio's Semaphore::new at boot; clap rejects it. + assert!( + Config::try_parse_from(["gitlawb-node", "--max-concurrent-git-pushes", "1048577"]) + .is_err() + ); + assert_eq!( + Config::parse_from(["gitlawb-node", "--max-concurrent-git-pushes", "1048576"]) + .max_concurrent_git_pushes, + 1_048_576 + ); + } + + #[test] + fn max_concurrent_reads_per_caller_defaults_and_rejects_out_of_range() { + assert_eq!( + Config::parse_from(["gitlawb-node"]).max_concurrent_reads_per_caller, + 16 + ); + assert_eq!( + Config::parse_from(["gitlawb-node", "--max-concurrent-reads-per-caller", "4"]) + .max_concurrent_reads_per_caller, + 4 + ); + // 0 would shed every read from a keyed caller; clap must reject it. + assert!( + Config::try_parse_from(["gitlawb-node", "--max-concurrent-reads-per-caller", "0"]) + .is_err() + ); + assert!(Config::try_parse_from([ + "gitlawb-node", + "--max-concurrent-reads-per-caller", + "1048577" + ]) + .is_err()); + assert_eq!( + Config::parse_from([ + "gitlawb-node", + "--max-concurrent-reads-per-caller", + "1048576" + ]) + .max_concurrent_reads_per_caller, + 1_048_576 + ); + } } diff --git a/crates/gitlawb-node/src/error.rs b/crates/gitlawb-node/src/error.rs index bdd5aec..a07235f 100644 --- a/crates/gitlawb-node/src/error.rs +++ b/crates/gitlawb-node/src/error.rs @@ -47,6 +47,9 @@ pub enum AppError { #[error("git service timed out: {0}")] Timeout(String), + #[error("server overloaded: {0}")] + Overloaded(String), + #[error("database error: {0}")] Db(#[from] sqlx::Error), @@ -147,6 +150,12 @@ impl IntoResponse for AppError { DB_UNAVAILABLE_CODE, DB_UNAVAILABLE_MESSAGE.into(), ), + // 503 with a Retry-After (attached after this match — the shared tail + // can't carry per-variant headers). This is the single place Overloaded + // becomes a response, so it can never ship a 503 without the retry hint. + AppError::Overloaded(msg) => { + (StatusCode::SERVICE_UNAVAILABLE, "overloaded", msg.clone()) + } AppError::Db(e) => (StatusCode::INTERNAL_SERVER_ERROR, "db_error", e.to_string()), AppError::Internal(e) => ( StatusCode::INTERNAL_SERVER_ERROR, @@ -160,7 +169,17 @@ impl IntoResponse for AppError { "message": message, })); - (status, body).into_response() + let mut resp = (status, body).into_response(); + // Overloaded advertises when to retry. It rides the shared tail above for + // its body/status, so the header is attached here rather than in a bespoke + // early return — keeping the variant handled in exactly one place. + if matches!(self, AppError::Overloaded(_)) { + resp.headers_mut().insert( + axum::http::header::RETRY_AFTER, + axum::http::HeaderValue::from_static("1"), + ); + } + resp } } @@ -182,4 +201,14 @@ mod tests { StatusCode::INTERNAL_SERVER_ERROR ); } + + #[test] + fn overloaded_maps_to_503_with_retry_after() { + let resp = AppError::Overloaded("x".into()).into_response(); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + resp.headers().get("retry-after").unwrap().to_str().unwrap(), + "1" + ); + } } diff --git a/crates/gitlawb-node/src/git/smart_http.rs b/crates/gitlawb-node/src/git/smart_http.rs index eeb35b9..206e7b2 100644 --- a/crates/gitlawb-node/src/git/smart_http.rs +++ b/crates/gitlawb-node/src/git/smart_http.rs @@ -6,7 +6,7 @@ use bytes::Bytes; use std::collections::HashSet; use std::path::Path; use std::process::Stdio; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::process::Command; @@ -14,21 +14,29 @@ use tokio::process::Command; /// or `?service=git-receive-pack` /// /// This is the ref advertisement — the first step of a clone or push. -pub async fn info_refs(repo_path: &Path, service: &str) -> Result { +/// +/// `git_bin` is injectable purely so the process-group teardown can be driven by a +/// fake `git` in tests (production passes `"git"`). `timeout` bounds the whole child +/// interaction: previously the advertisement ran a bare `Command::output()` with no +/// deadline and no teardown, so a hung git pinned its concurrency slot indefinitely +/// and a client disconnect orphaned the child (#174). It now shares +/// [`drive_git_child`]'s timeout + `process_group(0)` + [`KillGroupOnDrop`] teardown. +pub async fn info_refs( + git_bin: &str, + service: &str, + repo_path: &Path, + timeout: Duration, +) -> Result { validate_service(service)?; - let output = Command::new("git") + let mut command = Command::new(git_bin); + command .arg(service_to_command(service)) .arg("--stateless-rpc") .arg("--advertise-refs") - .arg(repo_path) - .output() - .await?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - bail!("git {service} --advertise-refs failed: {stderr}"); - } + .arg(repo_path); + // No request body — advertise-refs does not read stdin. + let stdout = drive_git_child(command, Bytes::new(), timeout, "advertise-refs").await?; let content_type = format!("application/x-{service}-advertisement"); @@ -38,7 +46,7 @@ pub async fn info_refs(repo_path: &Path, service: &str) -> Result { let mut body = Vec::new(); body.extend_from_slice(&pkt_service); body.extend_from_slice(flush); - body.extend_from_slice(&output.stdout); + body.extend_from_slice(&stdout); Ok(Response::builder() .status(StatusCode::OK) @@ -207,7 +215,24 @@ async fn run_git_service( command .arg(service_to_command(service)) .arg("--stateless-rpc") - .arg(repo_path) + .arg(repo_path); + drive_git_child(command, input, timeout, service).await +} + +/// Drive a spawned git child under `timeout` with process-group teardown, returning +/// its stdout. Shared core for [`run_git_service`] and [`info_refs`]: the caller +/// passes a `Command` with its args set; this adds piped stdio and `process_group(0)`. +/// On the deadline the whole group is torn down and reaped before returning +/// [`GitServiceTimeout`]; on a dropped future (client disconnect) the +/// [`KillGroupOnDrop`] guard fires. `input` is written to the child's stdin (empty +/// for the advertise-refs path, which has no request body); `what` labels errors. +async fn drive_git_child( + mut command: Command, + input: Bytes, + timeout: Duration, + what: &str, +) -> Result> { + command .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); @@ -294,7 +319,7 @@ async fn run_git_service( // the real cause. if !status.success() { let stderr = String::from_utf8_lossy(&err); - bail!("{service} failed: {stderr}"); + bail!("{what} failed: {stderr}"); } write_result.context("failed to write to git stdin")?; @@ -324,62 +349,75 @@ fn pkt_line(data: &str) -> Vec { format!("{len:04x}{data}").into_bytes() } -/// Build a packfile containing every object reachable from all refs EXCEPT the -/// given blob OIDs. Commits and trees are always included, so SHAs stay intact; -/// only the named blobs are dropped. -pub fn build_filtered_pack(repo_path: &Path, withheld: &HashSet) -> Result> { - // All reachable objects as "oid [path]" lines. - let rev = std::process::Command::new("git") +/// Run `rev-list --objects --all` under `timeout` and return the reachable object +/// ids minus the withheld blobs. Runs under [`drive_git_child`] (async, with the +/// `tokio::time::timeout` + `process_group(0)` + [`KillGroupOnDrop`] teardown), so a +/// hung/slow enumeration is duration-bounded and reaped on client disconnect just +/// like the pack-objects stage. A bare blocking `Command::output()` inside a +/// `spawn_blocking` is uncancellable, so a hung rev-list would pin the endpoint's +/// concurrency permit for the whole hang (#174). `git_bin` is injectable for the +/// same fake-git testing reason as `run_git_service`. +async fn rev_list_keep( + git_bin: &str, + repo_path: &Path, + withheld: &HashSet, + timeout: Duration, +) -> Result> { + let mut command = Command::new(git_bin); + command .args(["rev-list", "--objects", "--all"]) - .current_dir(repo_path) - .output()?; - if !rev.status.success() { - bail!( - "git rev-list failed: {}", - String::from_utf8_lossy(&rev.stderr) - ); - } + .current_dir(repo_path); + let stdout = drive_git_child(command, Bytes::new(), timeout, "rev-list").await?; let mut keep = Vec::new(); - for line in String::from_utf8_lossy(&rev.stdout).lines() { + for line in String::from_utf8_lossy(&stdout).lines() { let oid = line.split_whitespace().next().unwrap_or(""); if oid.is_empty() || withheld.contains(oid) { continue; } keep.push(oid.to_string()); } - let mut child = std::process::Command::new("git") + Ok(keep) +} + +/// Build a packfile containing every object reachable from all refs EXCEPT the +/// given blob OIDs. Commits and trees are always included, so SHAs stay intact; +/// only the named blobs are dropped. +/// +/// Both git stages — the `rev-list` enumeration and the streaming `pack-objects` +/// build — run under [`drive_git_child`] sharing one deadline, so a hung/slow git at +/// either stage is duration-bounded and its process group is reaped on client +/// disconnect. An outer `tokio::time::timeout` around a `spawn_blocking` cannot +/// cancel the blocking thread, so neither stage may live off the async side +/// (#174, KTD5). +pub async fn build_filtered_pack( + git_bin: &str, + repo_path: &Path, + withheld: &HashSet, + timeout: Duration, +) -> Result> { + // One deadline spans both git stages so a slow rev-list eats into the pack + // budget rather than granting each stage a fresh `timeout` (2x the permit hold). + let deadline = Instant::now() + timeout; + let keep = rev_list_keep( + git_bin, + repo_path, + withheld, + deadline.saturating_duration_since(Instant::now()), + ) + .await?; + let mut data = keep.join("\n").into_bytes(); + data.push(b'\n'); + let mut command = Command::new(git_bin); + command .args(["pack-objects", "--stdout"]) - .current_dir(repo_path) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()?; - // Feed the object ids on stdin, but always reap the child afterward even if - // the write fails or stdin is missing, so an error can't drop the Child - // unwaited and leak a zombie (#53). - let write_result: std::io::Result<()> = { - use std::io::Write as _; - match child.stdin.take() { - Some(mut stdin) => { - let mut data = keep.join("\n").into_bytes(); - data.push(b'\n'); - stdin.write_all(&data) - } - None => Err(std::io::Error::new( - std::io::ErrorKind::BrokenPipe, - "git pack-objects stdin unavailable", - )), - } - }; - let out = child.wait_with_output()?; - write_result.context("failed to write object ids to git pack-objects stdin")?; - if !out.status.success() { - bail!( - "git pack-objects failed: {}", - String::from_utf8_lossy(&out.stderr) - ); - } - Ok(out.stdout) + .current_dir(repo_path); + drive_git_child( + command, + Bytes::from(data), + deadline.saturating_duration_since(Instant::now()), + "pack-objects", + ) + .await } /// Serve a clone/fetch with the withheld blobs removed from the response pack. @@ -409,17 +447,13 @@ pub async fn upload_pack_excluding( repo_path: &Path, request_body: Bytes, withheld: &HashSet, + timeout: Duration, ) -> Result { - // build_filtered_pack shells out to git (rev-list, pack-objects) with - // blocking std::process I/O; run it off the async worker so a large repo's - // pack build does not stall the tokio runtime. - let pack = { - let repo_path = repo_path.to_path_buf(); - let withheld = withheld.clone(); - tokio::task::spawn_blocking(move || build_filtered_pack(&repo_path, &withheld)) - .await - .context("filtered-pack build task panicked")?? - }; + // The rev-list enumeration runs blocking off the runtime; the streaming + // pack-objects stage is duration-bounded and its process group is reaped on + // disconnect via drive_git_child (#174), so a hung build no longer pins its + // concurrency slot and a client disconnect no longer orphans the git child. + let pack = build_filtered_pack("git", repo_path, withheld, timeout).await?; // The client lists its capabilities on the first `want` line. Honor // side-band-64k when offered (every modern smart-HTTP client offers it); @@ -538,7 +572,9 @@ mod tests { let mut withheld = std::collections::HashSet::new(); withheld.insert(secret.clone()); - let pack = build_filtered_pack(&bare, &withheld).unwrap(); + let pack = build_filtered_pack("git", &bare, &withheld, Duration::from_secs(30)) + .await + .unwrap(); let ids = pack_object_ids(&pack); assert!(ids.contains(&public), "public blob must be in the pack"); assert!( @@ -600,7 +636,9 @@ mod tests { b"0098want 0000000000000000000000000000000000000000 \ side-band-64k ofs-delta agent=git/2\n00000009done\n", ); - let resp = upload_pack_excluding(&bare, req, &withheld).await.unwrap(); + let resp = upload_pack_excluding(&bare, req, &withheld, Duration::from_secs(30)) + .await + .unwrap(); let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); let ids = pack_object_ids(&extract_pack(&body)); assert!( @@ -650,14 +688,16 @@ mod tests { axum::extract::Query(q): axum::extract::Query>, ) -> Response { let service = q.get("service").cloned().unwrap_or_default(); - info_refs(&st.repo, &service).await.unwrap() + info_refs("git", &service, &st.repo, Duration::from_secs(30)) + .await + .unwrap() } async fn pack_handler( axum::extract::State(st): axum::extract::State>, body: Bytes, ) -> Response { - upload_pack_excluding(&st.repo, body, &st.withheld) + upload_pack_excluding(&st.repo, body, &st.withheld, Duration::from_secs(30)) .await .unwrap() } @@ -1225,6 +1265,114 @@ mod tests { ); } + // #174 (SC3): the info/refs advertisement is now duration-bounded — previously + // it ran a bare `Command::output()` with no deadline, so a hung git pinned its + // concurrency slot forever. A hung fake git must abort with GitServiceTimeout + // (which the handler maps to 504). The outer watchdog turns a missing timeout + // into a loud failure instead of hanging the suite. info_refs shares + // drive_git_child, so its disconnect/group-teardown is the same code proven by + // run_git_service_tears_down_group_when_future_dropped above. + #[cfg(unix)] + #[tokio::test] + async fn info_refs_times_out_a_hung_advertisement() { + let tmp = tempfile::TempDir::new().unwrap(); + // A fake git that hangs forever instead of advertising refs. + let git_bin = write_fake_git(tmp.path(), "#!/bin/sh\nsleep 300\n"); + + let result = tokio::time::timeout( + Duration::from_secs(10), + info_refs( + git_bin.to_str().unwrap(), + "git-upload-pack", + tmp.path(), + Duration::from_millis(200), + ), + ) + .await + .expect("info_refs must return within the watchdog — its own timeout must fire"); + let err = result.expect_err("a hung advertisement must return an error, not hang"); + assert!( + err.downcast_ref::().is_some(), + "a hung advertisement must abort with GitServiceTimeout (-> 504), got: {err}" + ); + } + + // #174 (SC3, KTD5): the filtered-pack build's streaming pack-objects stage is + // duration-bounded on the ASYNC side. The old build ran the whole thing in a + // spawn_blocking, so an outer tokio timeout could not cancel the blocking thread + // and a disconnect orphaned the git child. A fake git that returns objects fast + // on rev-list but hangs on pack-objects must now abort with GitServiceTimeout; + // the watchdog turns a missing bound into a loud failure. The stage runs under + // drive_git_child, so its disconnect/group-teardown is the same code proven by + // run_git_service_tears_down_group_when_future_dropped. + #[cfg(unix)] + #[tokio::test] + async fn build_filtered_pack_times_out_a_hung_pack_objects() { + let tmp = tempfile::TempDir::new().unwrap(); + // rev-list returns one oid fast; pack-objects hangs forever. + let body = "#!/bin/sh\ncase \"$1\" in\n rev-list) echo deadbeefdeadbeefdeadbeefdeadbeefdeadbeef ;;\n pack-objects) sleep 300 ;;\n *) exit 1 ;;\nesac\n"; + let git_bin = write_fake_git(tmp.path(), body); + let withheld = HashSet::new(); + + let result = tokio::time::timeout( + Duration::from_secs(10), + build_filtered_pack( + git_bin.to_str().unwrap(), + tmp.path(), + &withheld, + Duration::from_millis(200), + ), + ) + .await + .expect( + "build_filtered_pack must return within the watchdog — the pack-objects \ + stage must be timeout-bounded, not an uncancellable spawn_blocking", + ); + let err = result.expect_err("a hung pack-objects must return an error, not hang"); + assert!( + err.downcast_ref::().is_some(), + "a hung pack-objects must abort with GitServiceTimeout, got: {err}" + ); + } + + // #174 (SC3, KTD5): the filtered-pack build's rev-list ENUMERATION stage is now + // duration-bounded on the ASYNC side too. It previously ran inside a + // spawn_blocking via a bare `Command::output()`, which an outer tokio timeout + // cannot cancel, so a hung/slow rev-list pinned the endpoint's concurrency permit + // for the whole hang. A fake git that hangs on rev-list must now abort with + // GitServiceTimeout; the watchdog turns a missing bound into a loud failure. + // rev-list runs under drive_git_child, so its disconnect/group-teardown is the + // same code proven by run_git_service_tears_down_group_when_future_dropped. + #[cfg(unix)] + #[tokio::test] + async fn build_filtered_pack_times_out_a_hung_rev_list() { + let tmp = tempfile::TempDir::new().unwrap(); + // rev-list hangs forever; pack-objects would return fast if it were reached. + let body = "#!/bin/sh\ncase \"$1\" in\n rev-list) sleep 300 ;;\n pack-objects) printf '' ;;\n *) exit 1 ;;\nesac\n"; + let git_bin = write_fake_git(tmp.path(), body); + let withheld = HashSet::new(); + + let result = tokio::time::timeout( + Duration::from_secs(10), + build_filtered_pack( + git_bin.to_str().unwrap(), + tmp.path(), + &withheld, + Duration::from_millis(200), + ), + ) + .await + .expect( + "build_filtered_pack must return within the watchdog — the rev-list \ + stage must be timeout-bounded, not an uncancellable spawn_blocking", + ); + let err = result.expect_err("a hung rev-list must return an error, not hang"); + assert!( + err.downcast_ref::().is_some(), + "a hung rev-list must abort with GitServiceTimeout, got: {err}" + ); + } + // A request that runs to completion must DISARM the guard after reaping, so // no stray group SIGTERM fires. The fake exits non-zero (surfacing as Err) // but leaves a grandchild alive; the grandchild must survive. Goes RED if the diff --git a/crates/gitlawb-node/src/git/visibility_pack.rs b/crates/gitlawb-node/src/git/visibility_pack.rs index cb70e39..09a80ab 100644 --- a/crates/gitlawb-node/src/git/visibility_pack.rs +++ b/crates/gitlawb-node/src/git/visibility_pack.rs @@ -4,11 +4,155 @@ //! content is held back. use crate::db::VisibilityRule; -use crate::git::store; use crate::visibility::{visibility_check, Decision}; use anyhow::{Context, Result}; use std::collections::{BTreeSet, HashMap, HashSet}; use std::path::Path; +use std::process::Stdio; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +/// Fixed budget bounding the whole withheld-blob classification walk (#174 U3). +/// The walk is fast for a real repo; this bound exists to reap a hung or +/// pathologically slow git child so it cannot pin a served-git permit (the read +/// permit on the upload-pack serve path, the write permit on the receive-pack +/// post-push replication path) past the deadline. Every caller funnels through +/// `blob_paths`, so bounding here bounds both paths at one seam. Production callers +/// pass the operator-configured `GITLAWB_GIT_SERVICE_TIMEOUT_SECS` instead; this +/// fixed budget only backs the `git_bin`-less test wrappers. +#[cfg(test)] +const WALK_TIMEOUT: Duration = Duration::from_secs(600); + +/// Run one git child under a shared `deadline` with process-group teardown, +/// BLOCKING, and return its stdout. The child runs in its own process group; a +/// watchdog thread SIGTERMs (lets git clean up its `*.lock` files), then SIGKILLs, +/// the whole group if the deadline passes before the child is reaped, so a hung or +/// slow git can pin neither a served-git permit nor a blocking thread past the +/// deadline (jatmn's "retain admission until they are reaped"). This is the +/// blocking-side counterpart of `smart_http::drive_git_child`, needed because the +/// walk's callers run it inside `spawn_blocking`, which an async timeout cannot +/// cancel. Returns [`crate::git::smart_http::GitServiceTimeout`] on the deadline so +/// the serve handler maps it to 504. `git_bin` is injectable so a fake `git` can +/// drive the teardown in tests without mutating the process-global PATH; +/// `stdin_bytes` feeds children that read stdin (empty for the arg-only children). +#[cfg(unix)] +fn run_bounded_git( + git_bin: &str, + args: &[&str], + repo_path: &Path, + stdin_bytes: &[u8], + deadline: Instant, +) -> Result> { + use std::io::{Read, Write}; + use std::os::unix::process::CommandExt; + use std::sync::mpsc::RecvTimeoutError; + + let label = args.first().copied().unwrap_or("git"); + let mut child = std::process::Command::new(git_bin) + .args(args) + .current_dir(repo_path) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .process_group(0) + .spawn() + .with_context(|| format!("failed to spawn git {label}"))?; + // With process_group(0) the child leads its own group, so pgid == its pid. + let pgid = child.id() as i32; + + // Watchdog: on the deadline, SIGTERM the whole group, poll until it exits, + // escalate to SIGKILL after a grace, and report whether it killed. Signalled to + // stand down when the child is reaped in time. Kept off the main thread because + // the main thread's stdout drain is exactly what blocks until a hung child is + // torn down. + let (done_tx, done_rx) = mpsc::channel::<()>(); + // Set true the instant the main thread reaps the child, so the watchdog never + // SIGTERMs a pgid whose leader is already reaped and whose pid the kernel may + // recycle (the reused-pgid hazard smart_http guards via disarm-after-wait). + let reaped = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let watchdog = { + let reaped = reaped.clone(); + std::thread::spawn(move || -> bool { + use std::sync::atomic::Ordering; + let wait = deadline.saturating_duration_since(Instant::now()); + match done_rx.recv_timeout(wait) { + Ok(()) | Err(RecvTimeoutError::Disconnected) => false, + Err(RecvTimeoutError::Timeout) => { + // The child was reaped right as the deadline fired: do not signal + // its (possibly-recycled) pgid. + if reaped.load(Ordering::SeqCst) { + return false; + } + // SAFETY: kill(2) takes only integers and borrows no Rust memory; + // ESRCH on an already-gone group is ignored. + unsafe { libc::kill(-pgid, libc::SIGTERM) }; + for step in 0..200u32 { + if reaped.load(Ordering::SeqCst) || unsafe { libc::kill(-pgid, 0) } != 0 { + return true; // reaped, or ESRCH: every group member has exited + } + if step == 100 { + unsafe { libc::kill(-pgid, libc::SIGKILL) }; + } + std::thread::sleep(Duration::from_millis(10)); + } + // Survived SIGKILL past the cap: a wedged (D-state) git, the same + // condition smart_http's reap logs. Warn so an operator sees it. + tracing::warn!( + pgid, + "withheld-walk git survived SIGKILL past the watchdog cap (uninterruptible I/O?)" + ); + true + } + } + }) + }; + + // Feed stdin on a writer thread and drain stderr on a reader thread so the main + // thread can drain stdout concurrently; writing all of stdin (or draining one + // pipe) before the others can deadlock once a pipe buffer fills. + let mut stdin = child.stdin.take(); + let input = stdin_bytes.to_vec(); + let writer = std::thread::spawn(move || { + if let Some(mut s) = stdin.take() { + let _ = s.write_all(&input); + } + }); + let mut stderr = child.stderr.take().context("git stderr was not piped")?; + let err_reader = std::thread::spawn(move || { + let mut err = Vec::new(); + let _ = stderr.read_to_end(&mut err); + err + }); + let mut stdout = child.stdout.take().context("git stdout was not piped")?; + let mut out = Vec::new(); + // Blocking drain, unblocked by the child closing stdout on exit. The watchdog's + // SIGTERM/SIGKILL is what makes a hung child exit; a git wedged in uninterruptible + // (D-state) I/O survives even SIGKILL, so this drain and the wait below can block + // until the kernel returns, pinning the walk thread and its permit. That residual + // is unreachable in userspace (no signal reaps a D-state process) and matches the + // async `reap_group_on_timeout`, which likewise only warns and gives up there. + let read_result = stdout.read_to_end(&mut out); + let status = child.wait().context("git wait failed")?; + // Reaped: bar the watchdog from signalling the now-reaped (possibly recycled) + // pgid before it can fire, then stand it down. + reaped.store(true, std::sync::atomic::Ordering::SeqCst); + let err = err_reader.join().unwrap_or_default(); + let _ = writer.join(); + let _ = done_tx.send(()); + let killed = watchdog.join().unwrap_or(false); + read_result.context("failed to read git stdout")?; + // The watchdog runs off a wall clock that can race a child finishing right at the + // deadline. A child that exited on its own (success) is not a timeout even if the + // watchdog fired late; only a child that did not exit successfully is a genuine + // timeout, which keeps a walk completing at its budget from a spurious 504. + if killed && !status.success() { + return Err(crate::git::smart_http::GitServiceTimeout.into()); + } + if !status.success() { + anyhow::bail!("git {label} failed: {}", String::from_utf8_lossy(&err)); + } + Ok(out) +} /// Fail closed unless every ref ultimately resolves to a commit (a ref pointing /// directly at a blob or tree, or an annotated tag — even a nested one — of such @@ -21,19 +165,15 @@ use std::path::Path; /// Full peeling is why this is not `for-each-ref %(*objecttype)`, which /// dereferences only one tag level and so misclassifies a tag-of-a-tag-of-a- /// commit as a non-commit. -fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { - let refs = std::process::Command::new("git") - .args(["for-each-ref", "--format=%(refname)"]) - .current_dir(repo_path) - .output() - .context("git for-each-ref failed")?; - if !refs.status.success() { - anyhow::bail!( - "git for-each-ref failed: {}", - String::from_utf8_lossy(&refs.stderr) - ); - } - let refs_stdout = String::from_utf8_lossy(&refs.stdout); +fn assert_all_refs_are_commits(repo_path: &Path, git_bin: &str, deadline: Instant) -> Result<()> { + let refs_out = run_bounded_git( + git_bin, + &["for-each-ref", "--format=%(refname)"], + repo_path, + b"", + deadline, + )?; + let refs_stdout = String::from_utf8_lossy(&refs_out); let refnames: Vec<&str> = refs_stdout .lines() .map(str::trim) @@ -43,59 +183,25 @@ fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { return Ok(()); } - // Peel every ref in one `git cat-file --batch-check` pass: one - // `^{}` query per line, one output line per input line, in order. - // The stdin write runs on a separate thread so this thread can drain stdout - // concurrently. cat-file echoes the full query on a ` missing` line, - // so output scales with refname length (not a fixed size per ref); writing - // all of stdin before reading any stdout would deadlock both pipes once the - // child's stdout buffer fills. Dropping `stdin` at the end of the closure - // sends EOF. + // Peel every ref in one `git cat-file --batch-check` pass: one `^{}` + // query per line, one output line per input line, in order. cat-file echoes the + // full query on a ` missing` line, so output scales with refname length; + // run_bounded_git drains stdout concurrently with the stdin write, so the pipe + // cannot deadlock, and the whole peel is bounded by the shared walk deadline. let queries = refnames .iter() .map(|r| format!("{r}^{{}}")) .collect::>() .join("\n"); - use std::io::Write; - let mut child = std::process::Command::new("git") - .args(["cat-file", "--batch-check=%(objecttype)"]) - .current_dir(repo_path) - .stdin(std::process::Stdio::piped()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()) - .spawn() - .context("failed to spawn git cat-file")?; - // Feed stdin on a writer thread so this thread can drain stdout via - // wait_with_output concurrently; a None handle (the pipe vanished) becomes a - // broken-pipe write error. wait_with_output reaps the child unconditionally - // before any error is surfaced, so no path drops it unwaited (#53), and the - // writer is joined only after the drain so the join cannot deadlock. - let writer = child - .stdin - .take() - .map(|mut stdin| std::thread::spawn(move || stdin.write_all(queries.as_bytes()))); - let peel_result = child.wait_with_output(); - let write_result = match writer { - Some(handle) => handle - .join() - .map_err(|_| anyhow::anyhow!("git cat-file stdin writer thread panicked"))?, - None => Err(std::io::Error::new( - std::io::ErrorKind::BrokenPipe, - "git cat-file stdin unavailable", - )), - }; - // Surface a write error only if the process didn't already fail with a - // clearer status. - let peel = peel_result.context("git cat-file failed")?; - if !peel.status.success() { - anyhow::bail!( - "git cat-file --batch-check failed: {}", - String::from_utf8_lossy(&peel.stderr) - ); - } - write_result.context("failed to write to git cat-file stdin")?; - - let peel_stdout = String::from_utf8_lossy(&peel.stdout); + let peel_out = run_bounded_git( + git_bin, + &["cat-file", "--batch-check=%(objecttype)"], + repo_path, + queries.as_bytes(), + deadline, + )?; + + let peel_stdout = String::from_utf8_lossy(&peel_out); let types: Vec<&str> = peel_stdout.lines().map(str::trim).collect(); // A short read means at least one ref went unclassified — fail closed. if types.len() != refnames.len() { @@ -147,47 +253,46 @@ fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { /// Fails closed: if commit enumeration or any tree walk fails, returns an error so /// the caller aborts the serve/pin rather than producing a partial (under-withheld) /// set. -fn blob_paths(repo_path: &Path) -> Result> { - assert_all_refs_are_commits(repo_path)?; +fn blob_paths(repo_path: &Path, git_bin: &str, timeout: Duration) -> Result> { + // One deadline spans the whole walk (the ref check, the HEAD probe, rev-list, + // and every per-commit ls-tree), so a slow or hung walk is bounded as a unit + // rather than granting each git child a fresh timeout. + let deadline = Instant::now() + timeout; + assert_all_refs_are_commits(repo_path, git_bin, deadline)?; // Enumerate every reachable commit, not just ref tips. `--all` walks all refs; // append HEAD so a detached HEAD (reachable by rev-list/upload-pack but in no // ref) is still classified. When HEAD does not resolve (unborn branch on an - // empty repo) `--all` alone yields nothing, which is correct — no objects exist. - let head = store::head_commit(repo_path).context("resolve HEAD failed")?; + // empty repo) `--all` alone yields nothing, which is correct: no objects exist. + // The HEAD probe is a bounded `git rev-parse --verify HEAD` (a clean exit means + // HEAD resolves), replacing the previously unbounded `store::head_commit` child. + let head_resolves = run_bounded_git( + git_bin, + &["rev-parse", "--verify", "HEAD"], + repo_path, + b"", + deadline, + ) + .is_ok(); let mut rev_args = vec!["rev-list", "--all"]; - if head.is_some() { + if head_resolves { rev_args.push("HEAD"); } - let commits = std::process::Command::new("git") - .args(&rev_args) - .current_dir(repo_path) - .output() - .context("git rev-list --all failed")?; - if !commits.status.success() { - anyhow::bail!( - "git rev-list --all failed: {}", - String::from_utf8_lossy(&commits.stderr) - ); - } - let commits_stdout = String::from_utf8_lossy(&commits.stdout); + let commits_out = run_bounded_git(git_bin, &rev_args, repo_path, b"", deadline)?; + let commits_stdout = String::from_utf8_lossy(&commits_out); let mut out: HashSet<(String, String)> = HashSet::new(); for commit in commits_stdout.lines() { let commit = commit.trim(); if commit.is_empty() { continue; } - let listing = std::process::Command::new("git") - .args(["ls-tree", "-rz", commit]) - .current_dir(repo_path) - .output() - .context("git ls-tree -rz failed")?; - if !listing.status.success() { - anyhow::bail!( - "git ls-tree -rz {commit} failed: {}", - String::from_utf8_lossy(&listing.stderr) - ); - } + let listing_out = run_bounded_git( + git_bin, + &["ls-tree", "-rz", commit], + repo_path, + b"", + deadline, + )?; // `-z` NUL-delimits records and emits paths raw; plain `git ls-tree -r` // C-quotes any path with non-ASCII or special bytes (e.g. café.txt becomes // "secret/caf\303\251.txt"), and that quoted literal would not match a @@ -198,7 +303,7 @@ fn blob_paths(repo_path: &Path) -> Result> { // path (e.g. a non-UTF-8 directory name) with U+FFFD, and the mangled string // would no longer match its deny rule — the same under-withholding class, one // layer down. Fail closed instead so the caller aborts rather than leaks. - let Ok(listing_stdout) = std::str::from_utf8(&listing.stdout) else { + let Ok(listing_stdout) = std::str::from_utf8(&listing_out) else { anyhow::bail!( "git ls-tree -rz {commit} returned a non-UTF-8 path; \ refusing to produce a partial (under-withheld) set" @@ -229,6 +334,7 @@ fn blob_paths(repo_path: &Path) -> Result> { /// /// The whole-repo "/" gate is handled by the caller before this function runs: /// if "/" denies, the caller gets a 404 and never reaches the filtered serve. +#[cfg(test)] pub fn withheld_blob_oids( repo_path: &Path, rules: &[VisibilityRule], @@ -236,7 +342,33 @@ pub fn withheld_blob_oids( owner_did: &str, caller: Option<&str>, ) -> Result> { - let pairs = blob_paths(repo_path)?; + withheld_blob_oids_bounded( + repo_path, + "git", + WALK_TIMEOUT, + rules, + is_public, + owner_did, + caller, + ) +} + +/// [`withheld_blob_oids`] with an injectable `git_bin` and walk `timeout`. Served +/// handlers call this with the operator-configured git binary and +/// `GITLAWB_GIT_SERVICE_TIMEOUT_SECS`, so the whole walk is bounded by the same +/// budget as the other served-git ops and a fake `git` can drive its teardown in +/// tests. The `git_bin`-less wrapper above keeps the fixed [`WALK_TIMEOUT`] for the +/// classification tests that run against real git. +pub fn withheld_blob_oids_bounded( + repo_path: &Path, + git_bin: &str, + timeout: Duration, + rules: &[VisibilityRule], + is_public: bool, + owner_did: &str, + caller: Option<&str>, +) -> Result> { + let pairs = blob_paths(repo_path, git_bin, timeout)?; Ok(withheld_from_pairs( &pairs, rules, is_public, owner_did, caller, )) @@ -304,6 +436,7 @@ pub fn replicable_objects(all: Vec, withheld: &HashSet) -> Vec Result> { + allowed_blob_set_for_caller_bounded( + repo_path, git_bin, timeout, rules, is_public, owner_did, None, + ) +} + /// Reachable blob OIDs that visibility ALLOWS `caller` at some path. The /// caller-aware generalization of `replicable_blob_set` (which is the anonymous /// `caller = None` case). Used by `GET /ipfs/{cid}` to gate fail-closed against @@ -325,6 +473,7 @@ pub fn replicable_blob_set( /// elsewhere (its content is readable to this caller elsewhere). Trees and /// commits are NOT included here; the caller decides per object type whether /// the allow-set applies (it does not for trees/commits — KTD3). +#[cfg(test)] pub fn allowed_blob_set_for_caller( repo_path: &Path, rules: &[VisibilityRule], @@ -332,7 +481,29 @@ pub fn allowed_blob_set_for_caller( owner_did: &str, caller: Option<&str>, ) -> Result> { - let pairs = blob_paths(repo_path)?; + allowed_blob_set_for_caller_bounded( + repo_path, + "git", + WALK_TIMEOUT, + rules, + is_public, + owner_did, + caller, + ) +} + +/// [`allowed_blob_set_for_caller`] with an injectable `git_bin` and walk `timeout`, +/// for the `GET /ipfs/{cid}` gate. +pub fn allowed_blob_set_for_caller_bounded( + repo_path: &Path, + git_bin: &str, + timeout: Duration, + rules: &[VisibilityRule], + is_public: bool, + owner_did: &str, + caller: Option<&str>, +) -> Result> { + let pairs = blob_paths(repo_path, git_bin, timeout)?; let mut allowed = HashSet::new(); for (oid, path) in &pairs { if visibility_check(rules, is_public, owner_did, caller, path) == Decision::Allow { @@ -365,14 +536,28 @@ pub fn replicable_objects_fail_closed( /// owner plus any reader DID that `visibility_check` Allows at some path the /// blob appears at. Least-privilege: a reader of one private subtree is not a /// recipient of a blob that only lives in another. +#[cfg(test)] pub fn withheld_blob_recipients( repo_path: &Path, rules: &[VisibilityRule], is_public: bool, owner_did: &str, +) -> Result>> { + withheld_blob_recipients_bounded(repo_path, "git", WALK_TIMEOUT, rules, is_public, owner_did) +} + +/// [`withheld_blob_recipients`] with an injectable `git_bin` and walk `timeout`, for +/// the receive-pack encrypt-then-pin path. +pub fn withheld_blob_recipients_bounded( + repo_path: &Path, + git_bin: &str, + timeout: Duration, + rules: &[VisibilityRule], + is_public: bool, + owner_did: &str, ) -> Result>> { // One history walk feeds both the withheld set and the recipient mapping. - let pairs = blob_paths(repo_path)?; + let pairs = blob_paths(repo_path, git_bin, timeout)?; let withheld = withheld_from_pairs(&pairs, rules, is_public, owner_did, None); if withheld.is_empty() { return Ok(HashMap::new()); @@ -402,6 +587,155 @@ pub fn withheld_blob_recipients( #[cfg(test)] mod tests { use super::*; + + /// Write an executable fake `git` shell script into `dir` and return its path, + /// so a test can drive the walk's process-group teardown without a real git and + /// without mutating the process-global PATH (the crate's only injection seam). + #[cfg(unix)] + fn write_fake_git(dir: &Path, body: &str) -> String { + use std::os::unix::fs::PermissionsExt; + let p = dir.join("fakegit"); + std::fs::write(&p, body).unwrap(); + let mut perm = std::fs::metadata(&p).unwrap().permissions(); + perm.set_mode(0o755); + std::fs::set_permissions(&p, perm).unwrap(); + p.to_str().unwrap().to_string() + } + + /// #174 U3: the withheld-blob walk is bounded at the shared `blob_paths` seam, so + /// a hung git child cannot pin the caller's permit past the deadline. A fake git + /// that hangs on `rev-list` must make `blob_paths` return `GitServiceTimeout` + /// within the watchdog budget (not block for the child's lifetime), and the + /// child's process group must be reaped (its recorded leader PID gone). Every + /// caller (upload-pack serve, receive-pack replication) funnels through + /// `blob_paths`, so this seam-level proof covers both permit pools. Neutralize + /// the watchdog SIGTERM and this hangs past the recv budget (RED). + #[cfg(unix)] + #[test] + fn blob_paths_times_out_and_reaps_a_hung_walk() { + use std::time::Duration; + let tmp = TempDir::new().unwrap(); + // Fast on every stage except rev-list, which records its own (group-leader) + // PID and then hangs. `sleep 30` bounds the worst case if the watchdog is + // ever broken, so a regression cannot wedge the suite for 300s. + let body = "#!/bin/sh\ncase \"$1\" in\n rev-list) echo $$ > revlist.pid ; sleep 30 ;;\n rev-parse) echo deadbeef ;;\n *) : ;;\nesac\nexit 0\n"; + let git_bin = write_fake_git(tmp.path(), body); + + // Run the walk on a thread with a short budget; the recv_timeout succeeding + // is itself proof the walk did not block on the hung child. + let (tx, rx) = mpsc::channel(); + let path = tmp.path().to_path_buf(); + std::thread::spawn(move || { + let _ = tx.send(blob_paths(&path, &git_bin, Duration::from_millis(200))); + }); + let result = rx.recv_timeout(Duration::from_secs(10)).expect( + "blob_paths must return within the watchdog budget, not hang on a stuck git child", + ); + let err = result.expect_err("a hung rev-list must abort the walk with an error"); + assert!( + err.downcast_ref::() + .is_some(), + "a hung walk must abort with GitServiceTimeout (mapped to 504), got: {err}" + ); + + // The recorded process-group leader must be gone: the watchdog reaps the + // whole group before blob_paths returns, so no orphaned git lingers. + let pid: i32 = std::fs::read_to_string(tmp.path().join("revlist.pid")) + .expect("the fake git must have recorded its rev-list PID") + .trim() + .parse() + .expect("recorded PID must parse"); + let mut gone = false; + for _ in 0..200 { + // SAFETY: kill(2) with signal 0 only probes existence; ESRCH (-1) means + // the process is gone. Borrows no Rust memory. + if unsafe { libc::kill(pid, 0) } != 0 { + gone = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + gone, + "the hung git child (pid {pid}) must be reaped, not orphaned, after the walk aborts" + ); + } + + /// #174 (F1 status-gate, vetted by execution): a child that exits SUCCESSFULLY is + /// never reported as a timeout even when the watchdog fires, so a walk finishing + /// right at its deadline is not a spurious 504. The fake only exits when signalled + /// and exits 0 on SIGTERM, so with a deadline already elapsed the watchdog always + /// reaches its kill path (killed == true) yet the child's status is success. + /// Drop the `!status.success()` guard and this returns GitServiceTimeout (RED). + #[cfg(unix)] + #[test] + fn run_bounded_git_success_at_the_deadline_is_not_a_timeout() { + use std::time::{Duration, Instant}; + let tmp = TempDir::new().unwrap(); + let body = "#!/bin/sh\ntrap 'exit 0' TERM\nsleep 30 &\nwait\n"; + let git_bin = write_fake_git(tmp.path(), body); + let out = run_bounded_git( + &git_bin, + &["rev-list"], + tmp.path(), + b"", + Instant::now() + Duration::from_millis(100), + ); + assert!( + out.is_ok(), + "a child that exited successfully must not be reported as a timeout even if the watchdog fired: {out:?}" + ); + } + + /// #174 (F3, vetted by execution): a child that IGNORES SIGTERM is still reaped + /// via the watchdog's SIGKILL escalation, so it cannot pin the walk thread or its + /// permit. The fake traps SIGTERM and keeps sleeping; run_bounded_git must still + /// return (via SIGKILL at the grace step) with a timeout error and the group must + /// be gone. (A truly uninterruptible D-state child, which no signal can reap, is + /// the documented residual this teardown, like the async twin, cannot cover.) + #[cfg(unix)] + #[test] + fn run_bounded_git_reaps_a_sigterm_ignoring_child_via_sigkill() { + use std::time::{Duration, Instant}; + let tmp = TempDir::new().unwrap(); + let body = "#!/bin/sh\ntrap '' TERM\necho $$ > pid\nwhile true; do sleep 1; done\n"; + let git_bin = write_fake_git(tmp.path(), body); + let (tx, rx) = std::sync::mpsc::channel(); + let path = tmp.path().to_path_buf(); + std::thread::spawn(move || { + let _ = tx.send(run_bounded_git( + &git_bin, + &["rev-list"], + &path, + b"", + Instant::now() + Duration::from_millis(100), + )); + }); + let out = rx + .recv_timeout(Duration::from_secs(10)) + .expect("run_bounded_git must return via SIGKILL even for a SIGTERM-ignoring child"); + assert!( + out.is_err(), + "a SIGTERM-ignoring child killed by SIGKILL is a timeout, not a success: {out:?}" + ); + let pid: i32 = std::fs::read_to_string(tmp.path().join("pid")) + .unwrap() + .trim() + .parse() + .unwrap(); + let mut gone = false; + for _ in 0..300 { + if unsafe { libc::kill(pid, 0) } != 0 { + gone = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + gone, + "the SIGTERM-ignoring child (pid {pid}) must be reaped via SIGKILL, not left running" + ); + } use crate::db::VisibilityMode; use chrono::Utc; use std::process::Command; diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index aa0483d..9b646c2 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/crates/gitlawb-node/src/main.rs @@ -378,6 +378,21 @@ async fn main() -> Result<()> { sync_trigger_rate_limiter, peer_write_rate_limiter, shutdown_tx: shutdown_tx.clone(), + git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_git_ops)), + git_write_semaphore: Arc::new(tokio::sync::Semaphore::new( + config.max_concurrent_git_pushes, + )), + git_read_per_caller: rate_limit::PerCallerConcurrency::with_default_max_keys( + config.max_concurrent_reads_per_caller, + ), + // Per-source cap on the receive-pack advertisement, sized to an eighth of the + // write pool (min 1): a single source can hold at most this many write-pool + // slots via the anon advertisement, so saturating the pool takes ~8 distinct + // source IPs, each also rate-limited (#174). + git_push_advert_per_caller: rate_limit::PerCallerConcurrency::with_default_max_keys( + (config.max_concurrent_git_pushes / 8).max(1), + ), + git_bin: "git".to_string(), }; // Periodic peer-count poll for the metrics gauge. If p2p is disabled diff --git a/crates/gitlawb-node/src/rate_limit.rs b/crates/gitlawb-node/src/rate_limit.rs index d40e269..2228169 100644 --- a/crates/gitlawb-node/src/rate_limit.rs +++ b/crates/gitlawb-node/src/rate_limit.rs @@ -132,6 +132,96 @@ impl RateLimiter { } } +/// A bounded per-caller CONCURRENCY limiter — distinct from [`RateLimiter`], which +/// caps request RATE. Each caller key may hold at most `per_caller` in-flight +/// permits at once; beyond that [`try_acquire`](Self::try_acquire) returns `None` +/// and the caller sheds. Used to stop one caller (a single anonymous source-IP or +/// DID) monopolizing the served-git read pool (#174). +/// +/// The key map is self-bounding: a key is removed the instant its in-flight count +/// reaches zero, so it never holds more keys than there are concurrently-active +/// callers (itself bounded by the read semaphore). A `max_keys` reject-before-insert +/// backstop guarantees a key farm can never grow the map even if that invariant +/// weakened — a NEW key at the cap is rejected WITHOUT allocating an entry (INV-15). +/// +/// Uses a `std::sync::Mutex` (not the file's `tokio::sync::Mutex`) because the +/// permit's `Drop` must release synchronously; the critical section holds no await. +#[derive(Clone)] +pub struct PerCallerConcurrency { + state: Arc>>, + per_caller: usize, + max_keys: usize, +} + +/// RAII permit from [`PerCallerConcurrency::try_acquire`]. On drop it decrements +/// the caller's in-flight count and removes the key when it reaches zero. +pub struct PerCallerPermit { + state: Arc>>, + key: String, +} + +impl PerCallerConcurrency { + pub fn new(per_caller: usize, max_keys: usize) -> Self { + Self { + state: Arc::new(std::sync::Mutex::new(HashMap::new())), + per_caller: per_caller.max(1), + max_keys: max_keys.max(1), + } + } + + /// Convenience constructor with the default key bound. + pub fn with_default_max_keys(per_caller: usize) -> Self { + Self::new(per_caller, DEFAULT_MAX_KEYS) + } + + /// `Some(permit)` when the caller is under its cap and the map has room; + /// `None` (shed) otherwise. Reject-before-insert: a new key at `max_keys` is + /// rejected without allocating. + pub fn try_acquire(&self, key: &str) -> Option { + // Recover from a poisoned lock rather than panicking: the critical section + // is pure counter arithmetic and cannot itself panic, so a poisoned mutex + // would only ever come from an unrelated abort, and a slightly-off count + // self-heals as permits drop. A panic here would instead brick the limiter + // for every caller (each subsequent lock re-panics). + let mut map = self.state.lock().unwrap_or_else(|e| e.into_inner()); + match map.get_mut(key) { + Some(count) => { + if *count >= self.per_caller { + return None; + } + *count += 1; + } + None => { + if map.len() >= self.max_keys { + return None; + } + map.insert(key.to_string(), 1); + } + } + Some(PerCallerPermit { + state: self.state.clone(), + key: key.to_string(), + }) + } + + #[cfg(test)] + pub fn tracked_keys(&self) -> usize { + self.state.lock().unwrap().len() + } +} + +impl Drop for PerCallerPermit { + fn drop(&mut self) { + let mut map = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(count) = map.get_mut(&self.key) { + *count -= 1; + if *count == 0 { + map.remove(&self.key); + } + } + } +} + pub async fn rate_limit_by_did(request: Request, next: Next) -> Response { let limiter = request.extensions().get::().cloned(); @@ -288,6 +378,63 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response { mod tests { use super::*; + #[test] + fn per_caller_concurrency_caps_one_caller_and_frees_on_drop() { + let lim = PerCallerConcurrency::new(2, 100); + let p1 = lim.try_acquire("did:key:zA").expect("first under cap"); + let p2 = lim.try_acquire("did:key:zA").expect("second under cap"); + assert!( + lim.try_acquire("did:key:zA").is_none(), + "a third in-flight op for the same caller sheds (over the per-caller cap)" + ); + // A DIFFERENT caller is unaffected — the cap is per-caller, not global. + let _other = lim + .try_acquire("did:key:zB") + .expect("a different caller has its own budget"); + drop(p1); + assert!( + lim.try_acquire("did:key:zA").is_some(), + "freeing one in-flight slot lets the same caller back in" + ); + drop(p2); + } + + #[test] + fn per_caller_concurrency_map_is_self_bounding_and_reject_before_insert() { + // Self-bounding: acquire+drop many distinct keys — the map never grows + // because a key is removed the instant its in-flight count hits zero. + let lim = PerCallerConcurrency::new(4, 3); + for i in 0..50 { + let _p = lim.try_acquire(&format!("k{i}")); + } + assert_eq!( + lim.tracked_keys(), + 0, + "keys with zero in-flight ops are removed, so an acquire+drop flood leaves the map empty" + ); + // Reject-before-insert: HOLD max_keys distinct keys, then a new key sheds + // WITHOUT growing the map past the cap (INV-15 — a rejected request never + // allocates an entry). + let held: Vec<_> = (0..3) + .map(|i| lim.try_acquire(&format!("h{i}")).unwrap()) + .collect(); + assert_eq!( + lim.tracked_keys(), + 3, + "three distinct callers held concurrently" + ); + assert!( + lim.try_acquire("h3").is_none(), + "a new key at max_keys is rejected" + ); + assert_eq!( + lim.tracked_keys(), + 3, + "the rejected new key did not allocate an entry (reject-before-insert)" + ); + drop(held); + } + #[tokio::test] async fn allows_within_limit() { let limiter = RateLimiter::new(3, Duration::from_secs(60)); diff --git a/crates/gitlawb-node/src/state.rs b/crates/gitlawb-node/src/state.rs index 6a84b3c..9495212 100644 --- a/crates/gitlawb-node/src/state.rs +++ b/crates/gitlawb-node/src/state.rs @@ -88,6 +88,38 @@ pub struct AppState { /// * the libp2p swarm task /// * the gossip, sync, operator heartbeat, and rate-limit cleanup loops pub shutdown_tx: tokio::sync::watch::Sender, + /// Bounds concurrent served git READ operations (upload-pack + both info/refs + /// advertisements). A read handler acquires a permit before spawning git and + /// holds it for the op; when none are free the request is shed with a 503. + /// Writes draw from `git_write_semaphore` so a read flood cannot shed an + /// authenticated push at admission (#174). + pub git_read_semaphore: Arc, + /// Bounds concurrent `git-receive-pack` (push) operations, a pool separate + /// from `git_read_semaphore` so an anonymous READ flood can never shed an + /// authenticated push (#174). Sized by `max_concurrent_git_pushes`. Drawn from + /// by the `git-receive-pack` POST (owner-gated) and the receive-pack `info/refs` + /// advertisement (anon-reachable); the advertisement is additionally bounded per + /// source by `git_push_advert_per_caller` so no single source can monopolize this + /// pool and shed pushes. + pub git_write_semaphore: Arc, + /// Per-caller concurrency sub-cap on the read pool: each caller (keyed on the + /// resolved source IP, #174 U1) may hold at most `max_concurrent_reads_per_caller` + /// in-flight read ops, so one caller cannot monopolize `git_read_semaphore` + /// (#174). Applied by `git_upload_pack` and the upload-pack `info/refs` + /// advertisement. + pub git_read_per_caller: crate::rate_limit::PerCallerConcurrency, + /// Per-source concurrency sub-cap on the anon-reachable receive-pack `info/refs` + /// advertisement: each source IP may hold at most a small share of the write + /// pool, so a multi-source flood of push-handshake advertisements cannot + /// saturate `git_write_semaphore` and shed authenticated pushes (#174). Sized as + /// a fraction of `max_concurrent_git_pushes`, so filling the write pool takes many + /// distinct source IPs (each also braked by the per-IP push rate limiter). + pub git_push_advert_per_caller: crate::rate_limit::PerCallerConcurrency, + /// The `git` executable the served-git withheld-blob walk spawns. Production is + /// `"git"` (resolved via PATH); injectable so a fake `git` can drive the walk's + /// process-group teardown in handler tests without mutating the process-global + /// PATH (#174). + pub git_bin: String, } impl AppState { diff --git a/crates/gitlawb-node/src/test_support.rs b/crates/gitlawb-node/src/test_support.rs index 4fe52f6..765545f 100644 --- a/crates/gitlawb-node/src/test_support.rs +++ b/crates/gitlawb-node/src/test_support.rs @@ -82,6 +82,14 @@ fn build_state(db: Arc, pool: PgPool) -> AppState { sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)), peer_write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)), shutdown_tx: tokio::sync::watch::channel(false).0, + // Generous — no test drives the handler-level shed (git_permit is unit-tested). + git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(64)), + git_write_semaphore: Arc::new(tokio::sync::Semaphore::new(64)), + git_read_per_caller: crate::rate_limit::PerCallerConcurrency::with_default_max_keys(16), + git_push_advert_per_caller: crate::rate_limit::PerCallerConcurrency::with_default_max_keys( + 8, + ), + git_bin: "git".to_string(), } } @@ -187,6 +195,199 @@ mod tests { ); } + /// PR3 (#62): the served-git concurrency cap sheds at the HTTP layer before the + /// DB. The held `git_permit` acquire now sits after the per-source cap, so the + /// shed-before-DB property is carried by an explicit `available_permits() == 0` + /// early check at the top of the handler (the held permit remains the + /// authoritative bound further down). DB-free: an exhausted semaphore sheds + /// before any DB/disk access, so a lazy state works. Remove the early-shed block + /// from git_info_refs and this goes red (the request falls through to the DB and + /// returns something other than 503). + #[tokio::test] + async fn git_info_refs_sheds_with_503_when_semaphore_exhausted() { + let mut state = test_state_lazy(); + state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0)); + + let router = Router::new() + .route( + "/{owner}/{repo}/info/refs", + axum::routing::get(crate::api::repos::git_info_refs), + ) + .with_state(state); + let resp = router + .oneshot(anon_get( + "/alice/repo.git/info/refs?service=git-upload-pack", + )) + .await + .unwrap(); + + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "an exhausted git semaphore must shed info/refs with 503 before touching the DB" + ); + assert_eq!( + resp.headers() + .get("retry-after") + .and_then(|h| h.to_str().ok()), + Some("1"), + "the 503 shed must carry Retry-After" + ); + } + + /// PR3 (#62) sibling of the info/refs shed test: git-upload-pack carries the same + /// explicit `available_permits() == 0` early-shed check at the top, so an + /// exhausted semaphore must shed it with a 503 before any DB/disk work. + /// Anonymous-reachable, so no auth injection is needed. Remove the early-shed + /// block from git_upload_pack and this goes red. + #[tokio::test] + async fn git_upload_pack_sheds_with_503_when_semaphore_exhausted() { + let mut state = test_state_lazy(); + state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0)); + + let router = Router::new() + .route( + "/{owner}/{repo}/git-upload-pack", + axum::routing::post(crate::api::repos::git_upload_pack), + ) + .with_state(state); + let req = Request::builder() + .method(Method::POST) + .uri("/alice/repo.git/git-upload-pack") + .body(Body::from(&b"0000"[..])) + .unwrap(); + let resp = router.oneshot(req).await.unwrap(); + + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "an exhausted git semaphore must shed git-upload-pack with 503 before touching the DB" + ); + assert_eq!( + resp.headers() + .get("retry-after") + .and_then(|h| h.to_str().ok()), + Some("1"), + "the 503 shed must carry Retry-After" + ); + } + + /// PR3 (#62) receive-pack sibling of the info/refs shed test: the early shed + /// selects the WRITE pool for a git-receive-pack advertisement (phase one of a + /// push, #174 U2), so an exhausted write pool sheds the advert with 503 before + /// any DB/disk work — even while the read pool is free (only the write pool is + /// zeroed here). Flip the pool selection to the read pool, or remove the + /// early-shed block, and this goes red. + #[tokio::test] + async fn git_info_refs_receive_pack_sheds_with_503_when_write_pool_exhausted() { + let mut state = test_state_lazy(); + state.git_write_semaphore = Arc::new(tokio::sync::Semaphore::new(0)); + + let router = Router::new() + .route( + "/{owner}/{repo}/info/refs", + axum::routing::get(crate::api::repos::git_info_refs), + ) + .with_state(state); + let resp = router + .oneshot(anon_get( + "/alice/repo.git/info/refs?service=git-receive-pack", + )) + .await + .unwrap(); + + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "an exhausted WRITE pool must shed the receive-pack advertisement with 503 before touching the DB" + ); + assert_eq!( + resp.headers() + .get("retry-after") + .and_then(|h| h.to_str().ok()), + Some("1"), + "the 503 shed must carry Retry-After" + ); + } + + /// PR3 (#62) sibling for the push path: git-receive-pack requires an + /// AuthenticatedDid extension (production: require_signature injects it), so the + /// request carries one via signed_request_as — without it the Extension + /// extractor 500s before the handler body reaches git_permit. The permit is the + /// first statement, so an exhausted semaphore still sheds 503 before any DB + /// work. Remove the permit line from git_receive_pack and this goes red. + #[tokio::test] + async fn git_receive_pack_sheds_with_503_when_semaphore_exhausted() { + let mut state = test_state_lazy(); + state.git_write_semaphore = Arc::new(tokio::sync::Semaphore::new(0)); + + let router = Router::new() + .route( + "/{owner}/{repo}/git-receive-pack", + axum::routing::post(crate::api::repos::git_receive_pack), + ) + .with_state(state); + let owner = "did:key:zRECVSHEDOWNERAAAAAAAAAAAAAAAAAAAAAAAAAA"; + let resp = router + .oneshot(signed_request_as( + owner, + Method::POST, + "/alice/repo.git/git-receive-pack", + Body::from(&b"0000"[..]), + )) + .await + .unwrap(); + + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "an exhausted write pool must shed git-receive-pack with 503 before touching the DB" + ); + assert_eq!( + resp.headers() + .get("retry-after") + .and_then(|h| h.to_str().ok()), + Some("1"), + "the 503 shed must carry Retry-After" + ); + } + + /// #174 (SC1, load-bearing): a saturated READ pool must NOT shed an + /// authenticated push — the write pool is a separate budget. Read pool at zero, + /// write pool with capacity: the push proceeds PAST admission (it then errors on + /// the placeholder DB, but crucially it is not a 503). Route git-receive-pack + /// back to the read pool and this goes red — that is the isolation proof. + #[tokio::test] + async fn git_receive_pack_not_shed_by_exhausted_read_pool() { + let mut state = test_state_lazy(); + // Read pool exhausted as if a flood of anonymous clones held every slot. + state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0)); + // Write pool keeps its default capacity from test_state_lazy. + + let router = Router::new() + .route( + "/{owner}/{repo}/git-receive-pack", + axum::routing::post(crate::api::repos::git_receive_pack), + ) + .with_state(state); + let owner = "did:key:zRECVCROSSBOUNDARYAAAAAAAAAAAAAAAAAAAAA"; + let resp = router + .oneshot(signed_request_as( + owner, + Method::POST, + "/alice/repo.git/git-receive-pack", + Body::from(&b"0000"[..]), + )) + .await + .unwrap(); + + assert_ne!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "an exhausted READ pool must not shed a push — the write pool is a separate budget (#174)" + ); + } + /// N7: merge_pr is owner-only. A non-owner is rejected by require_repo_owner /// before any git work (so no on-disk repo is needed for the rejection). #[sqlx::test]