diff --git a/Cargo.lock b/Cargo.lock index dbf9a79..4f29c10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,7 @@ dependencies = [ "rusqlite", "serde_json", "thiserror", + "tokio", ] [[package]] diff --git a/alkstore-sqlite/Cargo.toml b/alkstore-sqlite/Cargo.toml index e6e26de..1ec64f7 100644 --- a/alkstore-sqlite/Cargo.toml +++ b/alkstore-sqlite/Cargo.toml @@ -11,6 +11,7 @@ parking_lot = "0.12" rusqlite = { version = "0.40", features = ["bundled", "functions"] } serde_json = "1" thiserror = "2" +tokio = { version = "1", features = ["rt", "sync"] } # file-id only supports unix and windows. Other targets (WASI, Redox, # illumos, etc.) get the `(0, 0)` fallback in substrate::watcher's @@ -19,4 +20,5 @@ thiserror = "2" file-id = "0.2" [dev-dependencies] -alkstore-contract-suite = { path = "../alkstore-contract-suite" } \ No newline at end of file +alkstore-contract-suite = { path = "../alkstore-contract-suite" } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } \ No newline at end of file diff --git a/alkstore-sqlite/src/lib.rs b/alkstore-sqlite/src/lib.rs index 6634c18..6fa01d6 100644 --- a/alkstore-sqlite/src/lib.rs +++ b/alkstore-sqlite/src/lib.rs @@ -1 +1,26 @@ +//! alkstore-sqlite — the SQLite engine: the core contract implemented +//! on rusqlite over the forked honker-core substrate carried in-tree +//! (ADR-001, ADR-011, ADR-013). +//! +//! Single-host by nature: file-backed, one machine; NFS +//! two-writers-unsupported (the lineage's honesty posture, inherited — +//! ADR-016). Durability knobs, pool sizing, and the watcher cadence +//! are engine-configuration concerns on [`SqliteOpts`] (ADR-008 §6) — +//! never contract surface. +//! +//! Long transactions park the writer: the writer slot is a lease, and +//! holding a caller-held [`alkstore::TxHandle`] across `await` points +//! holds it (ADR-007's honest model, surfaced so consumers budget +//! transactions). +//! +//! Every engine call round-trips a blocking thread (the +//! `spawn_blocking` seam — rusqlite connections are not +//! `Send`-across-await; the family REQ-TTY-01 posture). + +mod opts; +mod seam; +mod store; mod substrate; + +pub use opts::{DEFAULT_MAX_READERS, SqliteOpts}; +pub use store::{SqliteStore, open}; diff --git a/alkstore-sqlite/src/opts.rs b/alkstore-sqlite/src/opts.rs new file mode 100644 index 0000000..c1e98f7 --- /dev/null +++ b/alkstore-sqlite/src/opts.rs @@ -0,0 +1,68 @@ +//! The SQLite engine's option struct and the watcher-config plumbing +//! (ADR-008 §6 — constructors and their option structs live in the +//! engine crates; engine-configuration concerns never touch the +//! contract surface). +//! +//! `#[non_exhaustive]` deliberately does not apply (ADR-017 §3): opts +//! structs are consumer-*constructed*, and the attribute on them would +//! push every construction through a builder. New fields may be added +//! with `Default` fallbacks, per the core opts modules' posture. + +use std::time::Duration; + +use crate::substrate::WatcherConfig; + +/// The default reader-pool size (the shared watcher's poll cadence +/// rides [`SqliteOpts::poll_interval`]; this is the pool's own knob). +pub const DEFAULT_MAX_READERS: usize = 8; + +/// Construction options for the SQLite engine's +/// [`open`](crate::open) constructor. +/// +/// Durability knobs, pool sizing, and the watcher cadence are +/// engine-configuration concerns — they never appear on the contract +/// surface (ADR-008 §6); deployment.md carries the documented tuning +/// facts. +#[derive(Debug, Clone)] +pub struct SqliteOpts { + /// The watcher's `PRAGMA data_version` poll cadence. `None` = the + /// substrate's 1 ms shipping default (ADR-023 §4 — the default the + /// published wake-latency numbers were measured at); `Some(d)` sets + /// an explicit cadence (raise it for latency-tolerant deployments — + /// the documented idle cost at 1 ms is ~1000 poll reads/sec per + /// instance). Zero is rejected at `open` with + /// [`Error::Database`](alkstore::Error::Database) (the substrate's + /// `WatcherConfig::with_poll_interval` rejects zero; the mapping is + /// engine-layer work per ADR-012 §4's W-2 posture). + pub poll_interval: Option, + /// Upper bound on pooled reader connections (claims/lookups ride + /// these). Fresh readers open on demand up to the bound; `0` clamps + /// to 1. + pub max_readers: usize, +} + +impl Default for SqliteOpts { + fn default() -> Self { + Self { + poll_interval: None, + max_readers: DEFAULT_MAX_READERS, + } + } +} + +impl SqliteOpts { + /// Resolve this opts into the substrate's + /// [`WatcherConfig`](crate::substrate::WatcherConfig), mapping the + /// substrate's string error into the contract taxonomy's + /// [`Error::Database`](alkstore::Error) arm (the W-2 posture — + /// engine-layer mapping, ADR-012 §4). + pub(crate) fn watcher_config(&self) -> Result { + let config = WatcherConfig::default(); + match self.poll_interval { + None => Ok(config), + Some(d) => config + .with_poll_interval(d) + .map_err(alkstore::Error::database), + } + } +} diff --git a/alkstore-sqlite/src/seam.rs b/alkstore-sqlite/src/seam.rs new file mode 100644 index 0000000..4b715f6 --- /dev/null +++ b/alkstore-sqlite/src/seam.rs @@ -0,0 +1,45 @@ +//! The `spawn_blocking` seam (ADR-003, ADR-007) and the error +//! mapping — the two engine-wide postures established here that every +//! later mechanism task reuses. +//! +//! - Every engine call round-trips a blocking thread: rusqlite +//! connections are not `Send`-across-await, and the sync substrate +//! bridges at the trait seam (the family REQ-TTY-01 posture). +//! - Substrate errors map into the contract taxonomy's opaque +//! [`Error::Database`](alkstore::Error) fallback with the source +//! chain preserved (ADR-008 §5). No panics cross the seam: a panic +//! inside a bridged closure is the `JoinError` arm's `Database` +//! mapping, family-standard no-panics discipline maintained. + +use alkstore::Error; + +/// The blocking bridge every trait method's substrate round trip goes +/// through. Private: the seam is engine-internal — consumers meet it +/// only through the async trait surface. `cfg(test)`-gated until the +/// seam task (`sqlite-engine-seam-tx`) wires the trait impls; the +/// helper's shape and error path are pinned by this wave's smoke test. +#[cfg(test)] +pub(crate) async fn blocking(f: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + match tokio::task::spawn_blocking(f).await { + Ok(result) => result, + Err(e) => Err(database_error(format!("blocking task failed: {e}"))), + } +} + +/// Map a substrate string failure into the contract taxonomy. The +/// substrate is contract-blind — its fallible API surfaces strings +/// (the W-2 spawn posture, ADR-012 §4); the engine layer owns the +/// mapping, ADR-012 §2. +pub(crate) fn database_error(message: impl Into) -> Error { + Error::database(std::io::Error::other(message.into())) +} + +/// Map a `rusqlite::Error` into the contract taxonomy, source chain +/// preserved (ADR-008 §5 — no engine-specific variants minted). +pub(crate) fn sqlite_error(e: rusqlite::Error) -> Error { + Error::database(e) +} diff --git a/alkstore-sqlite/src/store.rs b/alkstore-sqlite/src/store.rs new file mode 100644 index 0000000..43f2e78 --- /dev/null +++ b/alkstore-sqlite/src/store.rs @@ -0,0 +1,220 @@ +//! The SQLite engine's store: the `open` constructor, the connection +//! architecture (engine-sqlite.md's "Connection architecture"), and +//! the [`Store`] trait wiring. +//! +//! Connection architecture: one writer connection behind the +//! substrate's `Writer` slot (WAL single-writer modeled honestly), a +//! pool of reader connections (fresh opens on demand up to +//! [`SqliteOpts::max_readers`]), and one `SharedUpdateWatcher` poll +//! thread over the db path. Every connection — writer and pooled +//! readers alike — runs the substrate's full bootstrap at open: +//! pragmas (WAL, `synchronous=NORMAL`, busy timeout), the notify +//! machinery, the substrate's SQL function set, and the schema +//! bootstrap. +//! +//! Watcher spawn is fallible in the substrate (ADR-012 §4's W-2); the +//! failure maps into [`Error::Database`](alkstore::Error::Database) +//! engine-side and `open` fails if the watcher cannot start. On +//! watcher death the substrate's `WatcherDeathGuard` closes every +//! subscriber (ADR-006) — the death-guard plumbing exists from open +//! onward; the consumer-visible receiver bridging arrives with +//! `listen`. +//! +//! Drop/close: the watcher's poll thread is joined (releasing its +//! connection and the db file handle) and the writer slot + reader +//! pool are closed; [`Drop`](SqliteStore::drop) delegates to +//! [`SqliteStore::close`]. +//! +//! Every engine call round-trips the `spawn_blocking` seam +//! ([`crate::seam`]); rusqlite connections are not +//! `Send`-across-await. Single-host by nature: file-backed, one +//! machine; NFS two-writers-unsupported (the lineage's honesty +//! posture, inherited — ADR-016). + +use alkstore::Store; +use std::path::PathBuf; +use std::sync::Arc; + +use crate::opts::SqliteOpts; +use crate::seam::{database_error, sqlite_error}; +use crate::substrate::{Readers, SharedUpdateWatcher, Writer}; + +/// The SQLite engine's store handle. Consumers hold +/// [`alkstore::Store`] — the trait is the contract surface; the type +/// exists so the engine can own its machinery and so tests can reach +/// the close path directly. +pub struct SqliteStore { + writer: Arc, + readers: Arc, + watcher: Arc, + db_path: PathBuf, +} + +impl std::fmt::Debug for SqliteStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SqliteStore") + .field("db_path", &self.db_path) + .finish_non_exhaustive() + } +} + +impl SqliteStore { + /// Explicit close: stops the watcher, disconnects every subscriber + /// (the death-guard close — receivers see them close), closes the + /// reader pool, and closes the writer slot. Idempotent. [`Drop`] + /// delegates here. + pub fn close(&self) { + let _ = self.watcher.close(); + self.readers.close(); + self.writer.close(); + } +} + +impl Drop for SqliteStore { + fn drop(&mut self) { + self.close(); + } +} + +/// Open a SQLite-backed store at `path`. +/// +/// The path is a **plain filesystem path** (ADR-023 §3): no URI +/// interpretation — a `?name=value` suffix is part of the literal +/// filename (`:memory:` works without URI mode). The full connection +/// architecture boots: the writer slot, the reader pool, and the +/// update watcher (whose spawn is fallible — open fails with +/// [`Error::Database`](alkstore::Error::Database) if it cannot start). +pub fn open(path: &str, opts: SqliteOpts) -> alkstore::Result> { + Ok(Box::new(open_store(path, opts)?)) +} + +fn open_store(path: &str, opts: SqliteOpts) -> alkstore::Result { + let config = opts.watcher_config()?; + + let writer_conn = crate::substrate::open_conn_bootstrapped(path) + .map_err(|e| match e { + crate::substrate::SchemaError::Sqlite(inner) => sqlite_error(inner), + }) + .map_err(|e| database_error(format!("failed to open/bootstrap writer connection: {e}")))?; + + let watcher = SharedUpdateWatcher::new_with_config(PathBuf::from(path), config) + .map_err(|e| database_error(format!("failed to spawn update watcher: {e}")))?; + + Ok(SqliteStore { + writer: Arc::new(Writer::new(writer_conn)), + readers: Arc::new(Readers::new(path.to_string(), opts.max_readers)), + watcher: Arc::new(watcher), + db_path: PathBuf::from(path), + }) +} + +impl Store for SqliteStore { + fn begin_tx( + &self, + ) -> alkstore::BoxedFuture<'_, alkstore::Result>> { + Box::pin(async { Err(database_error("begin_tx wiring lands with the seam task")) }) + } + + fn notify<'a>( + &'a self, + _channel: &str, + _payload: serde_json::Value, + ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> { + Box::pin(async { + Err(database_error( + "notify wiring lands with the mechanism tasks", + )) + }) + } + + fn listen<'a>( + &'a self, + _channel: &str, + ) -> alkstore::BoxedFuture<'a, alkstore::Result>> { + Box::pin(async { + Err(database_error( + "listen wiring lands with the mechanism tasks", + )) + }) + } + + fn stream<'a>( + &'a self, + _name: &str, + ) -> alkstore::BoxedFuture<'a, alkstore::Result>> { + Box::pin(async { + Err(database_error( + "stream wiring lands with the mechanism tasks", + )) + }) + } + + fn queue<'a>( + &'a self, + _name: &str, + _opts: alkstore::QueueOpts, + ) -> alkstore::BoxedFuture<'a, alkstore::Result>> { + Box::pin(async { + Err(database_error( + "queue wiring lands with the mechanism tasks", + )) + }) + } + + fn outbox<'a>( + &'a self, + _name: &str, + ) -> alkstore::BoxedFuture<'a, alkstore::Result>> { + Box::pin(async { + Err(database_error( + "outbox wiring lands with the mechanism tasks", + )) + }) + } + + fn try_lock<'a>( + &'a self, + _name: &str, + _owner: &str, + _ttl: i64, + ) -> alkstore::BoxedFuture<'a, alkstore::Result>>> { + Box::pin(async { Err(database_error("lock wiring lands with the mechanism tasks")) }) + } + + fn schedule<'a>( + &'a self, + _name: &str, + _spec: &str, + _queue: &str, + _payload: serde_json::Value, + _opts: alkstore::ScheduleOpts, + ) -> alkstore::BoxedFuture<'a, alkstore::Result> { + Box::pin(async { + Err(database_error( + "schedule wiring lands with the mechanism tasks", + )) + }) + } + + fn unschedule<'a>(&'a self, _name: &str) -> alkstore::BoxedFuture<'a, alkstore::Result> { + Box::pin(async { + Err(database_error( + "unschedule wiring lands with the mechanism tasks", + )) + }) + } + + fn run_schedules<'a>( + &'a self, + _stop: alkstore::StopToken, + ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> { + Box::pin(async { + Err(database_error( + "scheduler wiring lands with the mechanism tasks", + )) + }) + } +} + +#[cfg(test)] +mod open_tests; diff --git a/alkstore-sqlite/src/store/open_tests.rs b/alkstore-sqlite/src/store/open_tests.rs new file mode 100644 index 0000000..5a430dc --- /dev/null +++ b/alkstore-sqlite/src/store/open_tests.rs @@ -0,0 +1,415 @@ +//! The `open` constructor's acceptance tests (`sqlite-engine-open-opts`): +//! the connection architecture boots, opts flow, the plain-path +//! posture holds, and close/drop teardown releases the file. + +use std::path::PathBuf; +use std::time::Duration; + +use crate::opts::{DEFAULT_MAX_READERS, SqliteOpts}; +use crate::seam::blocking; +use crate::store::{open, open_store}; + +fn temp_dir(tag: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!( + "alkstore-open-{tag}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&d); + std::fs::create_dir_all(&d).unwrap(); + d +} + +fn cleanup(dir: &PathBuf) { + let _ = std::fs::remove_dir_all(dir); +} + +/// Acceptance: `open(path, opts)` boots the full connection +/// architecture — the writer slot takes writes, the reader pool +/// serves reads, and the substrate's bootstrap ran on every +/// connection (the `__alkstore_*` table family exists; the notify and +/// ops SQL functions respond on a fresh pooled reader). +#[tokio::test(flavor = "multi_thread")] +async fn open_boots_the_full_connection_architecture() { + let dir = temp_dir("boot"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + + let writer_conn = store.writer.acquire().unwrap(); + let mode: String = writer_conn + .pragma_query_value(None, "journal_mode", |r| r.get(0)) + .unwrap(); + assert!( + mode.eq_ignore_ascii_case("wal"), + "writer not in WAL: {mode}" + ); + store.writer.release(writer_conn); + + let reader_conn = store.readers.acquire().unwrap(); + let reader_mode: String = reader_conn + .pragma_query_value(None, "journal_mode", |r| r.get(0)) + .unwrap(); + assert!( + reader_mode.eq_ignore_ascii_case("wal"), + "pooled reader not in WAL: {reader_mode}" + ); + let notify_fn: i64 = reader_conn + .query_row("SELECT notify('__alkstore_probe', 'x')", [], |r| r.get(0)) + .unwrap(); + assert!(notify_fn >= 1, "pooled reader missing the function set"); + let publish_offset: i64 = reader_conn + .query_row( + "SELECT alkstore_stream_publish('__alkstore_probe', NULL, '{}')", + [], + |r| r.get(0), + ) + .unwrap(); + assert!(publish_offset >= 1, "pooled reader missing the ops surface"); + store.readers.release(reader_conn); + + assert!(store.db_path.exists(), "db file at the plain path"); + drop(store); + cleanup(&dir); +} + +/// Acceptance: the watcher is up after open — a commit from the +/// writer slot fans out to a raw substrate subscriber (the guard the +/// death plumbing rides; the consumer-visible receiver bridge lands +/// with `listen`). +#[tokio::test(flavor = "multi_thread")] +async fn open_leaves_the_watcher_running_and_fanning_out() { + let dir = temp_dir("watcher"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + + let (_id, rx) = store.watcher.subscribe(); + + let writer_conn = store.writer.acquire().unwrap(); + writer_conn + .execute("CREATE TABLE t(x INTEGER PRIMARY KEY)", []) + .unwrap(); + writer_conn.execute("INSERT INTO t VALUES (1)", []).unwrap(); + store.writer.release(writer_conn); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + let got = loop { + match rx.try_recv() { + Ok(()) => break true, + Err(std::sync::mpsc::TryRecvError::Empty) => { + assert!( + std::time::Instant::now() < deadline, + "watcher not fanning out after a commit" + ); + std::thread::sleep(Duration::from_millis(5)); + } + Err(std::sync::mpsc::TryRecvError::Disconnected) => break false, + } + }; + assert!( + got, + "subscriber channel disconnected instead of fanning out" + ); + + store.close(); + cleanup(&dir); +} + +/// Acceptance: watcher death handling wired — closing the shared +/// watcher disconnects every subscriber receiver (the death-guard +/// plumbing `listen` bridges into `recv() -> None`). The mechanism +/// lands with `listen`; the plumbing exists here. +#[tokio::test(flavor = "multi_thread")] +async fn closing_the_store_closes_watcher_subscribers() { + let dir = temp_dir("death"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + + let (_id, rx) = store.watcher.subscribe(); + store.close(); + assert!( + rx.recv().is_err(), + "subscriber receiver must close with the watcher (WatcherDeathGuard)" + ); + cleanup(&dir); +} + +/// Acceptance: `poll_interval` flows to the watcher's config (the +/// config-surface accounting test — asserting the configured cadence, +/// not wall-clock timing; ADR-023 §4's backlog row; `None` = the 1 ms +/// default, zero rejected at open). +#[tokio::test(flavor = "multi_thread")] +async fn poll_interval_flows_to_the_watcher_config() { + use crate::substrate::DEFAULT_WATCHER_POLL_INTERVAL; + + assert_eq!( + DEFAULT_WATCHER_POLL_INTERVAL, + Duration::from_millis(1), + "the substrate's shipping default is 1 ms (ADR-023 §4)" + ); + + assert_eq!( + SqliteOpts::default().poll_interval, + None, + "None = the substrate's 1 ms default" + ); + + let opts = SqliteOpts { + poll_interval: Some(Duration::from_millis(25)), + ..Default::default() + }; + let config = opts.watcher_config().unwrap(); + assert_eq!( + config.poll_interval, + Duration::from_millis(25), + "the knob must flow to the watcher's configured cadence" + ); + + let zero_err = SqliteOpts { + poll_interval: Some(Duration::ZERO), + ..Default::default() + } + .watcher_config() + .unwrap_err(); + assert!( + matches!(zero_err, alkstore::Error::Database(_)), + "the W-2 mapping lands in Error::Database, got: {zero_err:?}" + ); + let source = match &zero_err { + alkstore::Error::Database(source) => source.to_string(), + _ => unreachable!("matched above"), + }; + assert!( + source.contains("poll interval must be positive"), + "the substrate's zero-interval rejection must ride the source chain, got: {source}" + ); +} + +/// The 1 ms default resolves at open without an explicit interval, +/// and a generous explicit interval opens just the same. +#[tokio::test(flavor = "multi_thread")] +async fn open_succeeds_with_default_and_explicit_poll_intervals() { + let dir = temp_dir("cadence"); + for opts in [ + SqliteOpts::default(), + SqliteOpts { + poll_interval: Some(Duration::from_millis(50)), + ..Default::default() + }, + ] { + let path = dir.join(format!( + "cadence-{}.db", + opts.poll_interval.map(|d| d.as_millis()).unwrap_or(0) + )); + let store = open_store(path.to_str().unwrap(), opts).unwrap(); + assert!(store.db_path.exists()); + store.close(); + } + cleanup(&dir); +} + +/// Acceptance: the plain-path posture holds (ADR-023 §3 pin — a +/// `?name=value`-suffixed path creates a file with that literal +/// name; no URI flag anywhere in the open path). +#[tokio::test(flavor = "multi_thread")] +async fn uri_shaped_open_path_is_a_literal_filename() { + let dir = temp_dir("plain-path"); + let path = dir.join("plain.db?mode=memory&cache=shared"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + let writer_conn = store.writer.acquire().unwrap(); + writer_conn.execute("CREATE TABLE t(x)", []).unwrap(); + store.writer.release(writer_conn); + store.close(); + assert!( + path.exists(), + "the file must exist under its literal `?name=value` name (no URI interpretation)" + ); + cleanup(&dir); +} + +/// `:memory:` works without URI mode (ADR-023 §3): each connection is +/// its own private memory db, so the writer carries the bootstrap and +/// a fresh pooled reader opens the same way. +#[tokio::test(flavor = "multi_thread")] +async fn memory_paths_open_without_uri_mode() { + let store = open_store(":memory:", SqliteOpts::default()).unwrap(); + let reader_conn = store.readers.acquire().unwrap(); + let tables: i64 = reader_conn + .query_row( + "SELECT count(*) FROM sqlite_master + WHERE type = 'table' AND name LIKE '__alkstore_%'", + [], + |r| r.get(0), + ) + .unwrap(); + store.readers.release(reader_conn); + assert!(tables >= 7, ":memory: open must carry the bootstrap"); + store.close(); +} + +/// Acceptance: reader-pool sizing — `max_readers` bounds the pool +/// (excess acquirers block until release) and `0` clamps to 1 +/// (the substrate's max discipline). +#[tokio::test(flavor = "multi_thread")] +async fn reader_pool_size_is_bounded_by_opts() { + let dir = temp_dir("pool"); + let path = dir.join("store.db"); + + let store = open_store( + path.to_str().unwrap(), + SqliteOpts { + max_readers: 2, + ..Default::default() + }, + ) + .unwrap(); + + let c1 = store.readers.acquire().unwrap(); + let c2 = store.readers.acquire().unwrap(); + + let readers = store.readers.clone(); + let blocked = std::thread::spawn(move || readers.acquire()); + std::thread::sleep(Duration::from_millis(100)); + assert!( + !blocked.is_finished(), + "acquire above max_readers must block (pool bound enforced)" + ); + + store.readers.release(c2); + let c3 = blocked.join().unwrap().unwrap(); + store.readers.release(c3); + store.readers.release(c1); + store.close(); + + let store = open_store( + path.to_str().unwrap(), + SqliteOpts { + max_readers: 0, + ..Default::default() + }, + ) + .unwrap(); + let only = store.readers.acquire().unwrap(); + store.readers.release(only); + store.close(); + cleanup(&dir); +} + +/// The default reader count is the documented constant. +#[test] +fn default_max_readers_is_the_documented_constant() { + assert_eq!(DEFAULT_MAX_READERS, 8, "the documented default pool size"); + assert_eq!( + SqliteOpts::default().max_readers, + DEFAULT_MAX_READERS, + "`Default` resolves the reader-pool knob to the documented default" + ); +} + +/// A pool beyond its bound opens fresh readers on demand up to the +/// bound; reads on those readers see a writer's committed data +/// (the WAL cross-connection read path the reader pool exists for). +#[tokio::test(flavor = "multi_thread")] +async fn pooled_readers_see_writer_commits() { + let dir = temp_dir("read-visibility"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + + let writer_conn = store.writer.acquire().unwrap(); + writer_conn + .execute("CREATE TABLE t(x INTEGER)", []) + .unwrap(); + writer_conn + .execute_batch("INSERT INTO t VALUES (42);") + .unwrap(); + store.writer.release(writer_conn); + + let reader_conn = store.readers.acquire().unwrap(); + let x: i64 = reader_conn + .query_row("SELECT x FROM t", [], |r| r.get(0)) + .unwrap(); + store.readers.release(reader_conn); + assert_eq!(x, 42, "pooled reader reads the writer's committed row"); + + store.close(); + cleanup(&dir); +} + +/// Explicit close: the watcher joins (its thread exits), the file +/// handle is released, and the store is gone-but-reopenable. +#[tokio::test(flavor = "multi_thread")] +async fn explicit_close_releases_the_db_file_and_permits_reopen() { + let dir = temp_dir("close"); + let path = dir.join("store.db"); + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + store.close(); + + let reopened = open(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + drop(reopened); + cleanup(&dir); +} + +/// `Drop` stops the watcher and closes connections — the same teardown +/// as explicit close, reached through the drop path. +#[tokio::test(flavor = "multi_thread")] +async fn drop_runs_the_full_teardown() { + let dir = temp_dir("drop"); + let path = dir.join("store.db"); + { + let store = open_store(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + let (_id, rx) = store.watcher.subscribe(); + drop(store); + assert!( + rx.recv().is_err(), + "drop must run the same teardown as close (subscriber closed)" + ); + } + let reopened = open(path.to_str().unwrap(), SqliteOpts::default()).unwrap(); + drop(reopened); + cleanup(&dir); +} + +/// Acceptance: the `spawn_blocking` seam — every engine call's +/// substrate round trip rides this helper; smoked here through the +/// real bridge with a rusqlite-shaped closure (the pattern the +/// mechanism tasks reuse one-for-one). +#[tokio::test(flavor = "multi_thread")] +async fn the_blocking_bridge_round_trips_and_maps_errors() { + let ok = blocking(|| Ok::<_, alkstore::Error>(7i64)).await; + assert_eq!(ok.unwrap(), 7); + + let err = blocking(|| -> Result<(), alkstore::Error> { + Err(alkstore::Error::database(std::io::Error::other( + "smoke failure", + ))) + }) + .await; + assert!(matches!(err, Err(alkstore::Error::Database(_)))); + + let panicking: Result<(), alkstore::Error> = + blocking(|| -> Result<(), alkstore::Error> { panic!("bridge smoke panic") }).await; + assert!( + matches!(&panicking, Err(alkstore::Error::Database(_))), + "a panic inside a bridged closure maps to Database (no panics cross the seam), got {panicking:?}" + ); +} + +/// `open` on an unopenable path fails cleanly — the substrate's +/// bootstrap errors map into `Error::Database` at the constructor, not +/// panics. +#[tokio::test(flavor = "multi_thread")] +async fn open_fails_cleanly_on_an_unopenable_path() { + let result = open( + "/this/parent/does/not/exist/alkstore-open-fail.db", + SqliteOpts::default(), + ); + match result { + Err(alkstore::Error::Database(source)) => { + assert!( + !source.to_string().is_empty(), + "the source chain carries the substrate's detail" + ); + } + Ok(_) => panic!("expected Err(Database) from an unopenable path, got Ok"), + Err(e) => panic!("expected Err(Database) from an unopenable path, got {e}"), + } +} diff --git a/alkstore-sqlite/src/substrate/PROVENANCE.md b/alkstore-sqlite/src/substrate/PROVENANCE.md index 939fea5..443a295 100644 --- a/alkstore-sqlite/src/substrate/PROVENANCE.md +++ b/alkstore-sqlite/src/substrate/PROVENANCE.md @@ -118,6 +118,7 @@ fixes. Cherry-picks: empty at scaffold, append-only forever. | D-27 | re-derivation | wave-2 review fix — `scheduler_tick`'s fire enqueue resolved the schedule row's relative `expires_s` into the absolute row `expires_at` (`now + s` at the fire transaction, one clock read) instead of passing the relative value through; regression test pins a fired job's `expires_at = fire-instant + s` and its claimability. The pre-fix pass-through would have produced rows expired in 1970 under any `expires_s` schedule | `queue_ops.rs` (`scheduler_tick`, fire-expiry resolution + test) | ADR-020 §2 (relative `expires` → resolved absolute row value, resolved at enqueue against the single clock) — D-12's landed state carried the lineage's `expires_s` field shape without carrying the lineage's enqueue-side resolution; wave-2 review (task review-wave-2) re-derivation spot-check found it | ours | | D-28 | hygiene | wave-2 review fix — the cross-mechanism pressure test's lock-churn branch exercised a per-producer lock name (literal `"p{producer}"` had no interpolation, so all producers contended on one name) with its result swallowed via `.unwrap_or(0)`; now interpolates, releases periodically, propagates errors | `ops.rs` (pressure test, lock branch) | test-quality only (no lineage counterpart, no library code touched); `fork-provenance-and-floor`'s ported-surface pressure intent | ours | | D-29 | port delta | `SQLITE_OPEN_URI` dropped from `open_conn`'s flags (the watcher's own opens never carried it — the inherited posture was internally inconsistent); the path argument is a plain filesystem path, `?name=value` suffixes are literal filenames, no connection-semantics mutation outside the engine constructor's opts; pinned by test (`open_conn_treats_uri_shaped_path_as_plain_filename`) | `schema.rs` (`open_conn`) | ADR-023 §3 (plain-path open — the URI parameter surface is SQLite's, not this engine's; no consumer-inventory row names URI features) | ours | +| D-30 | port delta | `open_conn_bootstrapped` — new function stacking the engine's full per-connection open posture (pragmas, notify, `attach_alkstore_functions`, `bootstrap_schema`), and `Readers::acquire`'s open path switched from bare `open_conn` to it (fresh connections only — the pool never re-bootstraps a reused connection). The lineage's pools opened readers pragmas+notify only, the function set + schema living only on the writer; this engine's reader-slot operations (claim/ack, stream reads, lock ops, scheduler checks, leadership probes) run this substrate's SQL functions on pooled connections, so every pooled connection carries the full surface | `schema.rs` (`open_conn_bootstrapped`, `Readers::acquire` open path) | engine-sqlite.md's connection architecture ("each connection runs the substrate's bootstrap at open — the writer and each pooled reader") — engine-layer obligation of `sqlite-engine-open-opts`; landed with the engine's `open` wiring | ours | ### Cherry-picks diff --git a/alkstore-sqlite/src/substrate/mod.rs b/alkstore-sqlite/src/substrate/mod.rs index 5e9ae01..92bf8dd 100644 --- a/alkstore-sqlite/src/substrate/mod.rs +++ b/alkstore-sqlite/src/substrate/mod.rs @@ -53,5 +53,8 @@ mod queue_ops; mod schema; mod watcher; -pub(crate) use schema::{attach_notify, bootstrap_schema, open_conn}; -pub(crate) use watcher::SharedUpdateWatcher; +pub(crate) use schema::{ + Error as SchemaError, Readers, Writer, apply_default_pragmas, attach_notify, bootstrap_schema, + open_conn, open_conn_bootstrapped, +}; +pub(crate) use watcher::{DEFAULT_WATCHER_POLL_INTERVAL, SharedUpdateWatcher, WatcherConfig}; diff --git a/alkstore-sqlite/src/substrate/schema.rs b/alkstore-sqlite/src/substrate/schema.rs index fa469e7..55a4444 100644 --- a/alkstore-sqlite/src/substrate/schema.rs +++ b/alkstore-sqlite/src/substrate/schema.rs @@ -324,6 +324,27 @@ pub(crate) fn open_conn(path: &str, install_notify: bool) -> Result Result { + let conn = open_conn(path, true)?; + super::ops::attach_alkstore_functions(&conn)?; + bootstrap_schema(&conn)?; + Ok(conn) +} + pub(crate) struct Writer { slot: Mutex>, available: Condvar, @@ -411,7 +432,7 @@ impl Readers { *out += 1; drop(out); drop(pool); - match open_conn(&self.path, false) { + match open_conn_bootstrapped(&self.path) { Ok(conn) => { if self.closed.load(Ordering::Acquire) { *self.outstanding.lock() -= 1; @@ -577,6 +598,55 @@ mod writer_reader_tests { let _ = std::fs::remove_dir_all(&dir); } + #[test] + fn open_conn_bootstrapped_carries_the_full_surface_on_every_connection() { + let dir = std::env::temp_dir().join(format!( + "alkstore-bootstrapped-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("bootstrapped.db"); + + let first = open_conn_bootstrapped(path.to_str().unwrap()).unwrap(); + drop(first); + + for i in 0..3 { + let conn = open_conn_bootstrapped(path.to_str().unwrap()) + .unwrap_or_else(|e| panic!("bootstrapped open {i} failed: {e}")); + let mode: String = conn + .pragma_query_value(None, "journal_mode", |r| r.get(0)) + .unwrap(); + assert!(mode.eq_ignore_ascii_case("wal"), "open {i}: not WAL"); + let notify_fn: i64 = conn + .query_row("SELECT notify('__alkstore_probe', 'x')", [], |r| r.get(0)) + .unwrap_or(0); + assert!(notify_fn >= 1, "open {i}: notify function missing"); + let tables: i64 = conn + .query_row( + "SELECT count(*) FROM sqlite_master + WHERE type = 'table' AND name LIKE '__alkstore_%'", + [], + |r| r.get(0), + ) + .unwrap(); + assert!( + tables >= 7, + "open {i}: bootstrap schema missing (saw {tables} tables)" + ); + let publish_ready: i64 = conn + .query_row( + "SELECT alkstore_stream_publish('__alkstore_probe', NULL, '{}')", + [], + |r| r.get(0), + ) + .unwrap_or(0); + assert!(publish_ready >= 1, "open {i}: function set missing"); + drop(conn); + } + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn concurrent_opens_never_return_database_is_locked() { const ROUNDS: usize = 8; diff --git a/tasks/sqlite-engine-open-opts.md b/tasks/sqlite-engine-open-opts.md index 719b91c..1f2a938 100644 --- a/tasks/sqlite-engine-open-opts.md +++ b/tasks/sqlite-engine-open-opts.md @@ -1,7 +1,7 @@ --- id: sqlite-engine-open-opts name: SQLite engine — `open` constructor, `SqliteOpts`, connection architecture -status: pending +status: completed depends_on: [core-engine-value-constructors] scope: moderate risk: medium @@ -53,21 +53,21 @@ verify bootstrap ran, verify the watcher is up, verify opts flow). ## Acceptance Criteria -- [ ] `SqliteOpts` exists with `poll_interval: Option` +- [x] `SqliteOpts` exists with `poll_interval: Option` (None = 1 ms default) and reader-pool sizing; documented -- [ ] `open(path, opts)` boots the full connection architecture: +- [x] `open(path, opts)` boots the full connection architecture: writer slot, reader pool, watcher (fallible spawn mapped to `Database`), substrate bootstrap on every connection -- [ ] Plain-path posture holds: no URI flag re-introduced anywhere in +- [x] Plain-path posture holds: no URI flag re-introduced anywhere in the open path (pinning test: a `?name=value`-suffixed path creates a file with that literal name — ADR-023 §3's backlog row) -- [ ] `poll_interval` flows to the watcher config (config-surface +- [x] `poll_interval` flows to the watcher config (config-surface accounting test — assert the watcher's configured cadence, not wall-clock timing; ADR-023 §4's backlog row) -- [ ] Watcher death handling wired: subscribers' receivers close +- [x] Watcher death handling wired: subscribers' receivers close (mechanism lands with `listen`, but the death-guard plumbing exists here) -- [ ] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt clean +- [x] `cargo test -p alkstore-sqlite`, clippy `-D warnings`, fmt clean ## References @@ -79,8 +79,82 @@ verify bootstrap ran, verify the watcher is up, verify opts flow). ## Notes -> To be filled by implementation agent +- **Concrete type + boxed constructor** (the task's implementer's + choice): `open()` returns `Box`; the concrete + `SqliteStore` is also re-exported (pub) so tests and the later + mechanism tasks can reach the machinery — consumers hold the trait. +- **Substrate delta D-30** (registered in PROVENANCE.md): the engine + spec's "each connection runs the substrate's bootstrap" obligation + required reader opens to carry the full surface — the lineage + opened readers pragmas+notify only (functions + schema lived on the + writer alone). Added `open_conn_bootstrapped` (pragmas, notify, + `attach_alkstore_functions`, `bootstrap_schema`) and switched + `Readers::acquire`'s fresh-open path to it. Fresh connections only; + the pool never re-bootstraps reused connections (readers stay + read-mostly — writes ride the writer slot; the idempotent bootstrap + would be harmless but pointless per-acquire). +- **Zero `poll_interval` disposition**: `SqliteOpts::poll_interval = + Some(ZERO)` fails at `open` with `Error::Database` whose source + chain carries the substrate's "watcher poll interval must be + positive" (the W-2 mapping posture — engine-layer mapping of the + substrate's string into the taxonomy; no new variant, per ADR-008 + §5). Pinned by test at the source-chain level. +- **`max_readers: 0` clamps to 1** (the substrate's existing `max()` + discipline, inherited — not re-decided here). +- **Defaults**: `DEFAULT_MAX_READERS = 8` (the lineage bindings' + default, carried) — re-exported from the crate root like core's + opts constants. +- **Trait stubs**: all twelve `Store` methods return + `Err(Database("… wiring lands with the … task"))` — nothing panics + (family standard); the mechanism tasks replace them. `with_tx` is + core's provided method — it surfaces the `begin_tx` stub naturally. +- **`spawn_blocking` seam posture**: the private `blocking` helper + exists in `seam.rs` (tokio promoted to a real dep, `rt`+`sync`) but + is `#[cfg(test)]`-gated until `sqlite-engine-seam-tx` wires the + trait impls — clippy `-D warnings` rejects dead code, and an + ungated unused helper would trip the gate every intermediate commit. + Its shape is pinned by a smoke test (panics inside bridged closures + map to `Database` — no panics cross the seam) and the seam task + drops the gate. `database_error(msg)` (string→`Database`, io::Error + source) and `sqlite_error(e)` (`rusqlite::Error`→`Database`, source + chain preserved) are the two mappings the mechanism tasks reuse. +- **`SqliteStore` accessors deliberately absent**: the test module + lives under `store::` and reaches the private fields directly; + `close()` is pub (explicit close, idempotent, `Drop` delegates). + `Debug` is hand-implemented (`finish_non_exhaustive`) — no derive. +- **Watcher death plumbing verified at the substrate seam**: the open + tests subscribe through the store's `SharedUpdateWatcher` and + assert receivers disconnect on close (and on drop's teardown); the + file-replacement death path is already pinned substrate-side + (`shared_update_watcher_signals_subscribers_on_watcher_death`). +- **`store/open_tests.rs`**: `open` returns `Box`, so the + tests use `open_store` (crate-private, returns `SqliteStore`) — the + boxed public constructor itself is exercised by the reopen/close + tests. ## Summary -> To be filled on completion \ No newline at end of file +Stood up `alkstore-sqlite`'s public surface: `SqliteOpts` +(`poll_interval: Option` with None = the substrate's 1 ms +shipping default via `WatcherConfig::with_poll_interval`; `max_readers` +with `DEFAULT_MAX_READERS = 8`; not `#[non_exhaustive]` per ADR-017 +§3's opts exemption; `Default`+`Clone`), the `open(path, opts) -> +Result>` constructor booting engine-sqlite.md's full +connection architecture (writer slot, reader pool, +`SharedUpdateWatcher` with fallible spawn mapped W-2-style into +`Error::Database` — open fails if the watcher cannot start), and the +`spawn_blocking` seam's helper + error mappings. Required substrate +delta D-30 (`open_conn_bootstrapped` + reader-pool open path carries +the full per-connection bootstrap — pragmas/notify/functions/schema — +per engine-sqlite.md's "each connection runs the substrate's +bootstrap"; the lineage opened readers pragmas-only), registered in +PROVENANCE.md. Store wiring: `close()`/`Drop` join the watcher, +clear subscribers (death-guard), close pool + writer slot. Trait +methods are `Database`-error stubs pending the mechanism tasks. 14 +new engine tests (full-boot, watcher fanout, death-close, cadence +accounting ADR-023 §4, plain-path ADR-023 §3, `:memory:`, pool +bounding + 0-clamp, read visibility, close/reopen, drop teardown, +seam smoke incl. panic mapping, unopenable-path failure) + 1 substrate +test. Verified: `cargo test` (workspace: 25 core + 3 suite + 101 +sqlite), `cargo clippy --all-targets -- -D warnings`, `cargo fmt +--check` all clean. \ No newline at end of file