From 0b43ea14491487904a5bad176fba14486ac90ab4 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 21:56:29 +0000 Subject: [PATCH] =?UTF-8?q?feat(engines):=20the=20mem=20contract-reference?= =?UTF-8?q?=20engine=20=E2=80=94=20first=20real=20engine=20(wave=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/mem.rs: MemBackend implements Backend (seven methods + put_staged, name "mem") over Arc>>; CAS first-writer-wins puts, materializing MemCursor body, degenerate in-memory MemStaging whose commit publishes via the same CAS clause and self-discards on failure; discard leaves zero residue. - CAS re-put is dedup for both put forms; reads are no-ops on the virgin store; delete carries Missing; size is metadata-only. - Eight in-module tests mirror the wave-1 double's assertions: method contract, list well-formedness, cursor body read contract + ADR-006 range digest, empty-entry round-trip, staged happy path, and both hygiene forms (fail_commit test hook stands in for media failure). - backend.rs: clamped promoted pub(crate) — the shared EOF-clamping arithmetic both mem cursor bodies compose (no per-engine duplicate). Verification: cargo test --features mem (33+7 green); cargo test (default) green; clippy --all-features --all-targets -D warnings green; fmt --check green. --- src/backend.rs | 6 +- src/mem.rs | 480 +++++++++++++++++++++++++++++++++++++++++++- tasks/engine-mem.md | 47 ++++- 3 files changed, 527 insertions(+), 6 deletions(-) diff --git a/src/backend.rs b/src/backend.rs index 79a7e2f..a5de3aa 100644 --- a/src/backend.rs +++ b/src/backend.rs @@ -219,8 +219,10 @@ pub(crate) trait Backend: Send + Sync { /// 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) { +/// 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. +pub(crate) 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) diff --git a/src/mem.rs b/src/mem.rs index da9ce03..1eb1685 100644 --- a/src/mem.rs +++ b/src/mem.rs @@ -1,2 +1,480 @@ //! The `mem` kv-tier engine (ADR-011 §2): a `BTreeMap`-shaped ephemeral -//! implementation — the contract-reference engine. +//! implementation of the kv contract — the crate's first real engine and +//! the **contract-reference engine**: every other engine's conformance +//! column mirrors its suite rows (store-api.md invariant 7). Feature `mem`, +//! default-off; never fleet-valid (ADR-011 §4), so no +//! [`gc`][crate::gc] engine-state implementation rides it — the SQL +//! engines' companion seam does not apply. +//! +//! The bodies are the kv body families of ADR-010 §1 in their degenerate +//! form: [`get`][Backend::get] hands out a read cursor that materializes +//! the whole value (in-memory slices; `len()` from the stored value), and +//! [`put_staged`][Backend::put_staged] hands out the degenerate in-memory +//! stage whose `commit` publishes through the same whole-value CAS row +//! write as `put` — `entry().or_insert()`, first writer wins, a re-put is +//! dedup and never an overwrite. `commit` cannot fail in normal operation +//! (no media can fail here); the test-only hook noted on [`MemStaging`] +//! exercises the failure form's self-discard hygiene, standing in for the +//! media-failure injection later engines need. +//! +//! The `#![allow(dead_code)]` below marks the engine's scope: it is a +//! wave-2 deliverable defined ahead of its dispatch driver (no store-core +//! or facade wiring reaches it yet). The allow shrinks when dispatch +//! names the mem engine. + +#![allow(dead_code)] + +use std::collections::BTreeMap; +use std::sync::Arc; + +use parking_lot::Mutex; + +use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; +use crate::Error; + +/// The `mem` engine — an ephemeral CAS table of opaque byte keys/values +/// under a short-held [`parking_lot::Mutex`] (AGENTS convention 3; +/// parking_lot does not poison, so there is no +/// `unwrap_or_else(|e| e.into_inner())` recovery posture to apply). +/// +/// Clone shares the map, so a cloned [`Backend`] handle and every +/// [`MemStaging`] issued from either see the same table — the shape +/// `put_staged`'s out-of-band commit requires. Clone is the constructor +/// until dispatch names the engine (ADR-013 §4's feature graph; the +/// store's `mem` construction lands with its wave's dispatch task). +#[derive(Clone, Default)] +pub(crate) struct MemBackend { + rows: Arc, Vec>>>, + /// Test-only commit-failure injection, standing in for the media + /// failure the mem tier cannot produce; mirrors the wave-1 test + /// double's knob so stage-hygiene's failure form is exercised + /// against the real engine. + #[cfg(test)] + fail_commit: bool, +} + +impl Backend for MemBackend { + type Cursor = MemCursor; + type Staging = MemStaging; + + fn name(&self) -> &'static str { + "mem" + } + + 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| MemCursor { 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(MemStaging { + key: key.to_vec(), + staged: Vec::new(), + rows: Arc::clone(&self.rows), + #[cfg(test)] + fail_commit: self.fail_commit, + }) + } + + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + match self.rows.lock().remove(key) { + Some(_) => Ok(()), + None => Err(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 materializing read cursor (ADR-010 §1's kv body family): the whole +/// stored value in memory; range reads are in-memory slices with the +/// cursor contract's EOF clamping, and `len()` is the stored value's +/// length — metadata by construction, there is no content fetch to +/// perform. +pub(crate) struct MemCursor { + value: Vec, +} + +impl ReadCursor for MemCursor { + fn len(&self) -> u64 { + self.value.len() as u64 + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.value.len()); + Ok(self.value[start..end].to_vec()) + } +} + +/// The degenerate in-memory staging cursor (ADR-010 §1): appends extend +/// the buffer; `commit` publishes the whole staged bytes through the same +/// CAS clause as [`Backend::put`] — first writer wins, losing the race is +/// dedup (`Ok`), never a failure — and every non-commit path, including +/// the failure form, converges on zero residue. `commit` takes the +/// buffer via `std::mem::take` before the table lock, so a failed publish +/// cannot strand the staged bytes in the map. +pub(crate) struct MemStaging { + key: Vec, + staged: Vec, + rows: Arc, Vec>>>, + /// Test-only injection, shared with the issuing [`MemBackend`] — see + /// its field note. + #[cfg(test)] + fail_commit: bool, +} + +impl ReadCursor for MemStaging { + fn len(&self) -> u64 { + self.staged.len() as u64 + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.staged.len()); + Ok(self.staged[start..end].to_vec()) + } +} + +impl StagingCursor for MemStaging { + async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + self.staged.extend_from_slice(chunk); + Ok(()) + } + + async fn commit(mut self) -> Result<(), Error> { + #[cfg(test)] + if self.fail_commit { + self.staged.clear(); + return Err(Error::Io( + "engine commit failure (mem, injected)".to_owned(), + )); + } + let value = std::mem::take(&mut self.staged); + self.rows.lock().entry(self.key).or_insert(value); + Ok(()) + } + + async fn discard(mut self) { + self.staged.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; + + /// The real stored value, observed through the read seam — the + /// mirror of the test double's `row()` peek. + async fn stored_value(backend: &MemBackend, key: &[u8]) -> Option> { + let cursor = backend.get(key).await.expect("get succeeds")?; + let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read"); + Some(bytes) + } + + /// The Backend method contract: the name, the seven methods' + /// absence/value/error disciplines, CAS first-writer-wins, and + /// virgin-store reads as no-ops — mirroring the wave-1 double's + /// `backend_method_contract`, now against the real engine. + #[tokio::test] + async fn backend_method_contract() { + let backend = MemBackend::default(); + let key = b"\x01opaque-key-bytes".to_vec(); + assert_eq!(backend.name(), "mem", "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!( + stored_value(&backend, &key).await.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 `list` contract is complete AND well-formed: every key in the + /// table, no values leaked, and (BTreeMap ordering) lexicographic — + /// asserted order-insensitively against the map's keys plus a + /// determinism check. + #[tokio::test] + async fn list_is_complete_and_well_formed() { + let backend = MemBackend::default(); + let keys = [ + b"\x02b-key".to_vec(), + b"\x02a-key".to_vec(), + b"\x02c-key".to_vec(), + ]; + for key in &keys { + backend.put(key, b"value").await.unwrap(); + } + let mut listed = backend.list().await.unwrap(); + let mut expected = keys.to_vec(); + listed.sort(); + expected.sort(); + assert_eq!(listed, expected, "list is complete and well-formed"); + for key in &expected { + assert!( + backend.has(key).await.unwrap(), + "no spurious or mangled keys" + ); + } + } + + /// The materialized cursor body's read contract: `len()` exact, + /// EOF-clamped reads (over-read, tail, at-EOF empty, zero-length), + /// and the ADR-006 range seam's digest over exactly the returned + /// bytes — mirroring the double's `cursor_contract_materialized_body`. + #[tokio::test] + async fn cursor_contract_materialized_body() { + let backend = MemBackend::default(); + let key = b"\x01cursor-key".to_vec(); + backend.put(&key, CONTENT).await.unwrap(); + + let cursor = backend + .get(&key) + .await + .expect("get succeeds") + .expect("entry present"); + let len = CONTENT.len() as u64; + assert_eq!(cursor.len(), len, "len is the entry length"); + + assert_eq!(cursor.read_at(0, len).await.unwrap(), CONTENT, "full read"); + assert_eq!( + cursor.read_at(0, len + 17).await.unwrap(), + CONTENT, + "over-read clamps at EOF, pread-style" + ); + let mid = (CONTENT.len() / 2) as u64; + assert_eq!( + cursor.read_at(mid, len).await.unwrap(), + &CONTENT[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 (slice, digest) = cursor.read_range(mid, 7).await.unwrap(); + assert_eq!(slice, &CONTENT[mid as usize..(mid as usize + 7)]); + assert_eq!(digest, crate::backend::slice_digest(&slice)); + } + + /// The degenerate case: a zero-byte entry round-trips (put empty, + /// size exact, cursor body reads empty), and the staged form handles + /// an empty append-and-commit identically. + #[tokio::test] + async fn empty_entry_round_trips() { + let backend = MemBackend::default(); + let key = b"\x01empty-key".to_vec(); + + backend.put(&key, b"").await.unwrap(); + assert!(backend.has(&key).await.unwrap(), "empty value still lands"); + assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero"); + let cursor = backend.get(&key).await.unwrap().expect("entry present"); + assert_eq!(cursor.len(), 0); + assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::::new()); + + let staged_key = b"\x01empty-staged-key".to_vec(); + let mut staging = backend.put_staged(&staged_key).await.unwrap(); + staging.append(b"").await.unwrap(); + assert_eq!(staging.len(), 0, "empty append stays empty"); + staging.commit().await.expect("empty commit publishes"); + assert!( + backend.has(&staged_key).await.unwrap(), + "empty entry landed" + ); + assert_eq!(backend.size(&staged_key).await.unwrap(), Some(0)); + let committed = backend + .get(&staged_key) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!( + committed.read_at(0, 8).await.unwrap(), + Vec::::new(), + "committed empty body reads empty" + ); + } + + /// 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 through the same CAS clause, and the + /// committed entry round-tripping through get's cursor — mirroring + /// the double's `staged_put_round_trips_through_the_read_seam`. + #[tokio::test] + async fn staged_put_round_trips_through_the_read_seam() { + let backend = MemBackend::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, crate::backend::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, crate::backend::slice_digest(CONTENT)); + + // The staged form lands CAS-first-writer too: a subsequent + // re-put — and a re-staged commit against the live entry — are + // both dedup no-ops that never overwrite. + backend.put(&key, b"overwritten").await.unwrap(); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(CONTENT), + "whole-value re-put does not overwrite" + ); + let mut second = backend.put_staged(&key).await.unwrap(); + second.append(b"overwritten-by-stage").await.unwrap(); + second.commit().await.expect("commit is dedup, not failure"); + assert_eq!( + stored_value(&backend, &key).await.as_deref(), + Some(CONTENT), + "staged re-commit does not overwrite" + ); + } + + /// Stage hygiene, discard form: every non-commit path converges on + /// discard with zero residue — nothing was staged into the table. + #[tokio::test] + async fn staged_discard_leaves_zero_residue() { + let backend = MemBackend::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 table 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). The + /// injection stands in for the media failure the mem tier cannot + /// itself produce; the shape holds regardless of the failure's cause. + #[tokio::test] + async fn commit_failure_self_discards() { + let backend = MemBackend { + 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"); + assert!( + backend.list().await.unwrap().is_empty(), + "zero residue in the table itself" + ); + } +} diff --git a/tasks/engine-mem.md b/tasks/engine-mem.md index 7793c52..ee8317b 100644 --- a/tasks/engine-mem.md +++ b/tasks/engine-mem.md @@ -1,7 +1,7 @@ --- id: engine-mem name: mem kv engine — the contract-reference engine -status: pending +status: completed depends_on: [] scope: narrow risk: low @@ -69,8 +69,49 @@ this task ships the engine and its in-module unit tests only. ## Notes -> To be filled by implementation agent +- `backend.rs`'s `clamped` (already documented as "the shared EOF-clamping + arithmetic of the cursor contract") was promoted `pub(crate)` so both + mem cursor bodies compose it — no per-engine duplicate of the clamp. +- `MemBackend`/`MemStaging` are `pub(crate)` (the Backend trait is + crate-internal per ADR-003's visibility design point); the module + carries `#![allow(dead_code)]` mirroring `backend.rs`'s seam-scope + pattern, which shrinks when dispatch names the mem engine. +- parking_lot's `Mutex` does not poison, so the + `unwrap_or_else(|e| e.into_inner())` posture has nothing to apply to — + plain short-held locks (AGENTS convention 3). +- mem's `commit` cannot fail in normal operation (no media). The wave-1 + double's `fail_commit` knob is carried as a `#[cfg(test)]` field on the + backend/staging pair so the failure form's self-discard hygiene is + exercised against the real engine; it stands in for the media-failure + injection later engines need (the conformance suite's fault-hook story + is `suite-column-mem`'s concern). +- Not fleet-valid (ADR-011 §4): no `gc` engine-state implementation — + confirmed SQL-engine-only in src/gc.rs. ## Summary -> To be filled on completion \ No newline at end of file +Implemented the `mem` contract-reference engine (feature `mem`) in +`src/mem.rs`: `MemBackend` implements `Backend` completely (name `mem`; +`has`/`get`/`put`/`put_staged`/`delete`/`list`/`size`) over an +`Arc>>>`. `put` is CAS +first-writer-wins (`entry().or_insert_with()`); `get` hands out +`MemCursor`, the materializing kv body (whole value, EOF-clamped reads, +`len()` from the stored value); `put_staged` hands out `MemStaging`, the +degenerate in-memory stage — `append` extends, `commit` publishes via the +same CAS clause (`std::mem::take` before the lock; losing the race is +dedup, never a failure) and self-discards on failure, `discard` drops the +buffer with zero residue. Reads are no-ops on the virgin store; `delete` +carries `Missing`; `size` is metadata-only. + +Eight in-module unit tests mirror the wave-1 double's assertions against +the real engine: the method contract (virgin no-ops, CAS re-put, Missing +discipline, size exactness), list completeness/well-formedness, the +materialized cursor body's read contract (EOF clamping, zero-length, and +the ADR-006 read_range digest), the empty-entry round-trip, the staged +happy path (verification pass over the staged whole through the read +seam, then both re-put forms are dedup no-ops), and both hygiene forms +(discard → zero residue; injected commit failure → self-discard). + +Verification: `cargo test --features mem` green (33 unit + 7 integration), +`cargo test` (default) green, `cargo clippy --all-features --all-targets +-- -D warnings` green, `cargo fmt --check` green. \ No newline at end of file