From f63c8ca8ccb0c2fa6b4f334bab39777147852de3 Mon Sep 17 00:00:00 2001 From: Jonas Grosch Date: Thu, 1 Oct 2026 01:58:28 +0200 Subject: [PATCH] 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. Co-Authored-By: Claude Opus 5.5 --- 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 7998424c..2d5ff8f4 100644 --- a/crates/walgit-server/tests/sim.rs +++ b/crates/walgit-server/tests/sim.rs @@ -1597,8 +1597,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 ce85399e..4044ffaa 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 447af901..4a82da5e 100644 --- a/docs/ROUNDTRIPS.md +++ b/docs/ROUNDTRIPS.md @@ -54,6 +54,7 @@ right shape. This document is the thinking tool; apply it to every protocol chan | Lost CAS response resolution | fresh manifest GET; only if the exact segment descriptor is listed, GET and compare the claimed log bytes | normally +1 GET on this failure path compared with key/sequence-only resolution; no added successful-push requests. Missing/folded evidence remains unknown. Per-attempt nonce and actual frame sizes are computed locally. | `publish.rs::cas_landed` | | 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` | @@ -61,7 +62,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.