From 539dabcbf7dd42d90b7eb619118b433c1bb9b2b1 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 23:00:14 +0000 Subject: [PATCH] =?UTF-8?q?feat(store):=20the=20put=20seam=20=E2=80=94=20t?= =?UTF-8?q?wo=20internal=20paths,=20the=20three=20numbered=20rules,=20Batc?= =?UTF-8?q?h,=20kv-only=20rejection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `put(len: Option, stream: impl AsyncRead + Unpin + Send) -> (Key, Pin)`: known-length one-pass flow; unknown-length buffer-to-threshold flow with the mid-stream fresh-stage flush (ADR-003 option (d), three numbered rules: flush-and-continue, derivation over the eventual stored artifact as a whole via buffer form and StagingCursor read-back form, zero residue on stream error) - staged-put seam amended (operator sign-off): `put_staged()` keyless handout, `StagingCursor::commit(self, key)` publishes under the store-core-resolved key — the canonical key completes at stream end and the wave-1 handout shape could not express the commit-rename flow - pin-before-publish rides the arbitration lock on both paths and both tiers; dedup re-puts refcount the existing entry - Batch seam: `Store::batch()` + `put_in_batch()`, members publish pin- before-publish, group pins drop together on `Batch::commit` (new public verb) or drop - kv-only rejection (ADR-012 §6.2): known-length over-threshold rejects immediately (stream never read), unknown-length at mid-stream overflow; zero residue; error carrier `Io(String)` (named design point) - declared-len mismatch rejects `Verification` zero-residue (named design point); stream shape generic-no-dyn (named design point) - incremental BlobHasher (pinned byte-equal to Algorithm::derive); AnyStaging cursor dispatch; git-interop put-path leg against git hash-object; eleven put-path behavior tests Verification: cargo test (default/--features mem/--all-features) green; clippy --all-targets -D warnings green in all three feature shapes; cargo fmt --check green; cargo doc --no-deps at the 24-warning baseline. --- src/backend.rs | 59 +- src/facade.rs | 20 + src/hashing.rs | 105 ++- src/local.rs | 62 +- src/mem.rs | 55 +- src/store.rs | 1333 +++++++++++++++++++++++++++++++++++++- tasks/store-put-paths.md | 120 +++- tests/git_interop.rs | 74 +++ 8 files changed, 1723 insertions(+), 105 deletions(-) diff --git a/src/backend.rs b/src/backend.rs index a5de3aa..567fdae 100644 --- a/src/backend.rs +++ b/src/backend.rs @@ -108,6 +108,17 @@ pub(crate) trait ReadCursor: Send { /// on discard with zero residue (the stage-hygiene invariant, /// store-api.md §Write path; exact-count tested per engine). /// +/// **The key resolves at commit, not at handout (named design point, +/// the wave-2 amendment of the wave-1 handout shape).** The canonical +/// key is the derivation over preamble + content, and it completes at +/// stream end — after the staging cursor already exists (put paths +/// stream content once into the cursor while hashing; ADR-003's put +/// paths). So `commit` carries the store-core-resolved key and the +/// stage's identity before that point is only the engine-internal +/// stage file/tx (unique per `put_staged` handout). Publish is the +/// CAS first-writer-wins entry under that committed key exactly as +/// [`Backend::put`]. +/// /// Every engine implements the staged form: kv-tier bodies are the /// degenerate in-memory stage committed through the engine's whole-value /// row write (the store core selects the form by tier policy, ADR-010 @@ -121,15 +132,18 @@ pub(crate) trait StagingCursor: ReadCursor { /// "restart the stream" requirement anywhere). async fn append(&mut self, chunk: &[u8]) -> Result<(), Error>; - /// Publish the entry: the stage's bytes become the CAS entry in the + /// Publish the entry: the stage's bytes become the CAS entry under + /// `key` — the store-core-resolved canonical key (the derivation + /// completes at stream end; see the type's design point) — in the /// engine's put shape (commit-rename on `local`, the in-tx `lowrite` - /// batch on pg-lo, the row write on kv bodies). This is the staged - /// form's terminal step — the store core never re-uses the cursor - /// after the call. On failure the stage is **already discarded** — - /// zero residue by the stage-hygiene invariant; there is no - /// post-failure discard to drive (the consuming call took the - /// cursor). - async fn commit(self) -> Result<(), Error>; + /// batch on pg-lo, the row write on kv bodies). CAS first-writer-wins: + /// a losing race against an existing entry is dedup (`Ok`), never a + /// failure. This is the staged form's terminal step — the store core + /// never re-uses the cursor after the call. On failure the stage is + /// **already discarded** — zero residue by the stage-hygiene + /// invariant; there is no post-failure discard to drive (the + /// consuming call took the cursor). + async fn commit(self, key: &[u8]) -> Result<(), Error>; /// Discard the stage: every non-commit path converges here, zero /// residue. Driven explicitly by the store core (stream error, @@ -189,11 +203,12 @@ pub(crate) trait Backend: Send + Sync { async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error>; /// The staged put's cursor handout (ADR-010 §1). The store core - /// selects the form before content flows: the large tier's - /// stage-then-commit-rename contract, kv engines' degenerate - /// in-memory stage. See [`StagingCursor`] for the contract every - /// body converges on. - async fn put_staged(&self, key: &[u8]) -> Result; + /// selects the form before content flows (the cursor's identity + /// before commit is the engine-internal stage only — the canonical + /// key resolves at [`StagingCursor::commit`], the type's design + /// point). See [`StagingCursor`] for the contract every body + /// converges on. + async fn put_staged(&self) -> Result; /// Unconditional delete — arbitration has already happened above the /// trait (the facade decides and owns the refusal: `GcRefuse` never @@ -442,7 +457,6 @@ mod tests { } struct TestStaging { - key: Vec, staged: Vec, rows: Arc, Vec>>>, fail_commit: bool, @@ -464,13 +478,13 @@ mod tests { Ok(()) } - async fn commit(mut self) -> Result<(), Error> { + async fn commit(mut self, key: &[u8]) -> Result<(), Error> { if self.fail_commit { self.staged.clear(); return Err(Error::Io("engine commit failure (test double)".to_owned())); } let value = std::mem::take(&mut self.staged); - self.rows.lock().entry(self.key.clone()).or_insert(value); + self.rows.lock().entry(key.to_vec()).or_insert(value); Ok(()) } @@ -508,9 +522,8 @@ mod tests { Ok(()) } - async fn put_staged(&self, key: &[u8]) -> Result { + async fn put_staged(&self) -> Result { Ok(TestStaging { - key: key.to_vec(), staged: Vec::new(), rows: self.rows.clone(), fail_commit: self.fail_commit, @@ -603,7 +616,7 @@ mod tests { let backend = TestBackend::default(); let key = b"\x01staged-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(&CONTENT[..8]).await.unwrap(); staging.append(&CONTENT[8..16]).await.unwrap(); staging.append(&CONTENT[16..]).await.unwrap(); @@ -620,7 +633,7 @@ mod tests { assert_eq!(bytes, CONTENT, "staged artifact readable as a whole"); assert_eq!(digest, slice_digest(CONTENT)); - staging.commit().await.expect("commit publishes"); + staging.commit(&key).await.expect("commit publishes"); assert!(backend.has(&key).await.unwrap(), "entry published"); let cursor = backend @@ -650,7 +663,7 @@ mod tests { let backend = TestBackend::default(); let key = b"\x01discard-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"residue").await.unwrap(); staging.discard().await; @@ -676,9 +689,9 @@ mod tests { }; let key = b"\x01failure-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(CONTENT).await.unwrap(); - let err = staging.commit().await.unwrap_err(); + let err = staging.commit(&key).await.unwrap_err(); assert!(matches!(err, Error::Io(_)), "engine media failure"); assert!( diff --git a/src/facade.rs b/src/facade.rs index a1d5889..e069d00 100644 --- a/src/facade.rs +++ b/src/facade.rs @@ -178,6 +178,15 @@ impl Pin { } } +impl fmt::Debug for Pin { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Pin") + .field("key", &self.key.digest()) + .field("armed", &self.armed) + .finish() + } +} + impl Drop for Pin { fn drop(&mut self) { if self.armed { @@ -239,6 +248,17 @@ impl Batch { self.keys.is_empty() } + /// **Batch commit**: the group's pins drop together in one ordered + /// step — protection spans the whole batch or none of it, and ends + /// when the batch ends (ADR-005 §Marks 2; the same release batch + /// drop drives). A committed batch's members are no longer + /// batch-protected; durable protection beyond the write group is + /// the consumer's liveness seam (`Store::register_liveness_source` + /// or the pin-conversion ordering). + pub fn commit(mut self) { + self.release_members(); + } + /// The one-step release: every member refcount decremented inside /// a single arbitration-lock section, sorted into the keys' total /// order. Driven by [`Drop::drop`] (batch drop) and available for diff --git a/src/hashing.rs b/src/hashing.rs index 16cdce0..b841460 100644 --- a/src/hashing.rs +++ b/src/hashing.rs @@ -27,7 +27,7 @@ //! is absent from the crate entirely (ADR-002 Decision class 3). use sha1::Sha1; -use sha2::Sha256; +use sha2::{Digest as _, Sha256}; use crate::Error; @@ -129,6 +129,61 @@ where hasher.finalize().to_vec() } +/// The streaming form of the derivation: one hasher fed the preamble +/// at initialization (`"blob \0"`, the length known up front on +/// the known-length put path) and the content as it flows. The finish +/// is byte-identical to [`Algorithm::derive`] over the same input — +/// this is the same derivation, driven incrementally so content can +/// stream once (ADR-003's put paths; the wave-2 facade logic this +/// module's doc anticipated). +/// +/// Crate-internal: the derivation is a store-core concern; this type +/// is the put path's tool, not public API. +pub(crate) struct BlobHasher { + algorithm: Algorithm, + inner: InnerHasher, +} + +enum InnerHasher { + Sha256(Sha256), + Sha1(Sha1), +} + +impl BlobHasher { + /// A hasher initialized with the preamble for an artifact of `len` + /// bytes — the known-length path's one-pass form (the preamble is + /// known before content flows). + pub(crate) fn new(algorithm: Algorithm, len: u64) -> BlobHasher { + let mut hasher = BlobHasher { + algorithm, + inner: match algorithm { + Algorithm::GitBlobSha256 => InnerHasher::Sha256(Sha256::new()), + Algorithm::GitBlobSha1 => InnerHasher::Sha1(Sha1::new()), + }, + }; + hasher.update(&blob_preamble(len)); + hasher + } + + /// Feed the next chunk of content. + pub(crate) fn update(&mut self, chunk: &[u8]) { + match &mut self.inner { + InnerHasher::Sha256(inner) => inner.update(chunk), + InnerHasher::Sha1(inner) => inner.update(chunk), + } + } + + /// The derivation's answer over preamble + everything fed so far — + /// pinned equal to [`Algorithm::derive`] by the in-module tests. + pub(crate) fn finish(self) -> Digest { + let bytes = match self.inner { + InnerHasher::Sha256(inner) => inner.finalize().to_vec(), + InnerHasher::Sha1(inner) => inner.finalize().to_vec(), + }; + Digest::from_parts(self.algorithm, bytes) + } +} + /// An algorithm + digest pair — the hash of a blob under one of this /// crate's derivations (ADR-002; hashing-and-keys.md §The /// Key/Digest surface). @@ -294,6 +349,54 @@ mod tests { ); } + /// The streaming form is the same derivation: a `BlobHasher` fed + /// the content in arbitrary chunk boundaries finishes byte-identical + /// to [`Algorithm::derive`] — both algorithms, boundaries around the + /// known-answer vectors, and a multi-chunk bulk body. + #[test] + fn streaming_hasher_matches_known_length_derivation() { + for algorithm in [Algorithm::GitBlobSha256, Algorithm::GitBlobSha1] { + let bodies: [&[u8]; 4] = [b"", b"hello", &[0u8; 100_003], &[7u8; 9]]; + + for body in bodies { + let expected = algorithm.derive(body, body.len() as u64); + + // whole-body update. + let mut hasher = BlobHasher::new(algorithm, body.len() as u64); + hasher.update(body); + assert_eq!(hasher.finish(), expected, "whole-body, {algorithm:?}"); + + // byte-at-a-time. + let mut hasher = BlobHasher::new(algorithm, body.len() as u64); + for byte in body { + hasher.update(std::slice::from_ref(byte)); + } + assert_eq!(hasher.finish(), expected, "byte-at-a-time, {algorithm:?}"); + + // awkward fixed chunks (7, including past the end). + let mut hasher = BlobHasher::new(algorithm, body.len() as u64); + for chunk in body.chunks(7) { + hasher.update(chunk); + } + assert_eq!(hasher.finish(), expected, "chunks of 7, {algorithm:?}"); + } + + // The canonical 'hello' and empty vectors, streamed. + let expected = algorithm.derive(b"hello", 5).bytes().to_vec(); + let mut hasher = BlobHasher::new(algorithm, 5); + hasher.update(b"hel"); + hasher.update(b"lo"); + assert_eq!(hasher.finish().bytes().to_vec(), expected); + + let hasher = BlobHasher::new(algorithm, 0); + assert_eq!( + hasher.finish().bytes().to_vec(), + algorithm.derive(b"", 0).bytes().to_vec(), + "empty content streams with no update" + ); + } + } + /// Algorithm ordering equals registry-byte ordering, so /// digest-derived sort is the keys' byte total order. #[test] diff --git a/src/local.rs b/src/local.rs index 6b67041..250f8dd 100644 --- a/src/local.rs +++ b/src/local.rs @@ -266,16 +266,18 @@ impl ReadCursor for LocalCursor { /// The staged put's staging cursor (ADR-010 §1): backed by a stage file /// in the staging area, unique per stage — appends write through, the /// staged length tracks completed appends (the sole-writer protocol), -/// and `commit` renames the stage file into its sharded committed path — -/// publish is the rename. Every failure path converges on discard with -/// zero residue: a commit that cannot complete (media failure, CAS -/// dedup, parent creation) removes the stage file itself; [`StagingCursor::discard`] -/// removes it; and a cursor dropped without either is best-effort -/// removed by the [`Drop`] body — the engine's cancellation story, -/// covered by the in-module hygiene tests. +/// and `commit(key)` renames the stage file into its sharded committed +/// path under the store-core-resolved canonical key (the staged-put +/// seam's key-at-commit design point: the derivation completes at stream +/// end, so until commit the stage's identity is the engine-internal +/// stage file alone) — publish is the rename. Every failure path +/// converges on discard with zero residue: a commit that cannot complete +/// (media failure, CAS dedup, parent creation) removes the stage file +/// itself; [`StagingCursor::discard`] removes it; and a cursor dropped +/// without either is best-effort removed by the [`Drop`] body — the +/// engine's cancellation story, covered by the in-module hygiene tests. pub(crate) struct LocalStaging { root: Arc, - key: Vec, stage_path: Option, file: File, staged_len: AtomicU64, @@ -321,7 +323,7 @@ impl StagingCursor for LocalStaging { Ok(()) } - async fn commit(mut self) -> Result<(), Error> { + async fn commit(mut self, key: &[u8]) -> Result<(), Error> { let stage = match self.stage_path.take() { Some(stage) => stage, None => { @@ -329,7 +331,7 @@ impl StagingCursor for LocalStaging { } }; let root = Arc::clone(&self.root); - let key = std::mem::take(&mut self.key); + let key = key.to_vec(); let cleanup = stage.clone(); let result = blocking(move || { let committed = committed_path(&root, &key)?; @@ -454,10 +456,8 @@ impl Backend for LocalBackend { .await } - async fn put_staged(&self, key: &[u8]) -> Result { - let root = Arc::clone(&self.root); - let key = key.to_vec(); - let inner_root = Arc::clone(&root); + async fn put_staged(&self) -> Result { + let inner_root = Arc::clone(&self.root); let name = format!("stage-{}", stage_id()); #[cfg(test)] let fail_commit = self.fail_commit; @@ -473,7 +473,6 @@ impl Backend for LocalBackend { .map_err(io_err)?; Ok(LocalStaging { root: inner_root, - key, stage_path: Some(stage_path), file, staged_len: AtomicU64::new(0), @@ -590,9 +589,9 @@ mod tests { } async fn staged(backend: &LocalBackend, key: &[u8], content: &[u8]) -> Result<(), Error> { - let mut staging = backend.put_staged(key).await?; + let mut staging = backend.put_staged().await?; staging.append(content).await?; - staging.commit().await + staging.commit(key).await } /// The Backend method contract: the name, the seven methods' @@ -770,10 +769,13 @@ mod tests { assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::::new()); let staged_key = key_for(b"empty-staged"); - let mut staging = backend.put_staged(&staged_key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"").await.unwrap(); assert_eq!(staging.len(), 0, "empty append stays empty"); - staging.commit().await.expect("empty commit publishes"); + staging + .commit(&staged_key) + .await + .expect("empty commit publishes"); assert!( backend.has(&staged_key).await.unwrap(), "empty entry landed" @@ -800,7 +802,7 @@ mod tests { let (_root, backend) = new_backend().await; let key = key_for(b"staged-round-trip"); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(&CONTENT[..8]).await.unwrap(); staging.append(&CONTENT[8..24]).await.unwrap(); staging.append(&CONTENT[24..]).await.unwrap(); @@ -819,7 +821,7 @@ mod tests { let tail = staging.read_at(CONTENT.len() as u64, 32).await.unwrap(); assert_eq!(tail, Vec::::new(), "staged read clamps at staged EOF"); - staging.commit().await.expect("commit publishes"); + staging.commit(&key).await.expect("commit publishes"); assert!(backend.has(&key).await.unwrap(), "entry published"); let cursor = backend @@ -857,7 +859,7 @@ mod tests { let (root, backend) = new_backend().await; let key = key_for(b"discard-key"); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"residue").await.unwrap(); staging.discard().await; @@ -887,9 +889,9 @@ mod tests { }; let key = key_for(b"failure-key"); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(CONTENT).await.unwrap(); - let err = staging.commit().await.unwrap_err(); + let err = staging.commit(&key).await.unwrap_err(); assert!(matches!(err, Error::Io(_)), "engine media failure"); assert!( @@ -913,7 +915,7 @@ mod tests { let key = key_for(b"cancelled-key"); { - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"never published").await.unwrap(); // Dropped without commit or discard. } @@ -953,21 +955,23 @@ mod tests { for i in 0..N { let content = format!("discard-{i}"); let key = key_for(content.as_bytes()); - let mut staging = backend.put_staged(&key).await.unwrap(); + let _ = &key; + let mut staging = backend.put_staged().await.unwrap(); staging.append(content.as_bytes()).await.unwrap(); staging.discard().await; } for i in 0..N { let content = format!("commit-failure-{i}"); let key = key_for(content.as_bytes()); - let mut staging = failure_backend.put_staged(&key).await.unwrap(); + let mut staging = failure_backend.put_staged().await.unwrap(); staging.append(content.as_bytes()).await.unwrap(); - staging.commit().await.unwrap_err(); + staging.commit(&key).await.unwrap_err(); } for i in 0..N { let content = format!("cancelled-{i}"); let key = key_for(content.as_bytes()); - let mut staging = backend.put_staged(&key).await.unwrap(); + let _ = &key; + let mut staging = backend.put_staged().await.unwrap(); staging.append(content.as_bytes()).await.unwrap(); drop(staging); } diff --git a/src/mem.rs b/src/mem.rs index 86b93b7..3622be0 100644 --- a/src/mem.rs +++ b/src/mem.rs @@ -10,9 +10,12 @@ //! form: [`get`][Backend::get] hands out a read cursor that materializes //! the whole value (in-memory slices; `len()` from the stored value), and //! [`put_staged`][Backend::put_staged] hands out the degenerate in-memory -//! stage whose `commit` publishes through the same whole-value CAS row -//! write as `put` — `entry().or_insert()`, first writer wins, a re-put is -//! dedup and never an overwrite. `commit` cannot fail in normal operation +//! stage whose `commit(key)` publishes through the same whole-value CAS +//! row write as `put` — `entry().or_insert()`, first writer wins, a re-put +//! is dedup and never an overwrite. The key resolves at commit (the +//! staged-put seam's wave-2 amendment: the canonical derivation completes +//! at stream end, so the stage's identity before that point is the +//! engine-internal buffer alone). `commit` cannot fail in normal operation //! (no media can fail here); the test-only hook noted on [`MemStaging`] //! exercises the failure form's self-discard hygiene, standing in for the //! media-failure injection later engines need. @@ -75,9 +78,8 @@ impl Backend for MemBackend { Ok(()) } - async fn put_staged(&self, key: &[u8]) -> Result { + async fn put_staged(&self) -> Result { Ok(MemStaging { - key: key.to_vec(), staged: Vec::new(), rows: Arc::clone(&self.rows), #[cfg(test)] @@ -122,14 +124,15 @@ impl ReadCursor for MemCursor { } /// The degenerate in-memory staging cursor (ADR-010 §1): appends extend -/// the buffer; `commit` publishes the whole staged bytes through the same -/// CAS clause as [`Backend::put`] — first writer wins, losing the race is -/// dedup (`Ok`), never a failure — and every non-commit path, including -/// the failure form, converges on zero residue. `commit` takes the -/// buffer via `std::mem::take` before the table lock, so a failed publish -/// cannot strand the staged bytes in the map. +/// the buffer; `commit(key)` publishes the whole staged bytes through the +/// same CAS clause as [`Backend::put`] — under the store-core-resolved +/// canonical key (the staged-put seam's key-at-commit design point), +/// first writer wins, losing the race is dedup (`Ok`), never a failure — +/// and every non-commit path, including the failure form, converges on +/// zero residue. `commit` takes the buffer via `std::mem::take` before +/// the table lock, so a failed publish cannot strand the staged bytes +/// in the map. pub(crate) struct MemStaging { - key: Vec, staged: Vec, rows: Arc, Vec>>>, /// Test-only injection, shared with the issuing [`MemBackend`] — see @@ -155,7 +158,7 @@ impl StagingCursor for MemStaging { Ok(()) } - async fn commit(mut self) -> Result<(), Error> { + async fn commit(mut self, key: &[u8]) -> Result<(), Error> { #[cfg(test)] if self.fail_commit { self.staged.clear(); @@ -164,7 +167,7 @@ impl StagingCursor for MemStaging { )); } let value = std::mem::take(&mut self.staged); - self.rows.lock().entry(self.key).or_insert(value); + self.rows.lock().entry(key.to_vec()).or_insert(value); Ok(()) } @@ -338,10 +341,13 @@ mod tests { assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::::new()); let staged_key = b"\x01empty-staged-key".to_vec(); - let mut staging = backend.put_staged(&staged_key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"").await.unwrap(); assert_eq!(staging.len(), 0, "empty append stays empty"); - staging.commit().await.expect("empty commit publishes"); + staging + .commit(&staged_key) + .await + .expect("empty commit publishes"); assert!( backend.has(&staged_key).await.unwrap(), "empty entry landed" @@ -369,7 +375,7 @@ mod tests { let backend = MemBackend::default(); let key = b"\x01staged-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(&CONTENT[..8]).await.unwrap(); staging.append(&CONTENT[8..16]).await.unwrap(); staging.append(&CONTENT[16..]).await.unwrap(); @@ -386,7 +392,7 @@ mod tests { assert_eq!(bytes, CONTENT, "staged artifact readable as a whole"); assert_eq!(digest, crate::backend::slice_digest(CONTENT)); - staging.commit().await.expect("commit publishes"); + staging.commit(&key).await.expect("commit publishes"); assert!(backend.has(&key).await.unwrap(), "entry published"); let cursor = backend @@ -410,9 +416,12 @@ mod tests { Some(CONTENT), "whole-value re-put does not overwrite" ); - let mut second = backend.put_staged(&key).await.unwrap(); + let mut second = backend.put_staged().await.unwrap(); second.append(b"overwritten-by-stage").await.unwrap(); - second.commit().await.expect("commit is dedup, not failure"); + second + .commit(&key) + .await + .expect("commit is dedup, not failure"); assert_eq!( stored_value(&backend, &key).await.as_deref(), Some(CONTENT), @@ -427,7 +436,7 @@ mod tests { let backend = MemBackend::default(); let key = b"\x01discard-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(b"residue").await.unwrap(); staging.discard().await; @@ -455,9 +464,9 @@ mod tests { }; let key = b"\x01failure-key".to_vec(); - let mut staging = backend.put_staged(&key).await.unwrap(); + let mut staging = backend.put_staged().await.unwrap(); staging.append(CONTENT).await.unwrap(); - let err = staging.commit().await.unwrap_err(); + let err = staging.commit(&key).await.unwrap_err(); assert!(matches!(err, Error::Io(_)), "engine media failure"); assert!( diff --git a/src/store.rs b/src/store.rs index 2362724..9e585b2 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,16 +1,14 @@ //! The `Store` facade type (store-api.md §What this is): the public //! surface consumers program against — constructor modes, the -//! size-threshold dispatch seam, the read-path ops (`get`/`stat`/ -//! `read_range`/`has`/`list` with the fall-through general rule, -//! implemented below), and the GC protocol's ops (`sweep`, direct -//! `delete`, the liveness registration — arbitration-then-delete per -//! ADR-005/012, implemented below). The remaining op: the put seam -//! (`put`, the kv-only rejection behavior, the batching seam) lands -//! in `store-put-paths` against the pin registry minted here. -//! Direct `delete` is *not* a read path — it participates in the -//! delete-window arbitration. This module owns the engine representation -//! and the constructor-mode machinery; the pure routing function lives -//! in [`crate::dispatch`]. +//! size-threshold dispatch seam, the write path (`put`, the two internal +//! put paths behind one signature, the `Batch` batching seam, kv-only +//! rejection — implemented below), the read-path ops (`get`/`stat`/ +//! `read_range`/`has`/`list` with the fall-through general rule), +//! and the GC protocol's ops (`sweep`, direct `delete`, the liveness +//! registration — arbitration-then-delete per ADR-005/012, implemented +//! below). This module owns the engine representation and the +//! constructor-mode machinery; the pure routing function lives in +//! [`crate::dispatch`]. //! //! **Named design point — engine representation: concrete `Store`, //! crate-internal enum dispatch.** A `Store` @@ -118,6 +116,65 @@ //! and the engines receive only the physical byte encoding — they //! never see typed keys. //! +//! **Write-path design points (recorded for the review gate):** +//! +//! - **The stream parameter shape: `put(len, stream: impl AsyncRead + +//! Unpin + Send)`** — generic over the concrete reader, no boxing. +//! The family's no-`dyn` posture (wave-1 design point 3: no `dyn` +//! anywhere) carries to the write seam: the reader is monomorphized, +//! and a caller holding a boxed reader passes the box (a +//! `Box` is itself `AsyncRead + Unpin + +//! Send` under tokio's blanket impl). `Send` (and no `'static`: +//! nothing is spawned) keeps spawn-site bounds resolvable for the +//! ops layer (wave 7), matching the ADR-005/ops posture. A boxed +//! trait-object parameter was rejected as the *default* shape — it +//! would force an allocation on every call (including the known- +//! length path's) to carry a posture the family has already ruled +//! against. Callers who need erasure box at their call site. +//! - **The declared-len mismatch case (`Verification`).** A known- +//! length stream that yields fewer or more bytes than `len` rejects +//! with `Error::Verification` on either side: the derivation's +//! preamble carries the *declared* length, so a mismatched artifact's +//! canonical address can never equal the stored key (a later get +//! re-derives over the actual length and hash-checks — storing under +//! the declared-length derivation would break pool integrity by +//! construction). Zero residue on either side: the kv path buffers +//! and publishes nothing; the staged path drives +//! `crate::backend::backend`'s `StagingCursor::discard`. The error family is +//! ADR-pinned to five members (ADR-012 §6.5); `Verification` is the +//! put-path hash-check family member and this is a content-≠-key +//! fact (the ADR-006 whole-blob posture applied at put time). +//! - **The kv-only over-threshold rejection's error carrier (named +//! design point).** ADR-012 §6.2's put-time rejection rides +//! `Error::Io(String)` with a descriptive message: the pinned family +//! (ADR-012 §6.5) holds no dedicated variant, and `Missing`/ +//! `Verification`/`KeyInvalid`/`GcRefuse` are strictly narrower +//! families — the rejection is an operational/capability fact about +//! the store's medium posture, the bucket `Io(String)` covers +//! (stringly, so the message names mode, threshold, and the +//! rejected length). Recorded for the review gate: a dedicated +//! variant for embedders that need typed matchability would be a +//! new failure family — a new ADR. +//! - **The staged-put seam resolves the key at commit (the wave-2 +//! amendment of the wave-1 handout shape).** The canonical key is +//! the derivation over preamble + content and completes at stream +//! end — after the staging cursor exists (both large-tier put paths +//! stream content once into the cursor while hashing; ADR-003). The +//! `crate::backend::backend`'s `put_staged` handout is therefore +//! keyless and the staging cursor's `commit` carries the +//! store-core-resolved key (CAS first-writer-wins as the whole-value +//! put). The wave-1 notes' `put_staged(key)` handout predates this +//! task's flow realization; no ADR text pins the key's arrival +//! point, and the amendment is recorded for review-wave-2. +//! - **`Batch` surface: `batch()` + `put_in_batch(&mut Batch, ...)`.** +//! The batch-scoped pin group is minted by the store core +//! ([`crate::gc`]); a member put runs the lone-put flow and +//! contributes its pin to the group (`Batch::add` — the minted guard +//! is defused, the group's one-step release owns the protection). +//! Members publish pin-before-publish exactly as lone puts do; the +//! group's pins drop together on [`Batch::commit`] or drop (the +//! `gc-protocol` machinery drives). +//! //! **The fleet declaration defers to the engine waves.** ADR-012 §3 //! rules fleet mode a constructor declaration (`fleet: true`), never //! inferred — but it is a constructor *error* for every non-fleet-valid @@ -136,11 +193,10 @@ //! [`Store::large`]), and the GC state home ([`crate::gc::GcHome`]) //! the gc-protocol task fills. //! -//! The `#![allow(dead_code)]` below marks the work still ahead, not the -//! read path (which is driven below): write-side `tier_for`/`threshold` -//! and the GC home wait on `store-put-paths`/`gc-protocol`, and the kv -//! engine set is empty in default builds (mem is feature-gated; sqlite -//! is wave 3). The allow shrinks as those land. +//! The `#![allow(dead_code)]` below marks the work still ahead, not +//! the put path (which is driven below): the GC home's fleet clauses +//! and the kv engine set are empty in default builds (mem is +//! feature-gated; sqlite is wave 3). The allow shrinks as those land. #![allow(dead_code)] @@ -148,14 +204,18 @@ use std::collections::BTreeSet; use std::sync::Arc; use std::time::SystemTime; -use crate::backend::{Backend, ReadCursor}; +use tokio::io::{AsyncRead, AsyncReadExt}; + +use crate::backend::{Backend, ReadCursor, StagingCursor}; use crate::dispatch::{route, Tier}; use crate::facade::{ - Blob, EntryMeta, GcAbortCause, LivenessSource, ProtectCallback, Registration, SweepReport, + Batch, Blob, EntryMeta, GcAbortCause, LivenessSource, Pin, ProtectCallback, Registration, + SweepReport, }; use crate::gc::{Arbitration, GcHome}; +use crate::hashing::{Algorithm, BlobHasher}; use crate::keys::Key; -use crate::local::{LocalBackend, LocalCursor}; +use crate::local::{LocalBackend, LocalCursor, LocalStaging}; #[cfg(feature = "mem")] use crate::mem::{MemBackend, MemCursor}; @@ -259,7 +319,62 @@ impl ReadCursor for AnyCursor { } } +/// The write path's staging-cursor dispatch (ADR-010 §1: the store core +/// programs the abstraction): the large tier's stage bodies behind one +/// crate-internal value, the put seam's twin of [`AnyCursor`]. The kv +/// tier never stages on the facade path (the whole-value put is its +/// settled form); a kv-degenerate staged body variant arrives only if +/// the conformance suite demands the staged form through the facade. +/// Crate-internal — engine cursors never leak onto the public surface. +pub(crate) enum AnyStaging { + /// The large body family: a stage file under the stage-hygiene + /// protocol, published by commit-rename. + Large(LocalStaging), +} + +impl ReadCursor for AnyStaging { + fn len(&self) -> u64 { + match self { + Self::Large(cursor) => cursor.len(), + } + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + match self { + Self::Large(cursor) => cursor.read_at(offset, len).await, + } + } +} + +impl StagingCursor for AnyStaging { + async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + match self { + Self::Large(cursor) => cursor.append(chunk).await, + } + } + + async fn commit(self, key: &[u8]) -> Result<(), Error> { + match self { + Self::Large(cursor) => cursor.commit(key).await, + } + } + + async fn discard(self) { + match self { + Self::Large(cursor) => cursor.discard().await, + } + } +} + impl KvEngine { + #[allow(unused_variables)] // kv engine set empty until wave 3 (sqlite): no-mem builds never reference `key`/`value` + async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + match *self { + #[cfg(feature = "mem")] + KvEngine::Mem(ref engine) => engine.put(key, value).await, + } + } + #[allow(unused_variables)] // kv engine set empty until wave 3 (sqlite): no-mem builds never reference `key` async fn get(&self, key: &[u8]) -> Result, Error> { match *self { @@ -301,6 +416,15 @@ impl KvEngine { } impl LargeEngine { + /// The staged form's handout (ADR-010 §1): the put seam's large path + /// streams into this cursor while hashing; the key resolves at the + /// staging cursor's commit. + async fn put_staged(&self) -> Result { + match self { + LargeEngine::Local(engine) => Ok(engine.put_staged().await.map(AnyStaging::Large)?), + } + } + async fn get(&self, key: &[u8]) -> Result, Error> { match self { LargeEngine::Local(engine) => Ok(engine.get(key).await?.map(AnyCursor::Large)), @@ -598,6 +722,339 @@ impl Store { } } +/// The write path's streaming chunk granularity — every put loop reads +/// the stream in windows of this size. Sized inside the kv tier's +/// economics regime (bounded memory per iteration; engine bodies +/// aggregate upstream — pg-lo's measured 512 KiB `lowrite` knee is that +/// engine wave's concern). +const PUT_CHUNK: usize = 64 * 1024; + +/// The write path (store-api.md §Write path): `put` — the two internal +/// paths behind one signature, the `Batch` batching seam, and the +/// kv-only rejection point (ADR-012 §6.2), on the pin-before-publish +/// ordering (ADR-005). The named design points (stream shape, len +/// mismatch, kv-only carrier, key-at-commit amendment) are recorded in +/// the module doc; their behaviors are pinned by the tests below. +impl Store { + /// **`put(len, stream) → (Key, Pin)`** — the streaming put seam + /// (store-api.md §Write path; POC #3 finding A1): two internal + /// paths behind one signature, routing a pure function of content + /// length (ADR-003) — no content ever migrates tiers. + /// + /// **Known-length (`Some(len)`) — the encouraged path** (covers + /// git objects, `stat`-derived files, `Content-Length`-bearing + /// uploads, network offers): the preamble (`"blob \0"`) + /// initializes the derivation before content flows, so one pass + /// suffices — no staging needed for hashing. ≤ the threshold → the + /// kv tier's whole-value put; > the threshold → the large tier's + /// staged put (content streams into the staging cursor while + /// hashing; the derivation completes at stream end; commit-rename + /// publishes). A stream that yields fewer or more bytes than the + /// declared `len` rejects with `Error::Verification` — zero residue + /// either side (the module doc's named design point). + /// + /// **Unknown-length (`None`):** content buffers in memory up to the + /// dispatch threshold, ADR-003's option (d) semantics pinned in the + /// three numbered rules: + /// + /// 1. **Mid-stream threshold overflow** (non-restartable sources + /// are normal): the buffered prefix flushes *into a fresh stage + /// file* and the stream continues appending into it — there is + /// no "restart the stream" requirement anywhere. + /// 2. **Verification runs over the eventual stored artifact as a + /// whole**: the memory buffer for the sub-threshold kv commit; + /// the complete staged bytes for the large commit — the + /// derivation pass restarts over preamble + staged bytes (the + /// preamble is inside the hashed input, ADR-002), reading the + /// staged artifact back through the staging cursor's `ReadCursor` + /// seam. + /// 3. **Stream error before commit**: no entry lands and no stage + /// file remains — buffer discard on the failure path, exactly + /// like stage discard (the stage-hygiene invariant extended to + /// the buffered case). + /// + /// **Pin-before-publish (ADR-005):** the pin refcount is minted + /// under the arbitration lock *before* the entry becomes visible — + /// an entry cannot appear in a sweep's `list()` without already + /// carrying its pin. A dedup re-put acquires no new pool entry; the + /// returned `Pin` refcounts the existing one under the same lock. + /// + /// **kv-only mode (ADR-012 §6.2):** a known-length put over the + /// threshold rejects immediately; an unknown-length put rejects at + /// mid-stream threshold overflow — lengths are not known at + /// construction, so put time is the only possible rejection point. + /// Zero residue on either rejection. + pub async fn put( + &self, + len: Option, + stream: impl AsyncRead + Unpin + Send, + ) -> Result<(Key, Pin), Error> { + match len { + Some(len) => { + if self.mode == KvOnly && len > self.threshold { + return Err(Error::Io(format!( + "kv-only mode rejects an over-threshold put: declared len {len} > threshold {}", + self.threshold + ))); + } + self.put_known(len, stream).await + } + None => self.put_unknown(stream).await, + } + } + + /// **The batch-scoped pin group** (store-api.md §Write path; ADR-005 + /// §Marks 2). Multi-put writes — manifest writes are exactly this — + /// group under one batch-scoped pin: [`Store::put_in_batch`] runs + /// each member's lone-put flow and contributes its pin to the + /// group; the group's pins drop together on [`Batch::commit`] or + /// drop (no early single-pin release; member liveness never + /// outlives the batch — batch drop is the failure path). + pub fn batch(&self) -> Batch { + self.gc().new_batch() + } + + /// A member put inside a [`Batch`] — the batching put seam + /// (store-api.md §Write path). Member puts publish pin-before- + /// publish exactly as lone puts do: the two paths behind + /// [`Store::put`]'s one signature run, and the minted guard is + /// contributed to the group instead of returned. The batch's pins + /// drop together on [`Batch::commit`] or batch drop. + pub async fn put_in_batch( + &self, + batch: &mut Batch, + len: Option, + stream: impl AsyncRead + Unpin + Send, + ) -> Result { + let (key, pin) = self.put(len, stream).await?; + batch.add(pin); + Ok(key) + } + + /// The known-length internal path: one streaming pass, the preamble + /// initialized into the derivation up front. Routing by the declared + /// length (the pure function); a stream that yields fewer or more + /// bytes than `len` rejects `Verification` with zero residue on + /// either tier. + async fn put_known( + &self, + len: u64, + mut stream: R, + ) -> Result<(Key, Pin), Error> { + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, len); + let mut chunk = vec![0u8; PUT_CHUNK]; + let mut received = 0u64; + + match self.tier_for(len) { + Tier::Kv => { + let mut value: Vec = Vec::new(); + loop { + let n = stream + .read(&mut chunk) + .await + .map_err(|e| Error::Io(e.to_string()))?; + if n == 0 { + break; + } + received = received.saturating_add(n as u64); + if received > len { + return Err(Error::Verification); + } + hasher.update(&chunk[..n]); + value.extend_from_slice(&chunk[..n]); + } + if received != len { + return Err(Error::Verification); + } + let key = Key::from(hasher.finish()); + let pin = self.gc().acquire(&key).await; + self.kv().put(&key.as_bytes(), &value).await?; + Ok((key, pin)) + } + Tier::Large => { + let engine = match self.large() { + Some(engine) => engine, + None => return Err(Error::Io("no large tier in this mode".to_owned())), + }; + let mut stage = engine.put_staged().await?; + let fed: Result<(), Error> = 'stream: loop { + let n = match stream.read(&mut chunk).await { + Ok(n) => n, + Err(e) => break 'stream Err(Error::Io(e.to_string())), + }; + if n == 0 { + break 'stream Ok(()); + } + received = received.saturating_add(n as u64); + if received > len { + break 'stream Err(Error::Verification); + } + hasher.update(&chunk[..n]); + if let Err(e) = stage.append(&chunk[..n]).await { + break 'stream Err(e); + } + }; + match fed { + Err(err) => { + stage.discard().await; + Err(err) + } + Ok(()) if received != len => { + stage.discard().await; + Err(Error::Verification) + } + Ok(()) => { + let key = Key::from(hasher.finish()); + let pin = self.gc().acquire(&key).await; + stage.commit(&key.as_bytes()).await?; + Ok((key, pin)) + } + } + } + } + } + + /// The unknown-length internal path: buffer to the dispatch + /// threshold; on mid-stream overflow the buffered prefix flushes + /// into a fresh stage file and the stream continues appending + /// (rule 1) — kv-only mode rejects at that same overflow + /// (ADR-012 §6.2). At stream end, the derivation pass runs over the + /// eventual stored artifact as a whole: the buffer (kv form) or the + /// staged bytes read back through the cursor's `ReadCursor` seam + /// (staged form, rule 2). Every failure path discards (rule 3). + async fn put_unknown( + &self, + mut stream: R, + ) -> Result<(Key, Pin), Error> { + let mut buffered = Vec::with_capacity(PUT_CHUNK); + let mut stage: Option = None; + let mut streamed_total = 0u64; + let mut chunk = vec![0u8; PUT_CHUNK]; + + let fed: Result<(), Error> = 'feed: loop { + let n = match stream.read(&mut chunk).await { + Ok(n) => n, + Err(e) => break 'feed Err(Error::Io(e.to_string())), + }; + if n == 0 { + break 'feed Ok(()); + } + streamed_total = streamed_total.saturating_add(n as u64); + + if let Some(staged) = stage.as_mut() { + let append = staged.append(&chunk[..n]).await; + if let Err(e) = append { + if let Some(cursor) = stage.take() { + cursor.discard().await; + } + break 'feed Err(e); + } + continue; + } + + let overflow = match self.threshold() { + Some(t) => (buffered.len() as u64).saturating_add(n as u64) > t, + None => false, + }; + if !overflow { + buffered.extend_from_slice(&chunk[..n]); + continue; + } + match self.mode { + KvOnly => { + let t = self.threshold; + break 'feed Err(Error::Io(format!( + "kv-only mode rejects an over-threshold put: streamed {streamed_total} bytes past the threshold {t}" + ))); + } + _ => match self.large() { + Some(engine) => { + let mut staged = match engine.put_staged().await { + Ok(staged) => staged, + Err(e) => break 'feed Err(e), + }; + if let Err(e) = staged.append(&buffered).await { + staged.discard().await; + break 'feed Err(e); + } + if let Err(e) = staged.append(&chunk[..n]).await { + staged.discard().await; + break 'feed Err(e); + } + stage = Some(staged); + } + None => { + break 'feed Err(Error::Io("no large tier in this mode".to_owned())); + } + }, + } + }; + + match fed { + Err(err) => { + if let Some(cursor) = stage.take() { + cursor.discard().await; + } + Err(err) + } + Ok(()) => match stage.take() { + Some(staged) => { + // Rule 2, staged form: the derivation pass restarts + // over preamble + staged bytes, the artifact read + // back through the cursor's ReadCursor seam. + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, streamed_total); + let mut offset = 0u64; + let derived: Result<(), Error> = 'pass: loop { + if offset >= streamed_total { + break 'pass Ok(()); + } + let want = (streamed_total - offset).min(PUT_CHUNK as u64); + let bytes = match staged.read_at(offset, want).await { + Ok(bytes) => bytes, + Err(e) => break 'pass Err(e), + }; + if bytes.is_empty() { + break 'pass Err(Error::Io( + "staged artifact under-read during the derivation pass".to_owned(), + )); + } + hasher.update(&bytes); + offset += bytes.len() as u64; + }; + match derived { + Err(err) => { + staged.discard().await; + Err(err) + } + Ok(()) => { + let key = Key::from(hasher.finish()); + let pin = self.gc().acquire(&key).await; + staged.commit(&key.as_bytes()).await?; + Ok((key, pin)) + } + } + } + None => { + // Rule 2, buffer form: one derivation pass over the + // complete buffered artifact — the kv commit's whole + // stored value. + if buffered.len() as u64 != streamed_total { + return Err(Error::Io( + "buffered artifact length disagrees with the streamed total".to_owned(), + )); + } + let mut hasher = BlobHasher::new(Algorithm::GitBlobSha256, streamed_total); + hasher.update(&buffered); + let key = Key::from(hasher.finish()); + let pin = self.gc().acquire(&key).await; + self.kv().put(&key.as_bytes(), &buffered).await?; + Ok((key, pin)) + } + }, + } + } +} + /// **The delete-window commit batch size** — ~100/batch, iroh's proven /// shape (ADR-005 §Sweep; POC #1 finding 3's named shape adopted). const SWEEP_DELETE_BATCH: usize = 100; @@ -2190,4 +2647,842 @@ mod tests { "a malformed pool list is fatal through the sweep" ); } + + // ------------------------------------------------------------------ + // The put seam (store-put-paths): the two internal paths behind one + // signature, the three numbered unknown-length rules, pin-before- + // publish with dedup refcounting, the Batch seam, kv-only rejection + // (ADR-003/005/010/012 §6.2) — on the facade's write half. + // ------------------------------------------------------------------ + + use std::collections::VecDeque; + use std::io::Cursor; + use std::pin::Pin as TestPin; + use std::task::{Context, Poll}; + use tokio::io::ReadBuf; + + /// A stream cut at controlled chunk boundaries (`sizes`, in order; + /// tail data falls out whole) — the fixture that makes mid-stream + /// overflow, flush ordering, and staged continuation observable + /// past the put loop's own chunking. + struct Chunked { + data: Vec, + cursor: usize, + sizes: VecDeque, + } + + impl Chunked { + fn new(data: &[u8], sizes: &[usize]) -> Chunked { + Chunked { + data: data.to_vec(), + cursor: 0, + sizes: sizes.iter().copied().collect(), + } + } + } + + impl AsyncRead for Chunked { + fn poll_read( + mut self: TestPin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.cursor >= self.data.len() || buf.remaining() == 0 { + return Poll::Ready(Ok(())); + } + let remaining = self.data.len() - self.cursor; + let wanted = self + .sizes + .front() + .copied() + .unwrap_or(remaining) + .min(remaining) + .min(buf.remaining()); + buf.put_slice(&self.data[self.cursor..self.cursor + wanted]); + self.cursor += wanted; + if let Some(front) = self.sizes.front_mut() { + *front -= wanted; + if *front == 0 { + self.sizes.pop_front(); + } + } + Poll::Ready(Ok(())) + } + } + + /// A stream that delivers `fail_after` bytes then errors mid-stream — + /// the non-restartable source's failure (rule 3's shape). + struct Failing { + data: Vec, + cursor: usize, + fail_after: usize, + } + + impl Failing { + fn new(data: &[u8], fail_after: usize) -> Failing { + Failing { + data: data.to_vec(), + cursor: 0, + fail_after, + } + } + } + + impl AsyncRead for Failing { + fn poll_read( + mut self: TestPin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let message = "stream failure (test fixture)"; + if self.cursor >= self.fail_after { + return Poll::Ready(Err(std::io::Error::other(message))); + } + let wanted = (self.data.len() - self.cursor) + .min(self.fail_after - self.cursor) + .min(buf.remaining()); + if wanted == 0 { + return Poll::Ready(Err(std::io::Error::other(message))); + } + buf.put_slice(&self.data[self.cursor..self.cursor + wanted]); + self.cursor += wanted; + Poll::Ready(Ok(())) + } + } + + /// A stream nobody may ever read — the fixture proving kv-only's + /// known-length rejection happens *immediately*, before content + /// flows (ADR-012 §6.2). + struct Unreadable; + + impl AsyncRead for Unreadable { + fn poll_read( + self: TestPin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Ready(Err(std::io::Error::other( + "the stream must not be read (test fixture)", + ))) + } + } + + /// Files currently in the local tier's staging area — the residue + /// counter of the store-level stage-hygiene assertions. + fn store_stage_file_count(root: &Path) -> usize { + match fs::read_dir(root.join("staging")) { + Ok(entries) => entries.count(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0, + Err(e) => panic!("staging dir unreadable: {e}"), + } + } + + /// INV-1, kv tier, known-length path: `put(Some(len), stream)` + /// round-trips byte-identical, lands in the routed tier (kv), and + /// the pin rides out with the put. + #[tokio::test] + async fn known_length_put_round_trips_on_the_kv_tier() { + let root = temp_root("put-known-kv"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + let content = b"a known-length small blob"; + let canonical = key_for(content); + + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("kv put publishes"); + assert_eq!(key, canonical, "the canonical derivation addresses the put"); + assert_eq!(store.gc().pin_count(&canonical), 1, "the pin rode out"); + assert!( + store.kv().has(&canonical.as_bytes()).await.unwrap(), + "landed in the routed tier (kv)" + ); + assert!( + !store + .large() + .unwrap() + .has(&canonical.as_bytes()) + .await + .unwrap(), + "no large-tier copy on the kv route" + ); + let (len, blob) = store + .get(&canonical.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(len, content.len() as u64); + assert_eq!(blob.read_at(0, len).await.unwrap(), &content[..]); + + drop(pin); + } + + /// INV-1, large tier, known-length path: over-threshold known + /// content streams into the staged put with hashing in the same + /// pass, the derivation completing at stream end; commit-rename + /// publishes; zero stage files remain. + #[tokio::test] + async fn known_length_put_round_trips_on_the_large_tier_through_the_staged_put() { + let root = temp_root("put-known-large"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + let content = vec![42u8; 300]; + let canonical = key_for(&content); + + let (key, pin) = store + .put(Some(300), Cursor::new(&content[..])) + .await + .expect("large staged put publishes"); + assert_eq!(key, canonical); + assert_eq!(store.gc().pin_count(&canonical), 1); + assert!( + store + .large() + .unwrap() + .has(&canonical.as_bytes()) + .await + .unwrap(), + "landed in the routed tier (large, commit-rename)" + ); + assert!( + !store.kv().has(&canonical.as_bytes()).await.unwrap(), + "no kv copy on the large route" + ); + let (len, blob) = store + .get(&canonical.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(len, 300); + assert_eq!(blob.read_at(0, len).await.unwrap(), content); + assert_eq!( + store_stage_file_count(root.path()), + 0, + "commit-rename left zero stage files" + ); + + drop(pin); + } + + /// INV-1, both tiers, unknown-length path: sub-threshold content + /// commits through the buffered kv form; over-threshold content + /// flushed-through staging — both round-trip byte-identical and + /// both arrive under the canonical derivation over the eventual + /// stored artifact as a whole (rule 2 asserted by key equality: + /// anything but preamble + the wholeArtifact derivation would + /// derive a different key). + #[tokio::test] + async fn unknown_length_put_round_trips_on_both_tiers() { + let root = temp_root("put-unknown"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + + let small = b"small unknown-length content"; + let (key, pin) = store + .put(None, Cursor::new(&small[..])) + .await + .expect("buffered put publishes"); + assert_eq!(key, key_for(small), "rule 2, buffer form"); + assert!(store.kv().has(&key.as_bytes()).await.unwrap()); + drop(pin); + + let big: Vec = (0..200u32).map(|i| (i % 251) as u8).collect(); + let (key, pin) = store + .put(None, Chunked::new(&big, &[7; 30])) + .await + .expect("staged put publishes"); + assert_eq!(key, key_for(&big), "rule 2, staged-bytes form"); + assert!( + store.large().unwrap().has(&key.as_bytes()).await.unwrap(), + "the over-threshold unknown put landed large" + ); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(len, big.len() as u64); + assert_eq!(blob.read_at(0, len).await.unwrap(), big); + assert_eq!( + store_stage_file_count(root.path()), + 0, + "zero stage files on success" + ); + drop(pin); + } + + /// Rule 1 (ADR-003 option (d)): threshold exceeded mid-stream — the + /// buffered prefix flushes into a fresh stage file and the stream + /// *continues appending into it*; no restart anywhere, nothing + /// migrates tiers, and the flushed artifact derives canonically. + #[tokio::test] + async fn mid_stream_overflow_flushes_into_a_fresh_stage_and_the_stream_continues() { + let root = temp_root("put-rule1-flush"); + let store = Store::dual_tier_mem_local(root.path(), 8) + .await + .expect("dual-tier constructs"); + let content: Vec = (0..40u32).map(|i| (i % 199) as u8).collect(); + // Chunks of 3: two fill the buffer to 6; the third (would-be 9) + // crosses the threshold mid-stream and triggers the flush; the + // rest stream on into the stage. + let chunked = Chunked::new(&content, &[3; 14]); + + let (key, pin) = store + .put(None, chunked) + .await + .expect("mid-stream overflow flushes; the stream continues"); + assert_eq!( + key, + key_for(&content), + "the flushed artifact derives canonically" + ); + assert!( + !store.kv().has(&key.as_bytes()).await.unwrap(), + "nothing migrated to the kv tier on a large-route put" + ); + assert!( + store.large().unwrap().has(&key.as_bytes()).await.unwrap(), + "the stage holds the whole streamed artifact" + ); + let (_, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!( + blob.read_at(0, content.len() as u64).await.unwrap(), + content, + "the flushed prefix and the continued stream are one artifact" + ); + assert_eq!( + store_stage_file_count(root.path()), + 0, + "the stage published by rename; zero residue" + ); + drop(pin); + } + + /// The dispatch seam's boundary through the put path: an + /// unknown-length stream ending exactly at the threshold stays kv + /// (≤ threshold routes kv; the overflow is strictly past it). + #[tokio::test] + async fn unknown_length_exactly_at_threshold_stays_kv() { + let root = temp_root("put-boundary"); + let store = Store::dual_tier_mem_local(root.path(), 8) + .await + .expect("dual-tier constructs"); + let content = b"01234567"; // exactly 8 bytes + let chunked = Chunked::new(content, &[4, 4]); + + let (key, pin) = store + .put(None, chunked) + .await + .expect("at-threshold stream commits through the buffer"); + assert_eq!(key, key_for(content)); + assert!( + store.kv().has(&key.as_bytes()).await.unwrap(), + "exactly-at-threshold stays kv" + ); + assert!( + !store.large().unwrap().has(&key.as_bytes()).await.unwrap(), + "no large-tier copy on the boundary" + ); + drop(pin); + } + + /// Rule 3 (ADR-003): a stream error before commit leaves no entry + /// and no stage file — on the buffered kv form, on the flushed + /// staged form, and on both known-length tiers; the stage-hygiene + /// invariant extended to the buffered case. + #[tokio::test] + async fn stream_errors_before_commit_leave_no_entry_and_no_stage_file() { + let root = temp_root("put-rule3-error"); + let store = Store::dual_tier_mem_local(root.path(), 8) + .await + .expect("dual-tier constructs"); + let content: Vec = (0..40u32).map(|i| (i % 173) as u8).collect(); + + // Flushed staged form: the overflow flush ran, then the stream + // errored — the stage discards. + let err = store + .put(None, Failing::new(&content, 12)) + .await + .unwrap_err(); + assert!( + matches!(err, Error::Io(_)), + "stream failures surface as Io: {err}" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "no entry landed from the errored staging put" + ); + assert_eq!( + store_stage_file_count(root.path()), + 0, + "no stage file remains" + ); + + // Buffered kv form: error before the threshold — the buffer + // discards exactly like a stage. + let err = store + .put(None, Failing::new(&content, 5)) + .await + .unwrap_err(); + assert!(matches!(err, Error::Io(_))); + assert_eq!(store.list().await.unwrap(), Vec::::new()); + assert_eq!(store_stage_file_count(root.path()), 0); + + // Known-length large path: mid-stage failure discards. + let err = store + .put(Some(40), Failing::new(&content, 12)) + .await + .unwrap_err(); + assert!(matches!(err, Error::Io(_))); + assert_eq!(store.list().await.unwrap(), Vec::::new()); + assert_eq!(store_stage_file_count(root.path()), 0); + + // Known-length kv path: never published. + let err = store + .put(Some(40), Failing::new(&content, 5)) + .await + .unwrap_err(); + assert!(matches!(err, Error::Io(_))); + assert_eq!(store.list().await.unwrap(), Vec::::new()); + assert_eq!(store_stage_file_count(root.path()), 0); + } + + /// Pin-before-publish (ADR-005) on both put paths and both tiers: + /// a sweep running alongside an in-flight put never deletes the + /// protected entry — the pin is minted before the entry can appear + /// in any `list()` — and every completed schedule keeps exact + /// counts. The surviving pin protects; dropping it re-exposes. + #[tokio::test] + async fn pin_before_publish_holds_on_both_put_paths_and_both_tiers() { + let root = temp_root("put-pin"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let big: Vec = (0..200u32).map(|i| (i % 241) as u8).collect(); + let big_other: Vec = (0..288u32).map(|i| (i % 223) as u8).collect(); + let rounds: [(Vec, Option); 4] = [ + (b"pinned known small".to_vec(), Some(18)), + (b"still-pinned unknown small content".to_vec(), None), + (big, Some(200)), + (big_other, None), + ]; + + for (round, (content, len)) in rounds.iter().enumerate() { + let expected = key_for(content); + let (report, put) = tokio::join!(store.sweep(), async { + store.put(*len, Cursor::new(&content[..])).await + }); + let report = report.expect("sweeps are never errors"); + let (key, pin) = put.expect("put published"); + assert_eq!(key, expected, "round {round}"); + assert_eq!(report.aborted, None, "round {round}"); + assert_eq!( + report.candidates_staged, + report.deleted + report.cancelled_by_arbitration, + "round {round}: exact counts in every schedule" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "round {round}: the entry survived the race" + ); + assert_eq!( + store.gc().pin_count(&expected), + 1, + "round {round}: the minted pin still protects" + ); + let (_, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!( + blob.read_at(0, content.len() as u64).await.unwrap(), + &content[..], + "round {round}: the published entry is byte-identical" + ); + + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "round {round}: unpinned, deleted"); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + } + + /// Dedup (store-api.md invariant 2; ADR-005): a re-put of identical + /// content acquires no new pool entry and refcounts the existing + /// one — across put paths (known vs unknown) and across both tiers; + /// the refcounts compose under the same lock and drop in sequence. + #[tokio::test] + async fn dedup_reputs_refcount_the_existing_entry_from_any_path() { + let root = temp_root("put-dedup"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let content = b"deduplicated content"; + let (known_key, first) = store + .put(Some(content.len() as u64), Cursor::new(&content[..])) + .await + .expect("known-path put"); + let (unknown_key, second) = store + .put(None, Cursor::new(&content[..])) + .await + .expect("unknown-path re-put"); + assert_eq!(known_key, unknown_key, "one derivation, one address"); + assert_eq!( + store.list().await.unwrap(), + vec![known_key.clone()], + "one pool entry" + ); + assert_eq!( + store.gc().pin_count(&known_key), + 2, + "the re-put refcounted the existing entry" + ); + + drop(first); + assert_eq!(store.gc().pin_count(&known_key), 1); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 0, "one refcount still protects"); + + drop(second); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1); + assert!(!store.has(&known_key.as_bytes()).await.unwrap()); + + // The same composition across the large tier: two staged puts + // of identical content — one entry, two refcounts. + let big = [91u8; 200]; + let (first_key, first) = store + .put(Some(200), Cursor::new(&big[..])) + .await + .expect("known staged put"); + let (_, second) = store + .put(None, Cursor::new(&big[..])) + .await + .expect("unknown staged re-put"); + assert_eq!( + store.list().await.unwrap(), + vec![first_key.clone()], + "one pool entry across both staged paths" + ); + assert_eq!(store.gc().pin_count(&first_key), 2); + drop(first); + drop(second); + } + + /// `Batch` (store-api.md §Write path; ADR-005 §Marks 2): multi-put + /// writes group under one batch-scoped pin — members publish pin- + /// before-publish exactly as lone puts do, both tiers, and the + /// group's pins drop together on batch commit. + #[tokio::test] + async fn batch_groups_multi_put_writes_and_members_drop_together() { + let root = temp_root("put-batch"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let small_a = b"manifest member a"; + let small_b = b"manifest member b"; + let big = vec![77u8; 200]; + + let mut batch = store.batch(); + assert!(batch.is_empty(), "a fresh batch protects nothing"); + let key_a = store + .put_in_batch(&mut batch, Some(17), Cursor::new(&small_a[..])) + .await + .expect("batch member a"); + let key_b = store + .put_in_batch(&mut batch, None, Cursor::new(&small_b[..])) + .await + .expect("batch member b"); + let key_big = store + .put_in_batch(&mut batch, Some(200), Cursor::new(&big[..])) + .await + .expect("batch member large"); + assert_eq!(batch.len(), 3); + assert_eq!(key_a, key_for(small_a)); + assert_eq!(key_b, key_for(small_b)); + assert_eq!(key_big, key_for(&big)); + + for key in [&key_a, &key_b, &key_big] { + assert_eq!( + store.gc().pin_count(key), + 1, + "each member's pin rides in the group" + ); + } + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the batch-scoped pin protects the whole group" + ); + assert!(store.has(&key_a.as_bytes()).await.unwrap()); + assert!(store.has(&key_big.as_bytes()).await.unwrap()); + + batch.commit(); + for key in [&key_a, &key_b, &key_big] { + assert_eq!( + store.gc().pin_count(key), + 0, + "no early single-pin release and no late release either" + ); + } + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 3, "unprotected after the batch ends"); + assert!(!store.has(&key_a.as_bytes()).await.unwrap()); + assert!(!store.has(&key_b.as_bytes()).await.unwrap()); + assert!(!store.has(&key_big.as_bytes()).await.unwrap()); + } + + /// Batch drop is the failure path: a member put error leaves the + /// landed members batch-protected and contributes nothing to the + /// group; member liveness never outlives the batch. + #[tokio::test] + async fn batch_member_put_failure_does_not_endanger_landed_members() { + let root = temp_root("put-batch-failure"); + let store = Store::dual_tier_mem_local(root.path(), 64) + .await + .expect("dual-tier constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let landed = b"the member that landed"; + let mut batch = store.batch(); + let key = store + .put_in_batch(&mut batch, Some(22), Cursor::new(&landed[..])) + .await + .expect("landed member published"); + assert_eq!(store.gc().pin_count(&key), 1); + + let content: Vec = (0..40u32).map(|i| (i % 157) as u8).collect(); + let err = store + .put_in_batch(&mut batch, None, Failing::new(&content, 12)) + .await + .unwrap_err(); + assert!( + matches!(err, Error::Io(_)), + "the failed member surfaces its error" + ); + assert_eq!(batch.len(), 1, "the failed member joined nothing"); + assert_eq!( + store.gc().pin_count(&key), + 1, + "landed members stay protected" + ); + assert_eq!( + store.list().await.unwrap(), + vec![key.clone()], + "the failed member published no entry" + ); + + drop(batch); + assert_eq!( + store.gc().pin_count(&key), + 0, + "batch drop released every member together" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1); + } + + /// kv-only mode's put-time rejection (ADR-012 §6.2): known-length + /// over-threshold rejects *immediately* (the stream is never read); + /// unknown-length rejects at mid-stream threshold overflow — zero + /// residue on either rejection, and at/exactly-threshold puts land. + #[tokio::test] + async fn kv_only_mode_rejects_over_threshold_puts_at_the_two_points() { + let store = Store::kv_only_mem(64); + store.register_liveness_source(Arc::new(SetSource::default())); + + let err = store.put(Some(65), Unreadable).await.unwrap_err(); + assert!( + err.to_string().contains("kv-only mode rejects"), + "the immediate rejection carries its carrier message: {err}" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "zero residue" + ); + + let at_threshold = vec![9u8; 64]; + let (key, pin) = store + .put(Some(64), Cursor::new(&at_threshold[..])) + .await + .expect("exactly-at-threshold known put lands"); + assert_eq!(key, key_for(&at_threshold)); + assert!(store.has(&key.as_bytes()).await.unwrap()); + drop(pin); + + let streamed = vec![3u8; 100]; + let err = store + .put(None, Chunked::new(&streamed, &[10; 10])) + .await + .unwrap_err(); + assert!( + err.to_string() + .contains("kv-only mode rejects an over-threshold put"), + "the overflow rejection is the mode's, not the stream's: {err}" + ); + assert_eq!( + store.list().await.unwrap(), + vec![key_for(&at_threshold)], + "zero residue" + ); + + let just_under = vec![5u8; 64]; + let (key, pin) = store + .put(None, Chunked::new(&just_under, &[8; 8])) + .await + .expect("at-threshold unknown stream lands through the buffer"); + assert_eq!(key, key_for(&just_under)); + assert!(store.has(&key.as_bytes()).await.unwrap()); + drop(pin); + } + + /// The declared-len mismatch (the put seam's named design point): + /// a known-length stream yielding fewer or more bytes than `len` + /// rejects with `Error::Verification` on either tier — zero residue + /// either way (nothing published, no stage file left). + #[tokio::test] + async fn declared_len_mismatch_rejects_verification_with_zero_residue() { + let root = temp_root("put-mismatch"); + let store = Store::dual_tier_mem_local(root.path(), 8) + .await + .expect("dual-tier constructs"); + let content = b"0123456789"; // 10 bytes + + assert_eq!( + store + .put(Some(5), Cursor::new(&content[..])) + .await + .unwrap_err(), + Error::Verification, + "more bytes than declared (kv route)" + ); + assert_eq!( + store + .put(Some(20), Cursor::new(&content[..])) + .await + .unwrap_err(), + Error::Verification, + "fewer bytes than declared (kv route)" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "zero residue on the kv mismatches" + ); + + let big = [5u8; 32]; + assert_eq!( + store + .put(Some(16), Cursor::new(&big[..])) + .await + .unwrap_err(), + Error::Verification, + "more bytes than declared (large route)" + ); + assert_eq!( + store + .put(Some(64), Cursor::new(&big[..])) + .await + .unwrap_err(), + Error::Verification, + "fewer bytes than declared (large route)" + ); + assert_eq!( + store.list().await.unwrap(), + Vec::::new(), + "zero residue on the staged mismatches" + ); + assert_eq!( + store_stage_file_count(root.path()), + 0, + "the discarded stages left zero stage files" + ); + } + + /// mem-only accepts every length on both put paths — no dispatch + /// threshold in effect (ADR-011 §4), nothing to reject; the sole + /// mem engine holds even over-default-threshold content. + #[tokio::test] + async fn mem_only_accepts_every_length_on_both_put_paths() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + assert_eq!(store.threshold(), None, "no threshold in effect"); + + let small = b"mem-only content"; + let (key, pin) = store + .put(Some(16), Cursor::new(&small[..])) + .await + .expect("known put"); + assert_eq!(key, key_for(small)); + drop(pin); + + let big = vec![1u8; (DEFAULT_DISPATCH_THRESHOLD + 1) as usize]; + let (key, pin) = store + .put(None, Cursor::new(&big[..])) + .await + .expect("over-default-threshold unknown put routes to the sole kv engine"); + assert_eq!(key, key_for(&big)); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(len, big.len() as u64); + assert_eq!(blob.read_at(0, len).await.unwrap(), big); + drop(pin); + } + + /// The zero-length degenerate on both put paths: empty content + /// round-trips through its canonical key (one entry, both paths + /// dedup into it), reading empty-never-error. + #[tokio::test] + async fn zero_length_puts_round_trip_on_both_paths() { + let root = temp_root("put-empty"); + let store = Store::dual_tier_mem_local(root.path(), 8) + .await + .expect("dual-tier constructs"); + let empty = b""; + + let (known, pin) = store + .put(Some(0), Cursor::new(&empty[..])) + .await + .expect("known empty put"); + assert_eq!(known, key_for(empty)); + assert!(store.kv().has(&known.as_bytes()).await.unwrap()); + drop(pin); + + let (unknown, pin) = store + .put(None, Cursor::new(&empty[..])) + .await + .expect("unknown empty put"); + assert_eq!(unknown, known, "empty content dedups into one entry"); + assert_eq!(store.list().await.unwrap(), vec![known.clone()]); + drop(pin); + + let (len, blob) = store + .get(&known.as_bytes()) + .await + .unwrap() + .expect("entry present"); + assert_eq!(len, 0); + assert!(blob.is_empty()); + assert_eq!(blob.read_at(0, 8).await.unwrap(), Vec::::new()); + } } diff --git a/tasks/store-put-paths.md b/tasks/store-put-paths.md index 7ce31a7..65d06d6 100644 --- a/tasks/store-put-paths.md +++ b/tasks/store-put-paths.md @@ -1,7 +1,7 @@ --- id: store-put-paths name: Store put seam — two internal paths, the three numbered rules, Batch, kv-only rejection -status: pending +status: completed depends_on: [gc-protocol] scope: moderate risk: medium @@ -67,22 +67,22 @@ internal paths behind one signature, with pin-before-publish from the ## Acceptance Criteria -- [ ] `put(len, stream) → (Key, Pin)` with both internal paths behind +- [x] `put(len, stream) → (Key, Pin)` with both internal paths behind the one signature; routing is the pure length function (no content ever migrates tiers) -- [ ] The three numbered rules each behavior-tested: mid-stream +- [x] The three numbered rules each behavior-tested: mid-stream overflow flushes into a fresh stage and the stream continues; the derivation pass runs over the eventual stored artifact as a whole (buffer form and staged-bytes form); stream error leaves no entry and no stage file -- [ ] Pin-before-publish holds on both paths and both tiers; dedup +- [x] Pin-before-publish holds on both paths and both tiers; dedup re-puts refcount the existing entry -- [ ] `Batch` groups multi-put writes; member pins drop together -- [ ] kv-only mode: known-length over-threshold rejects immediately; +- [x] `Batch` groups multi-put writes; member pins drop together +- [x] kv-only mode: known-length over-threshold rejects immediately; unknown-length rejects at mid-stream overflow; zero residue on both -- [ ] Stream-shape and len-mismatch design points decided and recorded -- [ ] `cargo test` green; `cargo test --features mem` and `cargo test +- [x] Stream-shape and len-mismatch design points decided and recorded +- [x] `cargo test` green; `cargo test --features mem` and `cargo test --all-features` green; clippy/fmt gates green ## References @@ -96,8 +96,108 @@ internal paths behind one signature, with pin-before-publish from the ## Notes -> To be filled by implementation agent +- **The staged-put seam was amended: the key resolves at commit, not + at handout.** Implementing the two large-tier flows surfaced a + contradiction in the wave-1 pin: `put_staged(key)` hands the staging + cursor out with the key, but both large-tier put paths only know the + canonical key when the derivation completes — at stream end (the + preamble covers a length, and the digest covers content that has + not streamed yet). Content streams once (non-restartable sources are + the normal case), so `commit(self)` publishing under the handout + key cannot express the specced flow ("the derivation completes at + stream end; commit-rename publishes"). Amended with operator + sign-off: `put_staged()` is a keyless handout and + `StagingCursor::commit(self, key)` publishes under the + store-core-resolved key (CAS first-writer-wins as the whole-value + put). No ADR text pins the key's arrival point (ADR-003/010 pin + handout/commit/discard mechanics only); the wave-1 task notes' + `put_staged(key)` shape was pre-flow realism. Recorded for + review-wave-2. Engines updated in place (mem, local, the wave-1 + test double) with their hygiene suites re-run green. +- **Put-path named design points (recorded in `src/store.rs` module + docs, for review-wave-2):** + - *Stream parameter shape:* `put(len, stream: impl AsyncRead + Unpin + + Send)` — generic, no boxing; the family's no-`dyn` posture + carried to the write seam (a boxed reader is itself `AsyncRead + + Unpin + Send` under tokio's blanket impl, so callers needing + erasure box at their call site). `Send` keeps spawn-site bounds + resolvable for the ops wave; no `'static` (nothing spawns). + - *Declared-len mismatch:* a known-length stream yielding fewer or + more bytes than `len` rejects `Error::Verification` on either + tier, zero residue either side — the preamble carries the + *declared* length, so a mismatched artifact's canonical address + can never equal the stored key (storing it would break pool + integrity by construction; `Verification` is the put-path + content-≠-key family member, ADR-006's whole-blob posture applied + at put time). + - *kv-only rejection carrier (a named decision beyond the two the + task listed):* `Error::Io(String)` with a descriptive message — + the ADR-pinned family (ADR-012 §6.5) holds no dedicated variant + and the rejection is an operational/capability fact about the + medium posture, the bucket `Io(String)` covers. A dedicated + variant for typed matchability would be a new failure family — + a new ADR. +- **Flows as implemented** (`src/store.rs`, write-path `impl Store`): + known-length initializes the derivation with the preamble before + content flows and streams once (kv: buffer + hash in one pass → + mint pin → whole-value put; large: stream into the staging cursor + while hashing → mint pin → `commit(key)`). Unknown-length buffers + to the threshold; a mid-stream overflow (strictly past it — + exactly-at stays kv) flushes the buffered prefix into a fresh stage + and the stream continues into it; at EOF the derivation pass runs + over the eventual stored artifact as a whole — the buffer (kv form) + or preamble + staged bytes read back through the + `StagingCursor: ReadCursor` seam (staged form) — then pin, then + commit. Every failure path (read error, mismatch, under-read) + discards: no entry, no stage file. +- **API shapes:** `Store::put(&self, len: Option, stream) → + Result<(Key, Pin), Error>`; the batching seam is `Store::batch() → + Batch` + `Store::put_in_batch(&mut Batch, len, stream) → + Result` (members run the lone-put flow; the minted guard + is contributed to the group via crate-internal `Batch::add`); a + public `Batch::commit()` was added consuming the batch with the + same one-step release as drop (store-api.md's "pins drop together + on batch commit or drop", as a named verb). `hashing.rs` gained + crate-internal `BlobHasher` — the same derivation driven + incrementally, pinned byte-equal to `Algorithm::derive` by tests + (chunk-boundary invariance, both algorithms, the git vectors). +- **git-interop leg added** (`tests/git_interop.rs`): the put seam + keys entries git-identically through both put paths on both tiers + against live `git hash-object` (INV-1's source-of-truth posture + extended from the derivation constructor to the facade seam). +- **Wave-1 note updates:** `put_staged`/`commit` doc text in + `backend.rs`, `mem.rs`, `local.rs` now carries the key-at-commit + design point; the store.rs module doc's remaining-deferral note + shrunk to the actual remaining gaps (fleet clauses, wave-3 kv + engine). ## Summary -> To be filled on completion \ No newline at end of file +Implemented the facade's put seam (`src/store.rs`): `put(len: +Option, stream: impl AsyncRead + Unpin + Send) → +Result<(Key, Pin), Error>` with the two internal paths behind the one +signature — the known-length one-pass flow and the unknown-length +buffer-then-maybe-flush flow implementing ADR-003's option (d) three +numbered rules — plus `batch()`/`put_in_batch()` (ADR-005 §Marks 2's +batch-scoped pin group) and kv-only's put-time rejection at its two +pinned points (ADR-012 §6.2). Pin-before-publish rides +`gc().acquire` under the arbitration lock before both tiers' +publishes; dedup re-puts refcount the existing entry; every failure +path converges on discard with zero stage files. The staged-put seam +was amended (keyless handout; `commit(self, key)`) after operator +sign-off — the one blocking finding of the task. Supporting changes: +staging-cursor dispatch enum `AnyStaging` beside the read path's +`AnyCursor`; incremental `BlobHasher` in `hashing.rs`; public +`Batch::commit`; a `Pin`/`Debug` impl; the git-interop put-seam leg. +Eleven put-path tests in `src/store.rs` (plus four fixtures) pin the +three rules, both-tier round-trips, the threshold boundary, pin- +before-publish under a racing sweep, dedup refcounting across paths +and tiers, batch grouping/commit/failure, kv-only's two rejection +points, len-mismatch zero-residue, mem-only's threshold-free posture, +and the empty-content degenerate. + +Verification: `cargo test` (default), `cargo test --features mem`, +`cargo test --all-features` all green (97 lib tests under mem plus +conformance and git-interop); `cargo clippy --all-targets -- +-D warnings` green in all three feature shapes; `cargo fmt --check` +green; `cargo doc --no-deps` at the pre-existing 24-warning baseline. \ No newline at end of file diff --git a/tests/git_interop.rs b/tests/git_interop.rs index ef95625..166a14c 100644 --- a/tests/git_interop.rs +++ b/tests/git_interop.rs @@ -162,6 +162,80 @@ fn external_oid_round_trips_into_pool_keys() { } } +/// The facade put seam against git (store-api.md §Invariants 1's +/// source-of-truth leg): both internal put paths (known-length, +/// unknown-length) key entries git-identically on both tiers, and the +/// get path round-trips git's content byte-exactly. mem-gated: the +/// wave-2 constructible shape (`dual_tier_mem_local`). +#[cfg(feature = "mem")] +#[tokio::test] +async fn put_seam_keys_match_git_hash_object_on_both_tiers_and_paths() { + use std::io::Cursor; + + use alkblobs::Store; + + let root = scratch_dir("store-root"); + let store = Store::dual_tier_mem_local(root.join("root"), 32) + .await + .expect("dual-tier store constructs"); + let small = b"small git-interop blob"; + let big = (0..200).map(|i| (i % 253) as u8).collect::>(); + + let (repo, oid) = git_hash_object(small, true); + drop_dir(&repo); + let expected_small: Key = Digest::from_git_oid_hex(&oid) + .expect("git oid decodes") + .into(); + let (repo, oid) = git_hash_object(&big, true); + drop_dir(&repo); + let expected_big: Key = Digest::from_git_oid_hex(&oid) + .expect("git oid decodes") + .into(); + + let (key, pin) = store + .put(Some(small.len() as u64), Cursor::new(&small[..])) + .await + .expect("known-length put"); + assert_eq!(key, expected_small, "kv put keys git-identical"); + drop(pin); + let (key, pin) = store + .put(None, Cursor::new(&small[..])) + .await + .expect("unknown-length put"); + assert_eq!(key, expected_small, "buffered form keys git-identical"); + let (len, blob) = store + .get(&expected_small.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(blob.read_at(0, len).await.unwrap(), &small[..]); + drop(pin); + + let (key, pin) = store + .put(Some(big.len() as u64), Cursor::new(&big[..])) + .await + .expect("known-length large put"); + assert_eq!(key, expected_big, "staged put keys git-identical"); + drop(pin); + let (key, pin) = store + .put(None, Cursor::new(&big[..])) + .await + .expect("unknown-length large put"); + assert_eq!( + key, expected_big, + "staged-bytes derivation keys git-identical" + ); + let (len, blob) = store + .get(&expected_big.as_bytes()) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!(blob.read_at(0, len).await.unwrap(), big); + drop(pin); + + let _ = fs::remove_dir_all(root); +} + /// Tagged-key coexistence (POC finding 5): git oids of both formats /// address the same flat key universe from one external producer. #[test]