diff --git a/src/sqlite.rs b/src/sqlite.rs index 7642cde..3e5833a 100644 --- a/src/sqlite.rs +++ b/src/sqlite.rs @@ -1,2 +1,890 @@ -//! The `sqlite` kv-tier engine (ADR-003): WAL + `synchronous=NORMAL` -//! durability, prepared statements, bounded reads. +//! The `sqlite` kv-tier engine (ADR-003/007): WAL + `synchronous=NORMAL` +//! engaged at open — the shipped durability tier (ADR-003; honker ships +//! the posture and iroh's store measured it) — `prepare_cached` statement +//! discipline, the materializing read-cursor body, and the degenerate +//! in-memory stage. Feature `sqlite`, **default-on** (ADR-013 §4): the +//! default build's kv engine and the crate's first SQL-backed engine. +//! +//! **Schema (POC #5 finding B5's row shape adapted to sqlite):** +//! `blobs (key BLOB PRIMARY KEY, value BLOB, size INTEGER NOT NULL)`. +//! The dedicated `size` column is what makes [`Backend::size`] +//! metadata-only — `SELECT size … WHERE key = ?` rides the key's index +//! and never loads the value blob (ADR-008's size discipline; +//! plan-tested below). `has` is an EXISTS probe, never a value load. +//! `put` is CAS first-writer-wins (`INSERT OR IGNORE`): a re-put of an +//! existing key is dedup, never an overwrite — on the whole-value put +//! and the staged commit alike. +//! +//! **Schema-creation timing (recorded design point):** the constructor +//! runs `CREATE TABLE IF NOT EXISTS` at open. The pool table then exists +//! from construction on — `put` needs no lazy ensure-schema path, and +//! the pragma posture is applied exactly once before any statement runs. +//! The virgin-store tolerance itself ("read paths treat a missing table +//! as absent/empty", POC #1 finding 2) is an engine-level property of +//! the read paths independent of that timing — unit-tested against a +//! tableless database file through the test-only constructor below. +//! The tolerance is deliberately read-only: a write against a tableless +//! database surfaces the engine error rather than silently creating +//! schema. +//! +//! **Single-writer posture (recorded design point):** one connection +//! behind a short-held [`parking_lot::Mutex`], shared by the backend and +//! every staging cursor it hands out (a staging cursor sees the same +//! table through the clone — `put_staged`'s out-of-band commit requires +//! the shared handle). A pool is not required solo. rusqlite is +//! synchronous, so every engine I/O operation — constructor, all seven +//! methods, the staged commit — runs in [`tokio::task::spawn_blocking`] +//! (AGENTS convention 3; store-api.md §Concurrency posture); the cursor +//! and staging read bodies are in-memory after materialization and need +//! no blocking. The WAL single-writer ceiling (~1.2k objects/s under +//! contention, POC #5 B3) is a **documented posture, not a defect** — +//! REQ-1's economics put single-node push rates an order below it. +//! +//! The constructor errors when WAL cannot be engaged (in-memory +//! databases included): the durability promise is verified at open, not +//! assumed. +//! +//! The `#![allow(dead_code)]` below has real scope: nothing constructs +//! the engine until wave 3's constructor wiring (`store-sqlite-construction`). + +#![allow(dead_code)] + +use std::path::PathBuf; +use std::sync::Arc; + +use parking_lot::Mutex; +use rusqlite::{params, Connection}; + +use crate::backend::{clamped, Backend, ReadCursor, StagingCursor}; +use crate::Error; + +fn db_err(e: rusqlite::Error) -> Error { + Error::Io(format!("sqlite: {e}")) +} + +/// The virgin-store tolerance's discriminant (POC #1 finding 2): a +/// statement naming a table that does not exist maps on the read paths +/// to absent/empty, never an error. rusqlite carries sqlite's message +/// through its Display, so the check is stringly by nature — the stable +/// surface sqlite itself guarantees. +fn is_no_such_table(e: &rusqlite::Error) -> bool { + e.to_string().contains("no such table") +} + +/// The `size` row's integer, clamped to the domain the trait probes in. +fn size_u64(raw: i64) -> u64 { + raw.max(0) as u64 +} + +/// Run a synchronous rusqlite body off the executor — the engine's one +/// blocking discipline (rusqlite is synchronous; store-api.md §Concurrency +/// posture). Every engine I/O operation below routes through this; the +/// in-memory cursor/staging read bodies do not. +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 `sqlite` engine — one connection behind a short-held +/// [`parking_lot::Mutex`] (parking_lot does not poison, so there is no +/// `unwrap_or_else(|e| e.into_inner())` recovery posture to apply). +#[derive(Clone)] +pub(crate) struct SqliteBackend { + /// The single connection (the recorded single-writer posture), + /// shared with every staging cursor handout through the Arc. + conn: Arc>, + /// Test-only commit-failure injection, standing in for the media + /// failure the sqlite 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 SqliteBackend { + /// Open (or create) the pool database file and set the shipped + /// durability posture: `journal_mode=WAL` + `synchronous=NORMAL`, + /// then `CREATE TABLE IF NOT EXISTS` for the pool table. WAL + /// engagement is verified at open (the pragma returns the mode it + /// actually set); a database that cannot enter WAL is a construction + /// error, not a silent posture downgrade. + pub(crate) async fn new(path: impl Into) -> Result { + let path: Arc = Arc::new(path.into()); + let conn = blocking(move || { + let conn = Connection::open(&*path).map_err(db_err)?; + let mode: String = conn + .query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0)) + .map_err(db_err)?; + if !mode.eq_ignore_ascii_case("wal") { + return Err(Error::Io(format!( + "sqlite: journal_mode is {mode}, WAL unavailable" + ))); + } + conn.execute_batch("PRAGMA synchronous = NORMAL;") + .map_err(db_err)?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS blobs ( + key BLOB PRIMARY KEY, + value BLOB, + size INTEGER NOT NULL + );", + ) + .map_err(db_err)?; + Ok(conn) + }) + .await?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + #[cfg(test)] + fail_commit: false, + }) + } + + /// Open a database file without schema creation — the virgin-store + /// tolerance test's instrument (constructor-vs-lazy is a recorded + /// design point; the read paths' no-tables-yet semantics are tested + /// against a genuinely tableless database file either way). + #[cfg(test)] + async fn new_tableless(path: impl Into) -> Result { + let path: Arc = Arc::new(path.into()); + let conn = blocking(move || Connection::open(&*path).map_err(db_err)).await?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + #[cfg(test)] + fail_commit: false, + }) + } + + /// The connection's live pragma value — the posture test's probe + /// (`synchronous` is per-connection, so the backend's own connection + /// must be asked). + #[cfg(test)] + async fn pragma(&self, name: &str) -> Result { + let conn = Arc::clone(&self.conn); + let name = name.to_owned(); + blocking(move || { + let guard = conn.lock(); + let value: rusqlite::types::Value = guard + .query_row(&format!("PRAGMA {name}"), [], |row| row.get(0)) + .map_err(db_err)?; + Ok(value) + }) + .await + } +} + +/// The materializing read cursor (ADR-010 §1's kv body family): the whole +/// stored value in memory — `get` loads the row once (the kv tier holds +/// only ≤ threshold content); `len()` reads the row's `size` column, so +/// the cursor's length claim is the same indexed fact [`Backend::size`] +/// probes; range reads are in-memory slices with the cursor contract's +/// EOF clamping. This body is also what the facade's verification-on-get +/// re-derives over (no engine change needed). +pub(crate) struct SqliteCursor { + value: Vec, + len: u64, +} + +impl ReadCursor for SqliteCursor { + fn len(&self) -> u64 { + self.len + } + + 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(key)` publishes the whole staged bytes through the +/// same `INSERT OR IGNORE` clause as [`Backend::put`] — under the +/// store-core-resolved canonical key (the staged-put seam's key-at-commit +/// design point), 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 connection lock, so a failed publish +/// cannot strand the staged bytes anywhere. +pub(crate) struct SqliteStaging { + staged: Vec, + conn: Arc>, + /// Test-only injection, shared with the issuing [`SqliteBackend`] — + /// see its field note. + #[cfg(test)] + fail_commit: bool, +} + +impl ReadCursor for SqliteStaging { + 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 SqliteStaging { + async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> { + self.staged.extend_from_slice(chunk); + Ok(()) + } + + async fn commit(mut self, key: &[u8]) -> Result<(), Error> { + let value = std::mem::take(&mut self.staged); + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + #[cfg(test)] + let fail_commit = self.fail_commit; + blocking(move || { + #[cfg(test)] + if fail_commit { + return Err(Error::Io( + "engine commit failure (sqlite, injected)".to_owned(), + )); + } + let guard = conn.lock(); + let size = value.len() as i64; + let written: rusqlite::Result<()> = (|| { + let mut stmt = guard.prepare_cached( + "INSERT OR IGNORE INTO blobs (key, value, size) VALUES (?1, ?2, ?3)", + )?; + stmt.execute(params![key, value, size])?; + Ok(()) + })(); + written.map_err(db_err) + }) + .await + } + + async fn discard(mut self) { + self.staged.clear(); + } +} + +impl Backend for SqliteBackend { + type Cursor = SqliteCursor; + type Staging = SqliteStaging; + + fn name(&self) -> &'static str { + "sqlite" + } + + async fn has(&self, key: &[u8]) -> Result { + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + blocking(move || { + let guard = conn.lock(); + let probed: rusqlite::Result = (|| { + let mut stmt = + guard.prepare_cached("SELECT EXISTS(SELECT 1 FROM blobs WHERE key = ?1)")?; + Ok(stmt.query_row(params![key], |row| row.get::<_, i64>(0))? != 0) + })(); + match probed { + Ok(present) => Ok(present), + Err(e) if is_no_such_table(&e) => Ok(false), + Err(e) => Err(db_err(e)), + } + }) + .await + } + + async fn get(&self, key: &[u8]) -> Result, Error> { + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + blocking(move || { + let guard = conn.lock(); + let loaded: rusqlite::Result> = (|| { + let mut stmt = + guard.prepare_cached("SELECT value, size FROM blobs WHERE key = ?1")?; + let row = stmt.query_row(params![key], |row| { + Ok((row.get::<_, Vec>(0)?, row.get::<_, i64>(1)?)) + }); + match row { + Ok((value, size)) => Ok(Some(SqliteCursor { + value, + len: size_u64(size), + })), + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e), + } + })(); + match loaded { + Ok(cursor) => Ok(cursor), + Err(e) if is_no_such_table(&e) => Ok(None), + Err(e) => Err(db_err(e)), + } + }) + .await + } + + async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + let value = value.to_vec(); + let size = value.len() as i64; + blocking(move || { + let guard = conn.lock(); + let written: rusqlite::Result<()> = (|| { + let mut stmt = guard.prepare_cached( + "INSERT OR IGNORE INTO blobs (key, value, size) VALUES (?1, ?2, ?3)", + )?; + stmt.execute(params![key, value, size])?; + Ok(()) + })(); + written.map_err(db_err) + }) + .await + } + + async fn put_staged(&self) -> Result { + Ok(SqliteStaging { + staged: Vec::new(), + conn: Arc::clone(&self.conn), + #[cfg(test)] + fail_commit: self.fail_commit, + }) + } + + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + blocking(move || { + let guard = conn.lock(); + let removed: rusqlite::Result = (|| { + let mut stmt = guard.prepare_cached("DELETE FROM blobs WHERE key = ?1")?; + stmt.execute(params![key]) + })(); + match removed { + Ok(0) => Err(Error::Missing), + Ok(_) => Ok(()), + Err(e) if is_no_such_table(&e) => Err(Error::Missing), + Err(e) => Err(db_err(e)), + } + }) + .await + } + + async fn list(&self) -> Result>, Error> { + let conn = Arc::clone(&self.conn); + blocking(move || { + let guard = conn.lock(); + let listed: rusqlite::Result>> = (|| { + let mut stmt = guard.prepare_cached("SELECT key FROM blobs")?; + let rows = stmt.query_map([], |row| row.get::<_, Vec>(0))?; + let mut keys = Vec::new(); + for row in rows { + keys.push(row?); + } + Ok(keys) + })(); + match listed { + Ok(keys) => Ok(keys), + Err(e) if is_no_such_table(&e) => Ok(Vec::new()), + Err(e) => Err(db_err(e)), + } + }) + .await + } + + async fn size(&self, key: &[u8]) -> Result, Error> { + let conn = Arc::clone(&self.conn); + let key = key.to_vec(); + blocking(move || { + let guard = conn.lock(); + let probed: rusqlite::Result> = (|| { + let mut stmt = guard.prepare_cached("SELECT size FROM blobs WHERE key = ?1")?; + let row = stmt.query_row(params![key], |row| row.get::<_, i64>(0)); + match row { + Ok(size) => Ok(Some(size)), + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e), + } + })(); + match probed { + Ok(size) => Ok(size.map(size_u64)), + Err(e) if is_no_such_table(&e) => Ok(None), + Err(e) => Err(db_err(e)), + } + }) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog"; + + /// A fresh temp database file per test; Drop removes the file and + /// its WAL sidecars idempotently. + struct TempDb(PathBuf); + + fn temp_db(tag: &str) -> TempDb { + 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 path = std::env::temp_dir().join(format!( + "alkblobs-sqlite-{tag}-{}-{nanos:x}-{seq}.db", + std::process::id() + )); + TempDb(path) + } + + impl TempDb { + fn path(&self) -> &Path { + &self.0 + } + + fn drop_sidecar(&self, suffix: &str) { + let base = self.0.to_string_lossy().into_owned(); + let _ = std::fs::remove_file(format!("{base}{suffix}")); + } + } + + impl Drop for TempDb { + fn drop(&mut self) { + self.drop_sidecar("-wal"); + self.drop_sidecar("-shm"); + let _ = std::fs::remove_file(&self.0); + } + } + + async fn new_backend() -> (TempDb, SqliteBackend) { + let db = temp_db("engine"); + let backend = SqliteBackend::new(db.path()).await.unwrap(); + (db, backend) + } + + /// The real stored value, observed through the read seam — the + /// mirror of the mem engine's helper. + async fn stored_value(backend: &SqliteBackend, 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) + } + + async fn staged(backend: &SqliteBackend, key: &[u8], content: &[u8]) -> Result<(), Error> { + let mut staging = backend.put_staged().await?; + staging.append(content).await?; + staging.commit(key).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 + /// (every other engine's conformance column mirrors the reference). + #[tokio::test] + async fn backend_method_contract() { + let (_db, backend) = new_backend().await; + let key = b"\x01opaque-key-bytes".to_vec(); + assert_eq!(backend.name(), "sqlite", "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, asserted order-insensitively against the + /// put set plus a round-trip through `has`. + #[tokio::test] + async fn list_is_complete_and_well_formed() { + let (_db, backend) = new_backend().await; + 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 materializing cursor body's read contract: `len()` exact (the + /// row's size column), 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_materialized_body() { + let (_db, backend) = new_backend().await; + 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 through put, + /// the staged form, and the committed cursor (size exact-zero, every + /// read empty). A value of zero bytes is an empty BLOB, never a NULL. + #[tokio::test] + async fn empty_entry_round_trips() { + let (_db, backend) = new_backend().await; + 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().await.unwrap(); + staging.append(b"").await.unwrap(); + assert_eq!(staging.len(), 0, "empty append stays empty"); + staging + .commit(&staged_key) + .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 — 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 (_db, backend) = new_backend().await; + let key = b"\x01staged-key".to_vec(); + + let mut staging = backend.put_staged().await.unwrap(); + staging.append(&CONTENT[..8]).await.unwrap(); + staging.append(&CONTENT[8..16]).await.unwrap(); + staging.append(&CONTENT[16..]).await.unwrap(); + + assert_eq!( + staging.len() as usize, + CONTENT.len(), + "staged length is the appended length" + ); + let (bytes, digest) = staging + .read_range(0, staging.len()) + .await + .expect("staged bytes readable"); + assert_eq!(bytes, CONTENT, "staged artifact readable as a whole"); + assert_eq!(digest, crate::backend::slice_digest(CONTENT)); + + staging.commit(&key).await.expect("commit publishes"); + assert!(backend.has(&key).await.unwrap(), "entry published"); + + let cursor = backend + .get(&key) + .await + .expect("get succeeds") + .expect("entry present"); + let (bytes, digest) = cursor + .read_range(0, cursor.len()) + .await + .expect("committed read"); + assert_eq!(bytes, CONTENT); + assert_eq!(digest, 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 — nothing was staged into the table. + #[tokio::test] + async fn staged_discard_leaves_zero_residue() { + let (_db, backend) = new_backend().await; + let key = b"\x01discard-key".to_vec(); + + let mut staging = backend.put_staged().await.unwrap(); + staging.append(b"residue").await.unwrap(); + staging.discard().await; + + assert!( + !backend.has(&key).await.unwrap(), + "discarded stage published nothing" + ); + assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed"); + assert!( + backend.list().await.unwrap().is_empty(), + "the table holds exactly zero entries after discard" + ); + } + + /// Stage hygiene, failure form: a failed commit is already + /// self-discarded — the buffer was taken before the connection lock, + /// so 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 (_db, backend) = new_backend().await; + let failing = SqliteBackend { + conn: Arc::clone(&backend.conn), + #[cfg(test)] + fail_commit: true, + }; + let key = b"\x01failure-key".to_vec(); + + let mut staging = failing.put_staged().await.unwrap(); + staging.append(CONTENT).await.unwrap(); + let err = staging.commit(&key).await.unwrap_err(); + assert!(matches!(err, Error::Io(_)), "engine media failure"); + + assert!( + !backend.has(&key).await.unwrap(), + "failed commit published nothing" + ); + assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue"); + assert!( + backend.list().await.unwrap().is_empty(), + "zero residue in the table itself" + ); + } + + /// The pragma posture: the shipped durability tier engaged at open — + /// journal_mode WAL (persistent) and synchronous NORMAL (1, + /// per-connection) — verified on the backend's own single connection. + #[tokio::test] + async fn opens_with_wal_and_synchronous_normal() { + let (_db, backend) = new_backend().await; + match backend.pragma("journal_mode").await.unwrap() { + rusqlite::types::Value::Text(mode) => { + assert_eq!(mode.to_lowercase(), "wal", "journal_mode is WAL"); + } + other => panic!("journal_mode probe returned {other:?}"), + } + match backend.pragma("synchronous").await.unwrap() { + rusqlite::types::Value::Integer(sync) => { + assert_eq!(sync, 1, "synchronous is NORMAL (1)"); + } + other => panic!("synchronous probe returned {other:?}"), + } + } + + /// A database that cannot enter WAL is a construction error — the + /// durability promise is verified at open, not assumed (in-memory + /// databases cannot hold the WAL posture). + #[tokio::test] + async fn construction_rejects_a_database_that_cannot_hold_wal() { + let opened = SqliteBackend::new(":memory:").await; + assert!( + opened.is_err(), + "in-memory databases cannot hold the WAL posture" + ); + } + + /// `size` reads the dedicated size column via the key's index, never + /// the value blob (metadata-only, B5's row shape): the probe is exact + /// for a large value and absent keys, and `EXPLAIN QUERY PLAN` proves + /// the probe searches the key's index rather than scanning/loading. + #[tokio::test] + async fn size_probe_is_exact_and_rides_the_key_index() { + let (_db, backend) = new_backend().await; + let key = b"\x01size-probe-key".to_vec(); + let large_value = vec![0xAB_u8; 256 * 1024]; + backend.put(&key, &large_value).await.unwrap(); + + assert_eq!( + backend.size(&key).await.unwrap(), + Some(large_value.len() as u64), + "size is metadata-only and exact" + ); + assert_eq!(backend.size(b"\x01absent").await.unwrap(), None); + + let conn = Arc::clone(&backend.conn); + let plan = blocking(move || { + let guard = conn.lock(); + let planned: rusqlite::Result> = (|| { + let mut stmt = guard + .prepare_cached("EXPLAIN QUERY PLAN SELECT size FROM blobs WHERE key = ?1")?; + let rows = stmt.query_map(params![key], |row| row.get::<_, String>(3))?; + let mut details = Vec::new(); + for row in rows { + details.push(row?); + } + Ok(details) + })(); + planned.map_err(db_err) + }) + .await + .unwrap(); + let joined = plan.join(" "); + assert!( + joined.contains("USING INDEX"), + "the size probe searches the key's index: {joined}" + ); + } + + /// Virgin-store no-tables-yet tolerance (POC #1 finding 2): every + /// read path against a genuinely tableless database file is a + /// no-op — absent/empty, never "table does not exist" — and delete + /// reports Missing. The tolerance is read-only: a write surfaces the + /// engine error and leaves zero residue. + #[tokio::test] + async fn virgin_store_no_tables_yet_reads_are_noops() { + let db = temp_db("tableless"); + let backend = SqliteBackend::new_tableless(db.path()).await.unwrap(); + let key = b"\x01opaque-key".to_vec(); + + assert!( + !backend.has(&key).await.unwrap(), + "has on a tableless db is a no-op read" + ); + assert!( + backend.get(&key).await.unwrap().is_none(), + "get is a no-op read" + ); + assert_eq!( + backend.size(&key).await.unwrap(), + None, + "size is a no-op read" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "list is empty on a tableless db" + ); + assert_eq!( + backend.delete(&key).await.unwrap_err(), + Error::Missing, + "delete on a tableless db reports Missing" + ); + + assert!( + backend.put(&key, CONTENT).await.is_err(), + "the tolerance is read-only: writes surface the engine error" + ); + assert!( + backend.list().await.unwrap().is_empty(), + "the failed write left zero residue" + ); + } +} diff --git a/tasks/engine-sqlite.md b/tasks/engine-sqlite.md index 3ef125f..b680e79 100644 --- a/tasks/engine-sqlite.md +++ b/tasks/engine-sqlite.md @@ -1,7 +1,7 @@ --- id: engine-sqlite name: sqlite kv engine — the default engine -status: pending +status: completed depends_on: [] scope: narrow risk: medium @@ -111,8 +111,54 @@ in-module unit tests only. ## Notes -> To be filled by implementation agent +Recorded design points landed in the module doc (`src/sqlite.rs`): + +- **Schema-creation timing:** the constructor runs `CREATE TABLE IF + NOT EXISTS` at open. The virgin-store tolerance itself is an + engine-level read-path property independent of that timing — + unit-tested against a genuinely tableless database file via a + `#[cfg(test)]` `new_tableless` constructor. The tolerance is + deliberately read-only: a write against a tableless database surfaces + the engine error rather than silently creating schema (tested). +- **WAL engagement is verified at open, not assumed:** the constructor + probes `PRAGMA journal_mode = WAL` and errors when WAL cannot be + engaged (in-memory databases included — tested). The pragma posture + (WAL persistent + synchronous NORMAL per-connection) is asserted in + the pragma test on the backend's own single connection. +- **Error mapping:** the no-tables-yet discriminant is a stringly check + on rusqlite's Display ("no such table") — stringly by nature, the + stable surface sqlite itself guarantees. `delete` maps it to + `Error::Missing` (rows-affected discipline holds for the virgin + form); `put` propagates it as `Error::Io`. +- **size metadata-only proof:** exactness is asserted for large/absent + keys and `EXPLAIN QUERY PLAN` asserts the probe searches the key's + index (`USING INDEX`) — never a scan or value load. +- Connection posture: one connection behind a short-held + `parking_lot::Mutex`, shared with staging handouts via `Arc` (a + staging commit must see the same table); the engine `Clone`s. All + rusqlite I/O in `spawn_blocking` via a local `blocking` helper (the + local.rs pattern); cursor/staging read bodies are in-memory after + materialization and need no blocking. +- `EXPLAIN QUERY PLAN` calls must bind the `?1` parameter even though + they do not execute — sqlite enforces the bind count at step time. ## Summary -> To be filled on completion \ No newline at end of file +The `sqlite` kv engine shipped complete behind feature `sqlite` +(default-on): the row shape `blobs (key BLOB PRIMARY KEY, value BLOB, +size INTEGER NOT NULL)` (POC #5 B5 adapted), `INSERT OR IGNORE` CAS +first-writer-wins on both put forms, the EXISTS `has` probe, the +materializing cursor body with `len()` from the size column, the +degenerate in-memory stage committed through the whole-value row write +with both hygiene forms, the indexed-column `size` probe, complete +`list`, and the no-tables-yet read tolerance. WAL + synchronous=NORMAL +set and verified at open; `prepare_cached` discipline throughout; every +rusqlite operation in `spawn_blocking`; single-writer posture recorded +in the module doc. Eleven in-module tests green (method contract, list +completeness, cursor contract, empty entry, staged round-trip incl. +both dedup forms, both hygiene forms, pragma posture, WAL-engagement +rejection, size/index plan, tableless virgin store). `cargo test` 58 +pass, `cargo clippy --all-targets -- -D warnings` clean, `cargo fmt +--check` clean, `cargo check --no-default-features` green. Engine-state +window family and store construction remain in their own tasks +(`engine-sqlite-window`, `store-sqlite-construction`). \ No newline at end of file