alkblobs/src/backend.rs

//! The Backend trait contract (ADR-003): opaque byte keys, byte-valued
//! entries, complete `list()`, and the I/O seams every engine implements —
//! the read cursor and the staged put (ADR-010 §1).
//!
//! **The crate's second stable seam, crate-internal (ADR-003; the visibility
//! design point).** The trait contract is stable — engine implementations
//! never guess it — but it does not leave the crate: every engine ships
//! in-crate behind a feature flag and no consumer ever implements a
//! Backend (ADR-004's boundary; backends-and-dispatch.md §What the Backend
//! trait is). Exporting it later is an additive change with no persistence
//! or wire consequence, so `pub(crate)` costs nothing now and keeps the
//! public surface the facade's alone. The module itself is public so this
//! contract is documented.
//!
//! **One trait, two I/O abstractions.** There is no per-tier trait split
//! (ADR-010 §1): every engine implements the same method set, and the two
//! tiers differ only in the bodies behind the crate-internal abstractions —
//! a **read cursor** whose kv bodies materialize the whole value (bounded
//! by tier policy) and whose large bodies stream (`local` pread loop,
//! pg-lo `lo_get` windows), and a **staging cursor** for the staged put.
//! The store core programs the abstractions, never the engine shape. Both
//! body families serve the ADR-006 range contract — a `read_range` returns
//! the slice plus a SHA-256 slice digest, fixed by convention (the
//! `slice_digest` function below) — at one seam.
//!
//! **Engine posture:** namespace-blind (ADR-005) and virgin-store reads are
//! no-ops (missing table = absent/empty; POC #1 finding 2). Verification
//! never rides the trait: engines store opaque bytes under opaque keys;
//! the canonical derivation and its hash-check are store-core concerns
//! (ADR-002/006). GC state never rides this trait either (ADR-010 §2) —
//! the crate-internal engine-state companion seam lives in [`crate::gc`].
//!
//! The `#![allow(dead_code)]` below still has real scope: `Backend`
//! trait items with no in-tree lib caller (`Backend::name` — the
//! engines' reporting identity, consumed today by the engines' and
//! the seam tests) plus the wave-1 seam tests' doubles. The kv
//! engines' whole-value `put` gained its in-tree callers in wave 3
//! (sqlite is the default build's kv engine and the facade's put seam
//! drives the kv tier through it); the allow shrinks as the remaining
//! callers land.

#![allow(dead_code)]

use sha2::{Digest as _, Sha256};

use crate::Error;

/// The slice-digest convention (ADR-006): plain SHA-256 over the slice
/// bytes, fixed by convention so callers and ops handlers agree without
/// negotiation. Per-range verification against the canonical digest is
/// impossible under the preamble (the length is inside the hashed input,
/// ADR-002); transport-integrity is all a slice digest carries.
///
/// Crate-internal: this function is the convention's one seam-side
/// definition — the read cursor's range reads and the staged put's
/// verification pass both digest through it, and the facade's
/// `read_range` re-exports the result behind it. Callers verify
/// out-of-band by SHA-256-hashing the slice bytes themselves; there is
/// nothing to negotiate.
pub(crate) fn slice_digest(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

/// The read cursor — the abstraction [`Backend::get`] returns, driven by
/// the store core (ADR-010 §1). One abstraction, two body families: kv
/// bodies materialize the whole value (bounded by tier policy; `len` reads
/// the row's size column); large bodies stream (`local`: a pread loop over
/// the committed file — immutable once committed; pg-lo: descriptorless
/// `lo_get(oid, off, len)` windows over companion rows). Both serve the
/// ADR-006 range contract at one seam: [`ReadCursor::read_range`] is the
/// slice + slice-digest return the facade's `read_range` sits behind.
///
/// Crate-internal — not public API: store consumers see the facade's
/// `(len, stream)` shapes with this cursor behind them (store-api.md;
/// ADR-010 §1). The store's stream utility adapts engine cursor bodies; it
/// never exposes them.
pub(crate) trait ReadCursor: Send {
    /// The entry's length in bytes — a metadata-only probe, never a
    /// content load (ADR-008's size discipline): the row's size column on
    /// kv bodies, a file stat or companion-row read on large bodies. For
    /// a staging cursor ([`StagingCursor`]), the staged length so far —
    /// staging is append-only, so it grows with each append.
    fn len(&self) -> u64;

    /// The body's byte-level read: the bytes at `[offset, offset + len)`
    /// clamped to the entry's end — a pread-style short read at EOF.
    /// `offset >= len()` (EOF) yields an empty vec, never an error;
    /// callers wanting strictness check against [`ReadCursor::len`]. The
    /// only error family member this raises is media failure
    /// ([`Error::Io`]).
    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error>;

    /// The range-read seam both body families serve (ADR-006): the slice
    /// plus its slice digest — [`slice_digest`]'s SHA-256 over exactly
    /// the returned bytes, fixed by convention. The provided body
    /// composes `read_at` with the convention; a body override is a
    /// per-engine optimization and must keep the digest over exactly the
    /// returned bytes.
    async fn read_range(&self, offset: u64, len: u64) -> Result<(Vec<u8>, [u8; 32]), Error> {
        let bytes = self.read_at(offset, len).await?;
        let digest = slice_digest(&bytes);
        Ok((bytes, digest))
    }
}

/// The staged put's staging cursor (ADR-010 §1) — the named per-engine
/// contract of the large tier: the engine hands one out from
/// [`Backend::put_staged`]; content is appended; the store's verification
/// pass reads the eventual stored artifact back as a whole through the
/// same read seam ([`ReadCursor`] is the supertrait — ADR-003's
/// buffered-put ruling: the second derivation pass runs over the complete
/// staged bytes); commit publishes the entry. Every failure path converges
/// on discard with zero residue (the stage-hygiene invariant,
/// store-api.md §Write path; exact-count tested per engine).
///
/// **The key resolves at commit, not at handout (named design point,
/// the wave-2 amendment of the wave-1 handout shape).** The canonical
/// key is the derivation over preamble + content, and it completes at
/// stream end — after the staging cursor already exists (put paths
/// stream content once into the cursor while hashing; ADR-003's put
/// paths). So `commit` carries the store-core-resolved key and the
/// stage's identity before that point is only the engine-internal
/// stage file/tx (unique per `put_staged` handout). Publish is the
/// CAS first-writer-wins entry under that committed key exactly as
/// [`Backend::put`].
///
/// Every engine implements the staged form: kv-tier bodies are the
/// degenerate in-memory stage committed through the engine's whole-value
/// row write (the store core selects the form by tier policy, ADR-010
/// §1 — but the conformance suite runs stage hygiene on every engine,
/// store-api.md invariant 7).
pub(crate) trait StagingCursor: ReadCursor {
    /// Append a chunk to the stage. Unknown-length put streams append as
    /// content flows; the unknown-length flow's buffered prefix flushes
    /// into a fresh stage on threshold overflow, and the stream continues
    /// appending into it (ADR-003's buffered-put semantics — there is no
    /// "restart the stream" requirement anywhere).
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error>;

    /// Publish the entry: the stage's bytes become the CAS entry under
    /// `key` — the store-core-resolved canonical key (the derivation
    /// completes at stream end; see the type's design point) — in the
    /// engine's put shape (commit-rename on `local`, the in-tx `lowrite`
    /// batch on pg-lo, the row write on kv bodies). CAS first-writer-wins:
    /// a losing race against an existing entry is dedup (`Ok`), never a
    /// failure. This is the staged form's terminal step — the store core
    /// never re-uses the cursor after the call. On failure the stage is
    /// **already discarded** — zero residue by the stage-hygiene
    /// invariant; there is no post-failure discard to drive (the
    /// consuming call took the cursor).
    async fn commit(self, key: &[u8]) -> Result<(), Error>;

    /// Discard the stage: every non-commit path converges here, zero
    /// residue. Driven explicitly by the store core (stream error,
    /// verification failure, dispatch abandonment). The store core never
    /// drops a staging cursor without driving it to commit or discard;
    /// engine Drop bodies are best-effort residue cleanup, an engine-task
    /// concern (waves 4–5 own their blocking discipline there).
    async fn discard(self);
}

/// The engine contract — the crate's second stable seam (ADR-003),
/// crate-internal per the module docs' visibility design point.
///
/// **Opaque byte keys, opaque byte values.** Typed `Key` converts at the
/// store boundary only (ADR-002); backends must survive digest-layer
/// evolution because nothing here knows what a digest is — a kv row, a
/// `local` filename, or an LO companion row has no schema-migration
/// story, so self-describing bytes are the durable choice (POC #1
/// finding 1's generic-over-`K` alternative is rejected).
///
/// **One trait, every engine, same method set** (ADR-010 §1 — no per-tier
/// trait split). The tiers differ only in the bodies behind
/// [`Self::Cursor`] and [`Self::Staging`]. `list` is complete *and*
/// well-formed by contract (ADR-003): a malformed list is as deadly as an
/// incomplete one and is silent until a sweep deletes the wrong things,
/// so list correctness is gate-tested per engine through exact-count
/// sweep outcomes (store-api.md invariants 5/7). Namespace-blind, and
/// virgin-store reads are no-ops. GC state never rides this trait
/// (ADR-010 §2) — the store core reaches the fleet machinery through the
/// crate-internal engine-state companion seam in [`crate::gc`].
pub(crate) trait Backend: Send + Sync {
    /// This engine's read-cursor body (kv: materializing; large: streaming).
    type Cursor: ReadCursor;

    /// This engine's staging-cursor body.
    type Staging: StagingCursor;

    /// The engine's name — the dispatch/conformance reporting identity.
    fn name(&self) -> &'static str;

    /// The pure existence probe (ADR-008's `has` discipline). Virgin
    /// store: `false` — reads are no-ops, never "table does not exist"
    /// errors.
    async fn has(&self, key: &[u8]) -> Result<bool, Error>;

    /// The entry's read cursor, or `None` when absent — absence is a
    /// value on this path, never `Error::Missing` (ADR-012 §6.5; the
    /// facade's `get` sits directly behind). Virgin store: `None`.
    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error>;

    /// The whole-value put — the kv tier's settled form (ADR-010 §1): a
    /// **CAS write, first writer wins** (`INSERT OR IGNORE` /
    /// `ON CONFLICT DO NOTHING` / `entry().or_insert()`). An already
    /// present key is a no-op — content is immutable, so a re-put is
    /// dedup, never overwrite. Hash verification is the store core's;
    /// this method never sees a digest.
    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error>;

    /// The staged put's cursor handout (ADR-010 §1). The store core
    /// selects the form before content flows (the cursor's identity
    /// before commit is the engine-internal stage only — the canonical
    /// key resolves at [`StagingCursor::commit`], the type's design
    /// point). See [`StagingCursor`] for the contract every body
    /// converges on.
    async fn put_staged(&self) -> Result<Self::Staging, Error>;

    /// Unconditional delete — arbitration has already happened above the
    /// trait (the facade decides and owns the refusal: `GcRefuse` never
    /// fires below the facade, the error module's altitude discipline).
    /// `Err(Error::Missing)` when the key is absent (the facade's
    /// `delete` carries `Missing`, ADR-012 §6.5); virgin store:
    /// `Err(Error::Missing)`.
    async fn delete(&self, key: &[u8]) -> Result<(), Error>;

    /// Whole-engine key enumeration — complete and well-formed by
    /// contract (ADR-003; a non-enumerating engine must not ship, the
    /// rudolfs anti-lesson). Ordering unspecified; the store core sorts
    /// for its live-set diffs. Virgin store: empty.
    async fn list(&self) -> Result<Vec<Vec<u8>>, Error>;

    /// The metadata-only length probe (ADR-008): the indexed size column
    /// on kv engines, a file stat on `local`, a companion-row probe on
    /// pg-lo — never a value/content load. Store-level `stat` and pool
    /// accounting ride this method. `None` when absent; virgin store:
    /// `None`.
    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error>;
}

/// Clamp a `[offset, offset + len)` request by `u64` read parameters into
/// `usize` slice bounds against a body of `total` bytes — the shared
/// EOF-clamping arithmetic of the cursor contract. Crate-internal: the
/// engine cursor bodies behind [`Backend::Cursor`] compose this, so every
/// engine's materializing and staging bodies clamp identically.
///
/// The arithmetic runs in `u64` — the offset/len domain — and narrows to
/// `usize` only after clamping, so a truncated `as usize` cast can never
/// wrap a huge offset back onto a small slice position on narrow-pointer
/// targets.
pub(crate) fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
    let total = total as u64;
    let start = offset.min(total);
    let end = start.saturating_add(len).min(total);
    (start as usize, end as usize)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeMap;
    use std::sync::Arc;

    use parking_lot::Mutex;

    const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";

    /// Clamp a range request to a cursor body's whole length — the test
    /// doubles' shared read arithmetic.
    fn body_slice(whole: &[u8], offset: u64, len: u64) -> Vec<u8> {
        let (start, end) = clamped(offset, len, whole.len());
        whole[start..end].to_vec()
    }

    /// kv body family: the whole value materialized (the row read's shape).
    struct MaterializedCursor {
        value: Vec<u8>,
    }

    impl ReadCursor for MaterializedCursor {
        fn len(&self) -> u64 {
            self.value.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.value, offset, len))
        }
    }

    /// Large body family: a bounded pread-style assembly from segments —
    /// the value is never handed to the caller as one materialized unit,
    /// reads gather through the loop.
    struct StreamingCursor {
        whole: Vec<u8>,
        segment: usize,
    }

    impl ReadCursor for StreamingCursor {
        fn len(&self) -> u64 {
            self.whole.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            let (start, end) = clamped(offset, len, self.whole.len());
            let mut out = Vec::with_capacity(end - start);
            let mut pos = start;
            while pos < end {
                let next = (((pos / self.segment) + 1) * self.segment).min(end);
                out.extend_from_slice(&self.whole[pos..next]);
                pos = next;
            }
            Ok(out)
        }
    }

    /// The full cursor contract every body family must satisfy — run
    /// against both body-family doubles.
    async fn assert_cursor_contract<C: ReadCursor>(cursor: &C, whole: &[u8]) {
        assert_eq!(
            cursor.len() as usize,
            whole.len(),
            "len is the entry length, metadata-only"
        );
        let len = whole.len() as u64;

        assert_eq!(cursor.read_at(0, len).await.unwrap(), whole, "full read");
        assert_eq!(
            cursor.read_at(0, len + 17).await.unwrap(),
            whole,
            "over-read clamps at EOF, pread-style"
        );
        let mid = (whole.len() / 2) as u64;
        assert_eq!(
            cursor.read_at(mid, len).await.unwrap(),
            &whole[mid as usize..],
            "tail read clamps"
        );
        assert_eq!(
            cursor.read_at(len, 5).await.unwrap(),
            Vec::<u8>::new(),
            "read at EOF is empty, never an error"
        );
        assert_eq!(
            cursor.read_at(u64::MAX, 5).await.unwrap(),
            Vec::<u8>::new(),
            "an overflowing u64 offset clamps to EOF in u64 arithmetic — never wraps"
        );
        assert_eq!(
            cursor.read_at(0, u64::MAX).await.unwrap(),
            whole,
            "an overflowing u64 length clamps to the whole body"
        );
        assert_eq!(
            cursor.read_at(0, 0).await.unwrap(),
            Vec::<u8>::new(),
            "zero-length read"
        );

        let (start, end) = clamped(mid, 7.min(len.max(1)), whole.len());
        let (slice, digest) = cursor.read_range(mid, 7.min(len.max(1))).await.unwrap();
        assert_eq!(
            slice,
            &whole[start..end],
            "read_range slice == read_at bytes"
        );
        let expected: [u8; 32] = Sha256::digest(&slice).into();
        assert_eq!(
            digest, expected,
            "slice digest is SHA-256 over exactly the returned bytes"
        );

        let (full, full_digest) = cursor.read_range(0, len).await.unwrap();
        assert_eq!(full, whole);
        assert_eq!(full_digest, slice_digest(whole));
    }

    #[tokio::test]
    async fn cursor_contract_materialized_body() {
        assert_cursor_contract(
            &MaterializedCursor {
                value: CONTENT.to_vec(),
            },
            CONTENT,
        )
        .await;
    }

    #[tokio::test]
    async fn cursor_contract_streaming_body() {
        assert_cursor_contract(
            &StreamingCursor {
                whole: CONTENT.to_vec(),
                segment: 8,
            },
            CONTENT,
        )
        .await;
    }

    #[tokio::test]
    async fn cursor_contract_empty_entry() {
        assert_cursor_contract(&MaterializedCursor { value: Vec::new() }, b"").await;
        assert_cursor_contract(
            &StreamingCursor {
                whole: Vec::new(),
                segment: 8,
            },
            b"",
        )
        .await;
    }

    /// Both body families serve the ADR-006 range contract at one seam:
    /// identical slices and digests for identical ranges.
    #[tokio::test]
    async fn both_body_families_serve_the_range_seam_identically() {
        let materialized = MaterializedCursor {
            value: CONTENT.to_vec(),
        };
        let streaming = StreamingCursor {
            whole: CONTENT.to_vec(),
            segment: 9,
        };
        for (offset, len) in [(0u64, 0u64), (0, 5), (11, 9), (43, 4), (43, 400)] {
            assert_eq!(
                materialized.read_range(offset, len).await.unwrap(),
                streaming.read_range(offset, len).await.unwrap(),
                "range {offset}+{len} identical across body families"
            );
        }
    }

    /// The slice-digest convention is plain SHA-256 over the slice bytes
    /// (ADR-006), asserted against published SHA-256 vectors independent
    /// of this crate's own hashing (the POC lesson: known-answer tests
    /// are only as trustworthy as their source of truth — NIST's vectors
    /// here, git's for the canonical derivation).
    #[test]
    fn slice_digest_convention_matches_published_vectors() {
        assert_eq!(
            hex::encode(slice_digest(b"abc")),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
            "sha256 'abc' (NIST)"
        );
        assert_eq!(
            hex::encode(slice_digest(b"")),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "sha256 empty (NIST)"
        );
    }

    /// The test-double backend: a BTreeMap CAS table whose bodies are the
    /// shapes the trait shapes must compose — a materializing cursor body,
    /// a degenerate in-memory staging body, first-writer-wins puts.
    #[derive(Clone, Default)]
    struct TestBackend {
        rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
        fail_commit: bool,
    }

    impl TestBackend {
        fn row(&self, key: &[u8]) -> Option<Vec<u8>> {
            self.rows.lock().get(key).cloned()
        }
    }

    struct TestCursor {
        value: Vec<u8>,
    }

    impl ReadCursor for TestCursor {
        fn len(&self) -> u64 {
            self.value.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.value, offset, len))
        }
    }

    struct TestStaging {
        staged: Vec<u8>,
        rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
        fail_commit: bool,
    }

    impl ReadCursor for TestStaging {
        fn len(&self) -> u64 {
            self.staged.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.staged, offset, len))
        }
    }

    impl StagingCursor for TestStaging {
        async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
            self.staged.extend_from_slice(chunk);
            Ok(())
        }

        async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
            if self.fail_commit {
                self.staged.clear();
                return Err(Error::Io("engine commit failure (test double)".to_owned()));
            }
            let value = std::mem::take(&mut self.staged);
            self.rows.lock().entry(key.to_vec()).or_insert(value);
            Ok(())
        }

        async fn discard(mut self) {
            self.staged.clear();
        }
    }

    impl Backend for TestBackend {
        type Cursor = TestCursor;
        type Staging = TestStaging;

        fn name(&self) -> &'static str {
            "test-double"
        }

        async fn has(&self, key: &[u8]) -> Result<bool, Error> {
            Ok(self.rows.lock().contains_key(key))
        }

        async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
            Ok(self
                .rows
                .lock()
                .get(key)
                .cloned()
                .map(|value| TestCursor { value }))
        }

        async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
            self.rows
                .lock()
                .entry(key.to_vec())
                .or_insert_with(|| value.to_vec());
            Ok(())
        }

        async fn put_staged(&self) -> Result<Self::Staging, Error> {
            Ok(TestStaging {
                staged: Vec::new(),
                rows: self.rows.clone(),
                fail_commit: self.fail_commit,
            })
        }

        async fn delete(&self, key: &[u8]) -> Result<(), Error> {
            self.rows
                .lock()
                .remove(key)
                .map(|_| ())
                .ok_or(Error::Missing)
        }

        async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
            Ok(self.rows.lock().keys().cloned().collect())
        }

        async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
            Ok(self.rows.lock().get(key).map(|value| value.len() as u64))
        }
    }

    /// The Backend trait's method contract against the double: the seven
    /// methods' absence/value/error disciplines, CAS first-writer-wins,
    /// and virgin-store reads as no-ops.
    #[tokio::test]
    async fn backend_method_contract() {
        let backend = TestBackend::default();
        let key = b"\x01opaque-key-bytes".to_vec();
        assert_eq!(backend.name(), "test-double", "reporting identity");

        assert!(
            !backend.has(&key).await.unwrap(),
            "virgin store: has is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "virgin store: get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "virgin store: size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "virgin store: list is empty"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "virgin store: delete reports Missing"
        );

        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "put lands");
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64),
            "size is metadata-only and exact"
        );
        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            backend.row(&key).as_deref(),
            Some(CONTENT),
            "CAS first-writer wins: re-put is a no-op"
        );
        assert_eq!(
            backend.list().await.unwrap(),
            vec![key.clone()],
            "list is complete, keys not values"
        );

        backend.delete(&key).await.unwrap();
        assert!(!backend.has(&key).await.unwrap(), "delete lands");
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete reports Missing when absent"
        );
    }

    /// The staged put's happy path: handout, appends, the verification
    /// pass reading the staged artifact back as a whole through the read
    /// seam, commit publishing, and the committed entry round-tripping
    /// through get's cursor at the range seam.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let backend = TestBackend::default();
        let key = b"\x01staged-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(&CONTENT[..8]).await.unwrap();
        staging.append(&CONTENT[8..16]).await.unwrap();
        staging.append(&CONTENT[16..]).await.unwrap();

        assert_eq!(
            staging.len() as usize,
            CONTENT.len(),
            "staged length is the appended length"
        );
        let (bytes, digest) = staging
            .read_range(0, staging.len())
            .await
            .expect("staged bytes readable");
        assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
        assert_eq!(digest, slice_digest(CONTENT));

        staging.commit(&key).await.expect("commit publishes");
        assert!(backend.has(&key).await.unwrap(), "entry published");

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let (bytes, digest) = cursor
            .read_range(0, cursor.len())
            .await
            .expect("committed read");
        assert_eq!(bytes, CONTENT);
        assert_eq!(digest, slice_digest(CONTENT));

        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            backend.row(&key).as_deref(),
            Some(CONTENT),
            "staged commit also lands CAS-first-writer"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — nothing was staged into the pool.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let backend = TestBackend::default();
        let key = b"\x01discard-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"residue").await.unwrap();
        staging.discard().await;

        assert!(
            !backend.has(&key).await.unwrap(),
            "discarded stage published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the pool holds exactly zero entries after discard"
        );
    }

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — zero residue, entry unpublished, no post-failure
    /// discard left to drive (the consuming call took the cursor).
    #[tokio::test]
    async fn commit_failure_self_discards() {
        let backend = TestBackend {
            fail_commit: true,
            ..Default::default()
        };
        let key = b"\x01failure-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let err = staging.commit(&key).await.unwrap_err();
        assert!(matches!(err, Error::Io(_)), "engine media failure");

        assert!(
            !backend.has(&key).await.unwrap(),
            "failed commit published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
    }
}