alkblobs/tests/conformance/local_column.rs

//! 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<Vec<PathBuf>>,
}

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<Store, Error> {
        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<PathBuf>) {
        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::<u8>::new(),
            "a read at EOF is empty, never an error"
        );
        assert_eq!(
            blob.read_at(mid, 0).await.expect("the empty read succeeds"),
            Vec::<u8>::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::<Key>::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");
}