feat(engines): the sqlite kv engine — the default engine, first SQL-backed (wave 3)
Implements the sqlite kv-tier engine per ADR-003/007/010 §1 behind feature `sqlite` (default-on): the B5 row shape (key BLOB PK / value BLOB / size INTEGER NOT NULL), 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 with both hygiene forms, the indexed-column size probe (plan-tested), the no-tables-yet virgin-store read tolerance (tested against a tableless database file), WAL + synchronous=NORMAL verified at open, prepare_cached discipline, and every rusqlite operation in spawn_blocking under the recorded single-connection posture. Task: tasks/engine-sqlite.md (status: completed). The engine-state window family and store construction remain in their own tasks. Verification: - cargo test — 58 pass (11 new sqlite in-module tests) - cargo clippy --all-targets -- -D warnings — clean - cargo fmt --check — clean - cargo check --no-default-features — green
This commit is contained in:
glm-5.3-flash committed
1 parent
758ccfb359
commit
908fb5d2dd
2 files changed
+939
-5
No files matched your search
+890
-2
@@ -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<F, T>(body: F) -> Result<T, Error>
|
||||
where
|
||||
F: FnOnce() -> Result<T, Error> + 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<Mutex<Connection>>,
|
||||
/// 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<PathBuf>) -> Result<Self, Error> {
|
||||
let path: Arc<PathBuf> = 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<PathBuf>) -> Result<Self, Error> {
|
||||
let path: Arc<PathBuf> = 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<rusqlite::types::Value, Error> {
|
||||
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<u8>,
|
||||
len: u64,
|
||||
}
|
||||
|
||||
impl ReadCursor for SqliteCursor {
|
||||
fn len(&self) -> u64 {
|
||||
self.len
|
||||
}
|
||||
|
||||
async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, 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<u8>,
|
||||
conn: Arc<Mutex<Connection>>,
|
||||
/// 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<Vec<u8>, 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<bool, Error> {
|
||||
let conn = Arc::clone(&self.conn);
|
||||
let key = key.to_vec();
|
||||
blocking(move || {
|
||||
let guard = conn.lock();
|
||||
let probed: rusqlite::Result<bool> = (|| {
|
||||
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<Option<Self::Cursor>, Error> {
|
||||
let conn = Arc::clone(&self.conn);
|
||||
let key = key.to_vec();
|
||||
blocking(move || {
|
||||
let guard = conn.lock();
|
||||
let loaded: rusqlite::Result<Option<SqliteCursor>> = (|| {
|
||||
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<u8>>(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<Self::Staging, Error> {
|
||||
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<usize> = (|| {
|
||||
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<Vec<Vec<u8>>, Error> {
|
||||
let conn = Arc::clone(&self.conn);
|
||||
blocking(move || {
|
||||
let guard = conn.lock();
|
||||
let listed: rusqlite::Result<Vec<Vec<u8>>> = (|| {
|
||||
let mut stmt = guard.prepare_cached("SELECT key FROM blobs")?;
|
||||
let rows = stmt.query_map([], |row| row.get::<_, Vec<u8>>(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<Option<u64>, Error> {
|
||||
let conn = Arc::clone(&self.conn);
|
||||
let key = key.to_vec();
|
||||
blocking(move || {
|
||||
let guard = conn.lock();
|
||||
let probed: rusqlite::Result<Option<i64>> = (|| {
|
||||
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<Vec<u8>> {
|
||||
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::<u8>::new(),
|
||||
"read at EOF is empty, never an error"
|
||||
);
|
||||
assert_eq!(
|
||||
cursor.read_at(0, 0).await.unwrap(),
|
||||
Vec::<u8>::new(),
|
||||
"zero-length read"
|
||||
);
|
||||
|
||||
let (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::<u8>::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::<u8>::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<Vec<String>> = (|| {
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
+49
-3
@@ -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
|
||||
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`).
|
||||
Reference in new issue
Block a user