fix(wal): answer 409 when creating an existing repository

crates/walgit-config/src/lib.rs

@@ -1002,13 +1002,9 @@ impl Config {
             if path.is_empty() || path.iter().any(std::string::String::is_empty) {
                 continue;
             }
-            let value: toml::Value = v
-                .parse::<toml::Value>()
-                .unwrap_or(toml::Value::String(v.clone()));
             // Apply into a copy and type-check it alone: a bad/unknown key is
             // dropped (WARN) instead of failing every other override with it.
-            let mut trial = doc.clone();
-            let bad = {
+            let apply = |value: toml::Value| {
                 fn set(
                     cur: &mut toml::Table,
                     path: &[String],
@@ -1028,22 +1024,32 @@ impl Config {
                         .ok_or_else(|| format!("{key} is not a table"))?;
                     set(next, rest, value)
                 }
-                match set(&mut trial, &path, value) {
-                    Err(why) => Some(why),
-                    Ok(()) => trial.clone().try_into::<Config>().err().map(|e| {
-                        e.to_string()
-                            .lines()
-                            .next()
-                            .unwrap_or("invalid")
-                            .to_string()
-                    }),
+                let mut trial = doc.clone();
+                set(&mut trial, &path, value)?;
+                match trial.clone().try_into::<Config>() {
+                    Ok(_) => Ok(trial),
+                    Err(e) => Err(e
+                        .to_string()
+                        .lines()
+                        .next()
+                        .unwrap_or("invalid")
+                        .to_string()),
                 }
             };
-            if let Some(why) = bad {
-                ignored.push((k, why));
-            } else {
-                doc = trial;
-                touched = true;
+            // The TOML reading first, then the text itself: a key that takes text keeps a
+            // value that only looks like a number or boolean (a numeric bucket name, the
+            // `0` a duration documents as "off").
+            let literal = toml::Value::String(v.clone());
+            let applied = match v.parse::<toml::Value>() {
+                Ok(value) => apply(value).or_else(|why| apply(literal).map_err(|_| why)),
+                Err(_) => apply(literal),
+            };
+            match applied {
+                Ok(trial) => {
+                    doc = trial;
+                    touched = true;
+                }
+                Err(why) => ignored.push((k, why)),
             }
         }
         // `[placement]` is a host fact set as a GROUP: any WALGIT__PLACEMENT__* override
@@ -1079,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"
@@ -1354,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();
@@ -1514,6 +1532,47 @@ mod tests {
         .unwrap();
     }
 
+    /// A value that reads as a TOML number or boolean is still text to a key that takes
+    /// text: a numeric bucket name, or the `0` the duration keys document as "off".
+    #[test]
+    fn env_override_text_that_looks_like_a_toml_scalar_applies() {
+        let mut c = Config::default();
+        let ignored = c
+            .apply_env_report(
+                vec![
+                    ("WALGIT__STORE__BUCKET".to_string(), "20260915".to_string()),
+                    ("WALGIT__MAINTENANCE__HOST".to_string(), "true".to_string()),
+                    (
+                        "WALGIT__MAINTENANCE__FOLLOW_INTERVAL".to_string(),
+                        "0".to_string(),
+                    ),
+                    ("WALGIT__WAL__MAX_BATCH".to_string(), "7".to_string()),
+                ]
+                .into_iter(),
+            )
+            .unwrap();
+        assert!(ignored.is_empty(), "{ignored:?}");
+        assert_eq!(c.store.bucket, "20260915");
+        assert_eq!(c.maintenance.host.as_deref(), Some("true"));
+        assert_eq!(c.maintenance.follow_interval, Duration::ZERO);
+        assert_eq!(c.wal.max_batch, 7, "a TOML value still wins where it fits");
+        // A value that fits neither reading is still ignored, with the TOML reading's reason.
+        let ignored = c
+            .apply_env_report(
+                vec![
+                    ("WALGIT__SERVER__HTTP2".to_string(), "1".to_string()),
+                    ("WALGIT__STORE__BUCKET__NAME".to_string(), "b".to_string()),
+                ]
+                .into_iter(),
+            )
+            .unwrap();
+        assert_eq!(ignored.len(), 2, "{ignored:?}");
+        assert!(ignored[0].1.contains("integer"), "{:?}", ignored[0]);
+        assert!(ignored[1].1.contains("not a table"), "{:?}", ignored[1]);
+        assert!(c.server.http2);
+        assert_eq!(c.store.bucket, "20260915");
+    }
+
     #[test]
     fn removed_bundle_configuration_is_rejected() {
         for input in [

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> {

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

crates/walgit-server/src/bridge.rs

@@ -14,7 +14,8 @@
 //!   `…/manifest.pb` — the commit point itself as a notification. Accepted
 //!   shapes: a GCS Pub/Sub push envelope (`message.attributes.objectId`), an S3
 //!   event notification (`Records[].s3.object.key`), or a plain `{"key": "…"}`
-//!   / `{"repo": "owner/name"}`. A non-2xx here is meant to be redelivered;
+//!   (all bucket keys, `store.prefix` included) / `{"repo": "owner/name"}`
+//!   (whatever the prefix). A non-2xx here is meant to be redelivered;
 //! * the sweep (`events.sweep_interval`): every repo — the backstop. A sweep
 //!   that finds unpublished entries means the notifications are not flowing
 //!   and says so (`events_bridge_sweep_found_total`, warn).
@@ -258,8 +259,9 @@ impl Bridge {
     }
 }
 
-/// The object keys (or repository ids) a notification body names. Store-agnostic.
-fn notified_keys(v: &serde_json::Value) -> Vec<String> {
+/// The bucket object keys a notification body names, `store.prefix` included as the
+/// notifications carry it. Store-agnostic.
+fn notified_keys(v: &serde_json::Value, store_prefix: &str) -> Vec<String> {
     let mut keys = Vec::new();
     // GCS → Pub/Sub push envelope.
     let attrs = &v["message"]["attributes"];
@@ -299,8 +301,9 @@ fn notified_keys(v: &serde_json::Value) -> Vec<String> {
     if let Some(k) = v["key"].as_str() {
         keys.push(k.to_string());
     }
+    // A repository id names no object: use the key its manifest has in this bucket.
     if let Some(r) = v["repo"].as_str() {
-        keys.push(format!("repos/{r}/manifest.pb"));
+        keys.push(format!("{store_prefix}repos/{r}/manifest.pb"));
     }
     keys
 }
@@ -324,7 +327,7 @@ pub async fn http_notify(
     let v: serde_json::Value = serde_json::from_slice(&bytes)
         .map_err(|e| ApiError::BadRequest(format!("notify body: {e}")))?;
     let mut reports = Vec::new();
-    for key in notified_keys(&v) {
+    for key in notified_keys(&v, &bridge.store_prefix) {
         match bridge.object_finalized(&key).await {
             Ok(Some(report)) => reports.push(report),
             Ok(None) => {}
@@ -390,7 +393,25 @@ mod tests {
             .manifest_repo("prefix/repos/acme/monorepo/manifest.pb")
             .unwrap();
         assert_eq!(id.to_string(), "acme/monorepo");
+        // Every body shape names the same repository under a non-empty prefix.
+        for body in [
+            serde_json::json!({"repo": "acme/monorepo"}),
+            serde_json::json!({"key": "prefix/repos/acme/monorepo/manifest.pb"}),
+            serde_json::json!({"message": {"attributes": {
+                "eventType": "OBJECT_FINALIZE",
+                "objectId": "prefix/repos/acme/monorepo/manifest.pb"}}}),
+            serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put",
+                "s3": {"object": {"key": "prefix/repos/acme/monorepo/manifest.pb"}}}]}),
+        ] {
+            let ids: Vec<_> = notified_keys(&body, &b.store_prefix)
+                .iter()
+                .filter_map(|k| b.manifest_repo(k))
+                .map(|id| id.to_string())
+                .collect();
+            assert_eq!(ids, ["acme/monorepo"], "{body}");
+        }
         for other in [
+            "repos/t/r/manifest.pb",
             "prefix/repos/t/r/wal/abc.pack",
             "prefix/repos/t/r/events/cursor.json",
             "walgit-go/repos/t/r/manifest.pb",

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<middleware::Inflight>,
     pub caches: cache::ServerCaches,
-    pub metrics_handle: Arc<PrometheusHandle>,
+    /// Prometheus recorder handle. Absent when `telemetry.metrics` is disabled.
+    pub metrics_handle: Option<Arc<PrometheusHandle>>,
     /// 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<Arc<Self>> {
         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<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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

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<Arc<PrometheusHandle>> = OnceLock::new();
 
 /// Install the Prometheus recorder once per process and return a shared handle.
@@ -32,13 +30,15 @@ pub fn install() -> anyhow::Result<Arc<PrometheusHandle>> {
 }
 
 /// `GET /metrics`
-pub async fn metrics_route(State(st): State<Arc<AppState>>) -> impl IntoResponse {
-    let body = st.metrics_handle.render();
+pub async fn metrics_route(
+    Extension(handle): Extension<Arc<PrometheusHandle>>,
+) -> impl IntoResponse {
     (
         [(
             axum::http::header::CONTENT_TYPE,
             "text/plain; version=0.0.4",
         )],
-        body,
+        handle.render(),
     )
+        .into_response()
 }

crates/walgit-server/src/pack_lifecycle.rs

@@ -495,7 +495,7 @@ pub async fn run(
             _ => 2,
         };
         let global = matches!(work, Work::FirstFreeze | Work::Resegment);
-        let pin = handle.pin_local_objects().await;
+        let pin = handle.pin_local_objects().await?;
         let mut input = maintenance_input::prepare(
             handle.local().path().to_path_buf(),
             root,
@@ -701,7 +701,7 @@ pub async fn run(
     {
         let oid = gix_hash::ObjectId::from_hex(commit.oid.as_bytes())?;
         handle.local().set_segmented_midx_mode(true);
-        let pin = handle.pin_local_objects().await;
+        let pin = handle.pin_local_objects().await?;
         handle
             .local()
             .write_verified_midx(&final_ids, oid, pin)