feat(sqlite): add bounded background executor and connection policy

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"

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",

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

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<rusqlite::Error> 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<Connection, ConnectionError> {
+    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<Connection, ConnectionError> {
+    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;

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::<Result<Vec<_>, _>>()
+        .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());
+    }
+}

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<dyn FnOnce() + Send + 'static>;
+
+enum Message {
+    Run(Job),
+    Shutdown,
+}
+
+struct QueueState {
+    closed: bool,
+    usage: QueueUsage,
+    senders: Vec<mpsc::Sender<Message>>,
+}
+
+struct Shared {
+    config: ExecutorConfig,
+    state: Mutex<QueueState>,
+}
+
+/// 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<Shared>,
+    workers: Vec<JoinHandle<()>>,
+}
+
+#[derive(Clone)]
+pub struct ExecutorHandle {
+    shared: Arc<Shared>,
+}
+
+/// 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<Shared>,
+    kind: TaskKind,
+    bytes: usize,
+}
+
+impl Executor {
+    pub fn new(config: ExecutorConfig) -> Result<Self, ExecutorError> {
+        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<TaskPermit, ExecutorError> {
+        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;

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<mpsc::Sender<()>>);
+
+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::<Vec<_>>(),
+        (0..8).collect::<Vec<_>>()
+    );
+    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());
+}

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,
+};