From c7ecc2e77204d4e37eb6fd14a4761c46bbd0e05f Mon Sep 17 00:00:00 2001 From: ldm0 Date: Wed, 28 Oct 2026 21:58:46 +0800 Subject: [PATCH] feat(sqlite): add bounded background executor and connection policy --- Cargo.lock | 8 + Cargo.toml | 1 + moli-sqlite/Cargo.toml | 12 ++ moli-sqlite/src/connection.rs | 208 ++++++++++++++++++++ moli-sqlite/src/connection/tests.rs | 168 ++++++++++++++++ moli-sqlite/src/executor.rs | 294 ++++++++++++++++++++++++++++ moli-sqlite/src/executor/tests.rs | 217 ++++++++++++++++++++ moli-sqlite/src/lib.rs | 17 ++ 8 files changed, 925 insertions(+) create mode 100644 moli-sqlite/Cargo.toml create mode 100644 moli-sqlite/src/connection.rs create mode 100644 moli-sqlite/src/connection/tests.rs create mode 100644 moli-sqlite/src/executor.rs create mode 100644 moli-sqlite/src/executor/tests.rs create mode 100644 moli-sqlite/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 6c1156d2cf..d951a99269 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3506,6 +3506,14 @@ dependencies = [ "url", ] +[[package]] +name = "moli-sqlite" +version = "0.1.0" +dependencies = [ + "parking_lot", + "rusqlite", +] + [[package]] name = "moli-storage-key" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index dae2117009..85b0392303 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ members = [ "moli-shared-worker", "moli-storage-service", "moli-storage-key", + "moli-sqlite", "moli-streams", "moli-structured-clone", "moli-stylesheet-blocking", diff --git a/moli-sqlite/Cargo.toml b/moli-sqlite/Cargo.toml new file mode 100644 index 0000000000..b4e27c401a --- /dev/null +++ b/moli-sqlite/Cargo.toml @@ -0,0 +1,12 @@ +[package] +license.workspace = true +name = "moli-sqlite" +version = "0.1.0" +edition = "2024" + +[dependencies] +parking_lot = "0.12.5" +rusqlite = { version = "0.40.2", features = ["bundled"] } + +[lints] +workspace = true diff --git a/moli-sqlite/src/connection.rs b/moli-sqlite/src/connection.rs new file mode 100644 index 0000000000..b284940bf8 --- /dev/null +++ b/moli-sqlite/src/connection.rs @@ -0,0 +1,208 @@ +use std::{fmt, path::Path, time::Duration}; + +use rusqlite::{Connection, ErrorCode, OpenFlags}; + +#[derive(Debug, Clone, Copy)] +pub struct ConnectionOptions { + pub cache_kib: u32, + pub statement_cache_capacity: usize, +} + +impl Default for ConnectionOptions { + fn default() -> Self { + Self { + cache_kib: 2048, + statement_cache_capacity: 32, + } + } +} + +#[derive(Debug, Clone, Copy)] +pub enum OpenMode { + /// The caller has exclusively allocated the containing generation directory. + Create, + /// Never create a missing database while recovering an existing identity. + Existing, +} + +#[derive(Debug)] +pub enum ConnectionError { + Sqlite(rusqlite::Error), + InvalidCacheSize, + Configuration { + setting: &'static str, + expected: &'static str, + actual: String, + }, + SchemaVersion { + expected: u32, + actual: i64, + }, +} + +impl fmt::Display for ConnectionError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Sqlite(error) => error.fmt(f), + Self::InvalidCacheSize => { + f.write_str("SQLite cache size must be between 1 and i32::MAX KiB") + } + Self::Configuration { + setting, + expected, + actual, + } => write!(f, "SQLite {setting}: expected {expected}, got {actual}"), + Self::SchemaVersion { expected, actual } => { + write!( + f, + "SQLite schema version: expected {expected}, got {actual}" + ) + } + } + } +} + +impl std::error::Error for ConnectionError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Sqlite(error) => Some(error), + _ => None, + } + } +} + +impl From for ConnectionError { + fn from(error: rusqlite::Error) -> Self { + Self::Sqlite(error) + } +} + +/// Open only on the database's assigned background worker. +pub fn open_disk( + path: &Path, + mode: OpenMode, + options: ConnectionOptions, +) -> Result { + let mut flags = OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_NO_MUTEX; + if matches!(mode, OpenMode::Create) { + flags |= OpenFlags::SQLITE_OPEN_CREATE; + } + let connection = Connection::open_with_flags(path, flags)?; + configure_common(&connection, options)?; + connection.pragma_update(None, "locking_mode", "EXCLUSIVE")?; + connection.pragma_update(None, "journal_mode", "WAL")?; + connection.pragma_update(None, "synchronous", "NORMAL")?; + require_setting(&connection, "locking_mode", "exclusive")?; + require_setting(&connection, "journal_mode", "wal")?; + let synchronous: i64 = connection.pragma_query_value(None, "synchronous", |row| row.get(0))?; + if synchronous != 1 { + return Err(ConnectionError::Configuration { + setting: "synchronous", + expected: "NORMAL (1)", + actual: synchronous.to_string(), + }); + } + Ok(connection) +} + +/// A private database, retained by its owner until deletion or shutdown. +pub fn open_private_memory(options: ConnectionOptions) -> Result { + let connection = Connection::open_in_memory()?; + configure_common(&connection, options)?; + require_setting(&connection, "journal_mode", "memory")?; + Ok(connection) +} + +fn configure_common( + connection: &Connection, + options: ConnectionOptions, +) -> Result<(), ConnectionError> { + if options.cache_kib == 0 || options.cache_kib > i32::MAX as u32 { + return Err(ConnectionError::InvalidCacheSize); + } + // One connection per owner and a profile process lock: do not implicitly + // block a worker waiting for another connection to release this file. + connection.busy_timeout(Duration::ZERO)?; + connection.pragma_update(None, "temp_store", "MEMORY")?; + connection.pragma_update(None, "cache_size", -i64::from(options.cache_kib))?; + connection.set_prepared_statement_cache_capacity(options.statement_cache_capacity); + let temp_store: i64 = connection.pragma_query_value(None, "temp_store", |row| row.get(0))?; + if temp_store != 2 { + return Err(ConnectionError::Configuration { + setting: "temp_store", + expected: "MEMORY (2)", + actual: temp_store.to_string(), + }); + } + let cache_size: i64 = connection.pragma_query_value(None, "cache_size", |row| row.get(0))?; + if cache_size != -i64::from(options.cache_kib) { + return Err(ConnectionError::Configuration { + setting: "cache_size", + expected: "configured negative KiB budget", + actual: cache_size.to_string(), + }); + } + Ok(()) +} + +fn require_setting( + connection: &Connection, + setting: &'static str, + expected: &'static str, +) -> Result<(), ConnectionError> { + let actual: String = connection.pragma_query_value(None, setting, |row| row.get(0))?; + if actual.eq_ignore_ascii_case(expected) { + Ok(()) + } else { + Err(ConnectionError::Configuration { + setting, + expected, + actual, + }) + } +} + +pub fn require_schema_version( + connection: &Connection, + expected: u32, +) -> Result<(), ConnectionError> { + let actual: i64 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?; + if actual == i64::from(expected) { + Ok(()) + } else { + Err(ConnectionError::SchemaVersion { expected, actual }) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorClass { + Constraint, + Busy, + Resource, + Corruption, + Io, + Interrupted, + Other, +} + +/// Classification does not decide whether a transaction survived. After an +/// error, the owner must inspect `Connection::is_autocommit()` before attempting +/// request SAVEPOINT rollback or accepting another request in that transaction. +pub fn classify_error(error: &rusqlite::Error) -> ErrorClass { + match error.sqlite_error_code() { + Some(ErrorCode::ConstraintViolation) => ErrorClass::Constraint, + Some(ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked) => ErrorClass::Busy, + Some(ErrorCode::DiskFull | ErrorCode::OutOfMemory | ErrorCode::TooBig) => { + ErrorClass::Resource + } + Some(ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase) => ErrorClass::Corruption, + Some(ErrorCode::SystemIoFailure | ErrorCode::CannotOpen | ErrorCode::ReadOnly) => { + ErrorClass::Io + } + Some(ErrorCode::OperationInterrupted) => ErrorClass::Interrupted, + _ => ErrorClass::Other, + } +} + +#[cfg(test)] +mod tests; diff --git a/moli-sqlite/src/connection/tests.rs b/moli-sqlite/src/connection/tests.rs new file mode 100644 index 0000000000..0018f241e0 --- /dev/null +++ b/moli-sqlite/src/connection/tests.rs @@ -0,0 +1,168 @@ +use std::{ + fs, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, +}; + +use super::*; + +static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(1); + +struct TestDirectory(PathBuf); + +impl TestDirectory { + fn new() -> Self { + let path = std::env::temp_dir().join(format!( + "moli-sqlite-{}-{}", + std::process::id(), + NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).unwrap(); + Self(path) + } + + fn database(&self) -> PathBuf { + self.0.join("db.sqlite") + } +} + +impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn pragma_string(connection: &Connection, pragma: &str) -> String { + connection + .pragma_query_value(None, pragma, |row| row.get(0)) + .unwrap() +} + +#[test] +fn disk_policy_is_effective_and_committed_data_survives_reopen() { + let directory = TestDirectory::new(); + let path = directory.database(); + { + let connection = open_disk(&path, OpenMode::Create, ConnectionOptions::default()).unwrap(); + assert_eq!(pragma_string(&connection, "locking_mode"), "exclusive"); + assert_eq!(pragma_string(&connection, "journal_mode"), "wal"); + assert_eq!( + connection + .pragma_query_value(None, "synchronous", |r| r.get::<_, i64>(0)) + .unwrap(), + 1 + ); + assert_eq!( + connection + .pragma_query_value(None, "cache_size", |r| r.get::<_, i64>(0)) + .unwrap(), + -2048 + ); + connection.execute_batch("CREATE TABLE saved(value TEXT); BEGIN; INSERT INTO saved VALUES('committed'); COMMIT;").unwrap(); + } + let connection = open_disk(&path, OpenMode::Existing, ConnectionOptions::default()).unwrap(); + let value: String = connection + .query_row("SELECT value FROM saved", [], |row| row.get(0)) + .unwrap(); + assert_eq!(value, "committed"); +} + +#[test] +fn recovery_never_creates_a_missing_database_or_resets_corrupt_data() { + let directory = TestDirectory::new(); + let path = directory.database(); + assert!(open_disk(&path, OpenMode::Existing, ConnectionOptions::default()).is_err()); + assert!(!path.exists()); + let corrupt = b"this is deliberately not a SQLite database"; + fs::write(&path, corrupt).unwrap(); + let error = open_disk(&path, OpenMode::Existing, ConnectionOptions::default()).unwrap_err(); + assert!( + matches!(error, ConnectionError::Sqlite(ref error) if classify_error(error) == ErrorClass::Corruption) + ); + assert_eq!(fs::read(&path).unwrap(), corrupt); +} + +#[test] +fn private_memory_connections_and_temporary_tables_have_no_disk_paths() { + let first = open_private_memory(ConnectionOptions::default()).unwrap(); + let second = open_private_memory(ConnectionOptions::default()).unwrap(); + first.execute_batch("CREATE TABLE saved(value); INSERT INTO saved VALUES(1); CREATE TEMP TABLE scratch(value);").unwrap(); + assert_eq!(pragma_string(&first, "journal_mode"), "memory"); + assert_eq!( + first + .pragma_query_value(None, "temp_store", |r| r.get::<_, i64>(0)) + .unwrap(), + 2 + ); + let paths = first + .prepare("PRAGMA database_list") + .unwrap() + .query_map([], |row| row.get::<_, String>(2)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(paths, [String::new(), String::new()]); + assert!(second.prepare("SELECT * FROM saved").is_err()); +} + +#[test] +fn schema_version_must_match_without_implicitly_upgrading() { + let connection = open_private_memory(ConnectionOptions::default()).unwrap(); + assert!(matches!( + require_schema_version(&connection, 1), + Err(ConnectionError::SchemaVersion { actual: 0, .. }) + )); + connection.pragma_update(None, "user_version", 1).unwrap(); + require_schema_version(&connection, 1).unwrap(); + connection.pragma_update(None, "user_version", 2).unwrap(); + assert!(matches!( + require_schema_version(&connection, 1), + Err(ConnectionError::SchemaVersion { actual: 2, .. }) + )); + assert_eq!( + connection + .pragma_query_value(None, "user_version", |r| r.get::<_, i64>(0)) + .unwrap(), + 2 + ); +} + +#[test] +fn constraint_classification_preserves_transaction_state() { + let connection = open_private_memory(ConnectionOptions::default()).unwrap(); + connection.execute_batch("CREATE TABLE saved(value INTEGER UNIQUE); BEGIN; SAVEPOINT request; INSERT INTO saved VALUES(1);").unwrap(); + let error = connection + .execute("INSERT INTO saved VALUES(1)", []) + .unwrap_err(); + assert_eq!(classify_error(&error), ErrorClass::Constraint); + assert!(!connection.is_autocommit()); + connection + .execute_batch("ROLLBACK TO request; RELEASE request; INSERT INTO saved VALUES(2); COMMIT;") + .unwrap(); + assert_eq!( + connection + .query_row("SELECT value FROM saved", [], |row| row.get::<_, i64>(0)) + .unwrap(), + 2 + ); +} + +#[test] +fn invalid_cache_budget_fails_before_any_schema_is_created() { + let options = ConnectionOptions { + cache_kib: 0, + ..ConnectionOptions::default() + }; + assert!(matches!( + open_private_memory(options), + Err(ConnectionError::InvalidCacheSize) + )); + // A path is passed verbatim, never interpreted as a SQLite URI. + let directory = TestDirectory::new(); + let path = directory.0.join(Path::new("file:literal.sqlite")); + if cfg!(unix) { + let connection = open_disk(&path, OpenMode::Create, ConnectionOptions::default()).unwrap(); + drop(connection); + assert!(path.exists()); + } +} diff --git a/moli-sqlite/src/executor.rs b/moli-sqlite/src/executor.rs new file mode 100644 index 0000000000..8a6c6154cf --- /dev/null +++ b/moli-sqlite/src/executor.rs @@ -0,0 +1,294 @@ +use std::{ + fmt, + sync::{Arc, mpsc}, + thread::{self, JoinHandle}, +}; + +use parking_lot::Mutex; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Capacity { + pub tasks: usize, + pub bytes: usize, +} + +#[derive(Debug, Clone, Copy)] +pub struct ExecutorConfig { + pub workers: usize, + pub data: Capacity, + pub control: Capacity, +} + +impl Default for ExecutorConfig { + fn default() -> Self { + Self { + workers: thread::available_parallelism() + .map_or(1, usize::from) + .min(4), + data: Capacity { + tasks: 256, + bytes: 64 * 1024 * 1024, + }, + control: Capacity { + tasks: 64, + bytes: 256 * 1024, + }, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TaskKind { + Data, + /// Commit, abort, close, and other bounded lifecycle work only. + Control, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct QueueUsage { + pub data_tasks: usize, + pub data_bytes: usize, + pub control_tasks: usize, + pub control_bytes: usize, +} + +#[derive(Debug)] +pub enum ExecutorError { + InvalidConfiguration, + Capacity, + Closed, + Spawn(std::io::Error), + WorkerPanicked, +} + +impl fmt::Display for ExecutorError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidConfiguration => f.write_str("invalid SQLite executor configuration"), + Self::Capacity => f.write_str("SQLite executor capacity exhausted"), + Self::Closed => f.write_str("SQLite executor is closed"), + Self::Spawn(error) => write!(f, "failed to start SQLite worker: {error}"), + Self::WorkerPanicked => f.write_str("SQLite worker panicked"), + } + } +} + +impl std::error::Error for ExecutorError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Spawn(error) => Some(error), + _ => None, + } + } +} + +type Job = Box; + +enum Message { + Run(Job), + Shutdown, +} + +struct QueueState { + closed: bool, + usage: QueueUsage, + senders: Vec>, +} + +struct Shared { + config: ExecutorConfig, + state: Mutex, +} + +/// The partition owns this object and explicitly shuts it down off the V8 +/// thread. Cloneable handles never own/join worker threads. +pub struct Executor { + shared: Arc, + workers: Vec>, +} + +#[derive(Clone)] +pub struct ExecutorHandle { + shared: Arc, +} + +/// Accounts for input, work in progress, and the undelivered result. Move this +/// permit into the completion message; release it only after consumption or +/// cancellation. Reserve a byte bound before retaining the command's payload. +pub struct TaskPermit { + shared: Arc, + kind: TaskKind, + bytes: usize, +} + +impl Executor { + pub fn new(config: ExecutorConfig) -> Result { + if config.workers == 0 + || config.data.tasks == 0 + || config.data.bytes == 0 + || config.control.tasks == 0 + || config.control.bytes == 0 + { + return Err(ExecutorError::InvalidConfiguration); + } + let mut executor = Self { + shared: Arc::new(Shared { + config, + state: Mutex::new(QueueState { + closed: false, + usage: QueueUsage::default(), + senders: Vec::with_capacity(config.workers), + }), + }), + workers: Vec::with_capacity(config.workers), + }; + for index in 0..config.workers { + let (sender, receiver) = mpsc::channel(); + let worker = thread::Builder::new() + .name(format!("moli-sqlite-{index}")) + .spawn(move || { + while let Ok(Message::Run(job)) = receiver.recv() { + job(); + } + }) + .map_err(ExecutorError::Spawn)?; + executor.shared.state.lock().senders.push(sender); + executor.workers.push(worker); + } + Ok(executor) + } + + pub fn handle(&self) -> ExecutorHandle { + ExecutorHandle { + shared: self.shared.clone(), + } + } + + /// Reject new submissions, drain accepted work, then join all workers. + /// Storage owners enqueue connection cleanup before calling this method. + /// Already delivered results may outlive shutdown and retain their permits. + pub fn shutdown(&mut self) -> Result<(), ExecutorError> { + { + let mut state = self.shared.state.lock(); + if !state.closed { + state.closed = true; + // Sending under the same gate as submit orders shutdown after + // every accepted job, including concurrent submitters. + for sender in &state.senders { + let _ = sender.send(Message::Shutdown); + } + } + } + let mut failed = false; + for worker in self.workers.drain(..) { + failed |= worker.join().is_err(); + } + if failed { + Err(ExecutorError::WorkerPanicked) + } else { + Ok(()) + } + } +} + +impl Drop for Executor { + fn drop(&mut self) { + let _ = self.shutdown(); + } +} + +impl ExecutorHandle { + pub fn try_reserve(&self, kind: TaskKind, bytes: usize) -> Result { + let mut state = self.shared.state.lock(); + if state.closed { + return Err(ExecutorError::Closed); + } + let (capacity, tasks, used_bytes) = match kind { + TaskKind::Data => ( + self.shared.config.data, + state.usage.data_tasks, + state.usage.data_bytes, + ), + TaskKind::Control => ( + self.shared.config.control, + state.usage.control_tasks, + state.usage.control_bytes, + ), + }; + if tasks >= capacity.tasks || bytes > capacity.bytes.saturating_sub(used_bytes) { + return Err(ExecutorError::Capacity); + } + match kind { + TaskKind::Data => { + state.usage.data_tasks += 1; + state.usage.data_bytes += bytes; + } + TaskKind::Control => { + state.usage.control_tasks += 1; + state.usage.control_bytes += bytes; + } + } + Ok(TaskPermit { + shared: self.shared.clone(), + kind, + bytes, + }) + } + + pub fn usage(&self) -> QueueUsage { + self.shared.state.lock().usage + } + + pub fn worker_count(&self) -> usize { + self.shared.config.workers + } +} + +impl TaskPermit { + pub fn reserved_bytes(&self) -> usize { + self.bytes + } + + /// Equal keys run FIFO on one worker. A transaction can retain its database + /// connection between commands while that worker serves other databases. + pub fn submit( + self, + database_key: u64, + task: impl FnOnce(TaskPermit) + Send + 'static, + ) -> Result<(), ExecutorError> { + let shared = self.shared.clone(); + let job: Job = Box::new(move || task(self)); + let result = { + let state = shared.state.lock(); + if state.closed { + Err(Message::Run(job)) + } else { + let lane = (database_key % shared.config.workers as u64) as usize; + state.senders[lane] + .send(Message::Run(job)) + .map_err(|error| error.0) + } + }; + // A failed send owns a job/permit. Drop it outside the queue lock. + result.map_err(|_| ExecutorError::Closed) + } +} + +impl Drop for TaskPermit { + fn drop(&mut self) { + let mut state = self.shared.state.lock(); + match self.kind { + TaskKind::Data => { + state.usage.data_tasks -= 1; + state.usage.data_bytes -= self.bytes; + } + TaskKind::Control => { + state.usage.control_tasks -= 1; + state.usage.control_bytes -= self.bytes; + } + } + } +} + +#[cfg(test)] +mod tests; diff --git a/moli-sqlite/src/executor/tests.rs b/moli-sqlite/src/executor/tests.rs new file mode 100644 index 0000000000..da743b60ee --- /dev/null +++ b/moli-sqlite/src/executor/tests.rs @@ -0,0 +1,217 @@ +use std::{sync::mpsc, time::Duration}; + +use super::*; + +fn config() -> ExecutorConfig { + ExecutorConfig { + workers: 2, + data: Capacity { + tasks: 8, + bytes: 1024, + }, + control: Capacity { + tasks: 2, + bytes: 128, + }, + } +} + +fn submit(handle: &ExecutorHandle, key: u64, task: impl FnOnce(TaskPermit) + Send + 'static) { + handle + .try_reserve(TaskKind::Data, 16) + .unwrap() + .submit(key, task) + .unwrap(); +} + +struct ReleaseOnDrop(Option>); + +impl Drop for ReleaseOnDrop { + fn drop(&mut self) { + if let Some(sender) = self.0.take() { + let _ = sender.send(()); + } + } +} + +#[test] +fn same_database_jobs_are_fifo_and_shutdown_drains_every_accepted_job() { + let mut executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + let (sender, receiver) = mpsc::channel(); + for value in 0..8 { + let sender = sender.clone(); + submit(&handle, 42, move |_| sender.send(value).unwrap()); + } + executor.shutdown().unwrap(); + assert_eq!( + receiver.try_iter().collect::>(), + (0..8).collect::>() + ); + assert_eq!(handle.usage(), QueueUsage::default()); + assert!(matches!( + handle.try_reserve(TaskKind::Control, 0), + Err(ExecutorError::Closed) + )); + executor.shutdown().unwrap(); +} + +#[test] +fn another_worker_progresses_while_one_worker_is_busy() { + let mut executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + let (release, wait) = mpsc::channel(); + let release_guard = ReleaseOnDrop(Some(release)); + let (started, start) = mpsc::channel(); + submit(&handle, 0, move |_| { + let _ = started.send(()); + let _ = wait.recv(); + }); + start.recv_timeout(Duration::from_secs(5)).unwrap(); + let (sender, receiver) = mpsc::channel(); + submit(&handle, 1, move |_| { + sender + .send(thread::current().name().unwrap().to_owned()) + .unwrap() + }); + let result = receiver.recv_timeout(Duration::from_secs(5)); + drop(release_guard); + executor.shutdown().unwrap(); + assert_eq!(result.unwrap(), "moli-sqlite-1"); +} + +#[test] +fn undelivered_results_retain_bytes_and_cannot_starve_control_work() { + let mut settings = config(); + settings.data = Capacity { + tasks: 1, + bytes: 32, + }; + let mut executor = Executor::new(settings).unwrap(); + let handle = executor.handle(); + let (sender, receiver) = mpsc::channel(); + handle + .try_reserve(TaskKind::Data, 32) + .unwrap() + .submit(0, move |permit| { + assert!(sender.send(permit).is_ok()); + }) + .unwrap(); + let result = receiver.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(result.reserved_bytes(), 32); + assert_eq!(handle.usage().data_bytes, 32); + assert!(matches!( + handle.try_reserve(TaskKind::Data, 0), + Err(ExecutorError::Capacity) + )); + let (sender, receiver) = mpsc::channel(); + handle + .try_reserve(TaskKind::Control, 16) + .unwrap() + .submit(0, move |_| sender.send(()).unwrap()) + .unwrap(); + receiver.recv_timeout(Duration::from_secs(5)).unwrap(); + drop(result); + executor.shutdown().unwrap(); + assert_eq!(handle.usage(), QueueUsage::default()); +} + +#[test] +fn canceled_and_closed_submissions_release_their_capacity() { + let mut executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + let canceled = handle.try_reserve(TaskKind::Data, 24).unwrap(); + drop(canceled); + assert_eq!(handle.usage(), QueueUsage::default()); + let late = handle.try_reserve(TaskKind::Data, 24).unwrap(); + executor.shutdown().unwrap(); + assert!(matches!( + late.submit(0, |_| panic!("closed submission must not run")), + Err(ExecutorError::Closed) + )); + assert_eq!(handle.usage(), QueueUsage::default()); +} + +#[test] +fn byte_budget_rejects_overflow_and_control_has_its_own_limit() { + let executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + assert!(matches!( + handle.try_reserve(TaskKind::Data, usize::MAX), + Err(ExecutorError::Capacity) + )); + let first = handle.try_reserve(TaskKind::Control, 127).unwrap(); + assert!(matches!( + handle.try_reserve(TaskKind::Control, 2), + Err(ExecutorError::Capacity) + )); + let second = handle.try_reserve(TaskKind::Control, 1).unwrap(); + assert!(matches!( + handle.try_reserve(TaskKind::Control, 0), + Err(ExecutorError::Capacity) + )); + let data = handle.try_reserve(TaskKind::Data, 1024).unwrap(); + drop((first, second, data)); + assert_eq!(handle.usage(), QueueUsage::default()); +} + +#[test] +fn independent_transactions_can_wait_between_commands_on_the_same_worker() { + let mut executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + let (sender, receiver) = mpsc::channel(); + let connections = Arc::new(Mutex::new(std::collections::BTreeMap::new())); + for key in [0, 2] { + let connections = connections.clone(); + let sender = sender.clone(); + submit(&handle, key, move |_| { + let connection = crate::open_private_memory(Default::default()).unwrap(); + connection + .execute_batch("CREATE TABLE saved(value); BEGIN; INSERT INTO saved VALUES(1);") + .unwrap(); + connections.lock().insert(key, connection); + sender.send(key).unwrap(); + }); + } + // Both transactions have started although neither has committed yet. + assert_eq!(receiver.recv_timeout(Duration::from_secs(5)).unwrap(), 0); + assert_eq!(receiver.recv_timeout(Duration::from_secs(5)).unwrap(), 2); + let owners = connections.clone(); + submit(&handle, 0, move |_| { + let connection = owners.lock().remove(&0).unwrap(); + connection.execute_batch("COMMIT").unwrap(); + }); + submit(&handle, 2, move |_| { + let connection = connections.lock().remove(&2).unwrap(); + connection.execute_batch("ROLLBACK").unwrap(); + assert_eq!( + connection + .query_row("SELECT count(*) FROM saved", [], |row| row.get::<_, i64>(0)) + .unwrap(), + 0 + ); + }); + executor.shutdown().unwrap(); +} + +#[test] +fn shutdown_racing_with_submit_never_drops_an_accepted_task() { + let mut executor = Executor::new(config()).unwrap(); + let handle = executor.handle(); + let permit = handle.try_reserve(TaskKind::Data, 1).unwrap(); + let barrier = Arc::new(std::sync::Barrier::new(2)); + let gate = barrier.clone(); + let (sender, receiver) = mpsc::channel(); + let submitter = thread::spawn(move || { + gate.wait(); + permit.submit(0, move |_| sender.send(()).unwrap()) + }); + barrier.wait(); + executor.shutdown().unwrap(); + match submitter.join().unwrap() { + Ok(()) => receiver.recv_timeout(Duration::from_secs(5)).unwrap(), + Err(ExecutorError::Closed) => assert!(receiver.recv().is_err()), + Err(error) => panic!("unexpected submit error: {error}"), + } + assert_eq!(handle.usage(), QueueUsage::default()); +} diff --git a/moli-sqlite/src/lib.rs b/moli-sqlite/src/lib.rs new file mode 100644 index 0000000000..bddf91ccde --- /dev/null +++ b/moli-sqlite/src/lib.rs @@ -0,0 +1,17 @@ +//! SQLite connection policy and bounded, serial background execution. +//! +//! Storage owners define their schemas and transactions. They assign a stable +//! worker key to each database, and keep all access to its connection on that +//! worker. No JavaScript, profile layout, or quota policy belongs here. + +mod connection; +mod executor; + +pub use connection::{ + ConnectionError, ConnectionOptions, ErrorClass, OpenMode, classify_error, open_disk, + open_private_memory, require_schema_version, +}; +pub use executor::{ + Capacity, Executor, ExecutorConfig, ExecutorError, ExecutorHandle, QueueUsage, TaskKind, + TaskPermit, +};