diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index 8e0d9ebc..6a156eb3 100644 --- a/crates/walgit-config/src/lib.rs +++ b/crates/walgit-config/src/lib.rs @@ -1085,6 +1085,10 @@ impl Config { pub fn validate(&self) -> Result<()> { self.refs.validate()?; self.packs.validate()?; + anyhow::ensure!( + self.server.max_concurrent_per_repo > 0, + "server.max_concurrent_per_repo must be positive" + ); anyhow::ensure!( self.packfile_uri.uri_min_bytes.as_u64() > 0, "packfile_uri.uri_min_bytes must be positive" @@ -1360,6 +1364,14 @@ mod tests { assert_eq!(back.store.bucket, c.store.bucket); } + #[test] + fn zero_per_repo_concurrency_is_rejected() { + let mut c = Config::default(); + c.server.max_concurrent_per_repo = 0; + let err = c.validate().unwrap_err().to_string(); + assert!(err.contains("server.max_concurrent_per_repo must be positive")); + } + #[test] fn example_configurations_parse_and_validate() { Config::parse(include_str!("../../../walgit.example.toml")).unwrap(); diff --git a/crates/walgit-git/src/pack_segments.rs b/crates/walgit-git/src/pack_segments.rs index 25dbf72d..d02ef05f 100644 --- a/crates/walgit-git/src/pack_segments.rs +++ b/crates/walgit-git/src/pack_segments.rs @@ -637,16 +637,19 @@ fn append_index( } fn sort(input: &Path, output: &Path, work: &Path, resources: Resources) -> Result<(), GitError> { - run( - Command::new("sort") - .args(["--stable", "--unique", "--key=1,1", "--parallel=1"]) - .arg(format!("--buffer-size={}", resources.sort_memory_bytes)) - .arg("--temporary-directory") - .arg(work), - input, - output, - work, - ) + run(&mut sort_command(work, resources), input, output, work) +} + +/// GNU and BSD `sort` both read a bare `--buffer-size` number as KiB; the `b` +/// suffix keeps the budget in bytes instead of 1024 times larger. +fn sort_command(work: &Path, resources: Resources) -> Command { + let mut command = Command::new("sort"); + command + .args(["--stable", "--unique", "--key=1,1", "--parallel=1"]) + .arg(format!("--buffer-size={}b", resources.sort_memory_bytes)) + .arg("--temporary-directory") + .arg(work); + command } fn difference(a: &Path, b: &Path, output: &Path, work: &Path) -> Result<(), GitError> { diff --git a/crates/walgit-git/src/pack_segments_tests.rs b/crates/walgit-git/src/pack_segments_tests.rs index 5a9b9037..ea9883d5 100644 --- a/crates/walgit-git/src/pack_segments_tests.rs +++ b/crates/walgit-git/src/pack_segments_tests.rs @@ -305,6 +305,15 @@ async fn empty_groups_conserve_everything_and_invalid_budgets_fail() { ); } +#[test] +fn sort_budget_is_passed_in_bytes() { + let command = sort_command(Path::new("work"), resources()); + assert!( + command.get_args().any(|a| a == "--buffer-size=1048576b"), + "{command:?}" + ); +} + #[test] fn type_split_keeps_hint_bytes_and_rejects_missing_objects() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/walgit-server/src/lib.rs b/crates/walgit-server/src/lib.rs index 5f6c6154..f8017560 100644 --- a/crates/walgit-server/src/lib.rs +++ b/crates/walgit-server/src/lib.rs @@ -66,12 +66,12 @@ use std::future::Future; use std::net::SocketAddr; use std::sync::Arc; -use axum::Router; use axum::body::{Body, to_bytes}; use axum::extract::ConnectInfo; use axum::http::{Method, Request}; use axum::response::{IntoResponse, Response}; use axum::routing::get; +use axum::{Extension, Router}; use bytes::Bytes; use metrics_exporter_prometheus::PrometheusHandle; use walgit_store::DynStore; @@ -89,7 +89,8 @@ pub struct AppState { /// HTTP requests in flight (counted until the response body is done); on the watchdog line. pub inflight: Arc, pub caches: cache::ServerCaches, - pub metrics_handle: Arc, + /// Prometheus recorder handle. Absent when `telemetry.metrics` is disabled. + pub metrics_handle: Option>, /// Read-through LFS upstream client (`upstream.lfs`). pub lfs_upstream: lfs_upstream::Upstream, /// Startup prewarm state (gates /readyz when configured). @@ -112,7 +113,11 @@ impl AppState { ) -> anyhow::Result> { let registry = walgit_wal::Registry::new(store.clone(), cfg.clone()); let bridge = bridge::Bridge::new(&cfg, registry.clone()); - let metrics_handle = metrics::install()?; + let metrics_handle = if cfg.telemetry.metrics { + Some(metrics::install()?) + } else { + None + }; let tls = tls::load(&cfg)?; if let Some(t) = &tls { tracing::info!(fingerprint = %t.fingerprint, mode = ?cfg.server.tls.mode, "TLS terminated in-process"); @@ -148,7 +153,7 @@ pub fn router(state: Arc) -> Router { .gzip(true) .quality(tower_http::CompressionLevel::Fastest); // Nothing with content is public: the SPA shell/assets, installer, credential - // helper and metrics all sit behind `web::require_auth` (identity from a + // helper and enabled metrics all sit behind `web::require_auth` (identity from a // bearer token or a session cookie). The JSON API and git // endpoints do their own `require_read`/`require_write`; `/_auth/*` is the // login flow itself; `/services/public/*` is the installer a not-yet-signed-in @@ -162,11 +167,26 @@ pub fn router(state: Arc) -> Router { .with_state(()) .layer(web_compression.clone()), ) - .route("/metrics", get(metrics::metrics_route)) .layer(axum::middleware::from_fn_with_state( state.clone(), web::require_auth, )); + let metrics = if let Some(metrics_handle) = &state.metrics_handle { + Router::new() + .route("/metrics", get(metrics::metrics_route)) + .layer(Extension(metrics_handle.clone())) + .layer(axum::middleware::from_fn_with_state( + state.clone(), + web::require_auth, + )) + } else { + // Reserve the top-level name so the SPA's `/{owner}` route cannot turn a + // disabled scrape endpoint into a successful HTML response. + Router::new().route( + "/metrics", + get(|| async { axum::http::StatusCode::NOT_FOUND }), + ) + }; Router::new() .merge( @@ -180,6 +200,7 @@ pub fn router(state: Arc) -> Router { .layer(web_compression), ) .merge(gated) + .merge(metrics) .route("/healthz", get(health::healthz)) .route("/readyz", get(health::readyz)) // The SDK is a static artefact with no data in it; it must load from a diff --git a/crates/walgit-server/src/metrics.rs b/crates/walgit-server/src/metrics.rs index 264809e6..5cd09d48 100644 --- a/crates/walgit-server/src/metrics.rs +++ b/crates/walgit-server/src/metrics.rs @@ -1,14 +1,12 @@ -//! Prometheus metrics exporter. Installs a recorder up front (once per process) -//! and exposes the rendered scrape via `/metrics`. +//! Prometheus metrics exporter. When enabled, installs a recorder once per +//! process and exposes the rendered scrape via `/metrics`. use std::sync::{Arc, OnceLock}; -use axum::extract::State; +use axum::Extension; use axum::response::IntoResponse; use metrics_exporter_prometheus::PrometheusHandle; -use crate::AppState; - static HANDLE: OnceLock> = OnceLock::new(); /// Install the Prometheus recorder once per process and return a shared handle. @@ -32,13 +30,15 @@ pub fn install() -> anyhow::Result> { } /// `GET /metrics` -pub async fn metrics_route(State(st): State>) -> impl IntoResponse { - let body = st.metrics_handle.render(); +pub async fn metrics_route( + Extension(handle): Extension>, +) -> impl IntoResponse { ( [( axum::http::header::CONTENT_TYPE, "text/plain; version=0.0.4", )], - body, + handle.render(), ) + .into_response() } diff --git a/crates/walgit-server/tests/metrics.rs b/crates/walgit-server/tests/metrics.rs new file mode 100644 index 00000000..c07b55d2 --- /dev/null +++ b/crates/walgit-server/tests/metrics.rs @@ -0,0 +1,22 @@ +mod harness; + +use axum::http::StatusCode; + +use harness::Server; + +type TestResult = anyhow::Result<()>; + +#[tokio::test] +async fn metrics_config_controls_scrape_endpoint() -> TestResult { + let disabled = Server::start_with_tweak(|cfg| cfg.telemetry.metrics = false).await?; + assert!(disabled.state.metrics_handle.is_none()); + let response = reqwest::get(format!("{}/metrics", disabled.base_url)).await?; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + let enabled = Server::start_with_tweak(|cfg| cfg.telemetry.metrics = true).await?; + assert!(enabled.state.metrics_handle.is_some()); + let response = reqwest::get(format!("{}/metrics", enabled.base_url)).await?; + assert_eq!(response.status(), StatusCode::OK); + + Ok(()) +} diff --git a/crates/walgit-store/src/gcs.rs b/crates/walgit-store/src/gcs.rs index fbe4a62c..4dd3b4f7 100644 --- a/crates/walgit-store/src/gcs.rs +++ b/crates/walgit-store/src/gcs.rs @@ -1201,9 +1201,13 @@ fn is_retryable(e: &google_cloud_storage::Error) -> bool { ); } if let Some(code) = e.http_status_code() { - return matches!(code, 503 | 504 | 429 | 500); + // The statuses GCS documents as retryable. The client's own policy + // retries all of 5xx; 501 is left out because an unimplemented + // request will not start working on a second try. Same set as + // `s3::is_transient_status`. + return matches!(code, 408 | 429 | 500 | 502 | 503 | 504); } - e.is_connect() || e.is_io() || e.is_transient_and_before_rpc() + e.is_connect() || e.is_io() || e.is_timeout() || e.is_transient_and_before_rpc() } fn map_error(key: &str, e: google_cloud_storage::Error) -> StoreError { @@ -1384,6 +1388,32 @@ mod tests { assert!(is_retryable(&e)); } + fn http_error(status: u16) -> google_cloud_storage::Error { + google_cloud_gax::error::Error::http(status, http::HeaderMap::new(), bytes::Bytes::new()) + } + + #[test] + fn transient_http_statuses_are_retryable() { + for status in [408, 429, 500, 502, 503, 504] { + assert!( + is_retryable(&http_error(status)), + "{status} should be retryable" + ); + } + for status in [400, 403, 404, 409, 412, 501] { + assert!( + !is_retryable(&http_error(status)), + "{status} should be permanent" + ); + } + } + + #[test] + fn is_retryable_timeout() { + let e = google_cloud_gax::error::Error::timeout("attempt timed out"); + assert!(is_retryable(&e)); + } + #[test] fn is_not_retryable_not_found() { let e = google_cloud_gax::error::Error::service( diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs index d788cc2e..2138941b 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -8,14 +8,15 @@ //! ## Version tokens //! //! S3 `ETags` are used as opaque `Version` strings. Quotes are stripped -//! consistently on read and never stored. For non-multipart uploads the -//! `ETag` is the MD5 of the content; for multipart uploads it is a compound -//! hash. Callers never parse the token — equality comparison suffices. +//! consistently on read and never stored; conditional headers put them back +//! ([`entity_tag`]). For non-multipart uploads the `ETag` is the MD5 of the +//! content; for multipart uploads it is a compound hash. Callers never parse +//! the token — equality comparison suffices. //! //! ## Conditional PUT //! -//! `PutMode::Create` → `If-None-Match: *` (object must not exist). -//! `PutMode::Update(v)` → `If-Match: ` (CAS on current `ETag`). +//! `PutMode::Create` → `If-None-Match: *` (object must not exist). +//! `PutMode::Update(v)` → `If-Match: ""` (CAS on current `ETag`). //! On failure the SDK returns a `PreconditionFailed` service error; we fill //! `current` via a follow-up HEAD when the SDK doesn't include it. //! @@ -124,10 +125,10 @@ impl S3Store { let mut builder = self.client.get_object().bucket(&self.bucket).key(key); if let Some(v) = &opts.if_none_match { - builder = builder.if_none_match(v.as_str()); + builder = builder.if_none_match(entity_tag(v)); } if let Some(v) = &opts.if_match { - builder = builder.if_match(v.as_str()); + builder = builder.if_match(entity_tag(v)); } if let Some(r) = &opts.range { builder = builder.range(Self::range_header(r)); @@ -260,7 +261,7 @@ fn is_transient_code(code: &str) -> bool { /// HTTP statuses that mean the same, for services whose error codes we do not /// recognise (rustfs and other S3-compatible stores do not all use AWS codes). fn is_transient_status(status: u16) -> bool { - matches!(status, 429 | 500 | 502 | 503 | 504) + matches!(status, 408 | 429 | 500 | 502 | 503 | 504) } /// Whether an SDK failure is worth another attempt at walgit's layer. @@ -393,7 +394,7 @@ impl ObjectStore for S3Store { builder = builder.if_none_match("*"); } PutMode::Update(v) => { - builder = builder.if_match(v.as_str()); + builder = builder.if_match(entity_tag(v)); } } @@ -977,6 +978,20 @@ impl S3Store { } } +/// A version token as the entity-tag a conditional header carries. `ETag`s are +/// stored unquoted, but `If-Match` / `If-None-Match` take quoted entity-tags +/// (RFC 9110 §8.8.3, §13.1.1), the form every response's `ETag` header has. +/// AWS S3 and rustfs also accept a bare value; other S3-compatible stores do +/// not document it (Cloudflare R2 documents neither form). +fn entity_tag(v: &Version) -> String { + let s = v.as_str(); + if s.len() >= 2 && s.starts_with('"') && s.ends_with('"') { + s.to_owned() + } else { + format!("\"{s}\"") + } +} + fn static_credentials( access_key: &str, secret_key: &str, @@ -997,13 +1012,16 @@ fn static_credentials( // 7. Multipart: CreateMultipartUpload + UploadPart + CompleteMultipartUpload // supported. No conditional headers on Create/Complete (same as real S3). // 8. ETags: quoted, MD5 for single-PUT, compound for multipart. Quotes -// stripped consistently in our Version. +// stripped consistently in our Version, restored on If-Match/If-None-Match +// (rustfs honours both forms). // 9. force_path_style: required for rustfs local dev. #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; + use aws_sdk_s3::error::SdkError; use aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Error; use aws_sdk_s3::operation::put_object::PutObjectError; @@ -1141,6 +1159,119 @@ mod tests { )); } + /// A fake S3 that records each request head and answers a PUT with 200 and + /// a GET with 304, both with a quoted `ETag` like real S3. + async fn recording_s3() -> (String, Arc>>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let heads: Arc>> = Arc::default(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let recorded = heads.clone(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let recorded = recorded.clone(); + tokio::spawn(async move { + let mut buf = Vec::new(); + let mut chunk = [0u8; 8192]; + let end = loop { + let n = sock.read(&mut chunk).await.unwrap_or(0); + if n == 0 { + return; + } + buf.extend_from_slice(&chunk[..n]); + if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break i + 4; + } + }; + let head = String::from_utf8_lossy(&buf[..end]).into_owned(); + let body_len = head + .lines() + .filter_map(|l| l.split_once(':')) + .find(|(k, _)| k.eq_ignore_ascii_case("content-length")) + .and_then(|(_, v)| v.trim().parse::().ok()) + .unwrap_or(0); + while buf.len() < end + body_len { + let n = sock.read(&mut chunk).await.unwrap_or(0); + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + let resp = if head.starts_with("GET ") { + "HTTP/1.1 304 Not Modified\r\nETag: \"abc\"\r\nConnection: close\r\n\r\n" + } else { + "HTTP/1.1 200 OK\r\nETag: \"def\"\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + }; + recorded.lock().push(head); + let _ = sock.write_all(resp.as_bytes()).await; + }); + } + }); + (endpoint, heads) + } + + fn header<'a>(head: &'a str, name: &str) -> Option<&'a str> { + head.lines() + .filter_map(|l| l.split_once(':')) + .find(|(k, _)| k.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.trim()) + } + + #[test] + fn entity_tags_are_quoted_once() { + assert_eq!(entity_tag(&Version::new("abc")), "\"abc\""); + assert_eq!(entity_tag(&Version::new("\"abc\"")), "\"abc\""); + assert_eq!(entity_tag(&Version::new("abc-3")), "\"abc-3\""); + } + + /// Versions hold `ETag`s unquoted; on the wire a conditional header carries + /// the quoted entity-tag (RFC 9110), the form every `ETag` response has. + #[tokio::test] + async fn conditional_headers_carry_quoted_entity_tags() { + let (endpoint, heads) = recording_s3().await; + let store = S3Store { + client: client_for(&endpoint), + bucket: "b".into(), + http: reqwest::Client::new(), + multipart_threshold: u64::MAX, + multipart_part_size: 8 << 20, + }; + let put = store + .put( + "manifest.pb", + PutBody::Bytes(Bytes::from_static(b"x")), + PutMode::Update(Version::new("abc")).into(), + ) + .await + .unwrap(); + assert_eq!(put.version.as_str(), "def", "stored unquoted"); + let got = store + .get( + "manifest.pb", + GetOptions { + if_none_match: Some(Version::new("abc")), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(matches!(got, GetResult::NotModified { .. })); + let _ = store + .get( + "manifest.pb", + GetOptions { + if_match: Some(Version::new("abc")), + ..Default::default() + }, + ) + .await; + let heads = heads.lock().clone(); + assert_eq!(heads.len(), 3, "{heads:#?}"); + assert_eq!(header(&heads[0], "if-match"), Some("\"abc\"")); + assert_eq!(header(&heads[1], "if-none-match"), Some("\"abc\"")); + assert_eq!(header(&heads[2], "if-match"), Some("\"abc\"")); + } + #[test] fn transient_codes_are_recognised() { for code in [ @@ -1172,10 +1303,10 @@ mod tests { #[test] fn transient_statuses_are_recognised() { - for status in [429, 500, 502, 503, 504] { + for status in [408, 429, 500, 502, 503, 504] { assert!(is_transient_status(status), "{status} should be transient"); } - for status in [400, 403, 404, 409, 412] { + for status in [400, 403, 404, 409, 412, 501] { assert!(!is_transient_status(status), "{status} should be permanent"); } } diff --git a/justfile b/justfile index c7398eda..64b4e580 100644 --- a/justfile +++ b/justfile @@ -10,7 +10,7 @@ t15 := `if command -v timeout >/dev/null 2>&1; then echo "timeout 900"; elif com # The fast tier's package selections, shared by the build and the run of each line. fast_pkgs := "-p walgit-store -p walgit-git -p walgit-wal" -server_fast := "-p walgit-server --test web_api --test web_ui --test api_v1 --test static_http --test packfile_uri --test forward --test maintain --test routing_prefix --test lfs_upstream --test drain --test events --test follow --test policy" +server_fast := "-p walgit-server --test web_api --test web_ui --test api_v1 --test static_http --test packfile_uri --test forward --test maintain --test routing_prefix --test metrics --test lfs_upstream --test drain --test events --test follow --test policy" # Default: show available targets. default: