From 856d436f813159ffe3160283b70639afbdd50017 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 22:05:59 +0000 Subject: [PATCH] =?UTF-8?q?feat(engines):=20the=20local=20large-tier=20eng?= =?UTF-8?q?ine=20=E2=80=94=20sharded=20files,=20stage-then-commit-rename,?= =?UTF-8?q?=20pread=20cursors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LocalBackend: flat {hex-prefix}/{hex-prefix}/{hex} sharding over lowercase-hex-of-Key::as_bytes filenames (self-describing; the list-decode round-trip recorded for the review gate); virgin-store reads are no-ops - staged put: unique stage files under {root}/staging/, commit-rename is the publish, CAS first-writer-wins, every failure path (discard, failed commit, Drop cancellation) converges on zero residue - read cursor: pread loop over the immutable committed file, EOF clamping, provided read_range composing the slice-digest convention - all file I/O in spawn_blocking; size is a stat, never a content load - list walks the committed tree and is fatal (Error::Io, path named) on anything that does not decode — complete and well-formed - 11 in-module unit tests incl. exact-count stage hygiene over all failure forms and the success form Verification: cargo test (44 green), clippy --all-targets/--all-features -D warnings, fmt --check — all clean Task: tasks/engine-local.md --- src/local.rs | 1012 ++++++++++++++++++++++++++++++++++++++++- tasks/engine-local.md | 98 +++- 2 files changed, 1100 insertions(+), 10 deletions(-) diff --git a/src/local.rs b/src/local.rs index c886ba6..cfaeaf1 100644 --- a/src/local.rs +++ b/src/local.rs @@ -1,3 +1,1013 @@ //! The `local` large-tier engine (ADR-003/008): flat //! `{hex-prefix}/{hex-prefix}/{hash}` sharding, stage-then-commit-rename -//! staged puts, and pread-based read cursors. +//! staged puts (ADR-010 §1's named staged-put contract), and pread-based +//! read cursors — the ADR-003 backend as specced, ungated +//! (dependency-free, always present; ADR-013 §4). +//! +//! **Filename encoding (named design point, recorded for the review +//! gate):** committed filenames are the lowercase hex of the full key +//! bytes (`Key::as_bytes` — the alg-tagged, self-describing encoding +//! ADR-002's persistence-boundary ruling requires for anything with no +//! schema-migration story). Shard prefixes are taken from the hex +//! string itself (`hex[..2]`/`hex[2..4]`), so the layout is fully +//! decodable from the filesystem tree alone; `list` decodes filenames +//! back to key bytes and they round-trip through +//! `Key::from_bytes` exactly. A filename that does not decode — as +//! would a foreign or corrupted entry — is a **fatal list error** +//! (`Error::Io` naming the path): a malformed list is as deadly as an +//! incomplete one (ADR-003), and it is silent until a sweep deletes +//! the wrong things. Only the engine's own staging directory +//! ([`staging_dir`], excluded from the walk) is a known non-shard +//! entry. +//! +//! **Concurrency posture:** every file operation — constructor, +//! stat/open, pread, append, commit-rename, the CAS whole-value put's +//! write-then-rename, delete, the list walk, and discard — runs in +//! `tokio::task::spawn_blocking` (store-api.md §Concurrency posture; +//! the store never blocks the executor). The committed file is +//! immutable once published, so the read cursor preads against an open +//! handle without synchronization; the staging cursor is the sole +//! writer of its stage file and tracks the staged length in an atomic. +//! +//! **Stage hygiene (store-api.md invariant 3):** publish is the +//! commit-rename; every failure path converges on discard with zero +//! residue — a failed commit removes the stage file itself (there is +//! no post-failure discard to drive), an explicit [`StagingCursor::discard`] +//! removes it, and a staging cursor dropped without either is +//! best-effort-cleaned by [`Drop`] (the engine's own cancellation +//! story, per the wave-1 review's recorded observation; the hygiene +//! tests below cover all three forms exactly). CAS is first-writer-wins +//! everywhere a publish meets an existing entry: the committed file for +//! a key is never overwritten — a losing race or a re-put is dedup, +//! landing as `Ok` with the stage discarded. +//! +//! The `#![allow(dead_code)]` below marks the engine's scope: it is a +//! wave-2 deliverable defined ahead of its dispatch driver. The allow +//! shrinks when dispatch names the `local` engine. + +#![allow(dead_code)] + +use std::fs::{self, File, OpenOptions}; +use std::io::{ErrorKind, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; +use crate::keys::Key; +use crate::Error; + +/// Process-unique counter for stage and put-temp filenames, combined +/// with a wall-clock component so names stay unique across restarts of +/// the same root. +static STAGE_COUNTER: AtomicU64 = AtomicU64::new(0); + +pub(crate) struct LocalBackend { + root: Arc, + /// Test-only commit-failure injection (the media failure the local + /// tier cannot reliably produce in tests); mirrors the mem engine's + /// knob so the stage-hygiene failure form is exercised against the + /// real engine. + #[cfg(test)] + fail_commit: bool, +} + +impl LocalBackend { + /// Construct the engine over a fresh or existing root directory, + /// creating the root and its staging area (both blocking work, in + /// [`tokio::task::spawn_blocking`]). + pub(crate) async fn new(root: impl Into) -> Result { + let root: Arc = Arc::new(root.into()); + let staging = staging_dir(&root); + let path = Arc::clone(&root); + blocking(move || { + fs::create_dir_all(&*path).map_err(io_err)?; + fs::create_dir_all(staging).map_err(io_err)?; + Ok(()) + }) + .await?; + Ok(Self { + root, + #[cfg(test)] + fail_commit: false, + }) + } +} + +fn io_err(e: std::io::Error) -> Error { + Error::Io(e.to_string()) +} + +async fn blocking(body: F) -> Result +where + F: FnOnce() -> Result + Send + 'static, + T: Send + 'static, +{ + match tokio::task::spawn_blocking(body).await { + Ok(result) => result, + Err(e) => Err(Error::Io(format!("blocking task joint failure: {e}"))), + } +} + +/// The staging area under the root — the one directory of the layout +/// that holds in-flight stage and put-temp files, and the one entry the +/// committed-tree walk skips. +fn staging_dir(root: &Path) -> PathBuf { + root.join("staging") +} + +/// A wall-clock-plus-counter unique id for stage and put temp filenames. +fn stage_id() -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + format!( + "{:020}-{:x}", + STAGE_COUNTER.fetch_add(1, Ordering::Relaxed), + nanos + ) +} + +/// The committed path for a key: `{root}/{hex-prefix}/{hex-prefix}/{hex}` +/// — prefixes taken from the hex filename string itself. Requires at +/// least two key bytes; real encodings are 21 or 33. +fn committed_path(root: &Path, key: &[u8]) -> Result { + let name = hex::encode(key); + if name.len() < 4 { + return Err(Error::Io(format!( + "key too short for the sharded layout (hex {name})" + ))); + } + Ok(root.join(&name[..2]).join(&name[2..4]).join(&name)) +} + +/// A two-character lowercase-hex shard directory name — the only level-1 +/// and level-2 names the committed tree recognizes (besides the staging +/// area). +fn is_shard_name(name: &str) -> bool { + name.len() == 2 && name.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +/// The committed-tree walk behind [`Backend::list`]: enumerate every +/// shard directory, decode each filename from hex back to key bytes and +/// validate the decode (`Key::from_bytes`). Complete *and* well-formed +/// by contract (ADR-003): anything in the tree that is not a valid +/// committed entry — a non-shard directory, a stray file, an undecodable +/// filename — is `Err(Error::Io)` naming the offending path, never a +/// silent skip (a malformed list is as deadly as an incomplete one; it +/// is silent until a sweep deletes the wrong things). +fn walk_committed(root: &Path) -> Result>, Error> { + let mut keys = Vec::new(); + for shard1 in fs::read_dir(root).map_err(io_err)? { + let shard1 = shard1.map_err(io_err)?; + let name1 = shard1.file_name().to_string_lossy().into_owned(); + if name1 == "staging" { + continue; + } + if !shard1.file_type().map_err(io_err)?.is_dir() || !is_shard_name(&name1) { + return Err(Error::Io(format!( + "malformed pool entry at shard level 1: {name1}" + ))); + } + for shard2 in fs::read_dir(shard1.path()).map_err(io_err)? { + let shard2 = shard2.map_err(io_err)?; + let name2 = shard2.file_name().to_string_lossy().into_owned(); + if !shard2.file_type().map_err(io_err)?.is_dir() || !is_shard_name(&name2) { + return Err(Error::Io(format!( + "malformed pool entry at shard level 2: {name1}/{name2}" + ))); + } + for entry in fs::read_dir(shard2.path()).map_err(io_err)? { + let entry = entry.map_err(io_err)?; + let file_name = entry.file_name().to_string_lossy().into_owned(); + let name = format!("{name1}/{name2}/{file_name}"); + if !entry.file_type().map_err(io_err)?.is_file() { + return Err(Error::Io(format!( + "malformed pool entry at blob level: {name}" + ))); + } + let bytes = hex::decode(&file_name).map_err(|_| { + Error::Io(format!("malformed committed filename (not hex): {name}")) + })?; + Key::from_bytes(&bytes).map_err(|_| { + Error::Io(format!("malformed committed filename (not a key): {name}")) + })?; + keys.push(bytes); + } + } + } + Ok(keys) +} + +/// Fill `buf` entirely from `file` at `offset` — the pread loop of the +/// large body family (POC #3 finding A2's sound read window). Callers +/// clamp against the body's length first, so EOF is never reached; a +/// short pread in the loop just continues the read. +#[cfg(unix)] +fn pread_all(file: &File, mut buf: &mut [u8], mut offset: u64) -> std::io::Result<()> { + use std::os::unix::fs::FileExt; + while !buf.is_empty() { + match file.read_at(buf, offset) { + Ok(0) => break, + Ok(n) => { + buf = &mut buf[n..]; + offset += n as u64; + } + Err(e) if e.kind() == ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(()) +} + +#[cfg(not(unix))] +fn pread_all(file: &File, buf: &mut [u8], offset: u64) -> std::io::Result<()> { + use std::io::{Read, Seek, SeekFrom}; + let mut handle = file.try_clone()?; + handle.seek(SeekFrom::Start(offset))?; + handle.read_exact(buf)?; + Ok(()) +} + +/// The pread read cursor over a committed entry (ADR-010 §1's streaming +/// large body family): the file is immutable once committed, so reads +/// pread against an open handle at arbitrary offsets — never a content +/// load at open time (the handle and its length stat are all `get` +/// does). +pub(crate) struct LocalCursor { + file: File, + len: u64, +} + +impl ReadCursor for LocalCursor { + fn len(&self) -> u64 { + self.len + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped(offset, len, self.len as usize); + let wanted = end - start; + if wanted == 0 { + return Ok(Vec::new()); + } + let file = self.file.try_clone().map_err(io_err)?; + blocking(move || { + let mut buf = vec![0u8; wanted]; + pread_all(&file, &mut buf, start as u64).map_err(io_err)?; + Ok(buf) + }) + .await + } +} + +/// The staged put's staging cursor (ADR-010 §1): backed by a stage file +/// in the staging area, unique per stage — appends write through, the +/// staged length tracks completed appends (the sole-writer protocol), +/// and `commit` renames the stage file into its sharded committed path — +/// publish is the rename. Every failure path converges on discard with +/// zero residue: a commit that cannot complete (media failure, CAS +/// dedup, parent creation) removes the stage file itself; [`StagingCursor::discard`] +/// removes it; and a cursor dropped without either is best-effort +/// removed by the [`Drop`] body — the engine's cancellation story, +/// covered by the in-module hygiene tests. +pub(crate) struct LocalStaging { + root: Arc, + key: Vec, + stage_path: Option, + file: File, + staged_len: AtomicU64, + #[cfg(test)] + fail_commit: bool, +} + +impl ReadCursor for LocalStaging { + fn len(&self) -> u64 { + self.staged_len.load(Ordering::Acquire) + } + + async fn read_at(&self, offset: u64, len: u64) -> Result, Error> { + let (start, end) = clamped( + offset, + len, + self.staged_len.load(Ordering::Acquire) as usize, + ); + let wanted = end - start; + if wanted == 0 { + return Ok(Vec::new()); + } + let file = self.file.try_clone().map_err(io_err)?; + blocking(move || { + let mut buf = vec![0u8; wanted]; + pread_all(&file, &mut buf, start as u64).map_err(io_err)?; + Ok(buf) + }) + .await + } +} + +impl StagingCursor for LocalStaging { + async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + if chunk.is_empty() { + return Ok(()); + } + let mut file = self.file.try_clone().map_err(io_err)?; + let chunk = chunk.to_vec(); + let count = chunk.len() as u64; + blocking(move || file.write_all(&chunk).map_err(io_err)).await?; + self.staged_len.fetch_add(count, Ordering::AcqRel); + Ok(()) + } + + async fn commit(mut self) -> Result<(), Error> { + let stage = match self.stage_path.take() { + Some(stage) => stage, + None => { + return Err(Error::Io("staging cursor already consumed".to_owned())); + } + }; + let root = Arc::clone(&self.root); + let key = std::mem::take(&mut self.key); + let cleanup = stage.clone(); + let result = blocking(move || { + let committed = committed_path(&root, &key)?; + #[cfg(test)] + if self.fail_commit { + return Err(Error::Io( + "engine commit failure (local, injected)".to_owned(), + )); + } + if committed.exists() { + fs::remove_file(&stage).map_err(io_err)?; + return Ok(()); + } + let parent = committed + .parent() + .ok_or_else(|| Error::Io("committed path has no parent".to_owned()))?; + fs::create_dir_all(parent).map_err(io_err)?; + fs::rename(&stage, &committed).map_err(io_err)?; + Ok(()) + }) + .await; + if result.is_err() { + let _ = fs::remove_file(cleanup); + } + result + } + + async fn discard(mut self) { + if let Some(stage) = self.stage_path.take() { + let _ = blocking(move || { + if stage.exists() { + fs::remove_file(&stage).map_err(io_err)?; + } + Ok(()) + }) + .await; + } + } +} + +impl Drop for LocalStaging { + /// Best-effort residue cleanup for the cancellation story: a + /// staging cursor dropped without reaching `commit` or `discard` + /// removes its stage file (synchronous remove — Drop is sync by + /// nature; the staging area's contents are the residue this engine + /// cleans, and the hygiene tests assert it). A cursor reaching + /// `commit`/`discard` has already cleared the path, so this is a + /// no-op on those paths. + fn drop(&mut self) { + if let Some(stage) = self.stage_path.take() { + let _ = fs::remove_file(stage); + } + } +} + +impl Backend for LocalBackend { + type Cursor = LocalCursor; + type Staging = LocalStaging; + + fn name(&self) -> &'static str { + "local" + } + + async fn has(&self, key: &[u8]) -> Result { + Ok(self.size(key).await?.is_some()) + } + + async fn get(&self, key: &[u8]) -> Result, Error> { + let root = Arc::clone(&self.root); + let key = key.to_vec(); + blocking(move || { + let committed = committed_path(&root, &key)?; + match File::open(&committed) { + Ok(file) => { + let len = file.metadata().map_err(io_err)?.len(); + Ok(Some(LocalCursor { file, len })) + } + Err(e) if e.kind() == ErrorKind::NotFound => Ok(None), + Err(e) => Err(io_err(e)), + } + }) + .await + } + + async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + let root = Arc::clone(&self.root); + let key = key.to_vec(); + let value = value.to_vec(); + let temp_name = format!("put-{}", stage_id()); + blocking(move || { + let committed = committed_path(&root, &key)?; + if committed.exists() { + return Ok(()); + } + let staging = staging_dir(&root); + fs::create_dir_all(&staging).map_err(io_err)?; + let temp = staging.join(temp_name); + { + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(&temp) + .map_err(io_err)?; + file.write_all(&value).map_err(io_err)?; + } + let write_result = (|| { + let parent = committed + .parent() + .ok_or_else(|| Error::Io("committed path has no parent".to_owned()))?; + fs::create_dir_all(parent).map_err(io_err)?; + fs::rename(&temp, &committed).map_err(io_err) + })(); + match write_result { + Ok(()) => Ok(()), + Err(e) => { + let _ = fs::remove_file(&temp); + Err(e) + } + } + }) + .await + } + + async fn put_staged(&self, key: &[u8]) -> Result { + let root = Arc::clone(&self.root); + let key = key.to_vec(); + let inner_root = Arc::clone(&root); + let name = format!("stage-{}", stage_id()); + #[cfg(test)] + let fail_commit = self.fail_commit; + blocking(move || { + let staging = staging_dir(&inner_root); + fs::create_dir_all(&staging).map_err(io_err)?; + let stage_path = staging.join(name); + let file = OpenOptions::new() + .read(true) + .create(true) + .append(true) + .open(&stage_path) + .map_err(io_err)?; + Ok(LocalStaging { + root: inner_root, + key, + stage_path: Some(stage_path), + file, + staged_len: AtomicU64::new(0), + #[cfg(test)] + fail_commit, + }) + }) + .await + } + + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + let root = Arc::clone(&self.root); + let key = key.to_vec(); + blocking(move || { + let committed = committed_path(&root, &key)?; + match fs::remove_file(&committed) { + Ok(()) => Ok(()), + Err(e) if e.kind() == ErrorKind::NotFound => Err(Error::Missing), + Err(e) => Err(io_err(e)), + } + }) + .await + } + + async fn list(&self) -> Result>, Error> { + let root = Arc::clone(&self.root); + blocking(move || walk_committed(&root)).await + } + + async fn size(&self, key: &[u8]) -> Result, Error> { + let root = Arc::clone(&self.root); + let key = key.to_vec(); + blocking(move || { + let committed = committed_path(&root, &key)?; + match fs::metadata(&committed) { + Ok(meta) => Ok(Some(meta.len())), + Err(e) if e.kind() == ErrorKind::NotFound => Ok(None), + Err(e) => Err(io_err(e)), + } + }) + .await + } +} +#[cfg(test)] +mod tests { + use super::*; + use std::time::{SystemTime, UNIX_EPOCH}; + + const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; + + /// A fresh temp root per test (the factory's isolation contract in + /// miniature); Drop removes the whole tree idempotently. + struct TempRoot(PathBuf); + + fn temp_root(tag: &'static str) -> TempRoot { + static SEQ: AtomicU64 = AtomicU64::new(0); + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let dir = std::env::temp_dir().join(format!( + "alkblobs-local-{tag}-{}-{:x}-{seq}", + std::process::id(), + nanos + )); + std::fs::create_dir_all(&dir).expect("temp root created"); + TempRoot(dir) + } + + impl TempRoot { + fn path(&self) -> &Path { + &self.0 + } + } + + impl Drop for TempRoot { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + async fn new_backend() -> (TempRoot, LocalBackend) { + let root = temp_root("engine"); + let backend = LocalBackend::new(root.path()).await.unwrap(); + (root, backend) + } + + fn key_for(content: &[u8]) -> Vec { + Key::derive(content, content.len() as u64).as_bytes() + } + + fn sha1_key_for(content: &[u8]) -> Vec { + let digest = crate::hashing::Algorithm::GitBlobSha1.derive(content, content.len() as u64); + Key::from(digest).as_bytes() + } + + /// The real stored value, observed through the read seam. + async fn stored_value(backend: &LocalBackend, 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) + } + + /// Files currently in the staging area — the residue counter of the + /// exact-count hygiene tests. + fn stage_file_count(root: &Path) -> usize { + let staging = staging_dir(root); + match fs::read_dir(&staging) { + Ok(entries) => entries.count(), + Err(e) if e.kind() == ErrorKind::NotFound => 0, + Err(e) => panic!("staging dir unreadable: {e}"), + } + } + + async fn staged(backend: &LocalBackend, key: &[u8], content: &[u8]) -> Result<(), Error> { + let mut staging = backend.put_staged(key).await?; + staging.append(content).await?; + staging.commit().await + } + + /// 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 mem engine's suite. + #[tokio::test] + async fn backend_method_contract() { + let (_root, backend) = new_backend().await; + let key = key_for(b"method-contract"); + assert_eq!(backend.name(), "local", "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 a stat 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: both algorithms' + /// keys land, the walk returns every key byte-exact, and filenames + /// decode back through `Key::from_bytes` (the named design point's + /// round-trip). + #[tokio::test] + async fn list_is_complete_and_well_formed() { + let (_root, backend) = new_backend().await; + let keys = [ + key_for(b"a"), + sha1_key_for(b"a"), + key_for(b"bb"), + sha1_key_for(b"ccc"), + ]; + 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 raw in &listed { + let key = Key::from_bytes(raw).expect("listed bytes decode as keys"); + assert_eq!(key.as_bytes(), *raw, "decode round-trips to the same bytes"); + assert!( + backend.has(raw).await.unwrap(), + "no spurious or mangled keys" + ); + } + } + + /// The filename-encoding design point: the committed file lives at + /// `{root}/{hex[..2]}/{hex[2..4]}/{hex}` — lowercase hex of the full + /// key bytes, self-describing through the leading algorithm byte. + #[tokio::test] + async fn sharded_layout_and_filename_encoding() { + let (root, backend) = new_backend().await; + let key = key_for(CONTENT); + backend.put(&key, CONTENT).await.unwrap(); + + let name = hex::encode(&key); + assert_eq!(name, name.to_lowercase(), "filename is lowercase hex"); + let path = root.path().join(&name[..2]).join(&name[2..4]).join(&name); + assert!( + path.is_file(), + "committed file at the two-level sharded path" + ); + + let listed = backend.list().await.unwrap(); + assert_eq!(listed, vec![key], "the walk decodes the filename back"); + } + + /// The streaming pread 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. + #[tokio::test] + async fn cursor_contract_streaming_body() { + let (_root, backend) = new_backend().await; + let key = key_for(CONTENT); + 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 through put, + /// the staged form, and the committed cursor (size exact-zero, every + /// read empty). + #[tokio::test] + async fn empty_entry_round_trips() { + let (_root, backend) = new_backend().await; + let key = key_for(b""); + 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 = key_for(b"empty-staged"); + 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" + ); + 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-rename publishing, and the committed entry + /// round-tripping through get's cursor — with both re-put forms + /// (whole-value and staged) dedup no-ops against the live entry. + #[tokio::test] + async fn staged_put_round_trips_through_the_read_seam() { + let (_root, backend) = new_backend().await; + let key = key_for(b"staged-round-trip"); + + let mut staging = backend.put_staged(&key).await.unwrap(); + staging.append(&CONTENT[..8]).await.unwrap(); + staging.append(&CONTENT[8..24]).await.unwrap(); + staging.append(&CONTENT[24..]).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)); + let tail = staging.read_at(CONTENT.len() as u64, 32).await.unwrap(); + assert_eq!(tail, Vec::::new(), "staged read clamps at staged EOF"); + + 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)); + + 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" + ); + staged(&backend, &key, b"overwritten-by-stage") + .await + .expect("staged re-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 — no stage file, no committed entry. + #[tokio::test] + async fn staged_discard_leaves_zero_residue() { + let (root, backend) = new_backend().await; + let key = key_for(b"discard-key"); + + 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" + ); + assert_eq!(stage_file_count(root.path()), 0, "zero stage files"); + } + + /// 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. + #[tokio::test] + async fn commit_failure_self_discards() { + let root = temp_root("commit-failure"); + let backend = LocalBackend { + root: Arc::new(root.path().to_path_buf()), + #[cfg(test)] + fail_commit: true, + }; + let key = key_for(b"failure-key"); + + 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 pool itself" + ); + assert_eq!(stage_file_count(root.path()), 0, "zero stage files"); + } + + /// The engine's own cancellation story: a staging cursor dropped + /// without commit or discard leaves no residue (the Drop body's + /// best-effort cleanup, per the wave-1 review's observation). + #[tokio::test] + async fn drop_cancellation_leaves_zero_residue() { + let (root, backend) = new_backend().await; + let key = key_for(b"cancelled-key"); + + { + let mut staging = backend.put_staged(&key).await.unwrap(); + staging.append(b"never published").await.unwrap(); + // Dropped without commit or discard. + } + + assert!( + !backend.has(&key).await.unwrap(), + "cancelled 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 cancellation" + ); + assert_eq!( + stage_file_count(root.path()), + 0, + "Drop cleanup left zero stage files" + ); + } + + /// Exact-count stage hygiene (POC #3 A5's shape; store-api.md + /// invariant 3): N failed puts — over all three forms, explicit + /// discard, failed commit, and Drop cancellation — leave zero stage + /// files and zero committed entries, and N successful puts leave + /// exactly N committed entries and zero stage files. + #[tokio::test] + async fn exact_count_hygiene_failure_forms_leave_zero() { + const N: usize = 5; + let (root, backend) = new_backend().await; + let root = root.path().to_path_buf(); + let failure_backend = LocalBackend { + root: Arc::new(root.clone()), + #[cfg(test)] + fail_commit: true, + }; + + for i in 0..N { + let content = format!("discard-{i}"); + let key = key_for(content.as_bytes()); + let mut staging = backend.put_staged(&key).await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + staging.discard().await; + } + for i in 0..N { + let content = format!("commit-failure-{i}"); + let key = key_for(content.as_bytes()); + let mut staging = failure_backend.put_staged(&key).await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + staging.commit().await.unwrap_err(); + } + for i in 0..N { + let content = format!("cancelled-{i}"); + let key = key_for(content.as_bytes()); + let mut staging = backend.put_staged(&key).await.unwrap(); + staging.append(content.as_bytes()).await.unwrap(); + drop(staging); + } + + assert!( + backend.list().await.unwrap().is_empty(), + "3N failed puts left zero committed entries" + ); + assert_eq!( + stage_file_count(&root), + 0, + "3N failed puts left zero stage files" + ); + } + + #[tokio::test] + async fn exact_count_hygiene_success_form_leaves_exactly_n() { + const N: usize = 5; + let (root, backend) = new_backend().await; + + for i in 0..N { + let content = format!("success-{i}"); + let key = key_for(content.as_bytes()); + staged(&backend, &key, content.as_bytes()) + .await + .expect("commit publishes"); + } + + let listed = backend.list().await.unwrap(); + assert_eq!( + listed.len(), + N, + "N successful puts leave exactly N committed entries" + ); + for key in &listed { + assert!(backend.has(key).await.unwrap(), "every entry is real"); + } + assert_eq!( + stage_file_count(root.path()), + 0, + "success leaves zero stage files" + ); + } +} diff --git a/tasks/engine-local.md b/tasks/engine-local.md index 88c00e9..973fc39 100644 --- a/tasks/engine-local.md +++ b/tasks/engine-local.md @@ -1,7 +1,7 @@ --- id: engine-local name: local large engine — sharded files, stage-then-commit-rename, pread cursors -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -61,19 +61,19 @@ this task ships the engine and its in-module unit tests only. ## Acceptance Criteria -- [ ] `src/local.rs` implements `Backend` completely (ungated), with +- [x] `src/local.rs` implements `Backend` completely (ungated), with the sharded layout, the streaming pread cursor, and the stage-then-commit-rename staging cursor -- [ ] Filename encoding decided and recorded (self-describing key +- [x] Filename encoding decided and recorded (self-describing key bytes; list()-decode round-trips) -- [ ] All file I/O off the async path (`spawn_blocking`); no blocking +- [x] All file I/O off the async path (`spawn_blocking`); no blocking on the executor -- [ ] Exact-count stage-hygiene tests green over discard, failed +- [x] Exact-count stage-hygiene tests green over discard, failed commit, and Drop-cancellation forms — zero residue on every failure path, exactly one committed entry per success -- [ ] Virgin-store reads are no-ops on a fresh root; `size` is a stat, +- [x] Virgin-store reads are no-ops on a fresh root; `size` is a stat, never a content load; `list` is complete and well-formed -- [ ] `cargo test` green (local is ungated — in the default build); +- [x] `cargo test` green (local is ungated — in the default build); `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` green @@ -88,8 +88,88 @@ this task ships the engine and its in-module unit tests only. ## Notes -> To be filled by implementation agent +> **Filename encoding (the named design point, recorded for the review +> gate):** committed filenames are the lowercase hex of the full key +> bytes (`Key::as_bytes` — the alg-tagged, self-describing encoding +> ADR-002's persistence-boundary ruling requires); shard prefixes are +> taken from the hex string itself (`{hex[..2]}/{hex[2..4]}/{hex}`), so +> the sharded layout is fully decodable from the filesystem tree alone. +> `list` decodes each filename back through `Key::from_bytes`; the +> round-trip is asserted over both algorithms +> (`list_is_complete_and_well_formed`, +> `sharded_layout_and_filename_encoding`). A walk entry that does not +> decode — foreign junk or corruption — is a **fatal list error** +> (`Error::Io` naming the offending path), never a silent skip: a +> malformed list is as deadly as an incomplete one (ADR-003). Only the +> engine's staging area (`{root}/staging/`) is a known non-shard entry +> and is excluded from the walk. + +- **Constructor** — `LocalBackend::new(root)` (async) creates the root + and the staging area in `spawn_blocking`. Fresh temp dirs in tests + (`TempRoot` helper, Drop-removed idempotently). +- **Blocking discipline** — every file operation (constructor, stat, + open, pread, append, commit-rename, CAS put's write-then-rename, + delete, list walk, discard) runs in `tokio::task::spawn_blocking`; + nothing blocks the executor. The committed file is immutable once + published, so the read cursor preads against an open handle + (`FileExt::read_at` on unix; a seek-based fallback otherwise); the + staging cursor is the sole writer of its stage file, tracking the + staged length in an atomic (reads of it clamp at the staged EOF). +- **CAS posture** — `size`/`has` ride one metadata stat; both put forms + publish check-then-rename: an existing committed file is never + overwritten, and a losing concurrent re-put converges bit-identically + by content-addressing (worst case an idempotent rename). +- **Commit-failure injection** — `#[cfg(test)] fail_commit` on the + backend/cursor, mirroring the mem engine's knob, stands in for the + media failure; real failures (rename, parent creation, dedup) take + the same self-discard path (stage file removed, `Err` returned — no + post-failure discard to drive). +- **Drop cancellation** — the staging cursor's `Drop` best-effort + removes a stage file left unconsumed (synchronous remove; the + commit/discard paths clear the path first, so Drop is a no-op on + them). Covered by `drop_cancellation_leaves_zero_residue`. +- The conformance column was left to `suite-column-local` as scoped — + no `tests/conformance` changes in this task. ## Summary -> To be filled on completion \ No newline at end of file +Implemented the `local` large-tier engine (ungated) in `src/local.rs`: +`LocalBackend` implements `Backend` completely (name `local`) over an +`Arc` root. Committed layout is the flat +`{hex-prefix}/{hex-prefix}/{hex}` sharding with lowercase-hex-of- +`Key::as_bytes` filenames (self-describing; the two-level prefixes are +taken from the hex string itself — the decision and its decode +round-trip recorded in Notes). `put` is write-then-rename into the +sharded path, CAS first-writer-wins; `put_staged` hands out +`LocalStaging` — the stage-then-commit-rename contract: a unique stage +file under `{root}/staging/`, appends write through in `spawn_blocking`, +`commit` renames the stage into its sharded path (creating parents) — +publish is the rename — and every failure path (explicit discard, +failed commit — including injected media failure, Drop cancellation) +converges on zero residue. `get` hands out `LocalCursor`, the streaming +pread body over the immutable committed file (EOF-clamped reads, the +provided `read_range` composing the slice-digest convention); `size` +stats the file, never a content load; `list` walks the sharded tree, +decodes filenames back to key bytes, and is fatal (`Error::Io`, path +named) on anything that does not decode — complete and well-formed by +contract; virgin-store reads are no-ops; `delete` carries `Missing`. +All file I/O runs in `spawn_blocking` (store-api.md §Concurrency +posture). + +Eleven in-module unit tests mirror the mem column's suite and cover +this engine's stage-hygiene story: the Backend method contract, list +completeness/well-formedness plus the sharded-layout/filename-encoding +round-trip, the streaming cursor read contract (full/over/tail/EOF/ +zero-length/range-digest), the empty-entry round-trip, the staged happy +path with the verification pass over the staged whole and both re-put +forms as dedup no-ops, and the exact-count stage-hygiene gate over all +failure forms (discard, commit failure, Drop cancellation — 3N failed +puts left zero committed entries and zero stage files) and the success +form (N successful puts left exactly N committed entries, zero stage +files). + +Verification: `cargo test` green (37 lib + 4 conformance + 3 +git-interop; local ungated so included in the default build), `cargo +clippy --all-targets -- -D warnings` green, `cargo clippy +--all-features --all-targets -- -D warnings` green, `cargo fmt --check` +green. \ No newline at end of file