alkdev/alkblobs
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): the Backend trait, I/O seams, engine-state seam shape, and facade value types (ADR-003/010)

Browse Source 
The wave-1 contract surface: everything engine-blind that waves 2-6
implement against. No engine bodies, no facade behavior logic.

- src/backend.rs: the pub(crate) Backend trait — seven methods +
  put_staged, opaque byte keys/values, size metadata-only, list
  complete-and-well-formed, virgin-store reads as no-ops (ADR-003,
  ADR-008); the ReadCursor/StagingCursor I/O seams with the read_range
  slice-digest convention (ADR-006) at one seam; unit tests against
  both body-family doubles, a test-double engine, and the stage-hygiene
  invariant's two forms (discard; commit failure self-discards).
- src/gc.rs: the engine-state companion seam as EngineWindowState /
  EngineFleetState — ADR-012 §5's applicability table expressed
  structurally (no dead machinery); JointTx ownership rule pinned in
  the signature (store holds the tx; engine contributes mechanics).
- src/facade.rs: EntryMeta, Pin, Batch, SweepReport, GcAbortCause
  (x3, #[non_exhaustive]), PinState with lifecycle contracts
  (ADR-005, ADR-012 §1); re-exported from lib.rs.
- Design points pinned for the wave-1 review gate in
  tasks/core-trait-and-types.md: pub(crate) trait visibility,
  associated-type put-form shapes, AFIT + associated types (no dyn).

Verification: cargo test (26 lib + 3 git-interop), cargo test
--all-features, clippy --all-targets -D warnings (default and
--all-features), cargo fmt --check, cargo doc --no-deps — all green.
This commit is contained in:
glm-5.3-flash committed 2026-10-11 18:30:47 +00:00
1 parent fef05b4ec5
commit 0547c17876
5 files changed
+1295 -16
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/backend.rs
+688 -3
View File
@@ -1,3 +1,688 @@
//! The Backend trait contract (ADR-003): opaque byte keys and byte-valued
//! entries, complete `list()`, and the read-cursor / staged-put I/O seams
//! (ADR-010) with the crate-internal engine-state companion seam.
//! 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 marks the seam's scope: these items are
//! the wave-1 contract surface, defined ahead of their drivers (the store
//! core and engines of waves 2–5). It shrinks as those waves 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).
///
/// 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 in the
/// engine's put shape (commit-rename on `local`, the in-tx `lowrite`
/// batch on pg-lo, the row write on kv bodies). 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) -> 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 large tier's
/// stage-then-commit-rename contract, kv engines' degenerate
/// in-memory stage. See [`StagingCursor`] for the contract every
/// body converges on.
async fn put_staged(&self, key: &[u8]) -> 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.
fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
let start = (offset as usize).min(total);
let end = start.saturating_add(len as usize).min(total);
(start, end)
}
#[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(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 {
key: Vec<u8>,
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) -> 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(self.key.clone()).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, key: &[u8]) -> Result<Self::Staging, Error> {
Ok(TestStaging {
key: key.to_vec(),
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(&key).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().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(&key).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(&key).await.unwrap();
staging.append(CONTENT).await.unwrap();
let err = staging.commit().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");
}
}