From f86e20a217be70db8108e9fd44cbb09f8151c421 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 23:19:59 +0000 Subject: [PATCH] =?UTF-8?q?feat(tests):=20the=20conformance=20suite's=20me?= =?UTF-8?q?m=20column=20=E2=80=94=20the=20reference=20row=20set=20green=20?= =?UTF-8?q?(wave=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - rows.rs: the reference row set (INV-1…6 + two constructor-mode rows), factory-generic and engine-uniform, every row ADR-stamped and traced to the backlog: round-trips over the content matrix on both put paths, dedup (same/cross-path), stage hygiene (both paths, both injectable failure modes, kv-degenerate residue via sweep outcomes), sweep safety (full-protection-set happy shape, all three GcAbortCause abort shapes incl. the single-flight SweeperLock lease, the sweep-vs-pinned-put race), list correctness through exact sweep outcomes only, stat discipline, mem-only posture, and kv-only put-time rejection (threshold discovered by free probes; immediate/ mid-stream rejections and at-threshold-lands legs) - mem_column.rs: the column's factories — mem-only (reference host) and kv-only(mem) at a declared threshold — fresh isolated opens, no-op idempotent teardown; structural factory-contract tests cover isolation and teardown against a panicking row - README: the reference row set recorded (INV-7's basis — what other columns mirror), the row-lifecycle convention, column plan and current state updated - task suite-column-mem: status completed, notes and summary filled Verification: cargo test --features mem (97 lib + 18 conformance + 4 git-interop), cargo test (default), cargo test --all-features, clippy --all-features --all-targets -D warnings, fmt --check, and cargo doc --no-deps — all green --- tasks/suite-column-mem.md | 73 +- tests/conformance/README.md | 64 +- tests/conformance/main.rs | 16 +- tests/conformance/mem_column.rs | 240 ++++++ tests/conformance/rows.rs | 1221 +++++++++++++++++++++++++++++++ 5 files changed, 1601 insertions(+), 13 deletions(-) create mode 100644 tests/conformance/mem_column.rs create mode 100644 tests/conformance/rows.rs diff --git a/tasks/suite-column-mem.md b/tasks/suite-column-mem.md index 9f18db2..08104dc 100644 --- a/tasks/suite-column-mem.md +++ b/tasks/suite-column-mem.md @@ -1,7 +1,7 @@ --- id: suite-column-mem name: Conformance suite — factory binding and the mem column (the reference column) -status: pending +status: completed depends_on: [store-read-paths, store-put-paths, gc-protocol] scope: moderate risk: medium @@ -76,8 +76,75 @@ every other engine's rows mirror (store-api.md invariant 7; ADR-011 §2). ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-10-11). + +- **The factory binding was already made and recorded** by the + store-construction task: `open()` vends the concrete + `alkblobs::Store` facade (the skeleton's `Store` associated type + collapsed), recorded in `factory.rs` + README §The factory contract + before this task ran. Verified; no further shape correction was + needed. The wave-1 binding demonstrator (`binding.rs`) remains as + the proof-of-implementability tests. +- **The reference row set** lives in `tests/conformance/rows.rs` + (engine-uniform, factory-generic, every row ADR-stamped and traced + to the backlog): INV-1 (content matrix × both put paths, in-body), + INV-2 (same-path + cross-path legs), INV-3 (both paths × both + injectable failure modes — mid-stream `Io` failure and + declared-len `Verification` mismatch; the kv stage form is + degenerate, so residue is the pool itself, observed through exact + sweep outcomes), INV-4 (happy shape over the full protection set; + all three `GcAbortCause` abort shapes — `NoLivenessSources`, + `ProtectFailed`, `SweeperLock`, the last completing the + single-node family per ADR-013 §2/ADR-012 §5; the sweep-vs-pinned- + put race over both put paths), INV-5 (exact sweep outcomes, never + list comparison — honored by every row), INV-6, and the two + constructor-mode rows. +- **INV-6's slice through the facade**: stat's shape (len + + algorithm), purity, and `None`-on-miss; the engine-level + non-content-load half rides the tier `size` probes' own bodies + (ADR-008 §1) — noted in the row doc; the mismatch-based proof + needs a plant seam the factory contract deliberately lacks. +- **The kv-only row is threshold-agnostic by discovery**: a + known-length rejection reads nothing, so `Unreadable` probes + distinguish rejection from acceptance and binary search finds the + factory's rejection point (~64 free probes); the row then asserts + immediate/mid-stream rejections and both at-threshold-lands legs. + Configuration contract recorded in the README: threshold < u64::MAX + and materializable (≤ 8 MiB, `MAX_BOUNDARY_MATERIALIZATION`). + Any column's kv-only factory mirrors the row unchanged. +- **Mem factories** (`mem_column.rs`): `MemOnlyFactory` (the + reference host for the INV rows) and `KvOnlyMemFactory` at a + column-declared threshold (1024) — fresh isolated store per open by + construction, teardown a no-op (mem holds no external state), + trivially idempotent and panicking-row-safe. Structural wiring + tests (not backlog rows) demonstrate factory isolation and + teardown-against-panic: `factory_open_vends_fresh_isolated_stores`, + `factory_teardown_is_safe_against_a_panicking_row`. +- **Row lifecycle convention recorded**: the row body owns + open/assert/teardown; a failing assertion panics before teardown + (README §The factory contract grew the bullet). The column runs + under feature `mem`; rows.rs compiles ungated so the reference rows + stay type-checked in default builds. No SKIP machinery added. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion (2026-10-11). + +Landed the **mem column** — the conformance suite's contract-reference +column (ADR-011 §2). `tests/conformance/rows.rs` carries the reference +row set: INV-1…6 families (12 rows total, incl. all three sweep-abort +shapes and the sweep-vs-put race) plus the two constructor-mode rows +(mem-only posture; kv-only put-time rejection), each ADR-stamped and +traceable to the backlog. `tests/conformance/mem_column.rs` binds the +factories (`mem-only`, `kv-only(mem)`) and runs every row; the factory +contract's isolation and idempotent-teardown guarantees are +demonstrated structurally, including the panicking-row path. README +records the reference row set (INV-7's basis — what other columns +mirror), the row-lifecycle convention, and the new current state. The +wave-2 factory binding (concrete `Store` facade) was already made and +recorded upstream of this task; verified, nothing further to correct. + +Verification: `cargo test --features mem` green (97 lib + 18 +conformance + 4 git-interop), `cargo test` (default) and +`cargo test --all-features` green, `cargo clippy --all-features +--all-targets -- -D warnings` green, `cargo fmt --check` green. \ No newline at end of file diff --git a/tests/conformance/README.md b/tests/conformance/README.md index 02ac8d2..afc2b8e 100644 --- a/tests/conformance/README.md +++ b/tests/conformance/README.md @@ -105,6 +105,15 @@ row with its own owner stamp, still one per property. retired with it, replaced by the mem-gated binding demo factories in `main.rs`). A new engine column binds by implementing the trait; there is no longer a vendible type to choose. +- **The row body owns its lifecycle.** A row calls `open()` first — + a fresh, isolated store per call — drives its assertions, and calls + `teardown()` before returning; a failing assertion panics the row + *before* teardown runs (the early exit above), which is why + `teardown()` must make sense over the abandoned instance — dropped + or still alive at unwind. The mem column's structural tests + (`factory_teardown_is_safe_against_a_panicking_row`) exercise that + path; isolation itself (fresh state per open, nothing shared) is + exercised alongside (`factory_open_vends_fresh_isolated_stores`). ## The SKIP posture @@ -139,6 +148,43 @@ additions) needs a backlog entry *and* its stamp first. | INV-6 | `stat`/`size` returns entry length without fetching content | one stat-discipline family per engine column | store-api.md §Invariants 6; ADR-008 (`size` probe); ADR-009 (pg-lo companion-row probe) | wave 2 — mem | | INV-7 | Engine-conformance uniformity: every engine's bodies pass the same contract suite mem passes | the meta-gate — a column's green run *is* the row; no separate test | store-api.md §Invariants 7; ADR-010 §1; ADR-011 §2 (mem's reference role) | wave 2 — mem defines it | +## The reference row set (INV-7's basis) + +The mem column's rows (rows.rs) are the **reference row set** — the +required set every other engine's rows mirror (ADR-011 §2, mem's +contract-reference role; store-api.md §Invariants 7). A column mirrors +this inventory row-for-row against its own factory; an addition to the +set propagates to all columns (INV-7). The INV-row families are the +engine mirror targets — the round-trip family covers the tier the +column's engine serves (mem: the kv tier; local: the large tier). The +two constructor-mode rows mirror wherever that configuration is +constructible: a column with a kv-only factory runs the kv-only row +against it — the row discovers the mode's threshold by probing, since +known-length rejections read nothing, and requires the threshold +materializable for its at-threshold legs (≤ 8 MiB, +`MAX_BOUNDARY_MATERIALIZATION`). + +| Row (`rows.rs`) | Traces to | +|---|---| +| `round_trip_puts_and_gets_are_byte_identical` | INV-1 (content matrix, both put paths, in-body) | +| `dedup_identical_content_is_one_pool_entry` | INV-2 (same-path and cross-path legs) | +| `stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success` | INV-3 (both paths, both failure modes) | +| `sweep_counts_are_exact_across_the_full_protection_set` | INV-4 (happy shape: full protection set) | +| `sweep_without_liveness_sources_aborts_and_deletes_nothing` | INV-4 (abort shape: `NoLivenessSources`) | +| `sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing` | INV-4 (abort shape: `ProtectFailed`) | +| `sweep_is_single_flight_the_contended_second_aborts_as_data` | INV-4 (abort shape: `SweeperLock` lease) | +| `sweep_racing_a_pinned_put_never_deletes_the_protected_entry` | INV-4 (the sweep-vs-put race) | +| `list_correctness_is_observable_through_sweep_outcomes` | INV-5 (exact counts, never list comparison) | +| `stat_reports_len_and_algorithm_without_fetching_content` | INV-6 (the stat-discipline family) | +| `mem_only_mode_accepts_every_length_on_both_put_paths` | constructor mode (ADR-011 §4) | +| `kv_only_mode_rejects_over_threshold_puts_at_put_time` | constructor mode (ADR-012 §6.2) | + +Stamps live on the rows themselves — `grep -rn "Contract stamp:" +tests/conformance/` is the inventory. The INV-4 rows beyond the +families' first shapes (the `SweeperLock` abort) complete the three +`GcAbortCause` variants within the single-node exact-count family +(ADR-013 §2's lease, its in-process ADR-012 §5 form). + ## Column plan **mem is the contract-reference engine** (ADR-011 §2): wave 2 lands @@ -151,9 +197,15 @@ and first run in wave 6. ## Current state -The harness skeleton is compile-able with **zero rows**: the trait -shape, the stamp convention, and the backlog registry are in place; -the wave-2 factory binding is **made** — `open()` vends the concrete -`alkblobs::Store` facade (see §The factory contract), demonstrated by -the mem-gated binding demo factories. The rows arrive with wave 2's -mem column. \ No newline at end of file +The harness is green on the **mem column** (wave 2): the factory +binding is the concrete `alkblobs::Store` facade (see §The factory +contract — the binding demonstrator remains in `binding.rs`), the +reference row set is landed (§The reference row set — `rows.rs`, +every row ADR-stamped and traced to the backlog), the column's +factories (`mem-only`, the reference; `kv-only(mem)` at a +column-declared threshold) vend fresh isolated stores with idempotent +teardown (`mem_column.rs`, the structural factory-contract tests +included), and mem never skips. The column runs under feature `mem` +(`cargo test --features mem`); the row set compiles ungated so the +reference rows stay type-checked in default builds. The local column +grows next. \ No newline at end of file diff --git a/tests/conformance/main.rs b/tests/conformance/main.rs index c6034b0..4a3fb0c 100644 --- a/tests/conformance/main.rs +++ b/tests/conformance/main.rs @@ -13,11 +13,16 @@ //! `factory.rs` and the README. The wave-1 mock probe retired with it; //! the binding demonstrator ([`binding`], mem-gated — mem is the //! skip-free reference shape) proves the contract implementable -//! against the real type. Rows arrive with wave 2's columns (mem -//! first, the contract-reference engine). +//! against the real type. //! -//! Items beyond the tests below are the wave-2 drivers' surface, hence -//! the scoped `#![allow(dead_code)]` — it shrinks as rows land. +//! The rows landed with wave 2's mem column: [`rows`] holds the +//! reference row set (INV-1…6 plus the two constructor-mode rows, +//! ADR-stamped, traced to [`backlog`]) and [`mem_column`] is the +//! contract-reference column ([`README`](./README.md) §The reference +//! row set is the recorded inventory other columns mirror). +//! +//! Items beyond the tests below are the later columns' surface, hence +//! the scoped `#![allow(dead_code)]` — it shrinks as columns land. #![allow(dead_code)] @@ -25,6 +30,9 @@ mod backlog; #[cfg(feature = "mem")] mod binding; mod factory; +#[cfg(feature = "mem")] +mod mem_column; +mod rows; mod stamp; #[test] diff --git a/tests/conformance/mem_column.rs b/tests/conformance/mem_column.rs new file mode 100644 index 0000000..cf0f891 --- /dev/null +++ b/tests/conformance/mem_column.rs @@ -0,0 +1,240 @@ +//! The **mem column** — the contract-reference column (ADR-011 §2): +//! the mem engine's column of the reference row set ([`crate::rows`]), +//! the first column bound to the real facade, and INV-7's basis — +//! every other engine's rows mirror this set. +//! +//! The factories (README §The factory contract): `open()` vends the +//! concrete [`alkblobs::Store`] facade over a fresh, isolated backing +//! store per call — the mem engine is `BTreeMap`-shaped and shares +//! nothing across instances, so isolation is by construction. Teardown +//! is a no-op (the mem engine holds no external state), which makes it +//! trivially idempotent and safe against abandoned instances — the +//! panicking-row path included, which the structural tests below +//! exercise. Mem never skips (README §SKIP posture — the SKIP +//! machinery stays unneeded until the SQL waves). +//! +//! The INV rows run against the reference `mem-only` factory (the kv +//! tier, no dispatch threshold in effect); the two constructor-mode +//! rows run against the mode factories they name — `mem-only`, and +//! `kv-only(mem)` at the column-declared threshold, which the kv-only +//! row discovers by probing. + +#![allow(dead_code)] + +use std::io::Cursor; +use std::sync::Arc; + +use crate::factory::StoreFactory; +use crate::rows; +use alkblobs::{Error, Key, Store}; + +/// The reference factory: the mem-only mode (ADR-011 §4) — mem as the +/// sole engine, no dispatch threshold in effect. The column never +/// skips. +pub struct MemOnlyFactory; + +impl StoreFactory for MemOnlyFactory { + fn name(&self) -> &'static str { + "mem-only" + } + + async fn open(&self) -> Result { + Ok(Store::mem_only()) + } + + async fn teardown(&self) -> Result<(), Error> { + Ok(()) + } +} + +/// The kv-only(mem) factory (ADR-011 §4; ADR-012 §6.2): no large +/// tier; the threshold is the put-time rejection point the kv-only +/// row discovers by probing. +pub struct KvOnlyMemFactory { + /// The rejection point this column's configuration declares. The + /// kv-only row discovers it behaviorally and requires it + /// materializable for its boundary legs. + pub threshold: u64, +} + +impl StoreFactory for KvOnlyMemFactory { + fn name(&self) -> &'static str { + "kv-only(mem)" + } + + async fn open(&self) -> Result { + Ok(Store::kv_only_mem(self.threshold)) + } + + async fn teardown(&self) -> Result<(), Error> { + Ok(()) + } +} + +/// The column's declared kv-only threshold — small, so every leg of +/// the rejection row stays trivially materializable. +const KV_ONLY_THRESHOLD: u64 = 1024; + +const MEM: MemOnlyFactory = MemOnlyFactory; +const KV_ONLY: KvOnlyMemFactory = KvOnlyMemFactory { + threshold: KV_ONLY_THRESHOLD, +}; + +// ------------------------------------------------------------------ +// The reference row set, run against the mem column's factories. +// ------------------------------------------------------------------ + +#[tokio::test] +async fn inv1_round_trip_puts_and_gets_are_byte_identical() { + rows::round_trip_puts_and_gets_are_byte_identical(&MEM).await; +} + +#[tokio::test] +async fn inv2_dedup_identical_content_is_one_pool_entry() { + rows::dedup_identical_content_is_one_pool_entry(&MEM).await; +} + +#[tokio::test] +async fn inv3_stage_hygiene_zero_residue_on_failure_one_entry_on_success() { + rows::stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(&MEM).await; +} + +#[tokio::test] +async fn inv4_sweep_counts_are_exact_across_the_full_protection_set() { + rows::sweep_counts_are_exact_across_the_full_protection_set(&MEM).await; +} + +#[tokio::test] +async fn inv4_sweep_without_liveness_sources_aborts_and_deletes_nothing() { + rows::sweep_without_liveness_sources_aborts_and_deletes_nothing(&MEM).await; +} + +#[tokio::test] +async fn inv4_sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing() { + rows::sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing(&MEM).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn inv4_sweep_is_single_flight_the_contended_second_aborts_as_data() { + rows::sweep_is_single_flight_the_contended_second_aborts_as_data(&MEM).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn inv4_sweep_racing_a_pinned_put_never_deletes_the_protected_entry() { + rows::sweep_racing_a_pinned_put_never_deletes_the_protected_entry(&MEM).await; +} + +#[tokio::test] +async fn inv5_list_correctness_is_observable_through_sweep_outcomes() { + rows::list_correctness_is_observable_through_sweep_outcomes(&MEM).await; +} + +#[tokio::test] +async fn inv6_stat_reports_len_and_algorithm_without_fetching_content() { + rows::stat_reports_len_and_algorithm_without_fetching_content(&MEM).await; +} + +#[tokio::test] +async fn constructor_mem_only_accepts_every_length_on_both_put_paths() { + rows::mem_only_mode_accepts_every_length_on_both_put_paths(&MEM).await; +} + +#[tokio::test] +async fn constructor_kv_only_rejects_over_threshold_puts_at_put_time() { + rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&KV_ONLY).await; +} + +// ------------------------------------------------------------------ +// The factory contract's structural shapes (binding.rs' demonstrations +// carried to the real column factories): fresh isolated opens and +// idempotent teardown safe against a panicking row. These are not +// conformance rows — no backlog entry — they are the factory +// contract's own wiring-level assertions. +// ------------------------------------------------------------------ + +/// Every `open()` vends a fresh, isolated store: two stores live +/// concurrently without observable shared state, and a third open sees +/// neither island's entries. +#[tokio::test] +async fn factory_open_vends_fresh_isolated_stores() { + let a = MEM.open().await.expect("open vends a fresh store"); + let b = MEM.open().await.expect("open vends another fresh store"); + + let content_a = b"island a's content".to_vec(); + let content_b = b"island b's content".to_vec(); + let (key_a, _) = a + .put( + Some(content_a.len() as u64), + Cursor::new(content_a.as_slice()), + ) + .await + .expect("island a's put publishes"); + let (key_b, _) = b + .put(None, Cursor::new(content_b.as_slice())) + .await + .expect("island b's put publishes"); + + assert!(a.has(&key_a.as_bytes()).await.unwrap()); + assert!( + !a.has(&key_b.as_bytes()).await.unwrap(), + "island a sees nothing of island b" + ); + assert!(b.has(&key_b.as_bytes()).await.unwrap()); + assert!( + !b.has(&key_a.as_bytes()).await.unwrap(), + "island b sees nothing of island a" + ); + + let c = MEM.open().await.expect("a third open vends fresh"); + assert!( + !c.has(&key_a.as_bytes()).await.unwrap(), + "a third open shares nothing with the first" + ); + assert!( + !c.has(&key_b.as_bytes()).await.unwrap(), + "a third open shares nothing with the second" + ); +} + +/// A failing assertion panics the row before teardown runs — and +/// teardown must make sense over the abandoned instance anyway: it is +/// idempotent against the panicked row's path, and the next open vends +/// a fresh store with no residue (README §The factory contract). +#[tokio::test] +async fn factory_teardown_is_safe_against_a_panicking_row() { + let content = b"put before the assertion fails".to_vec(); + let abandoned = Key::derive(&content, content.len() as u64); + let factory = Arc::new(MemOnlyFactory); + + let task = { + let factory = Arc::clone(&factory); + tokio::spawn(async move { + let store = factory.open().await.expect("open vends the store"); + let (_, pin) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the put publishes"); + drop(pin); + panic!("a failing assertion panics the row (the factory contract's early exit)"); + }) + }; + assert!( + task.await.is_err(), + "the row's panic surfaced as its failure" + ); + + factory + .teardown() + .await + .expect("teardown makes sense over the abandoned instance"); + factory.teardown().await.expect("teardown is idempotent"); + + let store = factory + .open() + .await + .expect("the next open still vends a store"); + assert!( + !store.has(&abandoned.as_bytes()).await.unwrap(), + "no residue from the panicked row — isolation is factory-owned" + ); +} diff --git a/tests/conformance/rows.rs b/tests/conformance/rows.rs new file mode 100644 index 0000000..2b541cd --- /dev/null +++ b/tests/conformance/rows.rs @@ -0,0 +1,1221 @@ +//! The suite's **reference row set** — the INV-1…6 invariant families +//! and the two constructor-mode rows, landed by the mem column as the +//! required set every other engine's rows mirror (ADR-011 §2's +//! contract-reference role; store-api.md §Invariants 7; +//! tests/conformance/README.md §The reference row set is the recorded +//! inventory). +//! +//! Row anatomy (README §Row anatomy): one named, independently-runnable +//! property per row — an async function taking a [`StoreFactory`] +//! (never a store), asserting by panicking on violation (a failing row +//! is a test failure naming the property); the content matrix and both +//! put paths loop in-body. Each row owns its factory lifecycle: +//! `open()` at the top (a fresh, isolated store per call — factory.rs' +//! contract), the assertions, then `teardown()` before returning; a +//! failing assertion panics first and skips teardown, which is exactly +//! why `teardown()` is contract-required to be idempotent and safe +//! against an abandoned instance (the mem column's structural tests +//! exercise that path). +//! +//! The rows are engine-uniform by construction — none branches on the +//! factory or any engine. The one deliberate exception is the two +//! constructor-mode rows' **precondition** (`store.mode()`): a +//! mode-scoped property runs against the configuration it names, and +//! `mode()` is the facade contract's own introspection (ADR-011 §4), +//! not an engine branch. +//! +//! Every row is ADR-stamped ([`crate::stamp`] is the convention's +//! anchor; `grep -rn "Contract stamp:" tests/conformance/` is the +//! inventory) and traces to a backlog entry ([`crate::backlog`]). + +#![allow(dead_code)] + +use std::collections::BTreeSet; +use std::io::Cursor; +use std::pin::Pin as StdPin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; + +use parking_lot::Mutex; +use tokio::io::{AsyncRead, ReadBuf}; + +use crate::factory::StoreFactory; +use alkblobs::{ + Algorithm, EntryMeta, Error, GcAbortCause, Key, LivenessSource, ProtectCallback, Store, + StoreMode, DEFAULT_DISPATCH_THRESHOLD, +}; + +/// The shared fixtures the rows run on — set-backed liveness sources, +/// a set-backed protect callback, and the stream fixtures for the +/// failure legs. None of them knows an engine. +/// +/// A set-backed liveness source — the live-shared seam's test shape: +/// the store consults the same set the row mutates, so answers are +/// live-shared, never registration snapshots. +#[derive(Default)] +struct SetSource { + live: Mutex>, +} + +impl SetSource { + fn register(store: &Store) -> Arc { + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + source + } + + fn add(&self, key: Key) { + self.live.lock().insert(key); + } + + fn remove(&self, key: &Key) { + self.live.lock().remove(key); + } +} + +impl LivenessSource for SetSource { + fn live_keys(&self) -> Vec { + self.live.lock().iter().cloned().collect() + } + + fn contains(&self, key: &Key) -> bool { + self.live.lock().contains(key) + } +} + +/// A set-backed protect callback; `None` models the failed-to-answer +/// case (abort-as-data on the safe side). +#[derive(Default)] +struct SetProtect { + set: Mutex>>, +} + +impl SetProtect { + fn set_keys(&self, keys: Vec) { + *self.set.lock() = Some(keys); + } + + fn fail(&self) { + *self.set.lock() = None; + } +} + +impl ProtectCallback for SetProtect { + fn protect(&self) -> Option> { + self.set.lock().clone() + } +} + +/// A protect callback that blocks mid-consult — the fixture for the +/// sweep-lease contended window: it announces its entry on a channel, +/// waits on a second, then answers with an empty protection set. +struct LatchProtect { + entered: std::sync::mpsc::Sender<()>, + release: Mutex>>, +} + +impl ProtectCallback for LatchProtect { + fn protect(&self) -> Option> { + self.entered.send(()).ok(); + let receiver = self.release.lock().take(); + if let Some(receiver) = receiver { + let _ = receiver.recv(); + } + Some(Vec::new()) + } +} + +/// A stream cut at controlled chunk boundaries (`sizes`, in order; +/// tail data falls out whole) — makes multi-read loops, mid-stream +/// overflow, and staged continuation observable past the put loop's +/// own chunking. +struct Chunked { + data: Vec, + cursor: usize, + sizes: std::collections::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: StdPin<&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 (the failure-discard path's +/// injectable 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: StdPin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let message = "stream failure (conformance 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 before any byte flows. +struct Unreadable; + +impl AsyncRead for Unreadable { + fn poll_read( + self: StdPin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Ready(Err(std::io::Error::other( + "the Unreadable stream must never be read (conformance fixture)", + ))) + } +} + +/// The content matrix the round-trip family loops: the degenerate +/// empty, one byte, small text, and patterned binaries spanning the +/// put loop's multi-read range. Sizes are content-shape choices, not +/// engine or dispatch-internals knowledge. +fn matrix() -> Vec> { + vec![ + Vec::new(), + b"x".to_vec(), + b"the conformance round-trip content".to_vec(), + (0..137u32).map(|i| (i % 251) as u8).collect(), + (0..4096u32).map(|i| (i % 199) as u8).collect(), + (0..65537u32).map(|i| (i % 173) as u8).collect(), + ] +} + +fn patterned(len: usize, modulus: u32) -> Vec { + (0..len as u32).map(|i| (i % modulus) as u8).collect() +} + +/// One leg of the exact-count sweep assertions: a completed report's +/// counts are exact over the staged set (`candidates_staged == +/// deleted + cancelled_by_arbitration`). +fn assert_exact_counts(report: &alkblobs::SweepReport) { + assert_eq!( + report.candidates_staged, + report.deleted + report.cancelled_by_arbitration, + "the completed sweep's counts are exact over the staged set" + ); +} + +/// INV-1: whole-blob put/get round-trips byte-identical under the +/// canonical derivation — the content matrix loops both put paths +/// (known-length, unknown-length) in-body; every put's key is the +/// canonical derivation over the content (the git-interop test +/// remains the derivation's source of truth), and `get` returns the +/// pinned `(len, stream)` shape whole and windowed. +/// +/// Contract stamp: store-api.md §Invariants 1; ADR-002 (canonical +/// derivation; interop with real git is the source of truth); ADR-003 +/// (the two put paths behind one signature). +pub async fn round_trip_puts_and_gets_are_byte_identical(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + + for content in matrix() { + let len = content.len() as u64; + let canonical = Key::derive(&content, len); + for declared in [Some(len), None] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!(key, canonical, "the canonical derivation addresses the put"); + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!(got_len, len, "the tuple carries the entry length"); + assert_eq!(blob.len(), len, "the stream's metadata length agrees"); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "whole-blob round-trip is byte-identical" + ); + if len > 2 { + assert_eq!( + blob.read_at(1, len - 2) + .await + .expect("the inner window reads"), + content[1..(len - 2) as usize + 1], + "a bounded window reads the matching inner slice" + ); + } + drop(pin); + } + } + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-2: identical content put twice — through the same put path and +/// across the two paths — is one pool entry, and the re-put's `Pin` +/// protects exactly as a fresh put's does. One-pool-entry is asserted +/// through sweep outcomes (the INV-5 posture), never by comparing +/// enumerated lists. +/// +/// Contract stamp: store-api.md §Invariants 2; ADR-003 (whole-value +/// put is CAS first-writer-wins); ADR-005 (a re-put refcounts the +/// existing entry under the same arbitration lock). +pub async fn dedup_identical_content_is_one_pool_entry(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let content = b"deduplicated content".to_vec(); + let canonical = Key::derive(&content, content.len() as u64); + + let (key, first) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the fresh put publishes"); + let (reput, second) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the same-path re-put publishes"); + assert_eq!(key, canonical); + assert_eq!(reput, canonical, "one derivation, one address"); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + report.candidates_staged, 0, + "one pool entry, both refcounts composing under the same lock" + ); + drop(first); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the re-put's pin protects exactly as a fresh put's" + ); + drop(second); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "exactly one pool entry all along"); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + + let cross = b"deduplicated across the put paths".to_vec(); + let cross_canonical = Key::derive(&cross, cross.len() as u64); + let (key, first) = store + .put(Some(cross.len() as u64), Cursor::new(cross.as_slice())) + .await + .expect("the known-length put publishes"); + let (reput, second) = store + .put(None, Cursor::new(cross.as_slice())) + .await + .expect("the unknown-length re-put publishes"); + assert_eq!(key, cross_canonical); + assert_eq!(reput, cross_canonical, "one pool entry across the paths"); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 0, "both refcounts compose"); + drop(second); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the re-put's pin protects exactly as a fresh put's (other drop order)" + ); + drop(first); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "exactly one pool entry all along"); + assert!(!store.has(&cross_canonical.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-3: stage hygiene — every put-path failure leaves zero residue +/// and every success leaves exactly one committed entry, over both +/// put paths and both injectable failure modes (a mid-stream stream +/// failure and a declared-length mismatch). The kv tier's stage form +/// is degenerate (nothing stages outside a flush), so residue is the +/// pool itself, observed through exact-count sweep outcomes. +/// +/// Contract stamp: store-api.md §Invariants 3; ADR-010 §1 (the +/// staged-put contract: commit publishes, every failure path +/// discards); ADR-003 (stage-then-commit, the kv form degenerate). +pub async fn stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success( + factory: &F, +) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let content = patterned(200, 167); + let canonical = Key::derive(&content, content.len() as u64); + + let err = store + .put(Some(content.len() as u64), Failing::new(&content, 40)) + .await + .expect_err("the failed mid-stream put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + let err = store + .put(None, Failing::new(&content, 40)) + .await + .expect_err("the failed unknown-length put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + let err = store + .put(Some(100), Cursor::new(content.as_slice())) + .await + .expect_err("the over-fed mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + let err = store + .put(Some(400), Cursor::new(content.as_slice())) + .await + .expect_err("the under-fed mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "zero residue after every failure, both put paths, both failure modes" + ); + + let (key, pin) = store + .put(Some(content.len() as u64), Cursor::new(content.as_slice())) + .await + .expect("the known-length put publishes"); + assert_eq!(key, canonical); + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "exactly one committed entry on success" + ); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + + let (key, pin) = store + .put(None, Cursor::new(content.as_slice())) + .await + .expect("the unknown-length put publishes"); + assert_eq!(key, canonical); + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "exactly one committed entry on the other put path too" + ); + assert!(!store.has(&canonical.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, happy shape: with liveness correctly registered on a mixed +/// pool — a source-live entry, a pinned entry, a protect-callback- +/// named entry, and two dead entries (one per put path) — the sweep's +/// counts are exact and only the dead entries delete. The protected +/// entries stay byte-identical. +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-005 (three liveness +/// marks; delete windows with arbitrate-at-delete-time); ADR-012 §1 +/// (sweeps are never errors — the report is the outcome record). +pub async fn sweep_counts_are_exact_across_the_full_protection_set(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + let source = SetSource::register(&store); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let source_live = b"kept by the registered source".to_vec(); + let (key_live, _) = store + .put( + Some(source_live.len() as u64), + Cursor::new(source_live.as_slice()), + ) + .await + .expect("the source-live entry publishes"); + source.add(key_live.clone()); + + let pinned = b"kept by the pin".to_vec(); + let (key_pinned, pin_pinned) = store + .put(None, Cursor::new(pinned.as_slice())) + .await + .expect("the pinned entry publishes"); + + let protected = b"named by the protect callback".to_vec(); + let (key_protected, _) = store + .put( + Some(protected.len() as u64), + Cursor::new(protected.as_slice()), + ) + .await + .expect("the protect-named entry publishes"); + protect.set_keys(vec![key_protected.clone()]); + + let dead_one = b"a dead witness from the known-length path".to_vec(); + let (_, pin) = store + .put( + Some(dead_one.len() as u64), + Cursor::new(dead_one.as_slice()), + ) + .await + .expect("the first dead entry publishes"); + drop(pin); + let dead_two = b"a dead witness from the unknown-length path".to_vec(); + let (_, pin) = store + .put(None, Cursor::new(dead_two.as_slice())) + .await + .expect("the second dead entry publishes"); + drop(pin); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None, "correct liveness completes the sweep"); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (2, 2, 0), + "only the two dead entries stage and delete" + ); + assert_exact_counts(&report); + + assert!(store.has(&key_live.as_bytes()).await.unwrap()); + assert!(store.has(&key_pinned.as_bytes()).await.unwrap()); + assert!(store.has(&key_protected.as_bytes()).await.unwrap()); + for (key, content) in [ + (&key_live, &source_live), + (&key_pinned, &pinned), + (&key_protected, &protected), + ] { + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the protected entry is present"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content.as_slice(), + "the protected entry is byte-identical after the sweep" + ); + } + + drop(pin_pinned); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "with the pin dropped exactly the formerly pinned entry dies" + ); + assert!(!store.has(&key_pinned.as_bytes()).await.unwrap()); + assert!(store.has(&key_live.as_bytes()).await.unwrap()); + assert!(store.has(&key_protected.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, abort shape: the safe default — a sweep invoked with zero +/// registered liveness sources aborts with `NoLivenessSources` in a +/// completed `Ok(SweepReport)` and deletes nothing. +/// +/// Contract stamp: store-api.md §Invariants 4; store-api.md §Lifecycle +/// (the safe default); ADR-012 §1 (every abort is report data — never +/// an error, nothing deleted). +pub async fn sweep_without_liveness_sources_aborts_and_deletes_nothing( + factory: &F, +) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + let content = b"unprotected while no source is registered".to_vec(); + let (key, pin) = store + .put(None, Cursor::new(content.as_slice())) + .await + .expect("the entry publishes"); + drop(pin); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.aborted, + Some(GcAbortCause::NoLivenessSources), + "the safe default aborts as report data" + ); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "an aborted sweep stages and deletes nothing" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "the entry survives the abort" + ); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, abort shape: a protection source that fails to answer +/// aborts the sweep with `ProtectFailed` in a completed +/// `Ok(SweepReport)` — the flaky source skips the sweep rather than +/// risk deletion, and nothing is deleted. +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-005 §Marks 3 (a +/// failed protection source takes the safe side); ADR-012 §1 (abort +/// is report data, never an error). +pub async fn sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing( + factory: &F, +) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + protect.fail(); + + let content = b"dead while the protection source fails to answer".to_vec(); + let (key, pin) = store + .put(None, Cursor::new(content.as_slice())) + .await + .expect("the entry publishes"); + drop(pin); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.aborted, + Some(GcAbortCause::ProtectFailed), + "the failed protection source aborts as report data" + ); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "an aborted sweep stages and deletes nothing" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "the entry survives the abort" + ); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, abort shape: the single-node sweep is single-flight — the +/// in-process analog of the sweep lease (ADR-013 §2). A second +/// sweeper contending behind a lease holder aborts with +/// `SweeperLock` in a completed `Ok(SweepReport)`, deleting nothing; +/// the holder's own sweep completes with exact counts. +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-012 §1 (abort is +/// report data); ADR-013 §2 (the named exclusive sweep lease — one +/// sweeper per pool, holdership ends at release); ADR-012 §5 (the +/// in-process single-flighter is the non-SQL form of that lease). +pub async fn sweep_is_single_flight_the_contended_second_aborts_as_data( + factory: &F, +) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + store.register_protect_callback(Arc::new(LatchProtect { + entered: entered_tx, + release: Mutex::new(Some(release_rx)), + })); + + let content = b"dead under the contended sweeps".to_vec(); + let (key, pin) = store + .put(None, Cursor::new(content.as_slice())) + .await + .expect("the entry publishes"); + drop(pin); + + // The lease holder sweeps on its own task (its protect consult + // blocks that task's thread mid-callback); this task contends and + // then releases the holder. + let store = Arc::new(store); + let holder = tokio::spawn({ + let store = Arc::clone(&store); + async move { store.sweep().await } + }); + + while entered_rx.recv_timeout(Duration::from_millis(50)).is_err() { + tokio::task::yield_now().await; + } + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + second.aborted, + Some(GcAbortCause::SweeperLock), + "the contended sweeper aborts as report data" + ); + assert_eq!( + ( + second.candidates_staged, + second.deleted, + second.cancelled_by_arbitration + ), + (0, 0, 0), + "the contended sweep stages and deletes nothing" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "the contended sweep deleted nothing" + ); + release_tx.send(()).ok(); + + let report = holder + .await + .expect("the holder task joined") + .expect("sweeps are never errors"); + assert_eq!(report.aborted, None, "the lease holder completes"); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (1, 1, 0), + "the lease holder's counts are exact" + ); + assert_exact_counts(&report); + assert!( + !store.has(&key.as_bytes()).await.unwrap(), + "the holder's deletion landed after the release" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4, the race: a sweep racing an in-flight pinned put never +/// deletes the protected entry — the pin is minted before the entry +/// can appear in any `list()`, so every schedule keeps the counts +/// exact and the published entry whole and byte-identical; the +/// surviving pin keeps protecting, and dropping it re-exposes the +/// entry to exactly one deletion. Both put paths race in-body. +/// +/// Contract stamp: store-api.md §Invariants 4; ADR-005 (pin-before- +/// publish ordering; the delete window closes the sweep-vs-put race). +pub async fn sweep_racing_a_pinned_put_never_deletes_the_protected_entry( + factory: &F, +) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + SetSource::register(&store); + + let witness = b"the dead witness the first racing sweep deletes".to_vec(); + let (_, pin) = store + .put(None, Cursor::new(witness.as_slice())) + .await + .expect("the witness publishes"); + drop(pin); + + let rounds: [(Vec, bool); 4] = [ + (b"racing round a, known path".to_vec(), true), + (b"racing round b, unknown path".to_vec(), false), + (patterned(4096, 191), true), + (patterned(4096, 173), false), + ]; + for (round, (content, known)) in rounds.iter().enumerate() { + let declared = if *known { + Some(content.len() as u64) + } else { + None + }; + let (report, put) = tokio::join!(store.sweep(), async { + store.put(declared, Cursor::new(content.as_slice())).await + }); + let report = report.expect("sweeps are never errors"); + let (key, pin) = put.expect("the racing put publishes"); + assert_eq!( + key, + Key::derive(content, content.len() as u64), + "round {round}: the canonical derivation addresses the racing put" + ); + assert_eq!(report.aborted, None, "round {round}: the sweep completed"); + assert_exact_counts(&report); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the racing put's entry survived"); + assert_eq!(len, content.len() as u64, "round {round}"); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content.as_slice(), + "round {round}: the published entry is byte-identical" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "round {round}: the minted pin still protects" + ); + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "round {round}: unpinned, exactly one deletion" + ); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-5: `list()` correctness is observable only through GC — the +/// enumeration is complete and well-formed in both directions, proven +/// by exact sweep outcomes (neither more nor fewer candidates staged), +/// never by comparing enumerated lists. +/// +/// Contract stamp: store-api.md §Invariants 5; ADR-003 (`list()` is +/// complete and well-formed by contract — load-bearing for GC); +/// ADR-005 (the sweep's live-set diff rides `list()`). +pub async fn list_correctness_is_observable_through_sweep_outcomes(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + let source = SetSource::register(&store); + + let mut keys = Vec::new(); + for (index, content) in [ + b"kept live by the source, entry one".to_vec(), + b"kept live by the source, entry two".to_vec(), + b"dead at the first sweep, entry three".to_vec(), + b"dead at the first sweep, entry four".to_vec(), + b"dead at the first sweep, entry five".to_vec(), + ] + .into_iter() + .enumerate() + { + let declared = if index % 2 == 0 { + Some(content.len() as u64) + } else { + None + }; + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("the entry publishes"); + drop(pin); + keys.push(key); + } + source.add(keys[0].clone()); + source.add(keys[1].clone()); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (3, 3, 0), + "the enumeration staged exactly the three dead entries — complete and \ + well-formed in both directions" + ); + assert_exact_counts(&report); + assert!(store.has(&keys[0].as_bytes()).await.unwrap()); + assert!(store.has(&keys[1].as_bytes()).await.unwrap()); + for key in &keys[2..] { + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + source.remove(&keys[1]); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "the newly dead entry is found exactly" + ); + assert!(store.has(&keys[0].as_bytes()).await.unwrap()); + + source.remove(&keys[0]); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "the last entry is found exactly" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-6: `stat` returns the entry's length and hash algorithm +/// without fetching content — the cheap probe riding the trait's +/// `size` seam — over the content matrix on both put paths, pure +/// across repeated calls, and `None` on a miss. +/// +/// The non-fetching discipline's engine half (the size probe is a +/// metadata probe, never a value load) is asserted per engine by the +/// tier probes it rides; the facade-observable slice is the shape +/// this row owns. +/// +/// Contract stamp: store-api.md §Invariants 6; ADR-008 §1 (the trait +/// `size` probe — stat is metadata, never a content load). +pub async fn stat_reports_len_and_algorithm_without_fetching_content(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + + for content in matrix() { + let len = content.len() as u64; + let canonical = Key::derive(&content, len); + for declared in [Some(len), None] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + drop(pin); + assert_eq!(key, canonical); + let meta = store + .stat(&key.as_bytes()) + .await + .expect("stat succeeds") + .expect("the entry is present"); + assert_eq!( + meta, + EntryMeta { + len, + algorithm: Algorithm::GitBlobSha256, + }, + "stat carries the entry length and the key's algorithm" + ); + assert_eq!( + store + .stat(&key.as_bytes()) + .await + .expect("stat succeeds again"), + Some(meta), + "stat is a pure probe" + ); + } + } + + let absent = b"stat never sees this content".to_vec(); + let absent = Key::derive(&absent, absent.len() as u64); + assert_eq!( + store.stat(&absent.as_bytes()).await.expect("stat succeeds"), + None, + "a both-tier miss is None, never an error" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// The kv-only rejection carrier's message marker (ADR-012 §6.2's +/// put-time rejection rides `Error::Io(String)` with a descriptive +/// message — the pinned family holds no dedicated variant). +const KV_ONLY_REJECTION: &str = "kv-only mode rejects an over-threshold put"; + +/// The largest threshold the boundary legs of +/// [`kv_only_mode_rejects_over_threshold_puts_at_put_time`] +/// materialize (the at-threshold legs stream exactly `threshold` +/// bytes and the kv tier holds the whole value) — the configuration +/// contract a column's kv-only factory keeps. +const MAX_BOUNDARY_MATERIALIZATION: u64 = 8 * 1024 * 1024; + +/// One rejection point probe: a known-length put of `len` bytes whose +/// stream nobody may read. Over the threshold the put rejects before +/// the stream flows (`true`, the mode's message); at-or-below it the +/// put path reads the stream and fails there instead (`false`). Either +/// way nothing publishes and zero residue lands, so probing is free. +async fn probe_rejects(store: &Store, len: u64) -> bool { + match store.put(Some(len), Unreadable).await { + Err(err) => err.to_string().contains(KV_ONLY_REJECTION), + Ok(_) => panic!("the Unreadable probe stream can never publish"), + } +} + +/// The constructor mode row, kv-only: over-threshold puts reject at +/// put time (ADR-012 §6.2) — a known-length put rejects immediately +/// (its stream is never read), an unknown-length put rejects at +/// mid-stream threshold overflow — with zero residue on either +/// rejection, while at-threshold puts land on both paths. The row +/// discovers the configuration's rejection point by probing +/// (rejections read nothing, so probing is free) and requires it +/// materializable (see [`MAX_BOUNDARY_MATERIALIZATION`]). +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row of +/// the reference set); ADR-011 §4 (kv-only mode); ADR-012 §6.2 +/// (known-length immediate, unknown-length mid-stream); ADR-012 §6.5 +/// (the `Io(String)` rejection carrier). +pub async fn kv_only_mode_rejects_over_threshold_puts_at_put_time(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + assert_eq!( + store.mode(), + StoreMode::KvOnly, + "the row runs against a kv-only factory (ADR-011 §4's mode vocabulary)" + ); + SetSource::register(&store); + + // The rejection point: a known-length put over the threshold + // rejects before its stream is read, so an Unreadable probe + // stream distinguishes rejection (never read) from acceptance + // (the probe stream errors instead). Binary-search the smallest + // rejecting length. + assert!( + probe_rejects(&store, u64::MAX).await, + "the kv-only factory must declare a threshold below u64::MAX for this \ + row to apply" + ); + let mut low = 0u64; + let mut high = u64::MAX; + while high - low > 1 { + let mid = low + (high - low) / 2; + if probe_rejects(&store, mid).await { + high = mid; + } else { + low = mid; + } + } + let rejecting = high; + let threshold = rejecting - 1; + assert!( + threshold <= MAX_BOUNDARY_MATERIALIZATION, + "the kv-only factory's threshold must keep the boundary legs \ + materializable (≤ {MAX_BOUNDARY_MATERIALIZATION}); got {threshold}" + ); + + // Known-length, over-threshold: immediate — the stream never + // flows, and zero residue lands. + let err = store + .put(Some(u64::MAX), Unreadable) + .await + .expect_err("the over-threshold known-length put rejects"); + assert!( + err.to_string().contains(KV_ONLY_REJECTION), + "the immediate rejection carries the mode's message: {err}" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "zero residue after the immediate rejection" + ); + + // Unknown-length, over-threshold: rejects at mid-stream threshold + // overflow — the mode's rejection, not a stream error — and zero + // residue. + let content: Vec = (0..threshold) + .map(|i| (i % 199) as u8) + .chain(std::iter::once(7)) + .collect(); + assert_eq!(content.len() as u64, rejecting); + let err = store + .put(None, Chunked::new(&content, &[7; 64])) + .await + .expect_err("the over-threshold unknown-length put rejects at mid-stream"); + assert!( + err.to_string().contains(KV_ONLY_REJECTION) && err.to_string().contains("streamed"), + "the overflow rejection is the mode's, mid-stream: {err}" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + ( + report.candidates_staged, + report.deleted, + report.cancelled_by_arbitration + ), + (0, 0, 0), + "zero residue after the mid-stream rejection" + ); + + // Exactly-at-threshold lands on both put paths (≤ threshold routes + // kv; the rejection is strictly past it). + let at = (0..threshold).map(|i| (i % 251) as u8).collect::>(); + let at_canonical = Key::derive(&at, threshold); + let (key, pin) = store + .put(Some(threshold), Cursor::new(at.as_slice())) + .await + .expect("the at-threshold known-length put lands"); + assert_eq!(key, at_canonical); + drop(pin); + assert!(store.has(&at_canonical.as_bytes()).await.unwrap()); + + let other: Vec = (0..threshold).map(|i| (i % 173) as u8).collect(); + let other_canonical = Key::derive(&other, threshold); + let (key, pin) = store + .put(None, Chunked::new(&other, &[11; 512])) + .await + .expect("the at-threshold unknown-length put lands"); + assert_eq!(key, other_canonical); + drop(pin); + assert!(store.has(&other_canonical.as_bytes()).await.unwrap()); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// The constructor mode row, mem-only: the ephemeral mode reports its +/// mode (ADR-011 §4's vocabulary) and carries **no dispatch threshold +/// in effect** — every length is accepted on both put paths, +/// including content well past the shipped default threshold, and +/// round-trips byte-identical from the sole engine. +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row of +/// the reference set); ADR-011 §4 (mem-only mode — no dispatch +/// threshold in effect, never production). +pub async fn mem_only_mode_accepts_every_length_on_both_put_paths(factory: &F) { + let store = factory + .open() + .await + .expect("open vends a fresh, isolated store"); + assert_eq!( + store.mode(), + StoreMode::MemOnly, + "the row runs against a mem-only factory (ADR-011 §4's mode vocabulary)" + ); + + let content = patterned(DEFAULT_DISPATCH_THRESHOLD as usize + 7, 191); + let canonical = Key::derive(&content, content.len() as u64); + for declared in [Some(content.len() as u64), None] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("mem-only accepts the over-default-threshold put"); + assert_eq!(key, canonical, "one canonical entry across both paths"); + drop(pin); + } + + let (len, blob) = store + .get(&canonical.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!(len, content.len() as u64); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content, + "the sole engine holds the whole artifact byte-identical" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +}