From 81886b4663fdb3c11b2230f606d0a00860c90e7c Mon Sep 17 00:00:00 2001 From: Jonas Grosch Date: Thu, 1 Oct 2026 01:58:28 +0200 Subject: [PATCH 01/17] fix(wal): create answers 409 for an existing repository, warm handle or not MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `PUT /{owner}/{repo}` answered 201 for a repository that already existed whenever this instance held its handle: `Registry::create` returned the cached handle before the manifest Create PUT. A cold instance answered 409 for the same repository, so the answer depended on which host you hit. The Create PUT is now the only authority (ROUNDTRIPS §3, "let the conditional write be the read"): its 412 is AlreadyExists on every instance, and a warm handle no longer masks a repository deleted by another host (the create succeeds and replaces the stale handle). `open_or_create` (auto_create_on_push, import) treats losing the create race as success and opens the winner's repository — the same-instance race used to be absorbed by the warm-handle shortcut. Round trips: create of a new repository 1 → 1 (Create PUT); create of an existing one 0 (warm) / 1 (cold) → 1 on both; auto-create unchanged, its lost race adds only the open it needs. Pinned in the sim's healthy_request_round_trip_budgets and the ROUNDTRIPS table. (cherry picked from commit f63c8ca8ccb0c2fa6b4f334bab39777147852de3) Signed-off-by: Yordis Prieto --- crates/walgit-server/tests/e2e.rs | 60 +++++++++++++++++++++++++++ crates/walgit-server/tests/sim.rs | 23 ++++++++++- crates/walgit-wal/src/registry.rs | 17 ++++---- crates/walgit-wal/tests/wal.rs | 68 +++++++++++++++++++++++++++++++ docs/ROUNDTRIPS.md | 4 +- 5 files changed, 162 insertions(+), 10 deletions(-) diff --git a/crates/walgit-server/tests/e2e.rs b/crates/walgit-server/tests/e2e.rs index e37eea27..7cce50a3 100644 --- a/crates/walgit-server/tests/e2e.rs +++ b/crates/walgit-server/tests/e2e.rs @@ -298,6 +298,66 @@ async fn admin_create_list_delete() -> TestResult { Ok(()) } +/// `PUT /{owner}/{repo}` answers from the bucket: 201 once, then 409 — on the +/// instance holding the handle (warm) and on one that never opened it (cold); +/// concurrent creates of one name have one 201 and one 409, on one instance +/// and across two. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn create_answers_409_for_an_existing_repository() -> TestResult { + let (a, b) = Server::start_pair().await?; + let client = reqwest::Client::new(); + let put = |s: &Server, repo: &str| client.put(format!("{}/t/{repo}", s.base_url)).send(); + assert_eq!(put(&a, "r").await?.status(), 201); + assert_eq!(put(&a, "r").await?.status(), 409, "warm handle"); + assert_eq!(put(&b, "r").await?.status(), 409, "cold instance"); + // Deleted on B while A still holds a handle: the bucket, not A's map, answers. + assert_eq!( + client + .delete(format!("{}/t/r", b.base_url)) + .send() + .await? + .status(), + 204 + ); + assert_eq!( + put(&a, "r").await?.status(), + 201, + "re-create after a delete elsewhere" + ); + + for (n, (x, y)) in [(&a, &a), (&a, &b)].into_iter().enumerate() { + let repo = format!("race{n}"); + let (rx, ry) = tokio::join!(put(x, &repo), put(y, &repo)); + let mut codes = [rx?.status().as_u16(), ry?.status().as_u16()]; + codes.sort_unstable(); + assert_eq!(codes, [201, 409], "{repo}"); + } + Ok(()) +} + +/// `auto_create_on_push`: a push to a missing name creates it (its internal +/// create treats "exists" as fine), and a create after that is a 409. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn auto_create_on_push_creates_the_repository() -> TestResult { + let server = Server::start_with_tweak(|c| c.server.auto_create_on_push = true).await?; + let src = TestRepo::synthetic(1, 1)?; + for refspec in ["HEAD:refs/heads/main", "HEAD:refs/heads/second"] { + git_in(&src, &["push", &server.repo_url("t", "auto"), refspec])?; + } + let refs = server.ls_remote("t", "auto").await?; + assert!( + refs.contains("refs/heads/main") && refs.contains("refs/heads/second"), + "{refs}" + ); + let status = reqwest::Client::new() + .put(format!("{}/t/auto", server.base_url)) + .send() + .await? + .status(); + assert_eq!(status, 409); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] async fn concurrent_clones_and_pushes_with_telemetry() -> TestResult { // Regression: tracing spans entered across .await under a multi-threaded diff --git a/crates/walgit-server/tests/sim.rs b/crates/walgit-server/tests/sim.rs index d7a87d08..a0bdd9ea 100644 --- a/crates/walgit-server/tests/sim.rs +++ b/crates/walgit-server/tests/sim.rs @@ -1651,8 +1651,29 @@ async fn healthy_request_round_trip_budgets() -> Result<()> { cp.as_of.is_some() && cp.first_state_at.is_some(), "provenance times carried without a log read: {cp:?}" ); + // Create of an existing repository: the Create PUT's 412 is the answer, warm handle or + // cold, with no GET/HEAD in front of it — 1 request each. + let mut create_ops = Vec::new(); + for i in [0, cold] { + let before = c.instances[i].link.stats().ops.load(Ordering::Relaxed); + let again = c.instances[i] + .registry + .create(&c.id, ObjectFormat::Sha1) + .await; + let ops = c.instances[i].link.stats().ops.load(Ordering::Relaxed) - before; + ensure!( + matches!(again, Err(WalError::AlreadyExists)), + "create of an existing repository must be AlreadyExists, got {:?}", + again.map(|_| ()) + ); + ensure!( + ops == 1, + "create of an existing repository used {ops} requests, budget 1" + ); + create_ops.push(ops); + } eprintln!( - "healthy request counts: push={push_ops}, warm_refs={warm_ops}, cold_refs={cold_ops}, checkpoint={cp_ops}" + "healthy request counts: push={push_ops}, warm_refs={warm_ops}, cold_refs={cold_ops}, checkpoint={cp_ops}, create_existing={create_ops:?}" ); Ok(()) } diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index 9fc0147e..a815c8ce 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -194,19 +194,16 @@ impl Registry { } /// CAS-create manifest.pb (`PutMode::Create`). Err(AlreadyExists) on 412. + /// The Create PUT is the only authority, warm handle or not: a cached handle + /// is not evidence the manifest still exists (another host may have deleted + /// it) and must not turn a second create into a success. pub async fn create( &self, id: &RepoId, format: ObjectFormat, ) -> Result, WalError> { - if let Some(h) = self.repos.get(id) { - return Ok(h.clone()); - } let gate = self.opening.entry(id.clone()).or_default().clone(); let _g = gate.lock().await; - if let Some(h) = self.repos.get(id) { - return Ok(h.clone()); - } let prefix = id.store_prefix(); let prefixed = Prefixed::new(self.store.clone(), prefix); @@ -272,7 +269,8 @@ impl Registry { } } - /// Open or create. + /// Open or create. Losing a create race (another request or host created it + /// between our 404 and our Create) is success: open what the winner made. pub async fn open_or_create( &self, id: &RepoId, @@ -280,7 +278,10 @@ impl Registry { ) -> Result, WalError> { match self.open(id).await { Ok(h) => Ok(h), - Err(WalError::NotFound) => self.create(id, format).await, + Err(WalError::NotFound) => match self.create(id, format).await { + Err(WalError::AlreadyExists) => self.open(id).await, + other => other, + }, Err(e) => Err(e), } } diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index 8bd97b05..b3ac43d9 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -316,6 +316,74 @@ async fn test_two_registries_cross_sync() { ); } +/// A second create is `AlreadyExists` whether this registry holds the handle +/// (warm) or never saw the repository (cold): the Create PUT decides, not the map. +#[tokio::test] +async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let warm = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let cold = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("b"), 0)), + ); + let id = repo_id("test", "twice"); + + warm.create(&id, ObjectFormat::Sha1).await.unwrap(); + for registry in [&warm, &cold] { + assert!(matches!( + registry.create(&id, ObjectFormat::Sha1).await, + Err(walgit_wal::WalError::AlreadyExists) + )); + } + // The losing creates changed nothing: both still open the one repository. + assert_eq!(warm.open(&id).await.unwrap().manifest().revision, 1); + assert_eq!(cold.open(&id).await.unwrap().manifest().revision, 1); +} + +/// Concurrent creates of one name — on one registry and across two — have +/// exactly one winner; `open_or_create` (auto-create on push) treats losing +/// that race as success and opens the winner's repository. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_creates_have_one_winner_and_open_or_create_joins_it() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let a = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let b = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("b"), 0)), + ); + + for (n, (x, y)) in [(&a, &a), (&a, &b)].into_iter().enumerate() { + let id = repo_id("race", &format!("create{n}")); + let (rx, ry) = tokio::join!( + x.create(&id, ObjectFormat::Sha1), + y.create(&id, ObjectFormat::Sha1) + ); + let won = [rx.is_ok(), ry.is_ok()]; + assert_eq!(won.iter().filter(|w| **w).count(), 1, "{n}: one winner"); + for r in [rx, ry] { + if let Err(e) = r { + assert!(matches!(e, walgit_wal::WalError::AlreadyExists), "{n}: {e}"); + } + } + + let id = repo_id("race", &format!("auto{n}")); + let (hx, hy) = tokio::join!( + x.open_or_create(&id, ObjectFormat::Sha1), + y.open_or_create(&id, ObjectFormat::Sha1) + ); + assert_eq!(hx.unwrap().manifest().revision, 1, "{n}"); + assert_eq!(hy.unwrap().manifest().revision, 1, "{n}"); + } +} + #[tokio::test] async fn test_concurrent_different_refs() { let cache = tempfile::tempdir().unwrap(); diff --git a/docs/ROUNDTRIPS.md b/docs/ROUNDTRIPS.md index bed87112..513d7cb0 100644 --- a/docs/ROUNDTRIPS.md +++ b/docs/ROUNDTRIPS.md @@ -55,6 +55,7 @@ right shape. This document is the thinking tool; apply it to every protocol chan | Manifest CAS answered 412 | the same evidence check before dropping the own segment: a 412 is contention, or a client library retrying a write that already landed (S3 SDK standard mode, GCS uploads with a precondition). The segment is dropped only when the fresh manifest lists neither it nor a checkpoint at or past its seq (a fold drops descriptors); folded or otherwise unproven evidence stays unknown and keeps the segment | +1 manifest GET on the 412 path only (+1 log GET when our descriptor is listed); successful publishes unchanged | `publish.rs::slot_evidence`, `process_batch`, `publish_compact_classified`, `publish_settings_impl`, `classification.rs::seal_pack_replacement` | | Settings publish (D24) | refs sync (conditional GET) → log slot PUT → manifest CAS; readers pay nothing extra (settings ride inline on the manifest) | 3 rounds; read: 0 | `publish.rs::publish_settings_impl` | | Lease acquire | 1 GET → 1 CAS put (or 1 Create when absent) | 2 | `coord.rs::try_acquire` | +| Repository create (`PUT /{o}/{r}`, `walgit repo create`, auto-create on push) | 1 manifest Create PUT; its 412 *is* "exists" (409), whether this instance holds a handle or not; auto-create's lost race adds the open it would have done anyway | 1 (was 0 for a warm handle, which answered 201 for an existing repository) | `registry.rs::create`, `open_or_create` | | Publish, local commit (2026-08-23) | unchanged in round trips: after the manifest CAS the ref txns are applied to the local copy **before** the new manifest version is advertised, both under `sync_mutex` (the refs phase of every sync); the reverse order let a reader cache the old refs under the new version, and without the lock a concurrent sync replayed the same entry (two `update-ref`, a lock collision). A landed CAS is answered `ok` whatever the local apply does — the next sync replays (one conditional GET that then returns 200, no extra write). | 0 extra | `publish.rs::process_batch` | | Repository listing (`/api/v1/owners*`, `/services/api/owners*`, maintainer/bridge passes) | 0 within `LIST_TTL` (30 s, per instance); else delimited `repos/` → (delimited `repos//` ∥ owners) → (HEAD `manifest.pb` ∥ repos): 3 rounds | 1 + owners + repos | `registry.rs::list` | | Bundle removal (2026-09-11) | v2 capabilities and narrated fetch: removed optional list GET (1 → 0 extra); maintenance no longer reads/CASes a bundle list; direct import no longer composes a wrapper or reads/CASes a bundle list | no new store requests; checkpoint and push budgets unchanged | `smart.rs`, `maintain.rs`, `import_direct.rs` | @@ -62,7 +63,8 @@ right shape. This document is the thinking tool; apply it to every protocol chan | Orphan log slot (failure path only) | +1 fresh manifest GET, +HEAD per probe, +Create at next seq | — | `publish.rs::claim_log_slot` | `healthy_request_round_trip_budgets` in `crates/walgit-server/tests/sim.rs` pins the healthy MemoryStore -counts at push **5**, warm refs **1**, cold refs with one tail segment **2**, and checkpoint **4**. Cold open used to spend an +counts at push **5**, warm refs **1**, cold refs with one tail segment **2**, checkpoint **4**, and create of an +existing repository **1** (warm and cold). Cold open used to spend an extra unconditional manifest GET (3 requests, 3 sequential rounds); it now applies the manifest it already fetched directly (2 requests, 2 rounds). `claim_log_slot`, `cas_landed`, and `put_immutable_create` add probes only after Create/CAS failure, so the measured happy-path counts remain unchanged. From 17f637594f6627cb239aed216e9aee4512fa51c5 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 03:18:31 -0400 Subject: [PATCH 02/17] fix(wal): a created repository never inherits a stale local copy Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/registry.rs | 9 +++++++- crates/walgit-wal/tests/wal.rs | 38 +++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index a815c8ce..26db8503 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -235,7 +235,14 @@ impl Registry { .await { Ok(meta) => { - // Init local repo + // The Create landed, so nothing on disk belongs to this manifest: + // a copy left by a repository deleted elsewhere must not leak in. + let local_dir = id.local_dir(&self.cache_root); + if local_dir.exists() { + tokio::fs::remove_dir_all(&local_dir) + .await + .map_err(WalError::Io)?; + } let local = LocalRepo::init(&self.cache_root, id, format)?; let state = RepoState { diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index b3ac43d9..ae6c1766 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -344,6 +344,44 @@ async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { assert_eq!(cold.open(&id).await.unwrap().manifest().revision, 1); } +/// A create that lands proves nothing on disk belongs to the new manifest: the +/// copy left by a repository another host deleted must not leak into it. +#[tokio::test] +async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let registry = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let id = repo_id("test", "recreated"); + + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + assert_eq!(handle.local().refs().unwrap().refs.len(), 1); + + store + .delete( + &format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST), + None, + ) + .await + .unwrap(); + + let recreated = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + assert!(recreated.local().refs().unwrap().refs.is_empty()); +} + /// Concurrent creates of one name — on one registry and across two — have /// exactly one winner; `open_or_create` (auto-create on push) treats losing /// that race as success and opens the winner's repository. From ac1cef2d90fbf6fac1cf27bccc8ebd3f7a55ae59 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 04:00:10 -0400 Subject: [PATCH 03/17] fix(wal): recreate never deletes a local copy that in-flight work still uses Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/registry.rs | 65 ++++++++++++++++++++++++++----- crates/walgit-wal/tests/wal.rs | 28 +++++++++++-- 2 files changed, 81 insertions(+), 12 deletions(-) diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index 26db8503..b6a38c87 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -19,6 +19,47 @@ use crate::handle::RepoHandle; use crate::state::{RepoState, load_state, save_state}; use crate::store_proto::get_message; +/// Cache-root scratch directory for local copies superseded by a recreated +/// repository. Entries never end in `.git`, so no `RepoId::local_dir` can name one. +const STALE_COPIES_DIR: &str = "_stale"; + +/// Move a leftover local copy out of the way with an atomic rename to a unique +/// name, then reclaim it best-effort in the background. Holders of the old +/// handle keep working on the moved copy, isolated from the new repository. +/// If the rename fails, a best-effort in-place removal is the last resort so a +/// later `open` never inherits the stale refs. +fn set_stale_copy_aside(cache_root: &std::path::Path, id: &RepoId) -> Result<(), WalError> { + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let local_dir = id.local_dir(cache_root); + if !local_dir.exists() { + return Ok(()); + } + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos()); + let aside = cache_root.join(STALE_COPIES_DIR).join(format!( + "{}.{}.{}-{}-{nanos}", + id.owner(), + id.name(), + std::process::id(), + NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed), + )); + let moved = std::fs::create_dir_all(cache_root.join(STALE_COPIES_DIR)) + .and_then(|()| std::fs::rename(&local_dir, &aside)); + match moved { + Ok(()) => { + std::thread::spawn(move || { + let _ = std::fs::remove_dir_all(aside); + }); + Ok(()) + } + Err(e) => { + tracing::warn!(repo = %id, error = %e, "could not move the stale local copy aside"); + std::fs::remove_dir_all(&local_dir).map_err(WalError::Io) + } + } +} + pub struct Registry { store: DynStore, cfg: Arc, @@ -235,15 +276,21 @@ impl Registry { .await { Ok(meta) => { - // The Create landed, so nothing on disk belongs to this manifest: - // a copy left by a repository deleted elsewhere must not leak in. - let local_dir = id.local_dir(&self.cache_root); - if local_dir.exists() { - tokio::fs::remove_dir_all(&local_dir) - .await - .map_err(WalError::Io)?; - } - let local = LocalRepo::init(&self.cache_root, id, format)?; + // The Create landed, so nothing cached or on disk belongs to this + // manifest. Evict the stale handle first so `open` cannot hand it + // out; in-flight holders keep their own Arc on the moved copy. + // Nothing is cached until the new copy is fully built, so a failure + // below leaves `open` to rebuild from the manifest. + self.repos.remove(id); + self.invalidate_listing(); + let cache_root = self.cache_root.clone(); + let owned_id = id.clone(); + let local = tokio::task::spawn_blocking(move || { + set_stale_copy_aside(&cache_root, &owned_id)?; + Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) + }) + .await + .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??; let state = RepoState { manifest_version: Some(meta.version.as_str().to_string()), diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index ae6c1766..c12df943 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -344,8 +344,11 @@ async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { assert_eq!(cold.open(&id).await.unwrap().manifest().revision, 1); } -/// A create that lands proves nothing on disk belongs to the new manifest: the -/// copy left by a repository another host deleted must not leak into it. +/// A create that lands proves nothing cached or on disk belongs to the new +/// manifest: the copy left by a repository another host deleted must not leak +/// into it, and a request still working on the old copy must not stop the +/// create. A thread keeps writing into the old copy's object directory the whole +/// time, as an in-flight holder or its background work would. #[tokio::test] async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { let cache = tempfile::tempdir().unwrap(); @@ -378,8 +381,27 @@ async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { .await .unwrap(); - let recreated = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let stale_objects = handle.local().path().join("objects"); + let running = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let writer = { + let running = running.clone(); + std::thread::spawn(move || { + let mut n = 0u64; + while running.load(std::sync::atomic::Ordering::Relaxed) { + let _ = std::fs::write(stale_objects.join(format!("busy-{n}")), b"x"); + n += 1; + } + }) + }; + let recreated = registry.create(&id, ObjectFormat::Sha1).await; + running.store(false, std::sync::atomic::Ordering::Relaxed); + writer.join().unwrap(); + let recreated = recreated.unwrap(); + assert!(recreated.local().refs().unwrap().refs.is_empty()); + let opened = registry.open(&id).await.unwrap(); + assert!(Arc::ptr_eq(&opened, &recreated)); + assert!(!Arc::ptr_eq(&opened, &handle)); } /// Concurrent creates of one name — on one registry and across two — have From eb5ed5e61e2e829fd57347ba3b18c0cc4abaa2d0 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sun, 11 Oct 2026 23:19:43 +1100 Subject: [PATCH 04/17] =?UTF-8?q?Resolve=20a=20{"repo":=20=E2=80=A6}=20not?= =?UTF-8?q?ify=20under=20a=20store=20prefix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /_events/notify turns a plain {"repo": "o/r"} into the key repos/o/r/manifest.pb and resolves it like a bucket notification, which must start with store.prefix. With a non-empty prefix the key never matches: the notify is acked with an empty report and the bridge waits for its sweep (default 5 min), silently. Bucket notifications are fine: they carry the prefixed object key. Build the key the repository's manifest has in this bucket, prefix included. Keys in bucket notifications and {"key": …} stay bucket object keys; EVENTS.md now says so. The test harness now applies store.prefix the way walgit_store::from_config does for real backends, so a server test can run under a prefix. notify_wakes_the_bridge_under_a_store_prefix sends all four shapes to a prefixed server (before: {"repo": "t/r"} answered [] and woke nothing); the bridge unit test resolves them under a prefix. (cherry picked from commit 9a24f26da68c2cba6224c7b9e0007e3dbbd67725) Signed-off-by: Yordis Prieto --- crates/walgit-server/src/bridge.rs | 31 +++++++++++-- crates/walgit-server/tests/events.rs | 67 +++++++++++++++++++++++++++ crates/walgit-server/tests/harness.rs | 8 +++- docs/EVENTS.md | 4 +- 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/crates/walgit-server/src/bridge.rs b/crates/walgit-server/src/bridge.rs index b0a01903..c51ec2fb 100644 --- a/crates/walgit-server/src/bridge.rs +++ b/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 { +/// 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 { 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 { 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", diff --git a/crates/walgit-server/tests/events.rs b/crates/walgit-server/tests/events.rs index 6864dbeb..cf6110f0 100644 --- a/crates/walgit-server/tests/events.rs +++ b/crates/walgit-server/tests/events.rs @@ -206,6 +206,73 @@ async fn bridge_publishes_from_cursor_exactly_once() -> TestResult { Ok(()) } +/// With a bucket prefix (`store.prefix`), every notification shape wakes the bridge: bucket +/// notifications carry the prefixed object key, and a plain `{"repo": …}` names the repository +/// whatever the prefix. A key outside this walgit's prefix is acked and ignored. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn notify_wakes_the_bridge_under_a_store_prefix() -> TestResult { + use walgit_store::ObjectStoreExt; + let (url, captured) = webhook().await; + let server = Server::start_with_tweak(|c| { + bridge_cfg(&url, Duration::ZERO)(c); + c.store.prefix = "tenant-a".into(); + }) + .await?; + server.put_repo("t", "r").await?; + assert!( + server + .store + .exists("tenant-a/repos/t/r/manifest.pb") + .await?, + "the repository lives under the prefix" + ); + let src = TestRepo::synthetic(1, 1)?; + git_in(&src, &["commit", "--allow-empty", "-m", "a"])?; + git_in(&src, &["branch", "-M", "main"])?; + git_in( + &src, + &["remote", "add", "origin", &server.repo_url("t", "r")], + )?; + git_in(&src, &["push", "-u", "origin", "main"])?; + + let client = reqwest::Client::new(); + let notify = |body: serde_json::Value| { + let client = client.clone(); + let url = format!("{}/_events/notify", server.base_url); + async move { + let resp = client.post(url).json(&body).send().await?; + anyhow::ensure!(resp.status() == 200, "{body}: {}", resp.status()); + anyhow::Ok(resp.json::().await?) + } + }; + // Outside the prefix: not this walgit's object. + let report = notify(serde_json::json!({"key": "repos/t/r/manifest.pb"})).await?; + assert_eq!(report, serde_json::json!([])); + assert!(captured.lock().unwrap().is_empty()); + + let mut events = 0; + for body in [ + serde_json::json!({"repo": "t/r"}), + gcs_notification("tenant-a/repos/t/r/manifest.pb", "OBJECT_FINALIZE"), + serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put", "s3": {"object": {"key": "tenant-a/repos/t/r/manifest.pb"}}}]}), + serde_json::json!({"key": "tenant-a/repos/t/r/manifest.pb"}), + ] { + git_in(&src, &["commit", "--allow-empty", "-m", "next"])?; + git_in(&src, &["push"])?; + events += 1; + let report = notify(body.clone()).await?; + assert_eq!(report[0]["repo"], "t/r", "{body}: {report}"); + assert_eq!( + report[0]["emitted"], + if events == 1 { 2 } else { 1 }, + "{body}: {report}" + ); + } + let got = wait_for(&captured, 5).await; + assert_eq!(got.len(), 5); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn bridge_sweep_timer_publishes_without_notifications() -> TestResult { let (url, captured) = webhook().await; diff --git a/crates/walgit-server/tests/harness.rs b/crates/walgit-server/tests/harness.rs index 78d0d6b5..bc1f6c05 100644 --- a/crates/walgit-server/tests/harness.rs +++ b/crates/walgit-server/tests/harness.rs @@ -109,7 +109,13 @@ impl Server { cfg.server.public_url = Some(base_url.clone()); cfg.validate().context("config validate")?; - let dyn_store: DynStore = store.clone(); + // `store.prefix` scopes every key, as `walgit_store::from_config` does for real backends. + let prefix = cfg.store_prefix(); + let dyn_store: DynStore = if prefix.is_empty() { + store.clone() + } else { + Arc::new(walgit_store::Prefixed::new(store.clone(), prefix)) + }; let state = AppState::new(Arc::new(cfg), dyn_store).await?; let registry = state.registry.clone(); diff --git a/docs/EVENTS.md b/docs/EVENTS.md index 6ca6b97f..8f52b709 100644 --- a/docs/EVENTS.md +++ b/docs/EVENTS.md @@ -97,7 +97,9 @@ Wake-ups (both idempotent; they only ever call `catch_up`): itself as the notification. Accepted bodies: a GCS Pub/Sub push envelope (`message.attributes.eventType = OBJECT_FINALIZE`, `objectId`), an S3 event notification (`Records[].eventName = ObjectCreated:*`, `s3.object.key`; MinIO, rustfs and Ceph emit the same shape), or your own glue's `{"key": "repos/o/r/manifest.pb"}` - / `{"repo": "o/r"}`. Everything else is acked and ignored; a webhook failure answers 503 so the notifier + / `{"repo": "o/r"}`. Keys are bucket object keys as notifications carry them, `store.prefix` included + (`{"key": "/repos/o/r/manifest.pb"}`); `{"repo": …}` names the repository whatever the prefix. + Everything else is acked and ignored; a webhook failure answers 503 so the notifier redelivers. Authenticated like every route (`require_read`): give the notifier a token. - The sweep (`events.sweep_interval`, default 5 min): `list` + one conditional manifest GET per repo. Not needed for correctness; it is the backstop *and the health check* — a sweep that publishes anything means From 4ac4096961a0d2ec2b4802769e6aeb7b27197d6e Mon Sep 17 00:00:00 2001 From: kevin9327 <5299031+kevin9327@users.noreply.github.com> Date: Tue, 15 Sep 2026 21:40:02 +0900 Subject: [PATCH 05/17] Apply env overrides whose text reads as a TOML number or boolean WALGIT__ values are parsed as TOML and fall back to a string only when they do not parse. A value that parses as a number or boolean but lands on a key that takes text or a duration failed the type check and was dropped as "unknown in this build": WALGIT__STORE__BUCKET=20260915 left the host on the file's bucket, WALGIT__MAINTENANCE__FOLLOW_INTERVAL=0 (the documented "off") kept following every 30 s. Try the TOML reading first and the literal text second; only a value that fits neither is ignored, with the TOML reading's reason. (cherry picked from commit 32e060166eebcd902ee3e9eb19c8a56581726fb6) Signed-off-by: Yordis Prieto --- crates/walgit-config/src/lib.rs | 85 +++++++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 19 deletions(-) diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index e3cdaf15..8e0d9ebc 100644 --- a/crates/walgit-config/src/lib.rs +++ b/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::() - .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::().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::() { + 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::() { + 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 @@ -1514,6 +1520,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 [ From d216b78b4daab759aaf7d8084e8150e367d7402d Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 30 Sep 2026 11:08:46 -0700 Subject: [PATCH 06/17] Reject zero per-repo concurrency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A zero-sized per-repository semaphore can never grant a permit, so a valid-looking configuration caused every Git request for a repository to wait indefinitely. Require `server.max_concurrent_per_repo` to be positive and add a regression test for the invalid zero value. ✅ `cargo test -p walgit-config zero_per_repo_concurrency_is_rejected` ✅ `cargo fmt --all -- --check` ✅ `git diff --check` Closes #79 (cherry picked from commit 77ade75a8f024ae60b6dc6e3c60cd5f58a5cbdc0) Signed-off-by: Yordis Prieto --- crates/walgit-config/src/lib.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) 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(); From 83f9125d8797e6d099f0644680610137e4084cc0 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 30 Sep 2026 10:20:20 -0700 Subject: [PATCH 07/17] Fix telemetry metrics configuration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `telemetry.metrics = false` did not affect server startup or routing, so the recorder was always installed and `/metrics` remained available. Install the Prometheus recorder and authenticated route only when metrics are enabled. Reserve `/metrics` with an explicit 404 when disabled so the SPA owner route cannot handle it. Add enabled/disabled HTTP coverage to the fast suite. - ✅ `cargo test -p walgit-server --test metrics --test routing_prefix` - ✅ `cargo clippy -p walgit-server --all-targets -- -D warnings` - ✅ `cargo fmt --all -- --check` Closes #75 (cherry picked from commit 808139efb3755fa4e1cc02623063cff0ffa0dd10) Signed-off-by: Yordis Prieto --- crates/walgit-server/src/lib.rs | 28 +++++++++++++++++++++++---- crates/walgit-server/src/metrics.rs | 11 +++++++---- crates/walgit-server/tests/metrics.rs | 22 +++++++++++++++++++++ justfile | 2 +- 4 files changed, 54 insertions(+), 9 deletions(-) create mode 100644 crates/walgit-server/tests/metrics.rs diff --git a/crates/walgit-server/src/lib.rs b/crates/walgit-server/src/lib.rs index 5f6c6154..8bb2cff7 100644 --- a/crates/walgit-server/src/lib.rs +++ b/crates/walgit-server/src/lib.rs @@ -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,25 @@ 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 state.cfg.telemetry.metrics { + Router::new() + .route("/metrics", get(metrics::metrics_route)) + .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 +199,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..44e982a4 100644 --- a/crates/walgit-server/src/metrics.rs +++ b/crates/walgit-server/src/metrics.rs @@ -1,5 +1,5 @@ -//! 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}; @@ -33,12 +33,15 @@ pub fn install() -> anyhow::Result> { /// `GET /metrics` pub async fn metrics_route(State(st): State>) -> impl IntoResponse { - let body = st.metrics_handle.render(); + let Some(handle) = &st.metrics_handle else { + return axum::http::StatusCode::NOT_FOUND.into_response(); + }; ( [( 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/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: From 0ec7d3e42b8b74fd419d7997319fe724beb6c513 Mon Sep 17 00:00:00 2001 From: ethanstoner Date: Fri, 18 Sep 2026 10:56:59 -0700 Subject: [PATCH 08/17] Retry GCS 408, 502 and client timeouts like S3 gcs::is_retryable treated HTTP 408 and 502 as permanent and ignored the client's own timeout errors, so a bad gateway on a manifest CAS failed the push on GCS while the same response on S3 is retried (#18). Use the status set GCS documents as retryable (408, 429, 500, 502, 503, 504), leave 501 out, and share the same set with s3::is_transient_status. (cherry picked from commit 939348a2d40fe59af8dc91c6de050b78b5b9e4a0) Signed-off-by: Yordis Prieto --- crates/walgit-store/src/gcs.rs | 34 ++++++++++++++++++++++++++++++++-- crates/walgit-store/src/s3.rs | 6 +++--- 2 files changed, 35 insertions(+), 5 deletions(-) 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 08409d53..2138941b 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -261,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. @@ -1303,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"); } } From cdeb9c82c0cd9f27f6c0601eb95646bfe45e84a4 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sun, 11 Oct 2026 23:18:04 +1100 Subject: [PATCH 09/17] Send S3 conditional headers with quoted entity-tags S3Store keeps ETags unquoted in Version (quotes are stripped on read) and put them on the wire as they are: `If-Match: abc` on the manifest CAS and on GETs, `If-None-Match: abc` on the conditional manifest GET. HTTP defines both headers over quoted entity-tags (RFC 9110 sections 8.8.3 and 13.1.1; the S3 API docs defer to RFC 7232), and the ETag header of every S3 response carries that form. AWS S3 (its CLI example sends a bare value) and rustfs accept the bare value too, but that is not documented for S3-compatible stores in general, and Cloudflare R2's docs are silent on it. A store that compares the tag strictly would refuse every manifest CAS (no push could ever commit) and never answer the conditional GET with 304. entity_tag() renders a Version as a quoted entity-tag (once; an already quoted value is left alone) and every conditional header uses it. Versions stay unquoted, so nothing stored or compared changes. Checked: the store contract suite passes against rustfs 1.0.1, which honours both forms; against Cloudflare R2, a quoted If-Match returned 200 on a match and 412 on a mismatch. A new unit test records the headers walgit sends to a fake S3; before this change it saw `If-Match: abc`. (cherry picked from commit 7143aad6f82624bd545847bc9bad6bed0abd820f) Signed-off-by: Yordis Prieto --- crates/walgit-store/src/s3.rs | 149 ++++++++++++++++++++++++++++++++-- 1 file changed, 140 insertions(+), 9 deletions(-) diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs index d788cc2e..08409d53 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)); @@ -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 [ From 0bf402da646d04b08bdc6d64208e2cfc31923482 Mon Sep 17 00:00:00 2001 From: Yad Konrad Date: Fri, 25 Sep 2026 10:25:21 -0400 Subject: [PATCH 10/17] Pass the pack lifecycle's sort budget in bytes GNU and BSD sort both read a bare --buffer-size number as KiB, so the lifecycle's byte budget (clamped to 64 MiB) reached sort as 64 GiB and never made it spill to the work directory. Measured on a 21 MB object list with a 1 MiB budget: GNU sort peaked at 46 MB RSS without the suffix and 7.2 MB with it; BSD sort at 87 MB and 3.6 MB. (cherry picked from commit 23320a1957f953272207dc2bfc30d51cc0b01eff) Signed-off-by: Yordis Prieto --- crates/walgit-git/src/pack_segments.rs | 23 +++++++++++--------- crates/walgit-git/src/pack_segments_tests.rs | 9 ++++++++ 2 files changed, 22 insertions(+), 10 deletions(-) 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(); From 1eb37d788edc17fba83fa1ed58b059c8e6c91a73 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Thu, 1 Oct 2026 10:57:54 -0700 Subject: [PATCH 11/17] Remove redundant metrics handler guard Use the optional Prometheus handle when constructing the router and inject the enabled handle into the route. The handler can now render without rechecking an impossible disabled state. (cherry picked from commit 094964f3a5056f6d8d004760ec51fcb849ff9325) Signed-off-by: Yordis Prieto --- crates/walgit-server/src/lib.rs | 5 +++-- crates/walgit-server/src/metrics.rs | 11 ++++------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/crates/walgit-server/src/lib.rs b/crates/walgit-server/src/lib.rs index 8bb2cff7..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; @@ -171,9 +171,10 @@ pub fn router(state: Arc) -> Router { state.clone(), web::require_auth, )); - let metrics = if state.cfg.telemetry.metrics { + 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, diff --git a/crates/walgit-server/src/metrics.rs b/crates/walgit-server/src/metrics.rs index 44e982a4..5cd09d48 100644 --- a/crates/walgit-server/src/metrics.rs +++ b/crates/walgit-server/src/metrics.rs @@ -3,12 +3,10 @@ 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,10 +30,9 @@ pub fn install() -> anyhow::Result> { } /// `GET /metrics` -pub async fn metrics_route(State(st): State>) -> impl IntoResponse { - let Some(handle) = &st.metrics_handle else { - return axum::http::StatusCode::NOT_FOUND.into_response(); - }; +pub async fn metrics_route( + Extension(handle): Extension>, +) -> impl IntoResponse { ( [( axum::http::header::CONTENT_TYPE, From 5d7664371f5db372fee93e50b1a6dd59fe5764de Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 06:14:08 -0400 Subject: [PATCH 12/17] fix(wal): recreate retires and drains the old handle before resetting its directory Signed-off-by: Yordis Prieto --- crates/walgit-server/src/pack_lifecycle.rs | 4 +- crates/walgit-wal/src/checkpoint.rs | 2 +- crates/walgit-wal/src/handle.rs | 49 ++++++++++++++++-- crates/walgit-wal/src/publish.rs | 24 ++++----- crates/walgit-wal/src/registry.rs | 58 +++++---------------- crates/walgit-wal/src/snapshots.rs | 1 + crates/walgit-wal/src/sync.rs | 7 +-- crates/walgit-wal/tests/wal.rs | 59 ++++++++++++++-------- 8 files changed, 118 insertions(+), 86 deletions(-) diff --git a/crates/walgit-server/src/pack_lifecycle.rs b/crates/walgit-server/src/pack_lifecycle.rs index 31cbe5b6..8ae0dec7 100644 --- a/crates/walgit-server/src/pack_lifecycle.rs +++ b/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) diff --git a/crates/walgit-wal/src/checkpoint.rs b/crates/walgit-wal/src/checkpoint.rs index 5834c207..62354a0c 100644 --- a/crates/walgit-wal/src/checkpoint.rs +++ b/crates/walgit-wal/src/checkpoint.rs @@ -259,7 +259,7 @@ async fn write_checkpoint_inner(handle: &RepoHandle) -> Result { diff --git a/crates/walgit-wal/src/handle.rs b/crates/walgit-wal/src/handle.rs index 611fa3b4..e6198ab8 100644 --- a/crates/walgit-wal/src/handle.rs +++ b/crates/walgit-wal/src/handle.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use std::sync::{ Arc, - atomic::{AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }; use std::time::Instant; @@ -54,6 +54,9 @@ pub struct RepoHandle { /// stuck behind a multi-GB materialization (only removals take the write /// lock, briefly). pub(crate) pack_mutex: TokioMutex<()>, + /// Set when `Registry::create` replaces this repository's local copy: the + /// path now belongs to the next generation, so nothing here may touch it. + retired: AtomicBool, // Current manifest (last known). Short critical sections, no await. pub(crate) manifest: PLRwLock>, @@ -152,6 +155,7 @@ impl RepoHandle { rw: Arc::new(TokioRwLock::new(())), sync_mutex: TokioMutex::new(()), pack_mutex: TokioMutex::new(()), + retired: AtomicBool::new(false), manifest: PLRwLock::new(Arc::new(manifest)), manifest_version: PLMutex::new(version), state: PLMutex::new(state), @@ -356,6 +360,35 @@ impl RepoHandle { &self.local } + /// Fails once the local copy was handed to a newer generation of this repository. + pub(crate) fn ensure_live(&self) -> Result<(), WalError> { + if self.retired.load(Ordering::Acquire) { + return Err(WalError::NotFound); + } + Ok(()) + } + + pub(crate) fn persist_state(&self) -> Result<(), WalError> { + self.ensure_live()?; + crate::state::save_state(self.local.path(), &self.state.lock().clone()) + } + + /// Retire the handle, then wait for every holder of its locks. Requests that start + /// later fail `ensure_live`, so once the guards are held nothing touches the directory. + pub(crate) async fn retire_and_drain( + &self, + ) -> ( + tokio::sync::OwnedRwLockWriteGuard<()>, + tokio::sync::MutexGuard<'_, ()>, + tokio::sync::MutexGuard<'_, ()>, + ) { + self.retired.store(true, Ordering::Release); + let write = self.rw.clone().write_owned().await; + let sync = self.sync_mutex.lock().await; + let pack = self.pack_mutex.lock().await; + (write, sync, pack) + } + pub fn store(&self) -> &Prefixed { &self.store } @@ -423,8 +456,12 @@ impl RepoHandle { /// Pin installed object files across an owned blocking maintenance task. /// Unlike a borrowed request guard, cancellation of its caller cannot release /// this pin while a detached blocking worker still copies the input files. - pub async fn pin_local_objects(&self) -> tokio::sync::OwnedRwLockReadGuard<()> { - self.rw.clone().read_owned().await + pub async fn pin_local_objects( + &self, + ) -> Result, WalError> { + let pin = self.rw.clone().read_owned().await; + self.ensure_live()?; + Ok(pin) } pub async fn sync_full(&self) -> Result, WalError> { @@ -595,6 +632,7 @@ impl RepoHandle { changed = false, entries_applied = 0u64, ); + self.ensure_live()?; self.touch(); // Phase 1 — refs (manifest freshness + ref state), under sync_mutex // only; sub-second. Phase 2 — packs, under pack_mutex only. Neither @@ -619,6 +657,7 @@ impl RepoHandle { self.rw.read().instrument(span.clone()), ) .await; + self.ensure_live()?; Ok(crate::sync::ReadGuard { _guard: read_guard, handle: self, @@ -628,6 +667,7 @@ impl RepoHandle { /// Manifest freshness check + ref state apply (never packs, never /// `rw.write()`: ref files and the gix handle are replaced atomically). async fn sync_refs_phase(&self, span: &tracing::Span) -> Result<(), WalError> { + self.ensure_live()?; if self.freshness_ttl_active() { return Ok(()); } @@ -666,6 +706,7 @@ impl RepoHandle { self.pack_mutex.lock(), ) .await; + self.ensure_live()?; if self.level_satisfied(level) { return Ok(()); } @@ -824,6 +865,7 @@ impl RepoHandle { /// Freshness check + refs apply, with `sync_mutex` and the write lock held /// by the caller. Packs are never touched here (see `sync_packs_phase`). pub(crate) async fn sync_locked_inner(&self, span: &tracing::Span) -> Result<(), WalError> { + self.ensure_live()?; let known = self.manifest_version.lock().clone(); let outcome = crate::sync::freshness_check(&self.store, known.as_ref()).await?; match outcome { @@ -931,6 +973,7 @@ impl RepoHandle { self.pack_mutex.lock(), ) .await; + self.ensure_live()?; // Read manifest fresh let Some((meta, manifest)) = diff --git a/crates/walgit-wal/src/publish.rs b/crates/walgit-wal/src/publish.rs index d6d7db4e..87ccfcd7 100644 --- a/crates/walgit-wal/src/publish.rs +++ b/crates/walgit-wal/src/publish.rs @@ -582,6 +582,9 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul // lock that applies refs. Release it before any store or bulk work. let (manifest, known_version, mut working_refs) = { let _sync = handle.sync_mutex.lock().await; + if let Err(e) = handle.ensure_live() { + return finish_all_errors(batch, e); + } let (manifest, version) = handle.manifest_pair(); let refs = match handle.local.ref_view() { Ok(v) => v, @@ -865,7 +868,6 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul // every waiter is answered `ok`. A failed apply leaves the version unadvertised, so the next // sync sees a change and replays the entry — the copy repairs itself. (Answering an error // here produced a durable push that git reported as failed — "0 winners", commit fetchable.) - let mut local_ok = true; { let _sync_guard = crate::lockwait::timed( "sync_mutex", @@ -875,6 +877,8 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul handle.sync_mutex.lock(), ) .await; + let live = handle.ensure_live().is_ok(); + let mut local_ok = live; let applied_seq = handle.state.lock().applied_seq; let already_applied = applied_seq >= last_seq; if applied_seq < committed.head_seq @@ -884,11 +888,9 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul tracing::warn!(error = %e, "landed push catch-up failed; next sync repairs it"); local_ok = false; } - for (req, _) in batch - .iter() - .zip(&verified) - .filter(|(_, v)| v.valid && !already_applied && committed.head_seq == last_seq) - { + for (req, _) in batch.iter().zip(&verified).filter(|(_, v)| { + live && v.valid && !already_applied && committed.head_seq == last_seq + }) { if let Err(e) = handle.local.apply_ref_txn(&req.txn, false) { tracing::warn!(repo = %handle.id, seq = last_seq, error = %e, "published (CAS ok), but applying the ref txn to the local copy failed; the next sync replays it"); metrics::counter!("walgit_publish_local_apply_failed_total").increment(1); @@ -921,9 +923,7 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul state.packs_revision = committed.revision; } } - if let Err(e) = - crate::state::save_state(handle.local.path(), &handle.state.lock().clone()) - { + if let Err(e) = handle.persist_state() { tracing::warn!(repo = %handle.id, error = %e, "published (CAS ok), but saving local state failed; the next sync repairs it"); } } else if !local_ok { @@ -1320,7 +1320,7 @@ pub(crate) async fn publish_compact_classified( state.packs_revision = committed.revision; } } - crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?; + handle.persist_state()?; sweep_burned(&handle.store, &slot).await; return Ok(seq); } @@ -1428,7 +1428,7 @@ pub(crate) async fn annotate_pack_impl( state.manifest_version = Some(meta.version.as_str().to_string()); state.revision = updated.revision; } - crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?; + handle.persist_state()?; return Ok(pack_ref); } Err(StoreError::PreconditionFailed { .. }) => { @@ -1590,7 +1590,7 @@ pub(crate) async fn publish_settings_impl( state.packs_revision = updated.revision; } } - crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?; + handle.persist_state()?; *handle.effective.lock() = None; sweep_burned(&handle.store, &slot).await; tracing::info!(repo = %handle.id, seq, revision, author, "settings published"); diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index b6a38c87..38b75521 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -19,47 +19,6 @@ use crate::handle::RepoHandle; use crate::state::{RepoState, load_state, save_state}; use crate::store_proto::get_message; -/// Cache-root scratch directory for local copies superseded by a recreated -/// repository. Entries never end in `.git`, so no `RepoId::local_dir` can name one. -const STALE_COPIES_DIR: &str = "_stale"; - -/// Move a leftover local copy out of the way with an atomic rename to a unique -/// name, then reclaim it best-effort in the background. Holders of the old -/// handle keep working on the moved copy, isolated from the new repository. -/// If the rename fails, a best-effort in-place removal is the last resort so a -/// later `open` never inherits the stale refs. -fn set_stale_copy_aside(cache_root: &std::path::Path, id: &RepoId) -> Result<(), WalError> { - static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); - let local_dir = id.local_dir(cache_root); - if !local_dir.exists() { - return Ok(()); - } - let nanos = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.as_nanos()); - let aside = cache_root.join(STALE_COPIES_DIR).join(format!( - "{}.{}.{}-{}-{nanos}", - id.owner(), - id.name(), - std::process::id(), - NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed), - )); - let moved = std::fs::create_dir_all(cache_root.join(STALE_COPIES_DIR)) - .and_then(|()| std::fs::rename(&local_dir, &aside)); - match moved { - Ok(()) => { - std::thread::spawn(move || { - let _ = std::fs::remove_dir_all(aside); - }); - Ok(()) - } - Err(e) => { - tracing::warn!(repo = %id, error = %e, "could not move the stale local copy aside"); - std::fs::remove_dir_all(&local_dir).map_err(WalError::Io) - } - } -} - pub struct Registry { store: DynStore, cfg: Arc, @@ -277,16 +236,25 @@ impl Registry { { Ok(meta) => { // The Create landed, so nothing cached or on disk belongs to this - // manifest. Evict the stale handle first so `open` cannot hand it - // out; in-flight holders keep their own Arc on the moved copy. + // manifest. Evict the stale handle so `open` cannot hand it out, then + // wait for its in-flight work before the directory is reset in place: + // holders keep the path, so the handle is retired and drained first. // Nothing is cached until the new copy is fully built, so a failure // below leaves `open` to rebuild from the manifest. - self.repos.remove(id); + let stale = self.repos.remove(id).map(|(_, handle)| handle); self.invalidate_listing(); + let _drained = match &stale { + Some(handle) => Some(handle.retire_and_drain().await), + None => None, + }; let cache_root = self.cache_root.clone(); let owned_id = id.clone(); let local = tokio::task::spawn_blocking(move || { - set_stale_copy_aside(&cache_root, &owned_id)?; + match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(WalError::Io(e)), + } Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) }) .await diff --git a/crates/walgit-wal/src/snapshots.rs b/crates/walgit-wal/src/snapshots.rs index f0b04903..5eec4f9b 100644 --- a/crates/walgit-wal/src/snapshots.rs +++ b/crates/walgit-wal/src/snapshots.rs @@ -154,6 +154,7 @@ pub(crate) async fn checkpoint_snapshot( impl RepoHandle { pub async fn fetch_view(&self) -> Result { let _sync = self.sync_mutex.lock().await; + self.ensure_live()?; let manifest = self.manifest(); let local = self.local.clone(); let refs = tokio::task::spawn_blocking(move || local.ref_view()) diff --git a/crates/walgit-wal/src/sync.rs b/crates/walgit-wal/src/sync.rs index d2711057..c2ad5f4d 100644 --- a/crates/walgit-wal/src/sync.rs +++ b/crates/walgit-wal/src/sync.rs @@ -473,6 +473,7 @@ pub(crate) async fn apply_delta( new_manifest: &Manifest, new_version: &Version, ) -> Result<(), WalError> { + handle.ensure_live()?; let local = &handle.local; let current_state = handle.state.lock().clone(); @@ -517,7 +518,7 @@ pub(crate) async fn apply_delta( state.packs_revision = new_manifest.revision; } } - crate::state::save_state(local.path(), &handle.state.lock().clone())?; + handle.persist_state()?; local.refresh_async().await?; Ok(()) } @@ -829,7 +830,7 @@ pub(crate) async fn reconcile_packs_inner( .map_err(|e| WalError::Corrupt(format!("pack checksum: {e}"))) }) .collect::, _>>()?; - let pin = handle.pin_local_objects().await; + let pin = handle.pin_local_objects().await?; let repo = local.clone(); let (refs, pin) = tokio::task::spawn_blocking(move || { let refs = repo.refs(); @@ -861,7 +862,7 @@ pub(crate) async fn reconcile_packs_inner( let mut state = handle.state.lock(); state.packs_revision = manifest.revision; } - crate::state::save_state(local.path(), &handle.state.lock().clone())?; + handle.persist_state()?; Ok(()) } diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index c12df943..8db377a2 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -22,7 +22,7 @@ use walgit_git::{IngestOptions, ObjectFormat, RepoId}; use walgit_proto::v1::{EntryKind, RefTransaction, RefUpdate}; use walgit_store::ObjectStore; use walgit_store::memory::MemoryStore; -use walgit_wal::Registry; +use walgit_wal::{Registry, WalError}; use std::io::Write; @@ -346,9 +346,9 @@ async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { /// A create that lands proves nothing cached or on disk belongs to the new /// manifest: the copy left by a repository another host deleted must not leak -/// into it, and a request still working on the old copy must not stop the -/// create. A thread keeps writing into the old copy's object directory the whole -/// time, as an in-flight holder or its background work would. +/// into it. The old handle is retired, its in-flight work finishes before the +/// directory is reset, and anything it starts afterwards fails instead of +/// writing into the new repository. #[tokio::test] async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { let cache = tempfile::tempdir().unwrap(); @@ -381,23 +381,42 @@ async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { .await .unwrap(); - let stale_objects = handle.local().path().join("objects"); - let running = Arc::new(std::sync::atomic::AtomicBool::new(true)); - let writer = { - let running = running.clone(); - std::thread::spawn(move || { - let mut n = 0u64; - while running.load(std::sync::atomic::Ordering::Relaxed) { - let _ = std::fs::write(stale_objects.join(format!("busy-{n}")), b"x"); - n += 1; - } - }) - }; - let recreated = registry.create(&id, ObjectFormat::Sha1).await; - running.store(false, std::sync::atomic::Ordering::Relaxed); - writer.join().unwrap(); - let recreated = recreated.unwrap(); + let in_flight = handle.sync_refs().await.unwrap(); + let mut create = tokio::spawn({ + let registry = registry.clone(); + let id = id.clone(); + async move { registry.create(&id, ObjectFormat::Sha1).await } + }); + // The create retires the handle before it waits; a request started now is refused. + loop { + match handle.sync_refs().await { + Err(WalError::NotFound) => break, + Ok(guard) => drop(guard), + Err(e) => panic!("unexpected {e}"), + } + tokio::task::yield_now().await; + } + assert!( + tokio::time::timeout(std::time::Duration::from_millis(100), &mut create) + .await + .is_err(), + "create must wait for the request still holding the old handle" + ); + assert_eq!(handle.local().refs().unwrap().refs.len(), 1); + + drop(in_flight); + let recreated = create.await.unwrap().unwrap(); + + assert!(matches!(handle.sync().await, Err(WalError::NotFound))); + let late = handle + .publish_push( + None, + make_txn(vec![("refs/heads/late", "", &c1)]), + HashMap::new(), + ) + .await; + assert!(late.is_err()); assert!(recreated.local().refs().unwrap().refs.is_empty()); let opened = registry.open(&id).await.unwrap(); assert!(Arc::ptr_eq(&opened, &recreated)); From 67d3df63fdd8e6f41903d25c9de790acf34c3596 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 06:30:34 -0400 Subject: [PATCH 13/17] fix(wal): drain retired handles without nested locks and rebuild interrupted resets Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/handle.rs | 24 ++++----- crates/walgit-wal/src/publish.rs | 39 ++++++++------- crates/walgit-wal/src/registry.rs | 83 +++++++++++++++++++++++++++---- crates/walgit-wal/tests/wal.rs | 46 +++++++++++++++++ 4 files changed, 152 insertions(+), 40 deletions(-) diff --git a/crates/walgit-wal/src/handle.rs b/crates/walgit-wal/src/handle.rs index e6198ab8..8f782155 100644 --- a/crates/walgit-wal/src/handle.rs +++ b/crates/walgit-wal/src/handle.rs @@ -373,20 +373,16 @@ impl RepoHandle { crate::state::save_state(self.local.path(), &self.state.lock().clone()) } - /// Retire the handle, then wait for every holder of its locks. Requests that start - /// later fail `ensure_live`, so once the guards are held nothing touches the directory. - pub(crate) async fn retire_and_drain( - &self, - ) -> ( - tokio::sync::OwnedRwLockWriteGuard<()>, - tokio::sync::MutexGuard<'_, ()>, - tokio::sync::MutexGuard<'_, ()>, - ) { + /// Retire the handle, then wait out every holder that began before it. Each lock is + /// taken and released in turn, never held while waiting for the next, so the drain adds + /// no lock-order edge (a queued `rw` writer behind a holder of `pack_mutex`/`sync_mutex` + /// that wants `rw.read` would deadlock). Every acquirer checks `ensure_live` after it + /// holds its lock, so anything that starts later bails before touching the directory. + pub(crate) async fn retire_and_drain(&self) { self.retired.store(true, Ordering::Release); - let write = self.rw.clone().write_owned().await; - let sync = self.sync_mutex.lock().await; - let pack = self.pack_mutex.lock().await; - (write, sync, pack) + drop(self.pack_mutex.lock().await); + drop(self.sync_mutex.lock().await); + drop(self.rw.write().await); } pub fn store(&self) -> &Prefixed { @@ -549,6 +545,7 @@ impl RepoHandle { arc.pack_mutex.lock(), ) .await; + arc.ensure_live()?; let manifest = arc.manifest(); let task = match arc.begin_task("history-pack", HashMap::new()) { Begin::Started(t) => Some(t), @@ -679,6 +676,7 @@ impl RepoHandle { self.sync_mutex.lock(), ) .await; + self.ensure_live()?; if self.freshness_ttl_active() { return Ok(()); } diff --git a/crates/walgit-wal/src/publish.rs b/crates/walgit-wal/src/publish.rs index 87ccfcd7..222802f8 100644 --- a/crates/walgit-wal/src/publish.rs +++ b/crates/walgit-wal/src/publish.rs @@ -1466,25 +1466,28 @@ pub(crate) async fn add_pack_impl( let checksum = gix_hash::ObjectId::from_hex(hex.as_bytes()) .map_err(|e| WalError::Corrupt(format!("bad pack name {name}: {e}")))?; let dest = handle.local.pack_path(&checksum); - std::fs::create_dir_all( - dest.parent() - .ok_or_else(|| WalError::Corrupt("destination has no parent".into()))?, - )?; - for (src, dst) in [(pack, dest.clone()), (idx, dest.with_extension("idx"))] { - if !dst.exists() && std::fs::hard_link(src, &dst).is_err() { - std::fs::copy(src, &dst)?; + let info = { + let _pin = handle.pin_local_objects().await?; + std::fs::create_dir_all( + dest.parent() + .ok_or_else(|| WalError::Corrupt("destination has no parent".into()))?, + )?; + for (src, dst) in [(pack, dest.clone()), (idx, dest.with_extension("idx"))] { + if !dst.exists() && std::fs::hard_link(src, &dst).is_err() { + std::fs::copy(src, &dst)?; + } } - } - if let Some(base) = &history_of { - handle.local.mark_history_pack(&checksum, base).await?; - } - handle.local.refresh_async().await?; - let info = handle - .local - .packs()? - .into_iter() - .find(|p| p.checksum == checksum) - .ok_or_else(|| WalError::Corrupt(format!("pack {hex} not visible after install")))?; + if let Some(base) = &history_of { + handle.local.mark_history_pack(&checksum, base).await?; + } + handle.local.refresh_async().await?; + handle + .local + .packs()? + .into_iter() + .find(|p| p.checksum == checksum) + .ok_or_else(|| WalError::Corrupt(format!("pack {hex} not visible after install")))? + }; publish_compact_impl(handle, info, Vec::new(), tier).await } diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index 38b75521..eb39b534 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -107,8 +107,21 @@ impl Registry { let local = if let Some(l) = LocalRepo::open(&self.cache_root, id)? { l } else { + // A directory without HEAD is an unfinished init or an interrupted reset: + // whatever it still holds is not this repository's. + let cache_root = self.cache_root.clone(); + let owned_id = id.clone(); let format = parse_object_format(&manifest.object_format); - LocalRepo::init(&self.cache_root, id, format)? + tokio::task::spawn_blocking(move || { + match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(WalError::Io(e)), + } + Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) + }) + .await + .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))?? }; // Load state @@ -243,17 +256,24 @@ impl Registry { // below leaves `open` to rebuild from the manifest. let stale = self.repos.remove(id).map(|(_, handle)| handle); self.invalidate_listing(); - let _drained = match &stale { - Some(handle) => Some(handle.retire_and_drain().await), - None => None, - }; + if let Some(handle) = &stale { + handle.retire_and_drain().await; + } let cache_root = self.cache_root.clone(); let owned_id = id.clone(); let local = tokio::task::spawn_blocking(move || { - match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(WalError::Io(e)), + // Without HEAD `LocalRepo::open` rejects the directory, so a removal that + // fails midway cannot be mistaken for the old repository: `open` rebuilds. + let dir = owned_id.local_dir(&cache_root); + for result in [ + std::fs::remove_file(dir.join("HEAD")), + std::fs::remove_dir_all(&dir), + ] { + match result { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(WalError::Io(e)), + } } Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) }) @@ -469,6 +489,9 @@ impl Registry { let Ok(_write) = handle.rw.try_write() else { continue; }; + if handle.ensure_live().is_err() { + continue; + } let path = handle.local.path().to_path_buf(); let bytes = dir_size(&path); self.repos.remove(id); @@ -537,3 +560,45 @@ fn disk_usage(path: &std::path::Path) -> Option<(u64, u64)> { let avail = st.f_bavail as u64 * st.f_frsize as u64; Some((total.saturating_sub(avail), total)) } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use walgit_store::memory::MemoryStore; + + use super::*; + + /// A holder of `pack_mutex` that then wants `rw.read` (segmented MIDX + /// reconciliation) must not be stuck behind the drain's `rw` writer. + #[tokio::test] + async fn retire_and_drain_does_not_queue_a_writer_behind_a_pack_mutex_holder() { + let cache = tempfile::tempdir().unwrap(); + let mut cfg = walgit_config::Config::default(); + cfg.cache.dir = cache.path().to_path_buf(); + let registry = Registry::new(MemoryStore::shared(), Arc::new(cfg)); + let id = RepoId::new("test", "drain").unwrap(); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + + let pack = handle.pack_mutex.lock().await; + let drain = tokio::spawn({ + let handle = handle.clone(); + async move { handle.retire_and_drain().await } + }); + while handle.ensure_live().is_ok() { + tokio::task::yield_now().await; + } + tokio::time::sleep(Duration::from_millis(50)).await; + + let pin = tokio::time::timeout(Duration::from_secs(5), handle.pin_local_objects()) + .await + .expect("rw.read must not wait behind the drain"); + assert!(matches!(pin, Err(WalError::NotFound))); + + drop(pack); + tokio::time::timeout(Duration::from_secs(5), drain) + .await + .expect("drain completes once the holder is gone") + .unwrap(); + } +} diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index 8db377a2..e14dbe30 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -423,6 +423,52 @@ async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { assert!(!Arc::ptr_eq(&opened, &handle)); } +/// A reset that died after invalidating the directory leaves refs and packs behind; +/// `open` must rebuild from empty instead of serving them under the new manifest. +#[tokio::test] +async fn open_discards_a_directory_whose_reset_was_interrupted() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let registry = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let id = repo_id("test", "interrupted"); + + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + let dir = handle.local().path().to_path_buf(); + drop(handle); + + store + .delete( + &format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST), + None, + ) + .await + .unwrap(); + let other = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("b"), 0)), + ); + other.create(&id, ObjectFormat::Sha1).await.unwrap(); + std::fs::remove_file(dir.join("HEAD")).unwrap(); + + let fresh = Registry::new(store, Arc::new(make_config(&cache.path().join("a"), 0))); + let opened = fresh.open(&id).await.unwrap(); + assert!(opened.local().refs().unwrap().refs.is_empty()); +} + /// Concurrent creates of one name — on one registry and across two — have /// exactly one winner; `open_or_create` (auto-create on push) treats losing /// that race as success and opens the winner's repository. From b155cee5dbc84f70eeac3ed8f98d2e1ecfca33b3 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 16:58:41 -0400 Subject: [PATCH 14/17] fix(wal): never reuse a local copy once its repository was recreated Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/handle.rs | 6 +- crates/walgit-wal/src/registry.rs | 69 ++++++++++++------- crates/walgit-wal/tests/wal.rs | 111 ++++++++++++++++++++++++++++++ 3 files changed, 162 insertions(+), 24 deletions(-) diff --git a/crates/walgit-wal/src/handle.rs b/crates/walgit-wal/src/handle.rs index 8f782155..89c817f0 100644 --- a/crates/walgit-wal/src/handle.rs +++ b/crates/walgit-wal/src/handle.rs @@ -379,12 +379,16 @@ impl RepoHandle { /// that wants `rw.read` would deadlock). Every acquirer checks `ensure_live` after it /// holds its lock, so anything that starts later bails before touching the directory. pub(crate) async fn retire_and_drain(&self) { - self.retired.store(true, Ordering::Release); + self.retire(); drop(self.pack_mutex.lock().await); drop(self.sync_mutex.lock().await); drop(self.rw.write().await); } + pub(crate) fn retire(&self) { + self.retired.store(true, Ordering::Release); + } + pub fn store(&self) -> &Prefixed { &self.store } diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index eb39b534..e0fb3442 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -19,6 +19,8 @@ use crate::handle::RepoHandle; use crate::state::{RepoState, load_state, save_state}; use crate::store_proto::get_message; +const RESET_MARKER: &str = "walgit-reset-pending"; + pub struct Registry { store: DynStore, cfg: Arc, @@ -104,25 +106,27 @@ impl Registry { crate::validate_manifest(&manifest)?; // Open or init local repo (LocalRepo joins owner/name.git onto the root). - let local = if let Some(l) = LocalRepo::open(&self.cache_root, id)? { - l - } else { - // A directory without HEAD is an unfinished init or an interrupted reset: - // whatever it still holds is not this repository's. - let cache_root = self.cache_root.clone(); - let owned_id = id.clone(); - let format = parse_object_format(&manifest.object_format); - tokio::task::spawn_blocking(move || { - match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(WalError::Io(e)), - } - Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) - }) - .await - .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))?? - }; + let reset_pending = id.local_dir(&self.cache_root).join(RESET_MARKER).exists(); + let local = + if let Some(l) = LocalRepo::open(&self.cache_root, id)?.filter(|_| !reset_pending) { + l + } else { + // A directory without HEAD, or still marked by a create, is an unfinished init + // or an interrupted reset: whatever it still holds is not this repository's. + let cache_root = self.cache_root.clone(); + let owned_id = id.clone(); + let format = parse_object_format(&manifest.object_format); + tokio::task::spawn_blocking(move || { + match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(WalError::Io(e)), + } + Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) + }) + .await + .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))?? + }; // Load state let mut state = load_state(local.path()); @@ -176,10 +180,13 @@ impl Registry { { return Err(WalError::NotFound); } - // Drop the handle first so no request on this instance publishes into a - // prefix that is being removed; in-flight requests hold their own Arc. - self.repos.remove(id); + // Drop and retire the handle first so no request on this instance publishes + // into a prefix that is being removed or touches a later generation's directory. + let stale = self.repos.remove(id).map(|(_, handle)| handle); self.invalidate_listing(); + if let Some(handle) = &stale { + handle.retire_and_drain().await; + } // Manifest first: it is the linearization point, so the repo disappears // atomically for readers; remaining objects are unreferenced garbage. prefixed.delete(keys::MANIFEST, None).await?; @@ -238,6 +245,18 @@ impl Registry { retired_packs: Vec::new(), }; + // Marked before the Create, so a reset that fails or never runs after the Create + // lands (crash, lost reply) leaves `open` to rebuild instead of serving the old copy. + let marker = id.local_dir(&self.cache_root).join(RESET_MARKER); + let marked = marker.clone(); + tokio::task::spawn_blocking(move || match std::fs::write(&marked, b"") { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(WalError::Io(e)), + }) + .await + .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??; + let buf = manifest.encode_to_vec(); match prefixed .put( @@ -306,7 +325,10 @@ impl Registry { self.invalidate_listing(); Ok(handle) } - Err(StoreError::PreconditionFailed { .. }) => Err(WalError::AlreadyExists), + Err(StoreError::PreconditionFailed { .. }) => { + let _ = tokio::fs::remove_file(&marker).await; + Err(WalError::AlreadyExists) + } Err(e) => Err(WalError::Store(e)), } } @@ -492,6 +514,7 @@ impl Registry { if handle.ensure_live().is_err() { continue; } + handle.retire(); let path = handle.local.path().to_path_buf(); let bytes = dir_size(&path); self.repos.remove(id); diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index e14dbe30..6ea55390 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -349,6 +349,77 @@ async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { /// into it. The old handle is retired, its in-flight work finishes before the /// directory is reset, and anything it starts afterwards fails instead of /// writing into the new repository. +#[tokio::test] +async fn open_rebuilds_a_copy_whose_create_landed_without_its_reset() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let cfg = Arc::new(make_config(&cache.path().join("a"), 0)); + let id = repo_id("test", "crashed"); + + let registry = Registry::new(store.clone(), cfg.clone()); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + let local_dir = handle.local().path().to_path_buf(); + drop(handle); + drop(registry); + + // Another host deletes and recreates; this host's own Create would have marked the + // copy and then died before resetting it. + let other = tempfile::tempdir().unwrap(); + let foreign = Registry::new(store.clone(), Arc::new(make_config(other.path(), 0))); + foreign.delete(&id).await.unwrap(); + foreign.create(&id, ObjectFormat::Sha1).await.unwrap(); + std::fs::write(local_dir.join("walgit-reset-pending"), b"").unwrap(); + + let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap(); + reopened.sync_refs().await.unwrap(); + assert!(reopened.local().refs().unwrap().refs.is_empty()); + assert!(!local_dir.join("walgit-reset-pending").exists()); +} + +#[tokio::test] +async fn a_refused_create_keeps_the_existing_local_copy() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let cfg = Arc::new(make_config(&cache.path().join("a"), 0)); + let id = repo_id("test", "kept"); + + let registry = Registry::new(store.clone(), cfg.clone()); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + assert!(matches!( + registry.create(&id, ObjectFormat::Sha1).await, + Err(WalError::AlreadyExists) + )); + let local_dir = handle.local().path().to_path_buf(); + assert!(!local_dir.join("walgit-reset-pending").exists()); + drop(handle); + drop(registry); + + let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap(); + assert_eq!(reopened.local().refs().unwrap().refs.len(), 1); +} + #[tokio::test] async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { let cache = tempfile::tempdir().unwrap(); @@ -423,6 +494,46 @@ async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() { assert!(!Arc::ptr_eq(&opened, &handle)); } +/// A handle dropped by `delete` is retired too, so a holder that outlives the delete +/// cannot write into the directory of a repository created afterwards. +#[tokio::test] +async fn a_handle_held_across_delete_and_create_cannot_touch_the_new_repository() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let registry = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let id = repo_id("test", "deleted"); + + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + + registry.delete(&id).await.unwrap(); + let recreated = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + + assert!(matches!(handle.sync().await, Err(WalError::NotFound))); + let late = handle + .publish_push( + None, + make_txn(vec![("refs/heads/late", "", &c1)]), + HashMap::new(), + ) + .await; + assert!(late.is_err()); + assert!(recreated.local().refs().unwrap().refs.is_empty()); +} + /// A reset that died after invalidating the directory leaves refs and packs behind; /// `open` must rebuild from empty instead of serving them under the new manifest. #[tokio::test] From e0d008ddc3c43c556afc292821fa2f1f8a6d525d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 17:13:47 -0400 Subject: [PATCH 15/17] fix(wal): serialize delete and eviction with open on the same repository Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/registry.rs | 60 ++++++++++++++++++++---- crates/walgit-wal/tests/wal.rs | 76 +++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 8 deletions(-) diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index e0fb3442..e823f54f 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -85,12 +85,12 @@ impl Registry { // Every request that touches a repo goes through here: tag the // enclosing `http.request` span (no-op outside one). tracing::Span::current().record("repo", tracing::field::display(id)); - if let Some(h) = self.repos.get(id) { + if let Some(h) = self.repos.get(id).filter(|h| h.ensure_live().is_ok()) { return Ok(h.clone()); } let gate = self.opening.entry(id.clone()).or_default().clone(); let _g = gate.lock().await; - if let Some(h) = self.repos.get(id) { + if let Some(h) = self.repos.get(id).filter(|h| h.ensure_live().is_ok()) { return Ok(h.clone()); } @@ -173,6 +173,10 @@ impl Registry { /// handle and the local copy. Err(NotFound) if the manifest does not exist. /// Other instances notice on their next freshness check (manifest GET -> 404). pub async fn delete(&self, id: &RepoId) -> Result<(), WalError> { + // Held to the end: an `open` racing the drain would cache a handle on a + // manifest and directory that are about to disappear. + let gate = self.opening.entry(id.clone()).or_default().clone(); + let _g = gate.lock().await; let prefixed = Prefixed::new(self.store.clone(), id.store_prefix()); if get_message::(&prefixed, keys::MANIFEST) .await? @@ -248,11 +252,17 @@ impl Registry { // Marked before the Create, so a reset that fails or never runs after the Create // lands (crash, lost reply) leaves `open` to rebuild instead of serving the old copy. let marker = id.local_dir(&self.cache_root).join(RESET_MARKER); + // A marker left by an earlier attempt stays: that Create may have landed. let marked = marker.clone(); - tokio::task::spawn_blocking(move || match std::fs::write(&marked, b"") { - Ok(()) => Ok(()), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(e) => Err(WalError::Io(e)), + let marked_here = tokio::task::spawn_blocking(move || { + if marked.exists() { + return Ok(false); + } + match std::fs::write(&marked, b"") { + Ok(()) => Ok(true), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(e) => Err(WalError::Io(e)), + } }) .await .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??; @@ -326,7 +336,9 @@ impl Registry { Ok(handle) } Err(StoreError::PreconditionFailed { .. }) => { - let _ = tokio::fs::remove_file(&marker).await; + if marked_here { + let _ = tokio::fs::remove_file(&marker).await; + } Err(WalError::AlreadyExists) } Err(e) => Err(WalError::Store(e)), @@ -505,6 +517,12 @@ impl Registry { // excludes leaked/long-lived request ReadGuards. The old code only // probed sync_mutex and immediately dropped it, so it could delete // packs underneath an active reader. + // The opening gate keeps `open` from rebuilding on this directory, or + // caching a handle on it, until it is gone. + let gate = self.opening.entry(id.clone()).or_default().clone(); + let Ok(_opening) = gate.try_lock() else { + continue; + }; let Ok(_sync) = handle.sync_mutex.try_lock() else { continue; }; @@ -514,9 +532,9 @@ impl Registry { if handle.ensure_live().is_err() { continue; } - handle.retire(); let path = handle.local.path().to_path_buf(); let bytes = dir_size(&path); + handle.retire(); self.repos.remove(id); let _ = std::fs::remove_dir_all(&path); total_bytes = total_bytes.saturating_sub(bytes); @@ -624,4 +642,30 @@ mod tests { .expect("drain completes once the holder is gone") .unwrap(); } + + /// An `open` in progress owns the directory it is building on; eviction must + /// neither remove it nor drop the handle that `open` is about to cache. + #[tokio::test] + async fn eviction_skips_a_repository_an_open_is_building() { + let cache = tempfile::tempdir().unwrap(); + let mut cfg = walgit_config::Config::default(); + cfg.cache.dir = cache.path().to_path_buf(); + cfg.cache.max_bytes = walgit_config::ByteSize::b(0); + cfg.cache.evict_idle_after = Duration::ZERO; + let registry = Registry::new(MemoryStore::shared(), Arc::new(cfg)); + let id = RepoId::new("test", "building").unwrap(); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + tokio::time::sleep(Duration::from_millis(5)).await; + + let gate = registry.opening.entry(id.clone()).or_default().clone(); + let opening = gate.lock().await; + let report = registry.evict_idle().await.unwrap(); + assert_eq!(report.evicted, 0); + assert!(handle.ensure_live().is_ok()); + assert!(handle.local().path().join("HEAD").exists()); + + drop(opening); + assert_eq!(registry.evict_idle().await.unwrap().evicted, 1); + assert!(handle.ensure_live().is_err()); + } } diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index 6ea55390..5ae1fdb8 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -349,6 +349,44 @@ async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() { /// into it. The old handle is retired, its in-flight work finishes before the /// directory is reset, and anything it starts afterwards fails instead of /// writing into the new repository. +#[tokio::test] +async fn an_open_racing_a_delete_sees_the_repository_gone() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let registry = Registry::new( + store.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let id = repo_id("test", "racing"); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + + let in_flight = handle.sync_refs().await.unwrap(); + let delete = tokio::spawn({ + let registry = registry.clone(); + let id = id.clone(); + async move { registry.delete(&id).await } + }); + loop { + match handle.sync_refs().await { + Err(WalError::NotFound) => break, + Ok(guard) => drop(guard), + Err(e) => panic!("unexpected {e}"), + } + tokio::task::yield_now().await; + } + let open = tokio::spawn({ + let registry = registry.clone(); + let id = id.clone(); + async move { registry.open(&id).await.map(|_| ()) } + }); + tokio::task::yield_now().await; + drop(in_flight); + + delete.await.unwrap().unwrap(); + assert!(matches!(open.await.unwrap(), Err(WalError::NotFound))); + assert!(matches!(registry.open(&id).await, Err(WalError::NotFound))); +} + #[tokio::test] async fn open_rebuilds_a_copy_whose_create_landed_without_its_reset() { let cache = tempfile::tempdir().unwrap(); @@ -387,6 +425,44 @@ async fn open_rebuilds_a_copy_whose_create_landed_without_its_reset() { assert!(!local_dir.join("walgit-reset-pending").exists()); } +#[tokio::test] +async fn a_refused_create_keeps_a_marker_an_earlier_create_left() { + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let cfg = Arc::new(make_config(&cache.path().join("a"), 0)); + let id = repo_id("test", "remarked"); + + let registry = Registry::new(store.clone(), cfg.clone()); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + let local_dir = handle.local().path().to_path_buf(); + let marker = local_dir.join("walgit-reset-pending"); + std::fs::write(&marker, b"").unwrap(); + + assert!(matches!( + registry.create(&id, ObjectFormat::Sha1).await, + Err(WalError::AlreadyExists) + )); + assert!(marker.exists()); + drop(handle); + drop(registry); + + let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap(); + assert!(!marker.exists()); + reopened.sync_refs().await.unwrap(); + assert_eq!(reopened.local().refs().unwrap().refs.len(), 1); +} + #[tokio::test] async fn a_refused_create_keeps_the_existing_local_copy() { let cache = tempfile::tempdir().unwrap(); From d74b7f79aa83d13c0021a41cae18e6c6c528f1d9 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 17:27:53 -0400 Subject: [PATCH 16/17] fix(wal): rebuild an interrupted local reset instead of serving it Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/registry.rs | 62 +++++++++++++++---------------- crates/walgit-wal/tests/wal.rs | 28 ++++++++++++++ 2 files changed, 58 insertions(+), 32 deletions(-) diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index e823f54f..77cd65dc 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -107,26 +107,20 @@ impl Registry { // Open or init local repo (LocalRepo joins owner/name.git onto the root). let reset_pending = id.local_dir(&self.cache_root).join(RESET_MARKER).exists(); - let local = - if let Some(l) = LocalRepo::open(&self.cache_root, id)?.filter(|_| !reset_pending) { - l - } else { - // A directory without HEAD, or still marked by a create, is an unfinished init - // or an interrupted reset: whatever it still holds is not this repository's. - let cache_root = self.cache_root.clone(); - let owned_id = id.clone(); - let format = parse_object_format(&manifest.object_format); - tokio::task::spawn_blocking(move || { - match std::fs::remove_dir_all(owned_id.local_dir(&cache_root)) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(WalError::Io(e)), - } - Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) - }) + let local = if let Some(l) = + LocalRepo::open(&self.cache_root, id)?.filter(|_| !reset_pending) + { + l + } else { + // A directory without HEAD, or still marked by a create, is an unfinished init + // or an interrupted reset: whatever it still holds is not this repository's. + let cache_root = self.cache_root.clone(); + let owned_id = id.clone(); + let format = parse_object_format(&manifest.object_format); + tokio::task::spawn_blocking(move || reset_local_copy(&cache_root, &owned_id, format)) .await .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))?? - }; + }; // Load state let mut state = load_state(local.path()); @@ -291,20 +285,7 @@ impl Registry { let cache_root = self.cache_root.clone(); let owned_id = id.clone(); let local = tokio::task::spawn_blocking(move || { - // Without HEAD `LocalRepo::open` rejects the directory, so a removal that - // fails midway cannot be mistaken for the old repository: `open` rebuilds. - let dir = owned_id.local_dir(&cache_root); - for result in [ - std::fs::remove_file(dir.join("HEAD")), - std::fs::remove_dir_all(&dir), - ] { - match result { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(WalError::Io(e)), - } - } - Ok::<_, WalError>(LocalRepo::init(&cache_root, &owned_id, format)?) + reset_local_copy(&cache_root, &owned_id, format) }) .await .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??; @@ -548,6 +529,23 @@ impl Registry { } } +/// HEAD goes first: without it `LocalRepo::open` rejects the directory, so a wipe that +/// stops midway is rebuilt by the next `open` instead of served as a repository. +fn reset_local_copy( + cache_root: &std::path::Path, + id: &RepoId, + format: ObjectFormat, +) -> Result { + let dir = id.local_dir(cache_root); + let ignore_missing = |result: std::io::Result<()>| match result { + Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(WalError::Io(e)), + _ => Ok(()), + }; + ignore_missing(std::fs::remove_file(dir.join("HEAD")))?; + ignore_missing(std::fs::remove_dir_all(&dir))?; + Ok(LocalRepo::init(cache_root, id, format)?) +} + fn parse_object_format(s: &str) -> ObjectFormat { match s { "sha256" => ObjectFormat::Sha256, diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index 5ae1fdb8..27b706df 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -425,6 +425,34 @@ async fn open_rebuilds_a_copy_whose_create_landed_without_its_reset() { assert!(!local_dir.join("walgit-reset-pending").exists()); } +#[cfg(unix)] +#[tokio::test] +async fn an_interrupted_rebuild_is_rebuilt_again() { + use std::os::unix::fs::PermissionsExt; + + let cache = tempfile::tempdir().unwrap(); + let store = MemoryStore::shared(); + let cfg = Arc::new(make_config(&cache.path().join("a"), 0)); + let id = repo_id("test", "rewiped"); + + let registry = Registry::new(store.clone(), cfg.clone()); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let local_dir = handle.local().path().to_path_buf(); + drop(handle); + drop(registry); + + std::fs::write(local_dir.join("walgit-reset-pending"), b"").unwrap(); + let stuck = local_dir.join("stuck"); + std::fs::create_dir(&stuck).unwrap(); + std::fs::write(stuck.join("file"), b"").unwrap(); + std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o555)).unwrap(); + + let failed = Registry::new(store.clone(), cfg.clone()).open(&id).await; + std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o755)).unwrap(); + assert!(failed.is_err()); + assert!(!local_dir.join("HEAD").exists()); +} + #[tokio::test] async fn a_refused_create_keeps_a_marker_an_earlier_create_left() { let cache = tempfile::tempdir().unwrap(); From 82fe56f634b34e891c672f441553549c82738f59 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 21:05:04 -0400 Subject: [PATCH 17/17] fix(wal): stop serving a cached copy when a create's outcome is unknown Signed-off-by: Yordis Prieto --- crates/walgit-wal/src/registry.rs | 10 +++++- crates/walgit-wal/tests/wal.rs | 58 +++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs index 77cd65dc..2066d1fa 100644 --- a/crates/walgit-wal/src/registry.rs +++ b/crates/walgit-wal/src/registry.rs @@ -322,7 +322,15 @@ impl Registry { } Err(WalError::AlreadyExists) } - Err(e) => Err(WalError::Store(e)), + Err(e) => { + // The Create may have landed behind a lost reply: the marker stays, and + // no cached handle may keep serving the copy the marker condemns. + if let Some((_, stale)) = self.repos.remove(id) { + self.invalidate_listing(); + stale.retire_and_drain().await; + } + Err(WalError::Store(e)) + } } } diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index 27b706df..6271b276 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -446,11 +446,21 @@ async fn an_interrupted_rebuild_is_rebuilt_again() { std::fs::create_dir(&stuck).unwrap(); std::fs::write(stuck.join("file"), b"").unwrap(); std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o555)).unwrap(); + if std::fs::write(stuck.join("probe"), b"").is_ok() { + // Root ignores directory permissions, so the interrupted wipe cannot be staged. + std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o755)).unwrap(); + return; + } let failed = Registry::new(store.clone(), cfg.clone()).open(&id).await; std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o755)).unwrap(); assert!(failed.is_err()); assert!(!local_dir.join("HEAD").exists()); + + let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap(); + assert!(reopened.local().path().join("HEAD").exists()); + assert!(!local_dir.join("walgit-reset-pending").exists()); + assert!(!stuck.exists()); } #[tokio::test] @@ -638,6 +648,54 @@ async fn a_handle_held_across_delete_and_create_cannot_touch_the_new_repository( assert!(recreated.local().refs().unwrap().refs.is_empty()); } +/// A Create whose reply was lost may have landed, so the warm copy of a repository another +/// host deleted must not keep serving its refs under the new manifest. +#[tokio::test] +async fn a_create_with_a_lost_reply_stops_serving_the_cached_copy() { + use walgit_store::fault::{FaultPlan, FaultStore}; + let cache = tempfile::tempdir().unwrap(); + let truth = MemoryStore::shared(); + let link = FaultStore::new(truth.clone(), "lost-create", 1); + let registry = Registry::new( + link.clone(), + Arc::new(make_config(&cache.path().join("a"), 0)), + ); + let id = repo_id("test", "lost-create"); + + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let work = WorkRepo::new(); + let c1 = work.commit("init", "data"); + let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + handle + .publish_push( + Some(ingested), + make_txn(vec![("refs/heads/main", "", &c1)]), + HashMap::new(), + ) + .await + .unwrap(); + + Registry::new( + truth.clone(), + Arc::new(make_config(&cache.path().join("b"), 0)), + ) + .delete(&id) + .await + .unwrap(); + + link.set(FaultPlan { + p_err_after: 1.0, + only_keys: Some(vec!["manifest.pb".into()]), + ..Default::default() + }); + assert!(registry.create(&id, ObjectFormat::Sha1).await.is_err()); + link.heal(); + + assert!(matches!(handle.sync().await, Err(WalError::NotFound))); + let reopened = registry.open(&id).await.unwrap(); + assert!(reopened.local().refs().unwrap().refs.is_empty()); +} + /// A reset that died after invalidating the directory leaves refs and packs behind; /// `open` must rebuild from empty instead of serving them under the new manifest. #[tokio::test]