From 45c456989f2fd8ecd23bff2edd2ceff536ed237e Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 23:30:31 +0000 Subject: [PATCH] =?UTF-8?q?feat(tests):=20the=20conformance=20suite's=20lo?= =?UTF-8?q?cal=20column=20=E2=80=94=20the=20dual-tier=20dispatch-coverage?= =?UTF-8?q?=20shape=20green=20(wave=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/conformance/local_column.rs: the first dual-tier column, run against the kv = mem + large = local factory (ADR-012 §6.3) — fresh temp-root store per open, per-instance tracked roots, idempotent teardown safe against abandoned/panicking rows (structural tests included); never skips (mem-gated — the dual-tier constructor rides feature mem until wave 3 provides an ungated kv engine) - the reference INV row set (rows.rs) mirrored against the dual-tier factory at a column-declared tuned threshold (1024), so the shared matrix and race fixtures exercise both tiers' bodies (INV-7) - dispatch rows grown: pure-function routing across put paths and re-puts with an on-disk placement oracle (tier residency is facade-invisible; with kv = mem the medium is the oracle), threshold boundary legs at the tuned and shipped-default thresholds with the constructor-tuned re-route leg, and fall-through per op with both-tier miss shapes (ADR-003/ADR-012 §6.1/§6.3/§6.5) - local-specific rows: on-disk stage-hygiene exact counts over both put paths (the real stage form, INV-3), pread windows over committed files with slice digests verified independently (ADR-006/ADR-010), virgin-store no-ops on a fresh root, and exact sweep counts over a both-tiers pool (INV-4/INV-5's both-tiers form) - rows.rs shared fixtures made pub(crate); README records the column's row inventory (§Columns beyond the reference set), the declared- threshold posture, and the new current state - stat-as-stat rides the mirrored INV-6 row's over-threshold legs; the Drop-cancellation hygiene form stays per-engine (recorded) Verification: cargo test (default) green; cargo test --features mem green (97 lib + 37 conformance + 4 git-interop); cargo test --all-features green; clippy --all-targets -D warnings green in both feature shapes (default + all-features); cargo fmt --check green; cargo doc --no-deps shows only the pre-existing warning class --- tasks/suite-column-local.md | 139 ++- tests/conformance/README.md | 85 +- tests/conformance/local_column.rs | 1322 +++++++++++++++++++++++++++++ tests/conformance/main.rs | 18 +- tests/conformance/rows.rs | 39 +- 5 files changed, 1560 insertions(+), 43 deletions(-) create mode 100644 tests/conformance/local_column.rs diff --git a/tasks/suite-column-local.md b/tasks/suite-column-local.md index 91e88f2..eb1855e 100644 --- a/tasks/suite-column-local.md +++ b/tasks/suite-column-local.md @@ -1,7 +1,7 @@ --- id: suite-column-local name: Conformance suite — the local column and the dispatch-coverage shape -status: pending +status: completed depends_on: [suite-column-mem] scope: moderate risk: medium @@ -51,16 +51,17 @@ local-specific properties. ## Acceptance Criteria -- [ ] The dual-tier `kv = mem` + `large = local` factory green with +- [x] The dual-tier `kv = mem` + `large = local` factory green with isolation + idempotent teardown -- [ ] The local column green: mem's row set mirrored (INV-1/2/3/4/5/6) +- [x] The local column green: mem's row set mirrored (INV-1/2/3/4/5/6) plus dispatch rows (pure-function routing, fall-through per op, threshold boundaries) and local-specific rows (pread ranges, virgin-store no-ops, stat-as-stat) -- [ ] Stage-hygiene exact counts asserted on disk (the real stage form) -- [ ] Every row ADR-stamped and traceable to the backlog -- [ ] `cargo test` green (local is ungated — the column runs in the - default build); clippy/fmt gates green +- [x] Stage-hygiene exact counts asserted on disk (the real stage form) +- [x] Every row ADR-stamped and traceable to the backlog +- [x] `cargo test` green (default build; the column itself runs under + `--features mem` — the kv = mem pairing gates it, see Notes); + clippy/fmt gates green ## References @@ -73,8 +74,128 @@ local-specific properties. ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-10-11). + +- **The dual-tier factory** (`tests/conformance/local_column.rs`): + `DualTierLocalFactory` (the dispatch-coverage shape, `kv = mem` + + `large = local`) — `open()` vends the concrete `Store` facade over a + fresh private temp root per call (mem shares nothing; the local + engine gets its own root), tracked per *factory instance*; + `teardown()` removes that instance's roots idempotently and is safe + against abandoned instances (the structural tests carry the panicking + row over a medium-backed factory). Every runner constructs its own + factory instance — a shared static's teardown would race + concurrently running rows' roots (the mem factory's no-op teardown + didn't have this hazard; the local one does, so instance ownership + is the rule, recorded in the module doc). +- **The mirrored set** (INV-7): all ten INV-family rows of the + reference set run against the dual-tier factory at a column-declared + tuned threshold (1024) — the shared content matrix and race fixtures + cross it, so both tiers' bodies are exercised (sub-threshold kv + rows; over-threshold stage-commit rows). The INV-3 reference row + mirrors with the *degenerate kv stage form* it owns; the **stage + form** is this column's own row (below). The mem-only and kv-only + constructor-mode rows do not mirror (their `mode()` preconditions + name modes this column is not; README records the rule). +- **The dispatch rows** (the coverage shape's reason to exist): + `routing_is_a_pure_function_of_length_across_put_paths_and_re_puts` + (at/below/over the threshold, same content, both put paths: same + placement, nothing moves or duplicates, one pool entry through all + paths, exact sweep counts); + `threshold_boundary_lengths_route_deterministically` (boundary legs + at the declared threshold on both put paths, plus the default + boundary's length re-routing with the constructor's tuning — run + against the tuned and the default-threshold factories); + `kv_tier_misses_fall_through_to_the_large_tier_per_op` (get/stat/ + has/read_range each served from the large tier when the kv tier + misses, the kv-first mirror-position leg, and both-tier miss shapes + per op — `None`/`None`/`false`/`Missing`). +- **Tier placement is observed through the medium, not the facade.** + Tier residency is facade-invisible by design (the facade is + engine-blind; ADR-012 §6.1's consult order is not a routing + query). With `kv = mem`, the medium itself is the oracle: the mem kv + tier never touches the factory's disk, so a committed file outside + the staging area under the column's temp root is exactly one + large-tier entry and its absence is exactly a kv-tier entry. That + makes the dispatch rows *column-owned* rows taking the concrete + factory (an engine-uniform reference row can never make these + assertions); README §Columns beyond the reference set records the + shape. The dual-tier threshold is likewise **declared + configuration, not a probe**: a dual-tier store never rejects, so + the kv-only row's probing trick has nothing to probe — the rows read + the threshold the factory declares (the value the constructor took). +- **The local-specific rows**: + `stage_hygiene_exact_counts_on_disk_over_both_put_paths` (INV-3's + real stage form — mid-stream `Io` and declared-len `Verification` + failures over threshold-zeroing content leave zero stage files and + zero committed entries on disk; success leaves exactly one committed + entry per path, zero stage files; the Drop-cancellation form needs + the engine internals and stays covered by the engine's in-module + tests, per the engine-local task); + `pread_windows_read_committed_files_mid_slices_and_eof_clamped` + (both put paths' committed files: mid-file slices, tail clamps, + EOF-empty, zero-length, over-read clamp, `read_range` slice + digest + verified against an independently computed SHA-256); + `virgin_store_reads_are_no_ops_on_a_fresh_root` (empty `list`, + per-op miss shapes, empty medium); + `sweep_counts_stay_exact_over_both_tiers` (INV-4/INV-5's both-tiers + form: the full protection set spanning tiers, dead witness per tier, + exact report counts *and* exact on-disk counts, protected entries + byte-identical across tiers). +- **stat-as-stat**: the "size as a stat, never a content load" + property is carried by the mirrored INV-6 row run at the tuned + threshold — its over-threshold legs stat entries served by the + local engine's file-stat size probe (ADR-008 §1), the same + facade-slice/engine-half posture the mem column recorded; no + separate row was needed. +- **Feature gating (amendment to the acceptance criterion as + written)**: the criterion expected "local is ungated — the column + runs in the default build", but the only constructible kv engine + until wave 3 (sqlite) is mem, so `Store::dual_tier_mem_local` rides + feature `mem` (the store-construction task's recorded default-build + note) — the column is `#[cfg(feature = "mem")]`-gated and runs under + `cargo test --features mem`, exactly as the mem column does. The + local *engine* itself remains ungated (in `src/`, default-build + compiled); the reference row set compiles ungated (type-checked in + default builds). The criterion line was amended to the factual + reading; the gate is the kv pairing's, not the local engine's. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion (2026-10-11). + +Landed the **local column** — the suite's first dual-tier column, run +against the `kv = mem` + `large = local` dispatch-coverage factory +(ADR-012 §6.3): fresh temp-root dual-tier store per `open()`, +per-instance tracked roots, idempotent teardown safe against aban- +doned/panicking rows (structural tests included), never skips. +`tests/conformance/rows.rs` gained `pub(crate)` shared fixtures; +`tests/conformance/local_column.rs` mirrors the reference INV row set +against the dual-tier factory at a tuned threshold (1024) so the +shared matrix and race fixtures exercise both tiers' bodies, and grows +the column's own rows: the three dispatch rows (pure-function routing +across put paths and re-puts with an on-disk placement oracle — tier +residency is facade-invisible, so the medium is the oracle; threshold +boundary legs at the tuned and shipped-default thresholds with the +constructor-tuned re-route leg; fall-through per op with both-tier +miss shapes) and the four local rows (on-disk stage-hygiene exact +counts over both put paths — the real stage form; pread windows over +committed files with EOF clamping and an independently verified slice +digest; virgin-store no-ops on a fresh root; exact sweep counts over a +both-tiers pool — also INV-5's both-tiers form). stat-as-stat rides +the mirrored INV-6 row's over-threshold legs. Every row is +ADR-stamped and traced (README's new column inventory section); +stat-as-stat's engine half and the Drop-cancellation hygiene form +remain per-engine (the mem column's INV-6 posture and the engine's +unit tests, respectively — recorded). The column is mem-gated (the +dual-tier constructor rides feature `mem` until wave 3 provides an +ungated kv engine); the criterion's "runs in the default build" +expectation is amended and recorded. + +Verification: `cargo test` green (42 lib + 2 conformance + 3 +git-interop), `cargo test --features mem` green (97 lib + 37 +conformance + 4 git-interop), `cargo test --all-features` green; `cargo +clippy --all-targets -- -D warnings` and `cargo clippy --all-features +--all-targets -- -D warnings` green; `cargo fmt --check` green; +`cargo doc --no-deps` shows only the pre-existing private-intra-doc- +links warning class (this task touches no public API). \ No newline at end of file diff --git a/tests/conformance/README.md b/tests/conformance/README.md index afc2b8e..94b369e 100644 --- a/tests/conformance/README.md +++ b/tests/conformance/README.md @@ -185,6 +185,49 @@ 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). +## Columns beyond the reference set (INV-7's mirror + extension) + +The constructor-mode rows of the reference set run wherever the mode +they name is constructible (README §The reference row set). A +multi-tier column mirrors the set and grows **its own rows** for what +its configuration carries that the reference set's single-tier host +did not: the dispatch-coverage shape (ADR-012 §6.3 — both tiers +load-bearing) and the engine's tier-specific properties. Those rows +are still conformance rows — ADR-stamped, named for the property, +traced — but they take the **column's concrete factory** instead of +the `StoreFactory` trait: they observe the backing medium, which an +engine-uniform row (unable to branch on any engine, and unable to see +any tier's residency through the facade) cannot. Tier placement is +facade-invisible by design; with `kv = mem`, the medium itself is the +oracle — the mem kv tier never touches the factory's disk, so a +committed file outside the staging area under the column's temp root +is exactly one large-tier entry, and its absence beside a served read +is exactly a kv-tier entry. + +**The local column's own rows** (tests/conformance/local_column.rs, +the dispatch-coverage and local-specific families): + +| Row | Traces to | +|---|---| +| `routing_is_a_pure_function_of_length_across_put_paths_and_re_puts` | ADR-003 §Dispatch policy (purity/no-migration); ADR-012 §6.3; INV-7 facade-contract row | +| `threshold_boundary_lengths_route_deterministically` | ADR-003 §Dispatch policy (at/below/above the threshold; constructor-tuned re-route); ADR-012 §6.3 | +| `kv_tier_misses_fall_through_to_the_large_tier_per_op` | ADR-012 §6.1 (fall-through per op); ADR-012 §6.5 ruling 5 (miss shapes); ADR-006 | +| `stage_hygiene_exact_counts_on_disk_over_both_put_paths` | INV-3's stage form (exact counts on disk); ADR-010 §1; ADR-003 | +| `pread_windows_read_committed_files_mid_slices_and_eof_clamped` | store-api.md §Read path; ADR-010 §1 (streaming cursor); ADR-006 | +| `virgin_store_reads_are_no_ops_on_a_fresh_root` | store-api.md §Error model (virgin-store semantics); ADR-012 §6.5 | +| `sweep_counts_stay_exact_over_both_tiers` | INV-4/INV-5's both-tiers form (exact counts + list correctness through outcomes); ADR-005; ADR-012 §1/§6.3/§6.7 | + +**The dual-tier threshold is declared configuration, not a +behavioral probe.** The kv-only reference row discovers its factory's +threshold by probing (rejections read nothing — free). A dual-tier +store never rejects a length, and tier residency is facade-invisible, +so there is nothing to probe: the dispatch rows take the threshold +the column's factory declares (the same value its constructor took — +configuration, like the mem column's kv-only threshold). The mem-only +and kv-only mode rows do not mirror onto the dual-tier factories (the +precondition `mode()` shapes they assert are not this column's mode); +the dispatch rows are the dual-tier mode's facade-contract coverage. + ## Column plan **mem is the contract-reference engine** (ADR-011 §2): wave 2 lands @@ -197,15 +240,33 @@ and first run in wave 6. ## Current state -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 +The harness is green on **two columns** (wave 2). The **mem column** +is the contract-reference column: 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 **local column** (the suite's first dual-tier column, the +dispatch-coverage shape ADR-012 §6.3 names) mirrors the reference INV +rows against its `kv = mem` + `large = local` factory at a +column-declared tuned threshold — the shared matrix and race fixtures +cross it, so both tiers' bodies are exercised — and grows the +dispatch-coverage rows (pure-function routing with an on-disk +placement oracle, fall-through per op, threshold boundaries at both +the tuned and the shipped-default threshold) plus the local-specific +rows (on-disk stage hygiene exact counts, pread windows, virgin-store +no-ops, the both-tiers sweep shape) — §Columns beyond the reference +set is the recorded inventory. Its factory instances each track and +remove their own temp roots (idempotent teardown, panicking-row-safe; +each runner owns its instance so concurrent rows never share roots). +mem-only/kv-only modes and the local column never skip, but the local +column is feature-gated on `mem`: the `dual_tier_mem_local` +constructor rides feature `mem` because mem is the only constructible +kv engine until wave 3 (the gate is the kv pairing's, not the local +engine's, which is ungated in `src/`) — the column runs under `cargo +test --features mem`, with the reference row set compiling ungated so +it stays type-checked in default builds. \ No newline at end of file diff --git a/tests/conformance/local_column.rs b/tests/conformance/local_column.rs new file mode 100644 index 0000000..e90ac41 --- /dev/null +++ b/tests/conformance/local_column.rs @@ -0,0 +1,1322 @@ +//! The **local column** — the suite's first dual-tier column: the +//! reference row set run against the `kv = mem` + `large = local` +//! factory (the dispatch-coverage test shape ADR-012 §6.3 names, both +//! tiers load-bearing), extended with the dual-tier and local-specific +//! properties (README §The reference row set; INV-7: every engine's +//! rows mirror the reference column). +//! +//! The factories (README §The factory contract): `open()` vends the +//! concrete [`alkblobs::Store`] facade over a fresh temp directory per +//! call (`kv = mem` — isolated by construction — and `large = local` +//! over that private root), so two opens never share state. Teardown +//! removes every root this factory instance vended (idempotent: a +//! repeated teardown is a no-op; safe against abandoned instances — +//! each root is private to its open, so a half-open instance's root is +//! removable the same way). The factory instance owns what it vends: +//! every runner constructs its own factory, so concurrent tests never +//! touch each other's roots. The local column never skips (README §SKIP +//! posture — a dependency-free column, like mem). +//! +//! The column is feature-gated on `mem`: the only constructible kv +//! engine until wave 3 (sqlite) is mem, so the dual-tier constructor +//! (`Store::dual_tier_mem_local`) rides that feature — the gate is the +//! *kv pairing's*, not the local engine's (which is ungated in `src/`). +//! +//! Two row populations live here: +//! +//! - the **mirrored rows** — every INV-family row of the reference set +//! ([`crate::rows`]) run against this column's dual-tier factory at +//! the column-declared tuned threshold, so the shared content matrix +//! and the race fixtures cross the threshold and both tiers' bodies +//! are exercised (sub-threshold kv rows; over-threshold local +//! stage-commit rows); +//! - the column's **own rows** (dispatch coverage + local-specific, +//! documented per row): tier placement, fall-through, boundary +//! routing, on-disk stage hygiene, pread windows, virgin-store +//! no-ops, and the both-tiers sweep shape. These rows take this +//! column's concrete factory — they observe the backing medium +//! (the large tier's committed tree) — which a reference row, generic +//! over engines and unable to branch on either, cannot; the medium +//! observation *is* the dispatch-coverage shape's point. With +//! `kv = mem`, medium absence proves kv holding: the mem tier never +//! touches this disk, so a committed file under `root` outside the +//! staging area is exactly one large-tier entry, and its absence +//! (the put published, reads served) is exactly one kv-tier entry. + +#![allow(dead_code)] + +use std::io::Cursor; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +use sha2::{Digest as _, Sha256}; + +use crate::factory::StoreFactory; +use crate::rows::{self, assert_exact_counts, patterned, Chunked, Failing, SetProtect, SetSource}; +use alkblobs::{Error, Key, Store, StoreMode, DEFAULT_DISPATCH_THRESHOLD}; + +// ------------------------------------------------------------------ +// The factories +// ------------------------------------------------------------------ + +/// The column's declared dual-tier threshold — small, so the reference +/// rows' shared content matrix crosses it and both tiers' bodies are +/// exercised (sub-threshold rows land kv; over-threshold rows +/// stage-commit into the local tier). +pub(crate) const TUNED_THRESHOLD: u64 = 1024; + +/// The dispatch-coverage factory: `kv = mem` + `large = local` at the +/// threshold this column declares (see [`TUNED_THRESHOLD`], and the +/// default-threshold sibling `DualTierLocalFactory::new( +/// DEFAULT_DISPATCH_THRESHOLD)` the boundary row runs against). +/// +/// **Why a fresh instance per runner (not a shared static):** teardown +/// removes the roots *this instance* vended; a shared instance's +/// teardown would race concurrently running rows' roots. Every runner +/// constructs its own and owns its lifecycle. +pub(crate) struct DualTierLocalFactory { + /// The declared dispatch threshold — the routing boundary this + /// column's rows run at (constructor-tunable; ADR-003). Tier + /// placement is facade-invisible by design (the facade is + /// engine-blind), so the rows that need the boundary read it here — + /// the same value the constructor takes, declared configuration + /// rather than behaviorally probed (a dual-tier store never + /// rejects, so the kv-only row's probing trick has nothing to + /// probe; recorded in README). + threshold: u64, + /// Every temp root this instance vended, for its own teardown. + roots: Mutex>, +} + +impl DualTierLocalFactory { + pub(crate) fn new(threshold: u64) -> DualTierLocalFactory { + DualTierLocalFactory { + threshold, + roots: Mutex::new(Vec::new()), + } + } + + /// A fresh temp directory for one open — the isolated backing of + /// the dual-tier shape (`kv = mem` shares nothing; the local engine + /// gets this private root). + fn fresh_temp_root() -> PathBuf { + use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + static SEQ: std::sync::atomic::AtomicU64 = AtomicU64::new(0); + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let dir = std::env::temp_dir().join(format!( + "alkblobs-local-column-{}-{:x}-{seq}", + std::process::id(), + nanos + )); + std::fs::create_dir_all(&dir).expect("temp root created"); + dir + } + + /// Open + track the vended root: the shape the column's own rows + /// use (they observe the backing medium, so they need the root + /// alongside the store). + pub(crate) async fn open_tracked(&self) -> Result<(Store, PathBuf), Error> { + let root = Self::fresh_temp_root(); + let store = Store::dual_tier_mem_local(root.clone(), self.threshold).await?; + self.roots + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(root.clone()); + Ok((store, root)) + } +} + +impl StoreFactory for DualTierLocalFactory { + fn name(&self) -> &'static str { + "dual-tier mem+local" + } + + async fn open(&self) -> Result { + self.open_tracked().await.map(|(store, _)| store) + } + + async fn teardown(&self) -> Result<(), Error> { + for root in self + .roots + .lock() + .unwrap_or_else(|e| e.into_inner()) + .drain(..) + { + match std::fs::remove_dir_all(&root) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(Error::Io(e.to_string())), + } + } + Ok(()) + } +} + +// ------------------------------------------------------------------ +// On-disk observation helpers (the dispatch-coverage shape's medium +// oracle; kv = mem never touches this disk, so these counts are +// exactly the large tier's committed/staging residency). +// ------------------------------------------------------------------ + +/// Every regular file under `root` outside any `staging` directory — +/// with `kv = mem`, exactly one per committed large-tier entry. +fn committed_file_count(root: &Path) -> usize { + fn walk(dir: &Path, out: &mut Vec) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let Ok(kind) = entry.file_type() else { + continue; + }; + let path = entry.path(); + if kind.is_dir() { + if path.file_name().and_then(|n| n.to_str()) == Some("staging") { + continue; + } + walk(&path, out); + } else if kind.is_file() { + out.push(path); + } + } + } + let mut files = Vec::new(); + walk(root, &mut files); + files.len() +} + +/// Files in the large tier's staging area — the residue counter of the +/// exact-count hygiene row (zero after every failure form and every +/// success form; a staged artifact lives there only mid-put). +fn staging_file_count(root: &Path) -> usize { + match std::fs::read_dir(root.join("staging")) { + Ok(entries) => entries.count(), + Err(_) => 0, + } +} + +/// The ADR-006 slice digest, computed independently of the crate — +/// SHA-256 over exactly the returned bytes (the out-of-band +/// verification the convention requires; callers verify the same way). +fn slice_digest(bytes: &[u8]) -> [u8; 32] { + Sha256::digest(bytes).into() +} + +// ------------------------------------------------------------------ +// The column's own rows — dispatch coverage (the coverage shape's +// reason to exist) and the local-specific properties. Every row takes +// this column's concrete factory (it observes the backing medium), +// is ADR-stamped per the convention, and traces its family. +// ------------------------------------------------------------------ + +/// Dispatch row: **routing is a pure function of length** — at, below, +/// and over the declared threshold, the same content lands on exactly +/// the same tier through both put paths and both re-put rounds (same +/// placement every time; nothing moves, nothing duplicates), three +/// puts of one content are one pool entry through all paths, and every +/// read is byte-identical from wherever that tier holds it. +/// +/// Tier placement is observed through the backing medium (the module +/// doc's oracle): with `kv = mem`, a committed file under `root` +/// outside the staging area is exactly one large-tier entry, and its +/// absence is exactly a kv-tier entry — the facade cannot and must not +/// expose tier residency (it is engine-blind); the dual-tier column +/// observes it through the medium the coverage shape runs on. +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row); +/// ADR-003 §Dispatch policy (routing is a pure function of length; +/// no migration; CAS first-writer-wins makes re-puts one entry); +/// ADR-005 (re-put refcounts); ADR-012 §6.3 (the dual-tier +/// dispatch-coverage shape this row exists to cover). +pub async fn routing_is_a_pure_function_of_length_across_put_paths_and_re_puts( + factory: &DualTierLocalFactory, +) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + SetSource::register(&store); + + for content in [ + patterned((threshold - 1) as usize, 191), + patterned(threshold as usize, 173), + patterned((threshold + 1) as usize, 167), + patterned(4 * threshold as usize + 9, 151), + ] { + let len = content.len() as u64; + let canonical = Key::derive(&content, len); + let expected_files = if len > threshold { 1 } else { 0 }; + + for (path_tag, declared) in [("the known path", Some(len)), ("the unknown path", 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 on {path_tag}" + ); + assert_eq!( + committed_file_count(&root), + expected_files, + "length {len} on {path_tag}: the placement is exactly the \ + routing function's answer and never moves or duplicates" + ); + assert_eq!( + staging_file_count(&root), + 0, + "a published put leaves no stage file on {path_tag}" + ); + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!( + got_len, len, + "the read serves the entry from its placed tier" + ); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "the entry is byte-identical from its placed tier across paths" + ); + drop(pin); + } + + 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 + ), + (1, 1, 0), + "identical content through both put paths is one pool entry" + ); + assert_eq!( + committed_file_count(&root), + 0, + "the sweep removed exactly the one entry" + ); + } + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Dispatch row: **threshold boundary lengths route deterministically** +/// — at/below/above the factory's declared threshold, on both put +/// paths, the placement is the routing function's answer every time; +/// and the default-threshold boundary re-routes with the tuning (the +/// constructor-tuned threshold deterministically places the default +/// boundary's length on the tier its own threshold names). Run against +/// two factories by the runner — the column's tuned threshold and the +/// shipped default — the same row shape at two declared thresholds +/// being the tunable re-route demonstration. +/// +/// Contract stamp: store-api.md §Invariants 7 (facade-contract row); +/// ADR-003 §Dispatch policy (default 128 KiB at-or-below routes kv; +/// constructor-tunable without API change); ADR-012 §6.3 (the +/// dual-tier dispatch-coverage shape). +pub async fn threshold_boundary_lengths_route_deterministically(factory: &DualTierLocalFactory) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + SetSource::register(&store); + + let boundary = [ + (threshold - 1, 191u32, 0u64), + (threshold, 173, 0), + (threshold, 167, 0), + (threshold + 1, 151, 1), + (threshold + 1, 139, 1), + ]; + let mut committed = 0u64; + for (len, modulus, lands) in boundary { + let content = patterned(len as usize, modulus); + for declared in [Some(content.len() as u64), None] { + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!( + key, + Key::derive(&content, len), + "the canonical derivation addresses the put" + ); + assert_eq!( + committed_file_count(&root) as u64, + committed + lands, + "boundary length {len}: at/below the threshold lands kv \ + (no large-tier file), over it lands large (one file), \ + both put paths" + ); + drop(pin); + } + committed += lands; + } + + // The default boundary re-routes with the constructor's tuning: + // the shipped default's length belongs to the tier this + // configuration's own threshold puts it on. + let at_default = patterned(DEFAULT_DISPATCH_THRESHOLD as usize, 199); + let expected_files = if DEFAULT_DISPATCH_THRESHOLD > threshold { + 1 + } else { + 0 + }; + let (key, pin) = store + .put( + Some(DEFAULT_DISPATCH_THRESHOLD), + Cursor::new(at_default.as_slice()), + ) + .await + .expect("the default-boundary put publishes"); + assert_eq!( + key, + Key::derive(&at_default, DEFAULT_DISPATCH_THRESHOLD), + "the canonical derivation addresses the put" + ); + assert_eq!( + committed_file_count(&root) as u64, + committed + expected_files, + "the default boundary's length routes to the tier this \ + configuration's tuning names" + ); + drop(pin); + + 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 + ), + (6, 6, 0), + "each leg's content was exactly one pool entry" + ); + assert_eq!( + committed_file_count(&root), + 0, + "the sweep removed every entry, tier notwithstanding" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Dispatch row: **a kv-tier miss falls through to the large tier on +/// every key-addressed op** — content the routing function placed on +/// the large tier (one committed file under `root`; the kv tier +/// misses) serves `get`, `stat`, `has`, and `read_range` whole, +/// exact, and digest-correct; the mirror-position leg proves the kv +/// tier is the *first* consult (sub-threshold content, no large-tier +/// file, reads still served); and both-tier misses resolve per the +/// error model's per-op shapes. +/// +/// Contract stamp: store-api.md §Read path (fall-through consult +/// order); ADR-012 §6.1 (every key-addressed op consults kv first, +/// then large; the general rule); ADR-012 §6.5 ruling 5 (both-tier +/// miss shapes: `None` on get/stat, `Err(Missing)` on read_range); +/// ADR-006 (the slice digest); ADR-012 §6.3 (the dispatch-coverage +/// shape). +pub async fn kv_tier_misses_fall_through_to_the_large_tier_per_op(factory: &DualTierLocalFactory) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + SetSource::register(&store); + + // Over-threshold content — the kv tier misses, the large tier + // serves every op. Both put paths hold the shape. + for (content, declared) in [ + ( + patterned(3 * threshold as usize + 41, 181), + Some(3 * threshold + 41), + ), + (patterned(3 * threshold as usize + 43, 197), None), + ] { + let len = content.len() as u64; + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + assert_eq!( + committed_file_count(&root), + 1, + "the content lives on exactly the large tier — the kv tier \ + misses" + ); + + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the large tier served the kv miss"); + assert_eq!(got_len, len); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "the fall-through read is byte-identical" + ); + assert_eq!( + store.stat(&key.as_bytes()).await.expect("stat succeeds"), + Some(alkblobs::EntryMeta { + len, + algorithm: key.algorithm(), + }), + "stat falls through and carries the length plus algorithm" + ); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "has falls through" + ); + let mid = len / 2; + let (slice, digest) = store + .read_range(&key.as_bytes(), mid, 17) + .await + .expect("read_range falls through"); + assert_eq!( + slice, + content[(mid as usize)..(mid as usize + 17)], + "the range slice is exact" + ); + assert_eq!( + digest, + slice_digest(&slice), + "the slice digest is SHA-256 over exactly the returned bytes" + ); + + // Sweep the entry away (the pin dropped): the miss shapes now + // resolve per op — the large tier misses too. + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + (report.candidates_staged, report.deleted), + (1, 1), + "exactly the one entry died" + ); + assert!( + store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "a both-tier miss on get is Ok(None)" + ); + assert_eq!( + store.stat(&key.as_bytes()).await.expect("stat succeeds"), + None, + "a both-tier miss on stat is Ok(None)" + ); + assert!( + !store.has(&key.as_bytes()).await.unwrap(), + "a both-tier miss on has is false" + ); + assert_eq!( + store + .read_range(&key.as_bytes(), 0, 4) + .await + .expect_err("the both-tier miss errs"), + Error::Missing, + "a both-tier miss on read_range is Err(Missing)" + ); + } + + // The mirror-position leg: sub-threshold content — the kv tier is + // the first consult and serves before the large tier is reached + // (no large-tier file ever lands). + let small = patterned((threshold - 1) as usize, 149); + let (key, pin) = store + .put(Some(small.len() as u64), Cursor::new(small.as_slice())) + .await + .expect("the sub-threshold put publishes"); + assert_eq!( + committed_file_count(&root), + 0, + "the kv tier holds the sub-threshold entry — the large tier was \ + never consulted on the write, never reached on the read" + ); + let (len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the kv tier served"); + assert_eq!(len, small.len() as u64); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + small + ); + + drop(pin); + + // Both-tier miss shapes for a key that never existed at all, per + // op, over the fresh both-tier store. + let absent = b"never put here".to_vec(); + let absent = Key::derive(&absent, absent.len() as u64); + assert!( + store + .get(&absent.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "never-put: get is Ok(None)" + ); + assert_eq!( + store.stat(&absent.as_bytes()).await.expect("stat succeeds"), + None, + "never-put: stat is Ok(None)" + ); + assert!( + !store.has(&absent.as_bytes()).await.unwrap(), + "never-put: has is false" + ); + assert_eq!( + store + .read_range(&absent.as_bytes(), 0, 4) + .await + .expect_err("never-put read_range errs"), + Error::Missing, + "never-put: read_range is Err(Missing)" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-3, the stage-hygiene row's **real stage form**: exact counts +/// **on disk** over both put paths — every failure form (the +/// mid-stream stream failure, the declared-length mismatch over and +/// under), zero stage files and zero committed entries after each; +/// every success form, exactly one committed entry and zero stage +/// files. (The mirrored reference row asserts this invariant's kv +/// form — the degenerate stage the kv tier degenerates to — through +/// sweep outcomes only; the local tier's stage file is the form this +/// row adds.) +/// +/// Contract stamp: store-api.md §Invariants 3 (zero stage files after +/// any failure, exactly one committed entry per success); ADR-010 §1 +/// (the staged-put contract: commit publishes, every failure path +/// converges on discard); ADR-003 §Dispatch policy (the three numbered +/// rules — the buffered case's failure paths extend the invariant); +/// ADR-012 §6.3 (the dispatch-coverage shape). +pub async fn stage_hygiene_exact_counts_on_disk_over_both_put_paths( + factory: &DualTierLocalFactory, +) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + SetSource::register(&store); + + let content = patterned(2 * threshold as usize + 17, 167); + let len = content.len() as u64; + + // Known path, mid-stream stream failure — the stage file existed, + // the failure discards it: zero stage files, zero committed. + let err = store + .put(Some(len), Failing::new(&content, 100)) + .await + .expect_err("the failed mid-stream put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (0, 0), + "zero residue on disk after the known path's stream failure" + ); + + // Known path, declared-length mismatch — over-fed and under-fed + // legs; each had staged (over-threshold content) and discards. + for declared in [len + 7, len - 7] { + let err = store + .put(Some(declared), Cursor::new(content.as_slice())) + .await + .expect_err("the mismatch rejects"); + assert_eq!(err, Error::Verification, "declared-len mismatch rejects"); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (0, 0), + "zero residue on disk after the mismatch on declared {declared}" + ); + } + + // Unknown path, mid-stream stream failure past the overflow — the + // buffered prefix had flushed into a stage file (the failure + // reached the real stage form), and the discard leaves zero. + let failing_content = patterned(2 * threshold as usize + 19, 173); + let err = store + .put( + None, + Failing::new(&failing_content, threshold as usize + 600), + ) + .await + .expect_err("the failed unknown-length put rejects"); + assert!( + matches!(err, Error::Io(_)), + "the stream failure surfaces as Io: {err}" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (0, 0), + "zero residue on disk after the unknown path's stream failure — \ + the staged form included" + ); + + // Success, known path: the stage commits to exactly one committed + // entry, zero stage files. + let (key, pin) = store + .put(Some(len), Cursor::new(content.as_slice())) + .await + .expect("the known path publishes"); + assert_eq!( + key, + Key::derive(&content, len), + "the canonical derivation addresses the put" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (1, 0), + "exactly one committed entry on disk, no stage file" + ); + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!(got_len, len); + assert_eq!( + blob.read_at(0, len).await.expect("the whole window reads"), + content, + "the committed entry is byte-identical" + ); + drop(pin); + + // Success, unknown path: the overflow staged, the commit renames — + // exactly one committed entry, zero stage files. + let other = patterned(2 * threshold as usize + 23, 151); + let (key, pin) = store + .put(None, Chunked::new(&other, &[300; 32])) + .await + .expect("the unknown path publishes"); + assert_eq!( + key, + Key::derive(&other, other.len() as u64), + "the canonical derivation addresses the put" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (2, 0), + "exactly one committed entry per success, both on disk, no \ + stage file anywhere" + ); + let (got_len, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + assert_eq!(got_len, other.len() as u64); + assert_eq!( + blob.read_at(0, got_len) + .await + .expect("the whole window reads"), + other, + "the committed entry is byte-identical" + ); + drop(pin); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + (report.candidates_staged, report.deleted), + (2, 2), + "the two successes were exactly two pool entries" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (0, 0), + "the pool and the disk are both back to zero residue" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Local-specific row: **pread windows read committed files** — +/// mid-file slices exact, tail clamped at EOF, a read at EOF empty +/// (never an error), zero-length reads empty, over-reads clamped, and +/// the ADR-006 slice digest carried over exactly the returned bytes — +/// over content the routing function placed on the large tier (both +/// put paths: the commit-rename form and the staged-then-commit form +/// both publish committed files). +/// +/// Contract stamp: store-api.md §Read path (the streaming large-tier +/// windows, pread-style clamping); ADR-010 §1 (the streaming read +/// cursor body); ADR-006 (the slice digest over exactly the returned +/// bytes); ADR-012 §6.3 (the dispatch-coverage shape); ADR-012 §6.1 +/// (the reads ride the fall-through consult). +pub async fn pread_windows_read_committed_files_mid_slices_and_eof_clamped( + factory: &DualTierLocalFactory, +) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + SetSource::register(&store); + + let mut committed = 0u64; + for (content, declared) in [ + ( + patterned(2 * threshold as usize + 29, 191), + Some(2 * threshold + 29), + ), + (patterned(2 * threshold as usize + 31, 197), None), + ] { + let len = content.len() as u64; + let (key, pin) = store + .put(declared, Cursor::new(content.as_slice())) + .await + .expect("put publishes"); + committed += 1; + assert_eq!( + committed_file_count(&root) as u64, + committed, + "each leg's entry is a committed file on the large tier" + ); + + let (_, blob) = store + .get(&key.as_bytes()) + .await + .expect("get succeeds") + .expect("the entry is present"); + + let mid = len / 2; + assert_eq!( + blob.read_at(mid, 13).await.expect("the mid slice reads"), + content[(mid as usize)..(mid as usize + 13)], + "a mid-file slice is exact" + ); + assert_eq!( + blob.read_at(len - 5, 100).await.expect("the tail reads"), + content[(len as usize - 5)..], + "the tail read clamps at the entry's end" + ); + assert_eq!( + blob.read_at(len, 8).await.expect("the EOF read succeeds"), + Vec::::new(), + "a read at EOF is empty, never an error" + ); + assert_eq!( + blob.read_at(mid, 0).await.expect("the empty read succeeds"), + Vec::::new(), + "a zero-length read is empty" + ); + assert_eq!( + blob.read_at(0, len + 64) + .await + .expect("the over-read reads"), + content, + "an over-read clamps to the whole entry" + ); + + let (slice, digest) = blob + .read_range(mid, 30) + .await + .expect("the ranged slice reads"); + assert_eq!( + slice, + content[(mid as usize)..(mid as usize + 30)], + "the ranged slice is exact" + ); + assert_eq!( + digest, + slice_digest(&slice), + "the slice digest is SHA-256 over exactly the returned bytes" + ); + let (whole, whole_digest) = blob + .read_range(0, len) + .await + .expect("the whole range reads"); + assert_eq!(whole, content); + assert_eq!( + whole_digest, + slice_digest(&content), + "the whole-body range digest matches" + ); + + drop(pin); + } + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!( + (report.candidates_staged, report.deleted), + (2, 2), + "both pread subjects were exactly one entry each" + ); + assert_eq!( + committed_file_count(&root), + 0, + "the sweep removed both committed files" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// Local-specific row: **virgin-store reads are no-ops on a fresh +/// root** — an empty `list`, absent reads per op (`None` on get/stat, +/// `false` on has, `Err(Missing)` on read_range), and nothing on the +/// fresh medium itself. +/// +/// Contract stamp: store-api.md §Error model (virgin-store semantics: +/// reads on a fresh store see absent/empty, never errors); ADR-003 +/// (a fresh root reads as absent/empty — the missing-table no-op +/// generalizes to the local tier's fresh walk); ADR-012 §6.5 ruling 5 +/// (the per-op miss shapes). +pub async fn virgin_store_reads_are_no_ops_on_a_fresh_root(factory: &DualTierLocalFactory) { + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + assert_eq!( + store.mode(), + StoreMode::DualTier, + "the column's factory is the dispatch-coverage shape" + ); + + let absent = b"nothing was ever put".to_vec(); + let absent = Key::derive(&absent, absent.len() as u64); + + assert_eq!( + store.list().await.expect("list succeeds"), + Vec::::new(), + "the virgin pool lists empty" + ); + assert!( + store + .get(&absent.as_bytes()) + .await + .expect("get succeeds") + .is_none(), + "absent get is Ok(None)" + ); + assert_eq!( + store.stat(&absent.as_bytes()).await.expect("stat succeeds"), + None, + "absent stat is Ok(None)" + ); + assert!( + !store.has(&absent.as_bytes()).await.unwrap(), + "absent has is false" + ); + assert_eq!( + store + .read_range(&absent.as_bytes(), 0, 4) + .await + .expect_err("absent read_range errs"), + Error::Missing, + "absent read_range is Err(Missing)" + ); + assert_eq!( + (committed_file_count(&root), staging_file_count(&root)), + (0, 0), + "the fresh medium holds nothing" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +/// INV-4 mirrored onto the dual-tier pool, the both-tiers form: the +/// full protection set spanning both tiers (a source-live kv entry, a +/// pinned large-tier entry, a protect-named large-tier entry) plus +/// one dead entry per tier — exact counts on the report **and** on the +/// medium: only the dead entries stage, delete, and vanish from the +/// committed tree, and the protected entries stay byte-identical. The +/// exact candidates also make this the INV-5 posture's both-tiers +/// form: list correctness (complete and well-formed across both +/// tiers' enumerations) is observable only through these outcomes. +/// +/// Contract stamp: store-api.md §Invariants 4 (exact counts with +/// correct liveness) and §Invariants 5 (list correctness through +/// sweep outcomes); ADR-005 (three liveness marks; the delete window's +/// arbitrate-at-delete-time); ADR-012 §1 (sweeps are never errors); +/// ADR-012 §6.3 (both tiers load-bearing in this shape); ADR-012 §6.7 +/// (the non-SQL window executes inline). +pub async fn sweep_counts_stay_exact_over_both_tiers(factory: &DualTierLocalFactory) { + let threshold = factory.threshold; + let (store, root) = factory + .open_tracked() + .await + .expect("open vends a fresh, isolated dual-tier store"); + let source = SetSource::register(&store); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let source_live = patterned((threshold - 1) as usize, 191); + 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 = patterned(2 * threshold as usize + 7, 173); + let (key_pinned, pin_pinned) = store + .put(None, Cursor::new(pinned.as_slice())) + .await + .expect("the pinned entry publishes"); + + let protected = patterned(2 * threshold as usize + 11, 167); + 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_small = patterned((threshold - 3) as usize, 149); + let (_, pin) = store + .put( + Some(dead_small.len() as u64), + Cursor::new(dead_small.as_slice()), + ) + .await + .expect("the kv-tier dead entry publishes"); + drop(pin); + let dead_large = patterned(2 * threshold as usize + 13, 151); + let (_, pin) = store + .put(None, Cursor::new(dead_large.as_slice())) + .await + .expect("the large-tier dead entry publishes"); + drop(pin); + + assert_eq!( + committed_file_count(&root), + 3, + "the three large-tier entries hold exactly three committed files \ + before the sweep" + ); + + 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 + ), + (2, 2, 0), + "exactly the two dead entries staged and deleted — the pool \ + spans both tiers" + ); + assert_exact_counts(&report); + assert_eq!( + committed_file_count(&root), + 2, + "the dead large-tier entry's file is gone; the protected ones \ + remain on the medium" + ); + + 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!(len, content.len() as u64); + assert_eq!( + blob.read_at(0, len).await.expect("the window reads"), + content.as_slice(), + "the protected entry is byte-identical after the sweep (both \ + tiers of them)" + ); + } + + 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 large-tier \ + 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()); + assert_eq!( + committed_file_count(&root), + 1, + "exactly the protect-named entry's file remains" + ); + + drop(store); + factory.teardown().await.expect("teardown is idempotent"); +} + +// ------------------------------------------------------------------ +// The runners — the column's tests: the reference row set mirrored +// (INV-7) against the tuned dual-tier factory, the column's own rows +// above, and the factory contract's structural shapes. +// ------------------------------------------------------------------ + +#[tokio::test] +async fn local_inv1_round_trip_puts_and_gets_are_byte_identical() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::round_trip_puts_and_gets_are_byte_identical(&factory).await; +} + +#[tokio::test] +async fn local_inv2_dedup_identical_content_is_one_pool_entry() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::dedup_identical_content_is_one_pool_entry(&factory).await; +} + +#[tokio::test] +async fn local_inv3_stage_hygiene_zero_residue_on_failure_one_entry_on_success() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(&factory).await; +} + +#[tokio::test] +async fn local_inv4_sweep_counts_are_exact_across_the_full_protection_set() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::sweep_counts_are_exact_across_the_full_protection_set(&factory).await; +} + +#[tokio::test] +async fn local_inv4_sweep_without_liveness_sources_aborts_and_deletes_nothing() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::sweep_without_liveness_sources_aborts_and_deletes_nothing(&factory).await; +} + +#[tokio::test] +async fn local_inv4_sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing(&factory).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn local_inv4_sweep_is_single_flight_the_contended_second_aborts_as_data() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::sweep_is_single_flight_the_contended_second_aborts_as_data(&factory).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn local_inv4_sweep_racing_a_pinned_put_never_deletes_the_protected_entry() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::sweep_racing_a_pinned_put_never_deletes_the_protected_entry(&factory).await; +} + +#[tokio::test] +async fn local_inv5_list_correctness_is_observable_through_sweep_outcomes() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::list_correctness_is_observable_through_sweep_outcomes(&factory).await; +} + +#[tokio::test] +async fn local_inv6_stat_reports_len_and_algorithm_without_fetching_content() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + rows::stat_reports_len_and_algorithm_without_fetching_content(&factory).await; +} + +#[tokio::test] +async fn dispatch_routing_is_a_pure_function_of_length_across_put_paths_and_re_puts() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + routing_is_a_pure_function_of_length_across_put_paths_and_re_puts(&factory).await; +} + +#[tokio::test] +async fn dispatch_threshold_boundary_lengths_route_deterministically() { + let tuned = DualTierLocalFactory::new(TUNED_THRESHOLD); + threshold_boundary_lengths_route_deterministically(&tuned).await; + let default = DualTierLocalFactory::new(DEFAULT_DISPATCH_THRESHOLD); + threshold_boundary_lengths_route_deterministically(&default).await; +} + +#[tokio::test] +async fn dispatch_kv_tier_misses_fall_through_to_the_large_tier_per_op() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + kv_tier_misses_fall_through_to_the_large_tier_per_op(&factory).await; +} + +#[tokio::test] +async fn local_stage_hygiene_exact_counts_on_disk_over_both_put_paths() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + stage_hygiene_exact_counts_on_disk_over_both_put_paths(&factory).await; +} + +#[tokio::test] +async fn local_pread_windows_read_committed_files_mid_slices_and_eof_clamped() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + pread_windows_read_committed_files_mid_slices_and_eof_clamped(&factory).await; +} + +#[tokio::test] +async fn local_virgin_store_reads_are_no_ops_on_a_fresh_root() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + virgin_store_reads_are_no_ops_on_a_fresh_root(&factory).await; +} + +#[tokio::test] +async fn local_inv4_sweep_counts_stay_exact_over_both_tiers() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + sweep_counts_stay_exact_over_both_tiers(&factory).await; +} + +// ------------------------------------------------------------------ +// The factory contract's structural shapes (the mem column's carried +// to a medium-backed factory): fresh isolated opens, and idempotent +// teardown safe against a panicking row — with the local column's +// on-disk forms. Not conformance rows — no backlog entry — the +// factory contract's own wiring-level assertions. +// ------------------------------------------------------------------ + +/// Every `open()` vends a fresh, isolated dual-tier store: no shared +/// state between islands, observable on the facade *and* on the fresh +/// medium (each open's root is private and starts empty). +#[tokio::test] +async fn factory_open_vends_fresh_isolated_dual_tier_stores() { + let factory = DualTierLocalFactory::new(TUNED_THRESHOLD); + + let (a, root_a) = factory.open_tracked().await.expect("open vends a store"); + let (b, root_b) = factory.open_tracked().await.expect("open vends another"); + assert_ne!(root_a, root_b, "two opens never share a backing root"); + + let content_a = patterned(2 * TUNED_THRESHOLD as usize + 5, 167); + let content_b = patterned(TUNED_THRESHOLD as usize - 5, 191); + 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" + ); + assert_eq!( + committed_file_count(&root_a), + 1, + "island a's medium holds exactly its own entry" + ); + assert_eq!( + committed_file_count(&root_b), + 0, + "island b's medium holds exactly its own entry (b's content \ + routes kv)" + ); + + let (c, root_c) = factory.open_tracked().await.expect("a third vends fresh"); + assert_eq!( + committed_file_count(&root_c), + 0, + "a third open's medium is virgin" + ); + 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" + ); + + factory + .teardown() + .await + .expect("teardown removes the roots"); + factory.teardown().await.expect("teardown repeats cleanly"); +} + +/// 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 on a fresh (empty) medium (README §The factory +/// contract). +#[tokio::test] +async fn factory_teardown_is_safe_against_a_panicking_row() { + let content = patterned(2 * TUNED_THRESHOLD as usize + 7, 173); + let abandoned = Key::derive(&content, content.len() as u64); + let factory = Arc::new(DualTierLocalFactory::new(TUNED_THRESHOLD)); + + let task = { + let factory = Arc::clone(&factory); + tokio::spawn(async move { + let (store, _) = factory.open_tracked().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, root) = factory + .open_tracked() + .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" + ); + assert_eq!( + committed_file_count(&root), + 0, + "and its medium carries nothing of the abandoned run" + ); + factory + .teardown() + .await + .expect("teardown removes the last root"); +} diff --git a/tests/conformance/main.rs b/tests/conformance/main.rs index 4a3fb0c..d9cc769 100644 --- a/tests/conformance/main.rs +++ b/tests/conformance/main.rs @@ -15,11 +15,17 @@ //! skip-free reference shape) proves the contract implementable //! against the real type. //! -//! 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). +//! The reference 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`]); the columns' records live in +//! their module docs ([`README`](./README.md) §The reference row set +//! is the recorded inventory other columns mirror). +//! +//! The columns: [`mem_column`] — the contract-reference column; +//! [`local_column`] — the suite's first dual-tier column (the +//! `kv = mem` + `large = local` dispatch-coverage shape ADR-012 §6.3 +//! names), which mirrors the reference set and grows the +//! dispatch-coverage and local-specific rows. //! //! Items beyond the tests below are the later columns' surface, hence //! the scoped `#![allow(dead_code)]` — it shrinks as columns land. @@ -31,6 +37,8 @@ mod backlog; mod binding; mod factory; #[cfg(feature = "mem")] +mod local_column; +#[cfg(feature = "mem")] mod mem_column; mod rows; mod stamp; diff --git a/tests/conformance/rows.rs b/tests/conformance/rows.rs index 2b541cd..53a7909 100644 --- a/tests/conformance/rows.rs +++ b/tests/conformance/rows.rs @@ -52,24 +52,25 @@ use alkblobs::{ /// /// 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. +/// live-shared, never registration snapshots. `pub(crate)`: the +/// suite's other columns reuse these fixtures. #[derive(Default)] -struct SetSource { +pub(crate) struct SetSource { live: Mutex>, } impl SetSource { - fn register(store: &Store) -> Arc { + pub(crate) fn register(store: &Store) -> Arc { let source = Arc::new(SetSource::default()); store.register_liveness_source(source.clone()); source } - fn add(&self, key: Key) { + pub(crate) fn add(&self, key: Key) { self.live.lock().insert(key); } - fn remove(&self, key: &Key) { + pub(crate) fn remove(&self, key: &Key) { self.live.lock().remove(key); } } @@ -87,16 +88,16 @@ impl LivenessSource for SetSource { /// A set-backed protect callback; `None` models the failed-to-answer /// case (abort-as-data on the safe side). #[derive(Default)] -struct SetProtect { +pub(crate) struct SetProtect { set: Mutex>>, } impl SetProtect { - fn set_keys(&self, keys: Vec) { + pub(crate) fn set_keys(&self, keys: Vec) { *self.set.lock() = Some(keys); } - fn fail(&self) { + pub(crate) fn fail(&self) { *self.set.lock() = None; } } @@ -129,15 +130,15 @@ impl ProtectCallback for LatchProtect { /// 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 { +/// own chunking. `pub(crate)`: the suite's other columns reuse it. +pub(crate) struct Chunked { data: Vec, cursor: usize, sizes: std::collections::VecDeque, } impl Chunked { - fn new(data: &[u8], sizes: &[usize]) -> Chunked { + pub(crate) fn new(data: &[u8], sizes: &[usize]) -> Chunked { Chunked { data: data.to_vec(), cursor: 0, @@ -177,15 +178,15 @@ impl AsyncRead for Chunked { /// 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 { +/// injectable shape). `pub(crate)`: the suite's other columns reuse it. +pub(crate) struct Failing { data: Vec, cursor: usize, fail_after: usize, } impl Failing { - fn new(data: &[u8], fail_after: usize) -> Failing { + pub(crate) fn new(data: &[u8], fail_after: usize) -> Failing { Failing { data: data.to_vec(), cursor: 0, @@ -247,14 +248,18 @@ fn matrix() -> Vec> { ] } -fn patterned(len: usize, modulus: u32) -> Vec { +/// A content-shape generator: `len` bytes cycling against `modulus` — +/// deterministic, size-tunable content. `pub(crate)`: the suite's +/// other columns reuse it. +pub(crate) 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) { +/// deleted + cancelled_by_arbitration`). `pub(crate)`: the suite's +/// other columns reuse it. +pub(crate) fn assert_exact_counts(report: &alkblobs::SweepReport) { assert_eq!( report.candidates_staged, report.deleted + report.cancelled_by_arbitration,