diff --git a/src/backend.rs b/src/backend.rs index 0b8ef5e..79a7e2f 100644 --- a/src/backend.rs +++ b/src/backend.rs @@ -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, 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; 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; + + /// 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, 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; + + /// 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>, 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, 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 { + 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, + } + + impl ReadCursor for MaterializedCursor { + fn len(&self) -> u64 { + self.value.len() as u64 + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, 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, + 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, 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(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::::new(), + "read at EOF is empty, never an error" + ); + assert_eq!( + cursor.read_at(0, 0).await.unwrap(), + Vec::::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, Vec>>>, + fail_commit: bool, + } + + impl TestBackend { + fn row(&self, key: &[u8]) -> Option> { + self.rows.lock().get(key).cloned() + } + } + + struct TestCursor { + value: Vec, + } + + impl ReadCursor for TestCursor { + fn len(&self) -> u64 { + self.value.len() as u64 + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + Ok(body_slice(&self.value, offset, len)) + } + } + + struct TestStaging { + key: Vec, + staged: Vec, + rows: Arc, Vec>>>, + 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, 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 { + Ok(self.rows.lock().contains_key(key)) + } + + async fn get(&self, key: &[u8]) -> Result, 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 { + 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>, Error> { + Ok(self.rows.lock().keys().cloned().collect()) + } + + async fn size(&self, key: &[u8]) -> Result, 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"); + } +} diff --git a/src/facade.rs b/src/facade.rs index 8b21ffa..69216b0 100644 --- a/src/facade.rs +++ b/src/facade.rs @@ -1,2 +1,283 @@ -//! The store facade value types: `EntryMeta`, `Pin`/`Batch`, -//! `SweepReport`, `GcAbortCause`, and `PinState`. +//! The store facade's value types (store-api.md §Public surface): the +//! shapes put/get/stat/sweep return and the pin lifecycle vocabulary. +//! Engine-blind — nothing here knows which engines a store runs (ADR-007/ +//! 008/010); these types are the contract wave 2's facade behavior +//! implements and every consumer programs against. +//! +//! The pin types ([`Pin`], [`Batch`]) declare the ADR-005 lifecycle +//! contract; the arbitration machinery behind drops and conversions is +//! wave 2's facade logic. The contracts are pinned here so the guard +//! semantics never change shape — wave 2 wires the behavior, not the +//! surface. + +use std::time::SystemTime; + +use crate::hashing::Algorithm; +use crate::keys::Key; + +/// The cheap stat probe's result (store-api.md §Read path): the entry's +/// length and its hash algorithm — the "type" at this layer is the hash +/// algorithm, there is no other (ADR-002). `stat` rides the Backend +/// trait's `size` method (ADR-008): an indexed-column probe on kv +/// engines, a file stat on `local`, a companion-row probe on pg-lo — +/// never a value/content load. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EntryMeta { + /// The entry's length in bytes. + pub len: u64, + /// The key's hash algorithm — ADR-002's two-member policy. + pub algorithm: Algorithm, +} + +/// The RAII put-path pin guard (ADR-005 §Marks 2; store-api.md §Write +/// path). +/// +/// **Lifecycle contract (pinned here; the machinery is wave 2):** +/// +/// - Acquired **before the entry becomes visible** — pin-before-publish: +/// an entry cannot appear in a sweep's `list()` without already +/// carrying its pin (or its converted registration, ADR-005's put-side +/// invariant). +/// - The guard holds a per-key refcount under the same arbitration lock +/// the sweep consults, until **dropped or converted into a registered +/// consumer reference** (a registered liveness source or a root tag). +/// - **Conversion ordering (ADR-005, pinned):** the pin's drop and its +/// replacement's registration into a liveness source are one ordered +/// step under the same arbitration lock — a consumer may not release a +/// pin before its registered source observes the replacement (the +/// local analog of the ops-layer pin-token contract). +/// - A dedup put's guard protects exactly as a fresh put's does: a +/// re-put acquires no new pool entry but refcounts the existing one +/// under the same arbitration lock (ADR-005). +/// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the +/// joint tx commits the expiry-carrying pin row and this guard drops +/// at response time (the row's expiry is then the protection); in +/// non-fleet modes the ops layer's token is this guard's in-process +/// handle — nothing expires outside fleet mode. +pub struct Pin { + key: Key, +} + +impl Pin { + /// Store-core constructor: pins are minted by the put path, never + /// held by consumers before a put exists — the guard's meaning is + /// the put's protection (wave 2 wires the refcount acquisition under + /// the arbitration lock). + #[allow(dead_code)] // wave 2's put path wires the call sites; exercised by the tests + pub(crate) fn new(key: Key) -> Pin { + Pin { key } + } + + /// The key whose entry the guard protects. + pub fn key(&self) -> &Key { + &self.key + } +} + +/// Batch-scoped pin group (ADR-005 §Marks 2; store-api.md): multi-put +/// writes — manifest writes are exactly this — group under one +/// batch-scoped pin. +/// +/// **Lifecycle contract (pinned here; the machinery is wave 2):** +/// +/// - Created by the store core's batching put seam; each put within the +/// batch contributes its pin (batch-internal puts publish +/// pin-before-publish exactly as lone puts do). +/// - **A batch's pins drop together on batch commit or batch drop**, in +/// one ordered step under the same arbitration lock the sweep +/// consults. No early single-pin release: protection spans the whole +/// batch or none of it. +/// - Batch drop is the failure path (stream error, caller abandonment): +/// every member pin releases together; member liveness never outlives +/// the batch. +pub struct Batch { + pins: Vec, +} + +impl Batch { + /// Store-core constructor (wave 2's batching seam wires the + /// arbitration-lock acquisition). + #[allow(dead_code)] // wave 2's batching path wires the call sites; exercised by the tests + pub(crate) fn new() -> Batch { + Batch { pins: Vec::new() } + } + + /// Contribute a put's pin to the group (store-core call). + #[allow(dead_code)] // wave 2's batching path drives this; exercised by the tests + pub(crate) fn add(&mut self, pin: Pin) { + self.pins.push(pin); + } + + /// How many pins the group currently protects. + pub fn len(&self) -> usize { + self.pins.len() + } + + /// Whether the group currently protects nothing. + pub fn is_empty(&self) -> bool { + self.pins.is_empty() + } +} + +/// A sweep's abort cause — report data only, never an error (ADR-012 §1: +/// sweeps are never errors; an aborted sweep returns `Ok(SweepReport)` +/// with `aborted: Some(cause)`, and there is no sweep-abort error +/// variant). +/// +/// Exactly three variants, ADR-pinned; new causes are new ADRs. +/// `#[non_exhaustive]` makes the ADR-012 §1 ruling — callers match +/// non-exhaustively — structural for downstream crates. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum GcAbortCause { + /// A protection source failed to answer (ADR-005's protect-callback + /// failure): the sweep aborted, nothing deleted. + ProtectFailed, + /// The fleet sweep lease was contended (ADR-008; ADR-013 §2's + /// exclusive-lease semantics): the second sweeper aborts — nothing + /// deleted; its next run takes the lock. + SweeperLock, + /// The safe default: a sweep invoked with zero registered liveness + /// sources aborts without deleting (store-api.md §Lifecycle). + NoLivenessSources, +} + +/// A key's pin classification (ADR-010 §2): the engine's only vocabulary +/// for pins — the engine never decides deletion, it only classifies, and +/// the store core receives the answer and applies the decision. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PinState { + /// An un-expired pin row exists for the key — the key is protected. + Protected, + /// Nothing pins the key. + Unprotected, +} + +/// A sweep's outcome record — always data, never an error carrier +/// (ADR-012 §1). Counts are exact; the test gate asserts sweep outcomes +/// against these fields (store-api.md invariants 4/7). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SweepReport { + /// Candidates staged into the delete window at the mark phase. + pub candidates_staged: u64, + /// Candidates actually deleted at the commit phase — plus, on the + /// sweep-as-executor path, the overdue staged-window candidates this + /// sweep re-arbitrated and executed (ADR-012 §6.7); `0` on any abort. + pub deleted: u64, + /// Candidates cancelled at delete time by arbitration — the put side + /// of the sweep-vs-put race won, and protection composed over the + /// candidate (ADR-005's delete-window protocol). + pub cancelled_by_arbitration: u64, + /// `Some(cause)` when the sweep aborted — nothing was deleted; + /// `None` on a completed sweep. + pub aborted: Option, + /// Wall-clock start of the run. + pub started_at: SystemTime, + /// Wall-clock finish — an aborted sweep also finishes; the report is + /// its outcome record either way. + pub finished_at: SystemTime, +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hashing::Digest; + + /// The stat value's shape: length plus the key's algorithm — there + /// is no other type at this layer (store-api.md §Read path). + #[test] + fn entry_meta_carries_len_and_algorithm() { + let key = Key::from(Digest::derive(b"meta", 4)); + let meta = EntryMeta { + len: 4, + algorithm: key.algorithm(), + }; + assert_eq!(meta.len, 4); + assert_eq!(meta.algorithm, Algorithm::GitBlobSha256); + assert_eq!(meta.algorithm.digest_len(), 32); + } + + /// The abort vocabulary's exhaustiveness shape: exactly three + /// variants, ADR-pinned (ADR-012 §1). The in-crate match below is + /// exhaustive without a wildcard — adding a variant breaks it, + /// which is the "new causes are new ADRs" discipline surfacing at + /// compile time in-crate; `#[non_exhaustive]` enforces the + /// wildcard discipline on downstream callers. + #[test] + fn gc_abort_cause_is_exactly_three_variants() { + let mut matched: Vec<&'static str> = Vec::new(); + for cause in [ + GcAbortCause::ProtectFailed, + GcAbortCause::SweeperLock, + GcAbortCause::NoLivenessSources, + ] { + match cause { + GcAbortCause::ProtectFailed => matched.push("protect"), + GcAbortCause::SweeperLock => matched.push("lease"), + GcAbortCause::NoLivenessSources => matched.push("sources"), + } + } + assert_eq!(matched, ["protect", "lease", "sources"]); + } + + /// Report field semantics: exact counts, `aborted: None` on a + /// completed run, and the abort shape — nothing deleted, the cause + /// carried, the report still a finished record. + #[test] + fn sweep_report_carries_exact_counts_and_abort_data() { + let now = SystemTime::now(); + let completed = SweepReport { + candidates_staged: 3, + deleted: 2, + cancelled_by_arbitration: 1, + aborted: None, + started_at: now, + finished_at: now, + }; + assert!(completed.aborted.is_none()); + assert!(completed.finished_at >= completed.started_at); + + let aborted = SweepReport { + deleted: 0, + cancelled_by_arbitration: 0, + aborted: Some(GcAbortCause::NoLivenessSources), + ..completed + }; + assert_eq!(aborted.candidates_staged, 3, "mark-phase count is kept"); + assert_eq!(aborted.deleted, 0, "an aborted sweep deletes nothing"); + assert_eq!( + aborted.aborted, + Some(GcAbortCause::NoLivenessSources), + "the safe default cause is report data" + ); + } + + /// Pin/Batch shape invariants: the guard holds its key and the batch + /// groups pins. Wave 2 wires the refcount machinery; the shape these + /// tests pin is what the guard carries and what drops together. + #[test] + fn pin_guard_holds_its_key_and_batch_groups_pins() { + let pin = Pin::new(Key::from(Digest::derive(b"guarded", 7))); + assert_eq!(pin.key().algorithm(), Algorithm::GitBlobSha256); + drop(pin); + + let mut batch = Batch::new(); + assert!(batch.is_empty(), "fresh batch protects nothing yet"); + batch.add(Pin::new(Key::from(Digest::derive(b"a", 1)))); + batch.add(Pin::new(Key::from(Digest::derive(b"b", 1)))); + assert_eq!(batch.len(), 2, "member pins counted"); + drop(batch); + } + + /// The arbitration vocabulary: two states, and the engine's reading + /// is classification only (ADR-010 §2). + #[test] + fn pin_state_is_the_two_variant_classification() { + assert_ne!(PinState::Protected, PinState::Unprotected); + for state in [PinState::Protected, PinState::Unprotected] { + match state { + PinState::Protected | PinState::Unprotected => {} + } + } + } +} diff --git a/src/gc.rs b/src/gc.rs index 8875e51..dc4684b 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -1,2 +1,192 @@ -//! GC and namespace mechanism (ADR-005/012): registered liveness, -//! pin-before-publish, delete windows, and sweep outcomes as data. +//! GC state machinery (ADR-005/012): the crate-internal engine-state +//! companion seam (ADR-010 §2, as reduced by ADR-012 §5) and — in wave 2 — +//! the sweep/liveness mechanism the store core drives through it. +//! +//! **The seam's shape is ADR-012 §5's per-engine applicability table, +//! expressed structurally — there is no dead machinery.** The window +//! family (GC schema + staged delete window) is `EngineWindowState`, +//! implemented by every SQL-backed engine: postgres, sqlite, pg-lo. The +//! fleet family (pin clauses + the sweep lease) is `EngineFleetState`, +//! implemented by the postgres host only — all fleet GC state hosts on +//! the fleet's kv-tier engine (ADR-012 §4), sqlite's pin/sweep-lock +//! bodies were dropped as dead machinery (ADR-012 §5), and a sqlite +//! fleet is a recorded non-option (ADR-010 §3). `local` and `mem` are +//! excluded by construction: they stage delete-window candidates +//! in-process and execute the ADR-005 protocol inline within the same +//! sweep, and they are never fleet engines. +//! +//! **Ownership rule (ADR-010 §2, verbatim posture):** the store core owns +//! all GC state — the in-process pin and liveness maps and the fleet +//! state's rows and locks. Engines execute clauses; the store decides. +//! The public-side `Backend` trait (module [`crate::backend`]) stays +//! "plain content-addressed rows, nothing else": pins, window rows, and +//! the sweep lease are invisible at the public trait, and backends never +//! learn liveness (ADR-005). Every clause is contract-tested per its +//! applicability row in waves 3–5; `local`/`mem` are contract-excluded. +//! +//! The `#![allow(dead_code)]` below marks the seam's scope: its clauses are +//! implemented by the SQL engines of waves 3–5 and are not called yet. It +//! shrinks as those waves land. + +#![allow(dead_code)] + +use std::time::SystemTime; + +use crate::backend::Backend; +use crate::facade::PinState; +use crate::Error; + +/// A delete-window candidate staged into the durable window: the key and +/// the time it becomes visible for deletion — exactly the +/// `gc_window (key, visible_after)` row pair (ADR-005's delete-window +/// protocol; ADR-008 §2). +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct WindowCandidate { + /// The opaque key bytes (the seam's boundary discipline: opaque byte + /// keys, like the `Backend` trait). + pub key: Vec, + /// When the candidate becomes deletable (the staged window's + /// expiry; the calling sweep passes `now` to [`EngineWindowState::window_take`] + /// and executes whatever has become overdue). + pub visible_after: SystemTime, +} + +/// A fleet pin row to commit inside the joint entry+pin transaction — +/// exactly the `gc_pins (key, owner, expiry)` row triple (ADR-008; +/// ADR-010 §2). `owner` is the **store-instance id** (ADR-011's +/// vocabulary): the row's identity is `(key, owner)`, a dedup re-put by +/// a different instance is a different row, and protection composes by +/// row — an entry is protected while any un-expired pin row exists. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct PinRow { + /// The opaque key bytes of the entry being pinned. + pub key: Vec, + /// The store-instance id minting the pin (ADR-011's vocabulary). + pub owner: Vec, + /// The row's expiry; the TTL and the renewal schedule (every TTL/3) + /// are the store's, not the clause's (ADR-012 §2's rename). + pub expiry: SystemTime, +} + +/// The engine-state companion seam's **window family**: the GC schema +/// clause and the staged delete window — on every SQL-backed engine +/// (ADR-010 §2; ADR-012 §5's table: postgres, sqlite, pg-lo). Postgres is +/// fleet-active on these when declared; sqlite stages durably +/// single-node; pg-lo hosts the window where the kv engine is not +/// SQL-backed — under the one-window-host rule (at most one SQL engine +/// per store instance carries an active window; the others' clauses are +/// dormant). Crate-internal — never exported; the store core calls the +/// clauses, engines execute the SQL, and each clause is contract-tested +/// per its applicability row. +pub(crate) trait EngineWindowState: Backend { + /// Create the GC table(s) this engine's applicability row names: + /// postgres — `gc_pins` + `gc_window` (the fleet host); sqlite and + /// pg-lo — `gc_window` only (ADR-012 §5). Called at store + /// construction; dialect-specific bodies, engine-neutral call. The + /// embedder's liveness table is never created here — the sweep reads + /// it and never owns it (ADR-008's schema-ownership rule). + async fn ensure_gc_schema(&self) -> Result<(), Error>; + + /// Stage candidates into this engine's durable delete window (idempotent + /// staging per `(key, deadline)` is the body's discipline; the store + /// stages each candidate once per sweep). Virgin store: creates the + /// window lazily or errors nothing — reads and no-op writes stay + /// no-ops (the Backend trait's engine posture carries here). + async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error>; + + /// Take overdue candidates (`visible_after <= now`): the rows leave + /// the window table and return as keys for the calling sweep to + /// re-arbitrate and delete — the executor is the next sweep (ADR-012 + /// §6.7). Empty when nothing is overdue; virgin store: empty. + async fn window_take(&self, now: SystemTime) -> Result>, Error>; +} + +/// The engine-state companion seam's **fleet family**: the pin clauses +/// and the sweep lease — the postgres host only (ADR-012 §4: all fleet +/// GC state hosts on the fleet's kv-tier engine, large tier included; +/// ADR-012 §5: sqlite's and pg-lo's pin/sweep-lock bodies are dropped). +/// Fleet activation is a constructor declaration (`fleet: true`), never +/// inferred from the engine choice (ADR-012 §3): under `fleet: false` +/// none of this runs and the ADR-005 in-process protocol applies +/// verbatim, regardless of engine. Crate-internal — never exported. +pub(crate) trait EngineFleetState: EngineWindowState { + /// The joint entry+pin transaction — the declared holder type whose + /// concrete payloads are per-engine bodies (waves 3–5: the pg + /// transaction the pooled client carries; in the pg-lo consolidation + /// posture, the same tx the LO writes ride, ADR-012 §4). + /// + /// **Ownership rule, pinned in this signature:** the store core + /// holds the tx from [`Self::begin_joint_tx`] until it hands the + /// value to [`Self::pin_commit`] (which consumes it as the joint + /// terminal step); the engine contributes mechanics into the tx and + /// never owns it. This is ADR-010 §2's ruling — "the store holds the + /// tx; the engine contributes mechanics" — which corrected ADR-009's + /// put-transaction ownership statement. + type JointTx: Send; + + /// Begin the joint tx (store-core call): open on the engine's + /// connection/session, uncommitted until [`Self::pin_commit`]. + async fn begin_joint_tx(&self) -> Result; + + /// The entry half the engine contributes into the held tx: its put + /// mechanics (row shape, CAS clause) under the same first-writer-wins + /// semantics as the public `put` (ADR-010 §2). For the pg-lo + /// consolidation posture this clause is the kv host's row write — + /// the large entry's content reaches the same tx through that + /// engine's staged `lowrite` bodies (an engine-task contribution + /// against this `JointTx` type, per ADR-012 §4's one-tx ruling). + async fn joint_put( + &self, + tx: &mut Self::JointTx, + key: &[u8], + value: &[u8], + ) -> Result<(), Error>; + + /// The joint terminal step: the pin-row inserts **and** the commit — + /// publish and protection land in one step (ADR-005's + /// pin-before-publish, fleet form). On any failure the entire tx + /// aborts: no pin rows, no published entry, zero residue — the + /// stage-hygiene invariant's tx form. Consumes the tx; the store + /// core never re-uses it after this call. + async fn pin_commit(&self, tx: Self::JointTx, entries: &[PinRow]) -> Result<(), Error>; + + /// Renew the `(key, owner)` pin row: `true` when a **live** + /// (un-expired) row existed and its expiry moved to this call's + /// successor time (the store computes cadence — every TTL/3 — and + /// passes the renewal's shape through the row update). An expired or + /// absent row is never resurrected: `false`, nothing updated — + /// ADR-012 §2's expiry semantics verbatim (a putter paused past TTL + /// falls to the delete-then-recover backstop; the ops handler reads + /// this bool as the renewal probe's `present` answer). + async fn pin_renew(&self, key: &[u8], owner: &[u8]) -> Result; + + /// Classify the key against un-expired pin rows: any un-expired + /// `(key, *)` row exists → [`PinState::Protected`]; else + /// [`PinState::Unprotected`]. The engine never decides deletion — it + /// only classifies (ADR-010 §2); the store core receives the answer + /// and applies the decision. + async fn pin_check(&self, key: &[u8]) -> Result; + + /// The full delete-time arbitration (ADR-008): the un-expired pin + /// rows plus the embedder's registered liveness tables, under the + /// `SELECT … FOR UPDATE` finger-lock keyed at delete time, folded + /// into one answer — `true` = liveness protects the key at this + /// instant (the caller refuses the direct delete / cancels the + /// window candidate); `false` = nothing does and the caller may + /// proceed. The engine never decides; the facade decides and owns + /// the refusal (`Error::GcRefuse` never fires below the facade). + async fn arbitrate(&self, key: &[u8]) -> Result; + + /// Acquire the sweep lease — the pg advisory lock with ADR-013 §2's + /// named exclusive semantics: atomic test-and-set, one sweeper per + /// pool, holdership ends at release, owner death, or engine-session + /// loss. `Ok(true)` = held; `Ok(false)` = contention — the second + /// sweeper's sweep aborts with `GcAbortCause::SweeperLock` as report + /// data (`Ok(SweepReport)`; sweeps are never errors, ADR-012 §1). + async fn sweep_lock_acquire(&self) -> Result; + + /// Release the sweep lease. Releasing an unheld lease is a no-op; + /// owner death and engine-session loss reap it without this call + /// (ADR-013 §2's holdership rule). + async fn sweep_lock_release(&self) -> Result<(), Error>; +} diff --git a/src/lib.rs b/src/lib.rs index 89210b8..714d152 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod keys; pub mod local; pub use error::Error; +pub use facade::{Batch, EntryMeta, GcAbortCause, Pin, PinState, SweepReport}; pub use hashing::{Algorithm, Digest}; pub use keys::Key; diff --git a/tasks/core-trait-and-types.md b/tasks/core-trait-and-types.md index 2e59d3e..c4d9b9e 100644 --- a/tasks/core-trait-and-types.md +++ b/tasks/core-trait-and-types.md @@ -1,7 +1,7 @@ --- id: core-trait-and-types name: Backend trait, I/O seams, engine-state seam shape, and facade value types -status: pending +status: completed depends_on: [core-error-model, core-hashing-keys] scope: moderate risk: medium @@ -83,21 +83,21 @@ the contract surface in code. ## Acceptance Criteria -- [ ] `Backend` trait with the seven methods, opaque byte keys, +- [x] `Backend` trait with the seven methods, opaque byte keys, `size` metadata-only discipline documented -- [ ] Read cursor + staging cursor abstractions compile and are +- [x] Read cursor + staging cursor abstractions compile and are unit-tested against a test-double body; slice-digest convention tested -- [ ] Engine-state companion trait with all clause families and the +- [x] Engine-state companion trait with all clause families and the per-engine applicability table in docs; crate-internal (not exported from lib.rs) -- [ ] Facade value types complete (`EntryMeta`, `Pin`, `Batch`, +- [x] Facade value types complete (`EntryMeta`, `Pin`, `Batch`, `SweepReport`, `GcAbortCause` ×3, `PinState`) with lifecycle contracts documented -- [ ] The three named design points (trait visibility, put-form +- [x] The three named design points (trait visibility, put-form shapes, async shape) decided and recorded in the task's Notes for the review gate -- [ ] `cargo test`, `cargo clippy --all-targets -- -D warnings`, +- [x] `cargo test`, `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` green ## References @@ -112,8 +112,130 @@ the contract surface in code. ## Notes -> To be filled by implementation agent +The three named design points, decided and pinned here for the wave-1 +review gate: + +1. **Trait visibility — `pub(crate)`.** The gate's first reading holds: + engines ship in-crate as features and no consumer ever implements a + Backend (ADR-004), so the trait does not export. The module stays + `pub` (only the doc surface; items are `pub(crate)`) so the + stable-seam contract is documented at `crate::backend`. Exporting + later is a purely additive change — no row, no filename, no wire + bytes move — so no one-way door is forced either way. The same + visibility applies to `ReadCursor`/`StagingCursor` (ADR-010: not + public API) and to the engine-state seam (ADR-010 §2: never + exported; it lives in `src/gc.rs`). +2. **Put-form shapes — associated types on the one trait.** `Backend` + carries `type Cursor: ReadCursor` and `type Staging: + StagingCursor`, with `put(key, value)` (whole-value form) and + `put_staged(key) -> Self::Staging` (staged form) both required of + every engine — one trait, no per-tier split (ADR-010 §1). The + staging cursor extends `ReadCursor` so the store's verification pass + reads the eventual stored artifact back *as a whole through the same + seam it serves after commit* (ADR-003's buffered-put ruling requires + a second pass over the staged bytes; a verifying store cannot retain + the content stream). `commit(self)` is the joint publish-and- + self-discard terminal step (failure paths converge inside it, zero + residue — no post-failure discard exists; the consuming call took + the cursor); `discard(self)` is the explicit failure path. kv-tier + staged bodies are the degenerate in-memory stage committed through + the row write — deliberate, because store-api.md invariant 7 + contract-tests stage hygiene on *every* engine. `list()` is pinned + as `Vec>` (complete + well-formed by contract; ordering + unspecified — the store sorts for its live-set diffs); + `delete` reports `Err(Missing)` on absent keys (the facade's `delete` + carries `Missing`, ADR-012 §6.5); `get` returns `None` on absence + (never `Missing`). +3. **Async shape — AFIT with associated types.** `async fn in trait` + (Rust ≥ 1.75, rust-version 1.88), `Send + Sync` supertraits, no + `async_trait` dependency, no `dyn` anywhere — engines are statically + known, in-crate (ADR-004), and the family posture is tokio. + `Cursor`/`Staging`/`JointTx` are associated types rather than RPITIT, + which keeps each engine's body types nameable (the `local` cursor + type, the pg tx payload) and the store core generic. Spawn-site + `Send` bounds (ops layer, wave 6) resolve via RTN. + +Two further shape decisions recorded for the gate: + +4. **The companion seam is two traits realizing ADR-012 §5's table + structurally** (no default-bodied dead machinery): + `EngineWindowState` (schema + window family — postgres, sqlite, + pg-lo) and `EngineFleetState: EngineWindowState` (pin clauses + the + sweep lease — postgres host only). A single trait with all clauses + would have forced the exactly-the-bodies-ADR-012-§5-dropped + dead-code pattern on sqlite/pg-lo impls, so "one companion trait" + reads as "one seam, expressed as the window trait plus its fleet + extension". Clause signature details pinned in `src/gc.rs` docs: + `ensure_gc_schema()` (per-engine table sets per the §5 rows), + `window_stage(&[WindowCandidate])`/`window_take(now) -> + Vec>` (the take removes rows; the calling sweep re-arbitrates + and deletes — ADR-012 §6.7), `pin_renew -> Result` (a live row + is renewed; an expired/absent row is never resurrected — ADR-012 + §2's expiry semantics; the bool is the ops renewal probe's + `present` answer), `pin_check -> PinState` (classification only), + `arbitrate -> bool` (true = liveness protects at this instant; the + caller refuses), `sweep_lock_acquire -> Result` (true = lease + held; false = contention, a `GcAbortCause::SweeperLock` report + abort, never an error — ADR-012 §1) and `sweep_lock_release` + (idempotent). `JointTx` is an associated type (engine-defined + concrete payload, `Send`); the ownership rule is pinned in the + signature: `begin_joint_tx` hands the store the tx, `joint_put(tx, + key, value)` contributes the entry half, `pin_commit(tx, + &[PinRow])` consumes the tx as the joint terminal step (pin rows + + commit together; failure aborts the whole tx — the stage-hygiene + invariant's tx form). The pg-lo consolidation posture's + stream-into-tx contribution is documented as that engine's + wave-5 body against the `JointTx` type, not invented here. +5. **`GcAbortCause` carries `#[non_exhaustive]`** — ADR-012 §1's "callers + match non-exhaustively" ruling made structural for downstream + crates (in-crate matches stay exhaustive, which is the compile-time + surface of "new causes are new ADRs"). `PinState`, `Pin`, and + `Batch` are plain public types: pins are minted `pub(crate)`-only + by the store core (consumers never fabricate a guard), and the + Pin/Batch bodies are inert until wave 2 wires the refcount/ + arbitration machinery — the lifecycle contracts are pinned on the + types now, per the task. ## Summary -> To be filled on completion \ No newline at end of file +Implemented the wave-1 contract surface (task `core-trait-and-types`): + +- `src/backend.rs` — the `Backend` trait (seven methods + `put_staged`, + opaque byte keys/values, `size` metadata-only discipline, `list` + complete-and-well-formed, namespace-blind, virgin-store reads as + no-ops, engine posture documented); crate-internal per design point + 1, stable-seam status on the module and trait docs. +- `src/backend.rs` — the I/O seams (ADR-010 §1): `ReadCursor` + (`len` metadata-only, `read_at` EOF-clamping, provided `read_range` + composing the slice with `slice_digest` — the ADR-006 SHA-256 + convention at one seam), and `StagingCursor: ReadCursor` + (`append`/`commit`-self-discard/`discard`). Unit tests: the slice + digest convention against published NIST SHA-256 vectors; the cursor + contract run against both body-family doubles (materialized kv, + segmented streaming large); the identical-range-across-families + one-seam assertion; the seven-method Backend contract against a + test-double engine (virgin-store no-ops, CAS first-writer-wins, + Missing discipline); the staged-put happy path (staged bytes readable + as a whole through the verification seam, committed entry + round-tripping); and both stage-hygiene forms (discard → zero + residue; commit failure → self-discard). +- `src/gc.rs` — the engine-state companion seam (ADR-010 §2 + ADR-012 + §5) as `EngineWindowState`/`EngineFleetState`, crate-internal, never + exported, with the full clause families and the per-engine + applicability table in docs; `WindowCandidate`/`PinRow` row shapes; + the `JointTx` ownership rule pinned in signatures. +- `src/facade.rs` + `lib.rs` re-exports — the facade value types: + `EntryMeta`, `Pin` (RAII lifecycle contract incl. the ADR-005 + conversion-ordering rule), `Batch` (batch-scoped pin group, drops + together), `SweepReport` (exact counts, abort-as-data), `GcAbortCause` + (exactly three variants, `#[non_exhaustive]`), `PinState`. Value-type + invariant tests cover the enum shape, report semantics, guard/batch + shape, and the two-variant classification. + +Design points 1–5 above are the review gate's named decisions. + +Verification: `cargo test` (26 lib tests + 3 git-interop tests), +`cargo test --all-features`, `cargo clippy --all-targets -- -D +warnings`, `cargo clippy --all-features --all-targets -- -D warnings`, +`cargo fmt --check`, and `cargo doc --no-deps` (warnings unchanged from +the pre-existing baseline) all green. \ No newline at end of file