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.