//! 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::PathBuf; use std::sync::{Arc, Mutex}; use crate::factory::StoreFactory; use crate::rows::{ self, assert_exact_counts, committed_file_count, patterned, slice_digest, staging_file_count, 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 live in `rows` (the medium-backed // columns' shared oracle); kv = mem never touches this disk, so the // counts below are exactly the large tier's committed/staging // residency. // ------------------------------------------------------------------ // ------------------------------------------------------------------ // 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"); }