feat(indexeddb): implement SQL transactions and chunked record storage

Cargo.lock

@@ -3035,6 +3035,7 @@ name = "moli-indexeddb"
 version = "0.1.0"
 dependencies = [
  "moli-crypto",
+ "moli-sqlite",
  "parking_lot",
  "rusqlite",
  "serde",

moli-indexeddb/Cargo.toml

@@ -7,13 +7,12 @@ edition = "2024"
 [dependencies]
 parking_lot = "0.12.5"
 moli-crypto = { path = "../moli-crypto" }
+moli-sqlite = { path = "../moli-sqlite" }
+rusqlite = { version = "0.40.2", features = ["bundled"] }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 strum = { version = "0.28.0", features = ["derive"] }
 unicode-id-start = "1.4.0"
 
-[dev-dependencies]
-rusqlite = { version = "0.40.2", features = ["bundled"] }
-
 [lints]
 workspace = true

moli-indexeddb/src/lib.rs

@@ -25,6 +25,7 @@ mod key_path_codec;
 mod manager;
 mod options;
 mod persistence;
+pub mod sqlite;
 #[cfg(test)]
 mod sqlite_schema_tests;
 mod state;

moli-indexeddb/src/sqlite/mod.rs

@@ -0,0 +1,446 @@
+//! The SQL database owner. Every method runs on its assigned background worker.
+//!
+//! This owner keeps only committed schema metadata in memory. One Web
+//! transaction owns the connection at a time; records and rollback state stay
+//! in SQLite. The renderer supplies already extracted index keys.
+
+mod records;
+mod schema;
+mod values;
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use rusqlite::{Connection, OptionalExtension};
+
+use crate::{
+    IndexOptions, IndexedDbError, Key, ObjectStoreOptions, TransactionHandle, TransactionMode,
+};
+
+pub const SCHEMA_VERSION: u32 = 1;
+pub const CHUNK_BYTES: usize = 256 * 1024;
+const REQUEST_SAVEPOINT: &str = "SAVEPOINT idb_request";
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DatabaseIdentity {
+    pub owner_key: Vec<u8>,
+    pub name: String,
+    pub generation: [u8; 16],
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct IndexMetadata {
+    pub id: i64,
+    pub options: IndexOptions,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StoreMetadata {
+    pub id: i64,
+    pub options: ObjectStoreOptions,
+    pub indexes: BTreeMap<String, IndexMetadata>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DatabaseMetadata {
+    pub name: String,
+    pub version: u64,
+    pub usage: u64,
+    pub stores: BTreeMap<String, StoreMetadata>,
+}
+
+struct ActiveTransaction {
+    handle: TransactionHandle,
+    mode: TransactionMode,
+    stores: BTreeSet<String>,
+    pending: Option<PendingWrite>,
+}
+
+struct PendingWrite {
+    id: u64,
+    store_id: i64,
+    key: Key,
+}
+
+/// A request SAVEPOINT may survive a round trip to the renderer for inline-key
+/// injection and final serialization. Its token prevents stale completions
+/// from writing into a later request or transaction.
+#[derive(Debug, Clone)]
+pub struct PreparedWrite {
+    pub id: u64,
+    pub key: Key,
+}
+
+#[derive(Debug, Clone)]
+pub struct IndexKeys {
+    pub name: String,
+    pub keys: Vec<Key>,
+}
+
+pub struct Database {
+    connection: Connection,
+    identity: DatabaseIdentity,
+    committed: DatabaseMetadata,
+    active: Option<ActiveTransaction>,
+    next_request: u64,
+}
+
+impl Database {
+    pub fn create(
+        mut connection: Connection,
+        identity: DatabaseIdentity,
+    ) -> Result<Self, IndexedDbError> {
+        if identity.owner_key.is_empty() {
+            return Err(IndexedDbError::Data(
+                "empty IndexedDB owner identity".into(),
+            ));
+        }
+        let existing: i64 =
+            connection.query_row("SELECT count(*) FROM sqlite_schema", [], |row| row.get(0))?;
+        if existing != 0 {
+            return Err(corruption(
+                "cannot create IndexedDB over an existing schema",
+            ));
+        }
+        {
+            let transaction = connection.transaction()?;
+            transaction.execute_batch(include_str!("schema.sql"))?;
+            transaction.execute(
+                "INSERT INTO database_info VALUES(1, 'indexeddb', ?1, 1)",
+                [&identity.owner_key],
+            )?;
+            transaction.execute(
+                "INSERT INTO database_metadata(singleton, name, generation, version, state, key_format, key_path_format, value_format, usage) VALUES(1, ?1, ?2, 0, 'creating', 1, 1, 1, ?3)",
+                rusqlite::params![identity.name, identity.generation.as_slice(), sql_integer(identity.name.len() as u64 + 8)?],
+            )?;
+            transaction.commit()?;
+        }
+        Self::open(connection, identity)
+    }
+
+    pub fn open(
+        connection: Connection,
+        identity: DatabaseIdentity,
+    ) -> Result<Self, IndexedDbError> {
+        moli_sqlite::require_schema_version(&connection, SCHEMA_VERSION)
+            .map_err(|error| corruption(&error.to_string()))?;
+        let saved: Option<Vec<u8>> = connection.query_row(
+            "SELECT owner_key FROM database_info WHERE singleton=1 AND backend='indexeddb' AND identity_format=1", [], |row| row.get(0),
+        ).optional()?;
+        if saved.as_deref() != Some(identity.owner_key.as_slice()) {
+            return Err(corruption("IndexedDB owner identity mismatch"));
+        }
+        let generation: Vec<u8> = connection.query_row(
+            "SELECT generation FROM database_metadata WHERE singleton=1 AND name=?1",
+            [&identity.name],
+            |row| row.get(0),
+        )?;
+        if generation != identity.generation {
+            return Err(corruption("IndexedDB generation mismatch"));
+        }
+        let committed = schema::read_metadata(&connection)?;
+        Ok(Self {
+            connection,
+            identity,
+            committed,
+            active: None,
+            next_request: 1,
+        })
+    }
+
+    pub fn identity(&self) -> &DatabaseIdentity {
+        &self.identity
+    }
+    pub fn committed(&self) -> &DatabaseMetadata {
+        &self.committed
+    }
+    pub fn is_transaction_active(&self) -> bool {
+        self.active.is_some() && !self.connection.is_autocommit()
+    }
+
+    pub fn begin(
+        &mut self,
+        handle: TransactionHandle,
+        mode: TransactionMode,
+        stores: BTreeSet<String>,
+        upgrade_version: Option<u64>,
+    ) -> Result<(), IndexedDbError> {
+        if self.active.is_some() || !self.connection.is_autocommit() {
+            return Err(IndexedDbError::InvalidState(
+                "database already has a started transaction".into(),
+            ));
+        }
+        let version = if mode == TransactionMode::VersionChange {
+            let version = upgrade_version
+                .ok_or_else(|| IndexedDbError::Version("upgrade version is required".into()))?;
+            if version <= self.committed.version {
+                return Err(IndexedDbError::Version(
+                    "upgrade version must increase".into(),
+                ));
+            }
+            Some(sql_integer(version)?)
+        } else {
+            if upgrade_version.is_some() || self.committed.version == 0 || stores.is_empty() {
+                return Err(IndexedDbError::InvalidState(
+                    "regular transaction requires a live database and nonempty scope".into(),
+                ));
+            }
+            for name in &stores {
+                if !self.committed.stores.contains_key(name) {
+                    return Err(IndexedDbError::NotFound(format!(
+                        "object store `{name}` was not found"
+                    )));
+                }
+            }
+            None
+        };
+        self.connection
+            .execute_batch(if mode == TransactionMode::ReadOnly {
+                "BEGIN DEFERRED"
+            } else {
+                "BEGIN IMMEDIATE"
+            })?;
+        if let Some(version) = version
+            && let Err(error) = self.connection.execute(
+                "UPDATE database_metadata SET version=?1, state='live' WHERE singleton=1",
+                [version],
+            )
+        {
+            let _ = self.connection.execute_batch("ROLLBACK");
+            return Err(error.into());
+        }
+        self.active = Some(ActiveTransaction {
+            handle,
+            mode,
+            stores,
+            pending: None,
+        });
+        Ok(())
+    }
+
+    pub fn candidate_metadata(
+        &self,
+        transaction: TransactionHandle,
+    ) -> Result<DatabaseMetadata, IndexedDbError> {
+        self.transaction(transaction)?;
+        schema::read_metadata(&self.connection)
+    }
+
+    pub fn commit(
+        &mut self,
+        transaction: TransactionHandle,
+    ) -> Result<DatabaseMetadata, IndexedDbError> {
+        self.ready(transaction)?;
+        // Everything needed for publication is prepared before irreversible COMMIT.
+        let candidate = schema::read_metadata(&self.connection)?;
+        if let Err(error) = self.connection.execute_batch("COMMIT") {
+            let failure = error.into();
+            self.rollback_after_failure();
+            return Err(failure);
+        }
+        self.active = None;
+        self.committed = candidate;
+        Ok(self.committed.clone())
+    }
+
+    pub fn abort(&mut self, transaction: TransactionHandle) -> Result<(), IndexedDbError> {
+        if self
+            .active
+            .as_ref()
+            .is_none_or(|active| active.handle != transaction)
+        {
+            return Err(IndexedDbError::TransactionInactive(
+                "IndexedDB SQL transaction is not active".into(),
+            ));
+        }
+        let result = if self.connection.is_autocommit() {
+            Ok(())
+        } else {
+            self.connection.execute_batch("ROLLBACK")
+        };
+        self.active = None;
+        result.map_err(Into::into)
+    }
+
+    /// Shutdown runs on the database worker, before the profile lock is released.
+    pub fn close(mut self) -> Result<(), IndexedDbError> {
+        if !self.connection.is_autocommit() {
+            self.connection.execute_batch("ROLLBACK")?;
+        }
+        self.active = None;
+        self.connection.close().map_err(|(_, error)| error.into())
+    }
+
+    fn transaction(&self, handle: TransactionHandle) -> Result<&ActiveTransaction, IndexedDbError> {
+        self.active
+            .as_ref()
+            .filter(|tx| tx.handle == handle && !self.connection.is_autocommit())
+            .ok_or_else(|| {
+                IndexedDbError::TransactionInactive(
+                    "IndexedDB SQL transaction is not active".into(),
+                )
+            })
+    }
+
+    fn ready(&self, handle: TransactionHandle) -> Result<&ActiveTransaction, IndexedDbError> {
+        let transaction = self.transaction(handle)?;
+        if transaction.pending.is_some() {
+            return Err(IndexedDbError::InvalidState(
+                "a write request is awaiting its owner".into(),
+            ));
+        }
+        Ok(transaction)
+    }
+
+    fn writable(&self, handle: TransactionHandle) -> Result<(), IndexedDbError> {
+        if self.ready(handle)?.mode == TransactionMode::ReadOnly {
+            return Err(IndexedDbError::ReadOnly("transaction is readonly".into()));
+        }
+        Ok(())
+    }
+
+    fn store_id(&self, handle: TransactionHandle, name: &str) -> Result<i64, IndexedDbError> {
+        let transaction = self.transaction(handle)?;
+        if transaction.mode != TransactionMode::VersionChange && !transaction.stores.contains(name)
+        {
+            return Err(IndexedDbError::InvalidState(format!(
+                "transaction does not cover object store `{name}`"
+            )));
+        }
+        self.connection
+            .query_row(
+                "SELECT store_id FROM object_stores WHERE name=?1",
+                [name],
+                |row| row.get(0),
+            )
+            .optional()?
+            .ok_or_else(|| IndexedDbError::NotFound(format!("object store `{name}` was not found")))
+    }
+
+    fn request<T>(
+        &mut self,
+        transaction: TransactionHandle,
+        operation: impl FnOnce(&mut Self) -> Result<T, IndexedDbError>,
+    ) -> Result<T, IndexedDbError> {
+        self.writable(transaction)?;
+        self.open_request()?;
+        let outcome = operation(self);
+        self.finish_request(outcome)
+    }
+
+    fn open_request(&mut self) -> Result<(), IndexedDbError> {
+        if let Err(error) = self.connection.execute_batch(REQUEST_SAVEPOINT) {
+            if self.connection.is_autocommit() {
+                self.active = None;
+                return Err(IndexedDbError::TransactionInactive(format!(
+                    "SQLite aborted the transaction: {error}"
+                )));
+            }
+            return Err(error.into());
+        }
+        Ok(())
+    }
+
+    fn finish_request<T>(
+        &mut self,
+        outcome: Result<T, IndexedDbError>,
+    ) -> Result<T, IndexedDbError> {
+        match outcome {
+            Ok(value) => match self.connection.execute_batch("RELEASE idb_request") {
+                Ok(()) => Ok(value),
+                Err(error) => {
+                    self.rollback_after_failure();
+                    Err(IndexedDbError::TransactionInactive(format!(
+                        "request release failed: {error}"
+                    )))
+                }
+            },
+            Err(error) => {
+                self.rollback_request()?;
+                Err(error)
+            }
+        }
+    }
+
+    fn rollback_request(&mut self) -> Result<(), IndexedDbError> {
+        if self.connection.is_autocommit() {
+            self.active = None;
+            return Err(IndexedDbError::TransactionInactive(
+                "SQLite aborted the transaction".into(),
+            ));
+        }
+        if let Err(error) = self
+            .connection
+            .execute_batch("ROLLBACK TO idb_request; RELEASE idb_request")
+        {
+            self.rollback_after_failure();
+            return Err(IndexedDbError::TransactionInactive(format!(
+                "request rollback failed: {error}"
+            )));
+        }
+        Ok(())
+    }
+
+    fn rollback_after_failure(&mut self) {
+        if !self.connection.is_autocommit() {
+            let _ = self.connection.execute_batch("ROLLBACK");
+        }
+        self.active = None;
+    }
+
+    fn adjust_usage(&self, delta: i128) -> Result<(), IndexedDbError> {
+        let current: i64 = self.connection.query_row(
+            "SELECT usage FROM database_metadata WHERE singleton=1",
+            [],
+            |row| row.get(0),
+        )?;
+        let next = i64::try_from(i128::from(current) + delta)
+            .ok()
+            .filter(|value| *value >= 0)
+            .ok_or_else(|| corruption("IndexedDB logical usage overflow"))?;
+        self.connection.execute(
+            "UPDATE database_metadata SET usage=?1 WHERE singleton=1",
+            [next],
+        )?;
+        Ok(())
+    }
+
+    fn next_id(&self, column: &'static str) -> Result<i64, IndexedDbError> {
+        let id: i64 = self.connection.query_row(
+            &format!("SELECT {column} FROM database_metadata WHERE singleton=1"),
+            [],
+            |row| row.get(0),
+        )?;
+        let next = id
+            .checked_add(1)
+            .filter(|_| id > 0)
+            .ok_or_else(|| corruption("IndexedDB ID counter exhausted"))?;
+        self.connection.execute(
+            &format!("UPDATE database_metadata SET {column}=?1 WHERE singleton=1"),
+            [next],
+        )?;
+        Ok(id)
+    }
+}
+
+fn sql_integer(value: u64) -> Result<i64, IndexedDbError> {
+    i64::try_from(value).map_err(|_| {
+        IndexedDbError::Data("IndexedDB integer exceeds SQLite's supported range".into())
+    })
+}
+
+fn corruption(message: &str) -> IndexedDbError {
+    IndexedDbError::Corruption(message.into())
+}
+
+impl From<rusqlite::Error> for IndexedDbError {
+    fn from(error: rusqlite::Error) -> Self {
+        match moli_sqlite::classify_error(&error) {
+            moli_sqlite::ErrorClass::Constraint => Self::Constraint(error.to_string()),
+            moli_sqlite::ErrorClass::Corruption => Self::Corruption(error.to_string()),
+            _ => Self::Io(error.to_string()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests;

moli-indexeddb/src/sqlite/records.rs

@@ -0,0 +1,410 @@
+use super::*;
+use crate::{
+    IndexedDbValue,
+    transaction::MAX_AUTO_INCREMENT_KEY,
+    usage::{indexed_db_value_usage_bytes, key_usage_bytes},
+};
+
+impl Database {
+    pub fn prepare_write(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        key: Option<Key>,
+        add_only: bool,
+    ) -> Result<PreparedWrite, IndexedDbError> {
+        self.writable(transaction)?;
+        let store_id = self.store_id(transaction, store_name)?;
+        let id = self.next_request;
+        self.next_request = id
+            .checked_add(1)
+            .ok_or_else(|| IndexedDbError::InvalidState("write request token exhausted".into()))?;
+        // The SAVEPOINT includes both generated and explicit-key high-water updates.
+        self.open_request()?;
+        let outcome = (|| {
+            let (auto_increment, generator): (bool, i64) = self.connection.query_row(
+                "SELECT auto_increment, generator FROM object_stores WHERE store_id=?1",
+                [store_id],
+                |row| Ok((row.get(0)?, row.get(1)?)),
+            )?;
+            let (key, next_generator) = match key {
+                Some(key) => {
+                    key.encode_ordered()?;
+                    let mut next = generator as u64;
+                    if auto_increment && let Key::Number(number) = &key {
+                        next = next.max(
+                            number
+                                .value()
+                                .floor()
+                                .clamp(0.0, MAX_AUTO_INCREMENT_KEY as f64)
+                                as u64,
+                        );
+                    }
+                    (key, next)
+                }
+                None if auto_increment
+                    && generator >= 0
+                    && (generator as u64) < MAX_AUTO_INCREMENT_KEY =>
+                {
+                    let next = generator as u64 + 1;
+                    (Key::from(next as i64), next)
+                }
+                None if auto_increment => {
+                    return Err(IndexedDbError::Constraint(
+                        "auto_increment key generator exceeded 2^53".into(),
+                    ));
+                }
+                None => {
+                    return Err(IndexedDbError::InvalidState(
+                        "a key is required when auto_increment is disabled".into(),
+                    ));
+                }
+            };
+            if auto_increment {
+                self.connection.execute(
+                    "UPDATE object_stores SET generator=?1 WHERE store_id=?2",
+                    [sql_integer(next_generator)?, store_id],
+                )?;
+            }
+            if add_only {
+                let exists: bool = self.connection.query_row(
+                    "SELECT EXISTS(SELECT 1 FROM records WHERE store_id=?1 AND key_sort=?2)",
+                    rusqlite::params![store_id, key.encode_ordered()?],
+                    |row| row.get(0),
+                )?;
+                if exists {
+                    return Err(IndexedDbError::Constraint(
+                        "record already exists for key".into(),
+                    ));
+                }
+            }
+            Ok(key)
+        })();
+        match outcome {
+            Ok(key) => {
+                self.active.as_mut().expect("validated transaction").pending = Some(PendingWrite {
+                    id,
+                    store_id,
+                    key: key.clone(),
+                });
+                Ok(PreparedWrite { id, key })
+            }
+            Err(error) => self.finish_request(Err(error)),
+        }
+    }
+
+    pub fn cancel_write(
+        &mut self,
+        transaction: TransactionHandle,
+        request: u64,
+    ) -> Result<(), IndexedDbError> {
+        self.pending_write(transaction, request)?;
+        self.active.as_mut().expect("validated transaction").pending = None;
+        self.rollback_request()
+    }
+
+    pub fn finish_write(
+        &mut self,
+        transaction: TransactionHandle,
+        request: u64,
+        value: IndexedDbValue,
+        indexes: Vec<IndexKeys>,
+        quota: Option<u64>,
+    ) -> Result<Key, IndexedDbError> {
+        self.pending_write(transaction, request)?;
+        let pending = self
+            .active
+            .as_mut()
+            .expect("validated transaction")
+            .pending
+            .take()
+            .expect("validated request");
+        let outcome = self.write_prepared(pending, value, indexes, quota);
+        self.finish_request(outcome)
+    }
+
+    fn pending_write(
+        &self,
+        transaction: TransactionHandle,
+        request: u64,
+    ) -> Result<(), IndexedDbError> {
+        if self
+            .transaction(transaction)?
+            .pending
+            .as_ref()
+            .is_none_or(|pending| pending.id != request)
+        {
+            return Err(IndexedDbError::InvalidState(
+                "stale IndexedDB write completion".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn write_prepared(
+        &mut self,
+        pending: PendingWrite,
+        value: IndexedDbValue,
+        indexes: Vec<IndexKeys>,
+        quota: Option<u64>,
+    ) -> Result<Key, IndexedDbError> {
+        let sort = pending.key.encode_ordered()?;
+        let old: Option<(i64, Vec<u8>, i64)> = self
+            .connection
+            .query_row(
+                "SELECT record_id, key_value, usage FROM records WHERE store_id=?1 AND key_sort=?2",
+                rusqlite::params![pending.store_id, sort],
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+            )
+            .optional()?;
+        let (record, key_value, old_usage) = if let Some((record, key, usage)) = old {
+            self.remove_record_contents(record)?;
+            (record, key, usage)
+        } else {
+            (
+                self.next_id("next_record_id")?,
+                pending.key.encode_value()?,
+                0,
+            )
+        };
+        // Equivalent keys retain the original record representative, including -0.
+        let primary_key = Key::decode_value(&key_value)?;
+        let usage = sql_integer(
+            key_usage_bytes(&primary_key).saturating_add(indexed_db_value_usage_bytes(&value)),
+        )?;
+        self.connection.execute(
+            "INSERT INTO records(record_id, store_id, key_sort, key_value, wire_size, usage) VALUES(?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(record_id) DO UPDATE SET wire_size=excluded.wire_size, usage=excluded.usage",
+            rusqlite::params![record, pending.store_id, sort, key_value, sql_integer(value.wire_bytes().len() as u64)?, usage],
+        )?;
+        self.write_value(record, &value)?;
+        self.write_indexes(pending.store_id, record, &sort, indexes)?;
+        self.adjust_usage(i128::from(usage) - i128::from(old_usage))?;
+        if let Some(quota) = quota {
+            let requested: i64 = self.connection.query_row(
+                "SELECT usage FROM database_metadata WHERE singleton=1",
+                [],
+                |row| row.get(0),
+            )?;
+            if requested as u64 > quota {
+                return Err(IndexedDbError::QuotaExceeded {
+                    quota,
+                    requested: requested as u64,
+                });
+            }
+        }
+        Ok(pending.key)
+    }
+
+    fn write_indexes(
+        &self,
+        store: i64,
+        record: i64,
+        primary_sort: &[u8],
+        indexes: Vec<IndexKeys>,
+    ) -> Result<(), IndexedDbError> {
+        let definitions = self.index_definitions(store)?;
+        let mut supplied = BTreeMap::new();
+        for index in indexes {
+            if supplied.insert(index.name, index.keys).is_some() {
+                return Err(IndexedDbError::Data(
+                    "duplicate index-key declaration".into(),
+                ));
+            }
+        }
+        if supplied.len() != definitions.len() || supplied.keys().ne(definitions.keys()) {
+            return Err(IndexedDbError::InvalidState(
+                "write must supply keys for every current index".into(),
+            ));
+        }
+        for (name, (id, unique, multi_entry)) in definitions {
+            self.insert_index_keys(
+                id,
+                unique,
+                multi_entry,
+                record,
+                primary_sort,
+                supplied.remove(&name).expect("validated index keys"),
+            )?;
+        }
+        Ok(())
+    }
+
+    fn index_definitions(
+        &self,
+        store: i64,
+    ) -> Result<BTreeMap<String, (i64, bool, bool)>, IndexedDbError> {
+        let mut statement = self.connection.prepare_cached("SELECT name, index_id, is_unique, multi_entry FROM indexes WHERE store_id=?1 ORDER BY name")?;
+        Ok(statement
+            .query_map([store], |row| {
+                Ok((row.get(0)?, (row.get(1)?, row.get(2)?, row.get(3)?)))
+            })?
+            .collect::<Result<_, _>>()?)
+    }
+
+    fn insert_index_keys(
+        &self,
+        index: i64,
+        unique: bool,
+        multi_entry: bool,
+        record: i64,
+        primary: &[u8],
+        keys: Vec<Key>,
+    ) -> Result<(), IndexedDbError> {
+        if !multi_entry && keys.len() > 1 {
+            return Err(IndexedDbError::Data(
+                "a non-multiEntry index can have at most one key".into(),
+            ));
+        }
+        let mut seen = BTreeSet::new();
+        let mut insert = self
+            .connection
+            .prepare_cached("INSERT INTO index_entries VALUES(?1, ?2, ?3, ?4, ?5, ?6)")?;
+        for key in keys {
+            let sort = key.encode_ordered()?;
+            if seen.insert(sort.clone()) {
+                insert.execute(rusqlite::params![
+                    index,
+                    sort,
+                    key.encode_value()?,
+                    primary,
+                    record,
+                    unique
+                ])?;
+            }
+        }
+        Ok(())
+    }
+
+    /// A batch obtained and decoded by the versionchange owner. References and
+    /// unique flags come from this database, never from the caller's payload.
+    pub fn backfill_index_record(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        index_name: &str,
+        primary_key: &Key,
+        keys: Vec<Key>,
+    ) -> Result<(), IndexedDbError> {
+        self.versionchange(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        self.request(transaction, |owner| {
+            let (index, unique, multi_entry) = owner
+                .index_definitions(store)?
+                .remove(index_name)
+                .ok_or_else(|| {
+                IndexedDbError::NotFound(format!("index `{index_name}` was not found"))
+            })?;
+            let primary = primary_key.encode_ordered()?;
+            let record: i64 = owner.connection.query_row(
+                "SELECT record_id FROM records WHERE store_id=?1 AND key_sort=?2",
+                rusqlite::params![store, primary],
+                |row| row.get(0),
+            )?;
+            owner.connection.execute(
+                "DELETE FROM index_entries WHERE index_id=?1 AND record_id=?2",
+                [index, record],
+            )?;
+            owner.insert_index_keys(index, unique, multi_entry, record, &primary, keys)
+        })
+    }
+
+    pub fn get(
+        &self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        key: &Key,
+        max_bytes: u64,
+    ) -> Result<Option<IndexedDbValue>, IndexedDbError> {
+        self.ready(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        let record: Option<(i64, i64)> = self
+            .connection
+            .query_row(
+                "SELECT record_id, wire_size FROM records WHERE store_id=?1 AND key_sort=?2",
+                rusqlite::params![store, key.encode_ordered()?],
+                |row| Ok((row.get(0)?, row.get(1)?)),
+            )
+            .optional()?;
+        record
+            .map(|(record, wire)| {
+                if wire < 0 {
+                    return Err(corruption("negative wire byte length"));
+                }
+                self.read_value(record, wire as u64, max_bytes)
+            })
+            .transpose()
+    }
+
+    pub fn get_key(
+        &self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        key: &Key,
+    ) -> Result<Option<Key>, IndexedDbError> {
+        self.ready(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        let value: Option<Vec<u8>> = self
+            .connection
+            .query_row(
+                "SELECT key_value FROM records WHERE store_id=?1 AND key_sort=?2",
+                rusqlite::params![store, key.encode_ordered()?],
+                |row| row.get(0),
+            )
+            .optional()?;
+        value.map(|value| Key::decode_value(&value)).transpose()
+    }
+
+    pub fn count(
+        &self,
+        transaction: TransactionHandle,
+        store_name: &str,
+    ) -> Result<u64, IndexedDbError> {
+        self.ready(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        let count: i64 = self.connection.query_row(
+            "SELECT count(*) FROM records WHERE store_id=?1",
+            [store],
+            |row| row.get(0),
+        )?;
+        Ok(count as u64)
+    }
+
+    pub fn delete(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        key: &Key,
+    ) -> Result<(), IndexedDbError> {
+        self.writable(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        let sort = key.encode_ordered()?;
+        self.request(transaction, |owner| {
+            let old: Option<(i64, i64)> = owner
+                .connection
+                .query_row(
+                    "SELECT record_id, usage FROM records WHERE store_id=?1 AND key_sort=?2",
+                    rusqlite::params![store, sort],
+                    |row| Ok((row.get(0)?, row.get(1)?)),
+                )
+                .optional()?;
+            if let Some((record, usage)) = old {
+                owner.remove_record_contents(record)?;
+                owner
+                    .connection
+                    .execute("DELETE FROM records WHERE record_id=?1", [record])?;
+                owner.adjust_usage(-i128::from(usage))?;
+            }
+            Ok(())
+        })
+    }
+
+    pub fn clear(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+    ) -> Result<(), IndexedDbError> {
+        self.writable(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        self.request(transaction, |owner| owner.remove_store_records(store))
+    }
+}

moli-indexeddb/src/sqlite/schema.rs

@@ -0,0 +1,240 @@
+use super::*;
+use crate::{usage::key_path_usage_bytes, validate_index_options, validate_object_store_options};
+
+pub(super) fn read_metadata(connection: &Connection) -> Result<DatabaseMetadata, IndexedDbError> {
+    let (name, version, state, usage): (String, i64, String, i64) = connection.query_row(
+        "SELECT name, version, state, usage FROM database_metadata WHERE singleton=1 AND key_format=1 AND key_path_format=1 AND value_format=1", [],
+        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+    )?;
+    if version < 0
+        || usage < 0
+        || !matches!((state.as_str(), version), ("creating", 0) | ("live", 1..))
+    {
+        return Err(corruption("invalid IndexedDB database metadata"));
+    }
+    let mut stores = BTreeMap::new();
+    let mut statement = connection.prepare_cached(
+        "SELECT store_id, name, key_path, auto_increment FROM object_stores ORDER BY name",
+    )?;
+    let mut rows = statement.query([])?;
+    while let Some(row) = rows.next()? {
+        let id: i64 = row.get(0)?;
+        if id <= 0 {
+            return Err(corruption("invalid object store ID"));
+        }
+        let key_path: Option<Vec<u8>> = row.get(2)?;
+        let options = ObjectStoreOptions {
+            key_path: key_path
+                .map(|bytes| crate::KeyPath::decode_value(&bytes))
+                .transpose()?,
+            auto_increment: row.get(3)?,
+        };
+        validate_object_store_options(options.key_path.as_ref(), options.auto_increment)
+            .map_err(|_| corruption("invalid object store options"))?;
+        stores.insert(
+            row.get::<_, String>(1)?,
+            StoreMetadata {
+                id,
+                options,
+                indexes: BTreeMap::new(),
+            },
+        );
+    }
+    let mut statement = connection.prepare_cached("SELECT index_id, store_id, name, key_path, is_unique, multi_entry FROM indexes ORDER BY store_id, name")?;
+    let mut rows = statement.query([])?;
+    let store_names: BTreeMap<_, _> = stores
+        .iter()
+        .map(|(name, store)| (store.id, name.clone()))
+        .collect();
+    while let Some(row) = rows.next()? {
+        let id: i64 = row.get(0)?;
+        let store_id: i64 = row.get(1)?;
+        let store_name = store_names
+            .get(&store_id)
+            .ok_or_else(|| corruption("index refers to an unknown object store"))?;
+        if id <= 0 {
+            return Err(corruption("invalid index ID"));
+        }
+        let options = IndexOptions {
+            key_path: crate::KeyPath::decode_value(&row.get::<_, Vec<u8>>(3)?)?,
+            unique: row.get(4)?,
+            multi_entry: row.get(5)?,
+        };
+        validate_index_options(&options.key_path, options.multi_entry)
+            .map_err(|_| corruption("invalid index options"))?;
+        stores
+            .get_mut(store_name)
+            .expect("validated store ID")
+            .indexes
+            .insert(row.get(2)?, IndexMetadata { id, options });
+    }
+    if version == 0 && !stores.is_empty() {
+        return Err(corruption("unpublished IndexedDB contains object stores"));
+    }
+    Ok(DatabaseMetadata {
+        name,
+        version: version as u64,
+        usage: if version == 0 { 0 } else { usage as u64 },
+        stores,
+    })
+}
+
+pub(super) fn store_usage(name: &str, options: &ObjectStoreOptions) -> u64 {
+    name.len() as u64 + 9 + options.key_path.as_ref().map_or(0, key_path_usage_bytes)
+}
+
+pub(super) fn index_usage(name: &str, options: &IndexOptions) -> u64 {
+    name.len() as u64 + 2 + key_path_usage_bytes(&options.key_path)
+}
+
+impl Database {
+    pub(super) fn versionchange(
+        &self,
+        transaction: TransactionHandle,
+    ) -> Result<(), IndexedDbError> {
+        if self.ready(transaction)?.mode != TransactionMode::VersionChange {
+            return Err(IndexedDbError::InvalidState(
+                "schema changes require a versionchange transaction".into(),
+            ));
+        }
+        Ok(())
+    }
+
+    pub fn create_object_store(
+        &mut self,
+        transaction: TransactionHandle,
+        name: &str,
+        options: ObjectStoreOptions,
+    ) -> Result<(), IndexedDbError> {
+        self.versionchange(transaction)?;
+        if options
+            .key_path
+            .as_ref()
+            .is_some_and(|path| !path.is_valid())
+        {
+            return Err(IndexedDbError::Syntax("The key path is not valid.".into()));
+        }
+        let exists: bool = self.connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM object_stores WHERE name=?1)",
+            [name],
+            |row| row.get(0),
+        )?;
+        if exists {
+            return Err(IndexedDbError::Constraint(format!(
+                "object store `{name}` already exists"
+            )));
+        }
+        validate_object_store_options(options.key_path.as_ref(), options.auto_increment).map_err(
+            |_| {
+                IndexedDbError::InvalidAccess(
+                    "autoIncrement cannot use an empty or sequence key path".into(),
+                )
+            },
+        )?;
+        self.request(transaction, |owner| {
+            let id = owner.next_id("next_store_id")?;
+            let key_path = options
+                .key_path
+                .as_ref()
+                .map(crate::KeyPath::encode_value)
+                .transpose()?;
+            owner.connection.execute(
+                "INSERT INTO object_stores VALUES(?1, ?2, ?3, ?4, 0)",
+                rusqlite::params![id, name, key_path, options.auto_increment],
+            )?;
+            owner.adjust_usage(i128::from(store_usage(name, &options)))
+        })
+    }
+
+    pub fn delete_object_store(
+        &mut self,
+        transaction: TransactionHandle,
+        name: &str,
+    ) -> Result<(), IndexedDbError> {
+        self.versionchange(transaction)?;
+        let store = self.store_id(transaction, name)?;
+        self.request(transaction, |owner| {
+            let metadata = read_metadata(&owner.connection)?;
+            let metadata = metadata.stores.get(name).ok_or_else(|| corruption("missing store metadata"))?;
+            let definitions = store_usage(name, &metadata.options) + metadata.indexes.iter().map(|(name, index)| index_usage(name, &index.options)).sum::<u64>();
+            owner.remove_store_records(store)?;
+            owner.connection.execute("DELETE FROM index_entries WHERE index_id IN (SELECT index_id FROM indexes WHERE store_id=?1)", [store])?;
+            owner.connection.execute("DELETE FROM indexes WHERE store_id=?1", [store])?;
+            owner.connection.execute("DELETE FROM object_stores WHERE store_id=?1", [store])?;
+            owner.adjust_usage(-i128::from(definitions))
+        })
+    }
+
+    /// Backfill is subsequent ordered internal work in the same versionchange
+    /// transaction. No committed metadata is published until that work ends.
+    pub fn create_index(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        name: &str,
+        options: IndexOptions,
+    ) -> Result<IndexMetadata, IndexedDbError> {
+        self.versionchange(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        let exists: bool = self.connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM indexes WHERE store_id=?1 AND name=?2)",
+            rusqlite::params![store, name],
+            |row| row.get(0),
+        )?;
+        if exists {
+            return Err(IndexedDbError::Constraint(format!(
+                "index `{name}` already exists"
+            )));
+        }
+        if !options.key_path.is_valid() {
+            return Err(IndexedDbError::Syntax("The key path is not valid.".into()));
+        }
+        validate_index_options(&options.key_path, options.multi_entry).map_err(|_| {
+            IndexedDbError::InvalidAccess(
+                "multiEntry indexes cannot use a sequence key path".into(),
+            )
+        })?;
+        self.request(transaction, |owner| {
+            let id = owner.next_id("next_index_id")?;
+            owner.connection.execute(
+                "INSERT INTO indexes VALUES(?1, ?2, ?3, ?4, ?5, ?6)",
+                rusqlite::params![
+                    id,
+                    store,
+                    name,
+                    options.key_path.encode_value()?,
+                    options.unique,
+                    options.multi_entry
+                ],
+            )?;
+            owner.adjust_usage(i128::from(index_usage(name, &options)))?;
+            Ok(IndexMetadata { id, options })
+        })
+    }
+
+    pub fn delete_index(
+        &mut self,
+        transaction: TransactionHandle,
+        store_name: &str,
+        name: &str,
+    ) -> Result<(), IndexedDbError> {
+        self.versionchange(transaction)?;
+        let store = self.store_id(transaction, store_name)?;
+        self.request(transaction, |owner| {
+            let metadata = read_metadata(&owner.connection)?;
+            let index = metadata
+                .stores
+                .get(store_name)
+                .and_then(|store| store.indexes.get(name))
+                .ok_or_else(|| IndexedDbError::NotFound(format!("index `{name}` was not found")))?;
+            owner
+                .connection
+                .execute("DELETE FROM index_entries WHERE index_id=?1", [index.id])?;
+            owner.connection.execute(
+                "DELETE FROM indexes WHERE store_id=?1 AND index_id=?2",
+                [store, index.id],
+            )?;
+            owner.adjust_usage(-i128::from(index_usage(name, &index.options)))
+        })
+    }
+}

moli-indexeddb/src/sqlite/tests.rs

@@ -0,0 +1,776 @@
+use super::*;
+use crate::transaction::MAX_AUTO_INCREMENT_KEY;
+use crate::{IndexedDbExternalObject as External, IndexedDbValue, KeyPath};
+
+const TX: TransactionHandle = TransactionHandle::from_raw(1);
+
+fn identity() -> DatabaseIdentity {
+    DatabaseIdentity {
+        owner_key: b"partition/default/storage-key/top-site/bucket-17".to_vec(),
+        name: "database".into(),
+        generation: [42; 16],
+    }
+}
+
+fn empty() -> Database {
+    Database::create(
+        moli_sqlite::open_private_memory(Default::default()).unwrap(),
+        identity(),
+    )
+    .unwrap()
+}
+
+fn database() -> Database {
+    let mut database = empty();
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(1))
+        .unwrap();
+    for name in ["a", "b"] {
+        database
+            .create_object_store(
+                TX,
+                name,
+                ObjectStoreOptions {
+                    auto_increment: true,
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+    }
+    database.commit(TX).unwrap();
+    database
+}
+
+fn begin(database: &mut Database, mode: TransactionMode) {
+    database
+        .begin(TX, mode, BTreeSet::from(["a".into()]), None)
+        .unwrap();
+}
+
+fn put(
+    database: &mut Database,
+    key: Option<Key>,
+    bytes: &[u8],
+    indexes: Vec<IndexKeys>,
+    quota: Option<u64>,
+) -> Result<Key, IndexedDbError> {
+    let request = database.prepare_write(TX, "a", key, false)?;
+    database.finish_write(TX, request.id, bytes.to_vec().into(), indexes, quota)
+}
+
+fn index_keys(key: &str) -> Vec<IndexKeys> {
+    vec![IndexKeys {
+        name: "unique".into(),
+        keys: vec![key.into()],
+    }]
+}
+
+fn add_unique_index(database: &mut Database) {
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(2))
+        .unwrap();
+    database
+        .create_index(
+            TX,
+            "a",
+            "unique",
+            IndexOptions {
+                key_path: KeyPath::from("value"),
+                unique: true,
+                multi_entry: false,
+            },
+        )
+        .unwrap();
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn committed_schema_and_usage_publish_only_after_commit_and_same_database_is_serial() {
+    let mut database = empty();
+    assert_eq!(database.committed().version, 0);
+    assert_eq!(database.committed().usage, 0);
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(1))
+        .unwrap();
+    database
+        .create_object_store(TX, "a", ObjectStoreOptions::default())
+        .unwrap();
+    assert_eq!(database.committed().version, 0);
+    assert!(database.committed().stores.is_empty());
+    assert_eq!(
+        database.candidate_metadata(TX).unwrap().usage,
+        8 + 8 + 1 + 9
+    );
+    database.commit(TX).unwrap();
+    let committed = database.committed().clone();
+    assert_eq!(committed.version, 1);
+    begin(&mut database, TransactionMode::ReadOnly);
+    assert!(
+        database
+            .begin(
+                TransactionHandle::from_raw(2),
+                TransactionMode::ReadOnly,
+                BTreeSet::from(["a".into()]),
+                None
+            )
+            .is_err()
+    );
+    assert!(matches!(
+        database.prepare_write(TX, "a", Some(1.into()), false),
+        Err(IndexedDbError::ReadOnly(_))
+    ));
+    database.commit(TX).unwrap();
+    assert_eq!(database.committed(), &committed);
+}
+
+#[test]
+fn readonly_scope_and_transaction_identity_are_enforced_before_sql() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadOnly);
+    assert!(matches!(
+        database.count(TX, "b"),
+        Err(IndexedDbError::InvalidState(_))
+    ));
+    assert!(matches!(
+        database.count(TransactionHandle::from_raw(99), "a"),
+        Err(IndexedDbError::TransactionInactive(_))
+    ));
+    assert!(matches!(
+        database.clear(TX, "a"),
+        Err(IndexedDbError::ReadOnly(_))
+    ));
+    database.abort(TX).unwrap();
+}
+
+#[test]
+fn quota_failure_rolls_back_generated_key_value_and_usage_but_allows_next_request() {
+    let mut database = database();
+    let initial = database.committed().usage;
+    begin(&mut database, TransactionMode::ReadWrite);
+    assert!(matches!(
+        put(&mut database, None, &[1; 100], vec![], Some(initial)),
+        Err(IndexedDbError::QuotaExceeded { .. })
+    ));
+    assert!(database.is_transaction_active());
+    assert_eq!(database.count(TX, "a").unwrap(), 0);
+    assert_eq!(database.candidate_metadata(TX).unwrap().usage, initial);
+    assert_eq!(
+        put(&mut database, None, b"saved", vec![], None).unwrap(),
+        Key::from(1)
+    );
+    assert_eq!(database.committed().usage, initial);
+    database.commit(TX).unwrap();
+    assert_eq!(database.committed().usage, initial + 8 + 5);
+}
+
+#[test]
+fn unique_conflict_restores_old_external_objects_indexes_and_id_counters() {
+    let mut database = database();
+    add_unique_index(&mut database);
+    begin(&mut database, TransactionMode::ReadWrite);
+    let first = database.prepare_write(TX, "a", None, false).unwrap();
+    let value = IndexedDbValue::new(
+        vec![11],
+        vec![External::Blob {
+            bytes: vec![7; CHUNK_BYTES + 17],
+            mime_type: "a/b".into(),
+        }],
+    );
+    database
+        .finish_write(TX, first.id, value.clone(), index_keys("first"), None)
+        .unwrap();
+    put(&mut database, None, b"second", index_keys("second"), None).unwrap();
+    let usage = database.candidate_metadata(TX).unwrap().usage;
+    let next_object: i64 = database
+        .connection
+        .query_row("SELECT next_object_id FROM database_metadata", [], |row| {
+            row.get(0)
+        })
+        .unwrap();
+    let replacement = database
+        .prepare_write(TX, "a", Some(1.into()), false)
+        .unwrap();
+    let other = IndexedDbValue::new(
+        vec![9],
+        vec![External::File {
+            bytes: vec![8; CHUNK_BYTES * 2 + 1],
+            mime_type: "c/d".into(),
+            name: "replacement".into(),
+            last_modified: 3.5,
+        }],
+    );
+    assert!(matches!(
+        database.finish_write(TX, replacement.id, other, index_keys("second"), None),
+        Err(IndexedDbError::Constraint(_))
+    ));
+    assert_eq!(
+        database.get(TX, "a", &1.into(), u64::MAX).unwrap(),
+        Some(value)
+    );
+    assert_eq!(database.candidate_metadata(TX).unwrap().usage, usage);
+    assert_eq!(
+        database
+            .connection
+            .query_row("SELECT next_object_id FROM database_metadata", [], |row| {
+                row.get::<_, i64>(0)
+            })
+            .unwrap(),
+        next_object
+    );
+    assert_eq!(
+        database
+            .connection
+            .query_row("SELECT count(*) FROM index_entries", [], |row| row
+                .get::<_, i64>(0))
+            .unwrap(),
+        2
+    );
+    assert_eq!(
+        put(&mut database, None, b"third", index_keys("third"), None).unwrap(),
+        Key::from(3)
+    );
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn failed_generated_unique_write_and_owner_cancellation_reuse_the_key() {
+    let mut database = database();
+    add_unique_index(&mut database);
+    begin(&mut database, TransactionMode::ReadWrite);
+    put(&mut database, None, b"one", index_keys("same"), None).unwrap();
+    assert!(matches!(
+        put(&mut database, None, b"bad", index_keys("same"), None),
+        Err(IndexedDbError::Constraint(_))
+    ));
+    let canceled = database.prepare_write(TX, "a", None, false).unwrap();
+    assert_eq!(canceled.key, Key::from(2));
+    assert!(database.commit(TX).is_err());
+    assert!(database.count(TX, "a").is_err());
+    database.cancel_write(TX, canceled.id).unwrap();
+    let current = database.prepare_write(TX, "a", None, false).unwrap();
+    assert_eq!(current.key, Key::from(2));
+    assert!(matches!(
+        database.finish_write(TX, canceled.id, vec![0].into(), index_keys("new"), None),
+        Err(IndexedDbError::InvalidState(_))
+    ));
+    database
+        .finish_write(TX, current.id, vec![2].into(), index_keys("new"), None)
+        .unwrap();
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn explicit_key_advancement_and_generator_boundary_rollback_are_atomic() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    let usage = database.committed().usage;
+    assert!(
+        put(
+            &mut database,
+            Some(999.into()),
+            b"too large",
+            vec![],
+            Some(usage)
+        )
+        .is_err()
+    );
+    assert_eq!(
+        put(&mut database, None, b"one", vec![], None).unwrap(),
+        Key::from(1)
+    );
+    database
+        .connection
+        .execute(
+            "UPDATE object_stores SET generator=?1 WHERE name='a'",
+            [(MAX_AUTO_INCREMENT_KEY - 1) as i64],
+        )
+        .unwrap();
+    let request = database.prepare_write(TX, "a", None, false).unwrap();
+    assert_eq!(request.key, Key::from(MAX_AUTO_INCREMENT_KEY as i64));
+    database.cancel_write(TX, request.id).unwrap();
+    assert_eq!(
+        put(&mut database, None, b"last", vec![], None).unwrap(),
+        Key::from(MAX_AUTO_INCREMENT_KEY as i64)
+    );
+    assert!(matches!(
+        database.prepare_write(TX, "a", None, false),
+        Err(IndexedDbError::Constraint(_))
+    ));
+    assert!(database.is_transaction_active());
+    database.abort(TX).unwrap();
+    begin(&mut database, TransactionMode::ReadWrite);
+    assert_eq!(
+        put(&mut database, None, b"one again", vec![], None).unwrap(),
+        Key::from(1)
+    );
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn equal_zero_keys_preserve_record_representation_until_delete() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    let negative = Key::number(-0.0).unwrap();
+    let positive = Key::number(0.0).unwrap();
+    put(&mut database, Some(negative), b"first", vec![], None).unwrap();
+    let returned = put(
+        &mut database,
+        Some(positive.clone()),
+        b"second",
+        vec![],
+        None,
+    )
+    .unwrap();
+    let Key::Number(returned) = returned else {
+        panic!()
+    };
+    assert_eq!(returned.value().to_bits(), 0.0_f64.to_bits());
+    let Key::Number(stored) = database.get_key(TX, "a", &positive).unwrap().unwrap() else {
+        panic!()
+    };
+    assert_eq!(stored.value().to_bits(), (-0.0_f64).to_bits());
+    database.delete(TX, "a", &positive).unwrap();
+    put(
+        &mut database,
+        Some(positive.clone()),
+        b"third",
+        vec![],
+        None,
+    )
+    .unwrap();
+    let Key::Number(stored) = database.get_key(TX, "a", &positive).unwrap().unwrap() else {
+        panic!()
+    };
+    assert_eq!(stored.value().to_bits(), 0.0_f64.to_bits());
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn upgrade_abort_restores_schema_data_generator_and_committed_usage() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    put(&mut database, None, b"original", vec![], None).unwrap();
+    database.commit(TX).unwrap();
+    let original = database.committed().clone();
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(2))
+        .unwrap();
+    database.delete_object_store(TX, "a").unwrap();
+    database
+        .create_object_store(
+            TX,
+            "a",
+            ObjectStoreOptions {
+                auto_increment: true,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    put(&mut database, None, b"replacement", vec![], None).unwrap();
+    database.abort(TX).unwrap();
+    assert_eq!(database.committed(), &original);
+    begin(&mut database, TransactionMode::ReadWrite);
+    assert_eq!(
+        database
+            .get(TX, "a", &1.into(), u64::MAX)
+            .unwrap()
+            .unwrap()
+            .wire_bytes(),
+        b"original"
+    );
+    assert_eq!(
+        put(&mut database, None, b"next", vec![], None).unwrap(),
+        Key::from(2)
+    );
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn sqlite_automatic_rollback_is_fatal_instead_of_a_recoverable_request_error() {
+    let mut database = database();
+    // Install outside the tested transaction so the injected rollback does not
+    // also roll back the trigger's creation.
+    database.connection.execute_batch("CREATE TEMP TRIGGER fail_write BEFORE INSERT ON main.record_chunks BEGIN SELECT RAISE(ROLLBACK, 'injected transaction rollback'); END").unwrap();
+    begin(&mut database, TransactionMode::ReadWrite);
+    // A zero-length value has no record_chunks row and precedes the fault.
+    put(&mut database, None, b"", vec![], None).unwrap();
+    assert!(matches!(
+        put(&mut database, None, b"failure", vec![], None),
+        Err(IndexedDbError::TransactionInactive(_))
+    ));
+    assert!(!database.is_transaction_active());
+    assert!(matches!(
+        database.count(TX, "a"),
+        Err(IndexedDbError::TransactionInactive(_))
+    ));
+    database
+        .connection
+        .execute_batch("DROP TRIGGER fail_write")
+        .unwrap();
+    begin(&mut database, TransactionMode::ReadWrite);
+    assert_eq!(database.count(TX, "a").unwrap(), 0);
+    assert_eq!(
+        put(&mut database, None, b"fresh", vec![], None).unwrap(),
+        Key::from(1)
+    );
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn chunked_values_roundtrip_and_delivered_bytes_survive_abort_and_close() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    let value = IndexedDbValue::new(
+        vec![13; CHUNK_BYTES * 2 + 7],
+        vec![
+            External::Blob {
+                bytes: vec![],
+                mime_type: "empty/type".into(),
+            },
+            External::File {
+                bytes: vec![17; CHUNK_BYTES + 19],
+                mime_type: "file/type".into(),
+                name: "file.name".into(),
+                last_modified: f64::from_bits(0x7ff8_0000_0000_002a),
+            },
+            External::FileSystemHandle {
+                kind: crate::IndexedDbFileSystemHandleKind::Directory,
+                bucket: crate::IndexedDbFileSystemHandleBucket::Named { bucket_id: 17 },
+                path: vec!["root".into(), "file".into()],
+            },
+        ],
+    );
+    let request = database.prepare_write(TX, "a", None, false).unwrap();
+    database
+        .finish_write(TX, request.id, value.clone(), vec![], None)
+        .unwrap();
+    assert!(database.get(TX, "a", &1.into(), 16).is_err());
+    let delivered = database.get(TX, "a", &1.into(), u64::MAX).unwrap().unwrap();
+    assert_eq!(delivered.wire_bytes(), value.wire_bytes());
+    assert_eq!(
+        &delivered.external_objects()[0],
+        &value.external_objects()[0]
+    );
+    assert_eq!(
+        &delivered.external_objects()[2],
+        &value.external_objects()[2]
+    );
+    let External::File {
+        bytes,
+        name,
+        last_modified,
+        ..
+    } = &delivered.external_objects()[1]
+    else {
+        panic!()
+    };
+    assert_eq!(bytes, &vec![17; CHUNK_BYTES + 19]);
+    assert_eq!(name, "file.name");
+    assert_eq!(last_modified.to_bits(), 0x7ff8_0000_0000_002a);
+    database.abort(TX).unwrap();
+    database.close().unwrap();
+    assert_eq!(delivered.wire_bytes(), &vec![13; CHUNK_BYTES * 2 + 7]);
+}
+
+#[test]
+fn key_only_queries_do_not_decode_corrupt_wire_chunks() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    put(&mut database, None, b"payload", vec![], None).unwrap();
+    database
+        .connection
+        .execute("DELETE FROM record_chunks", [])
+        .unwrap();
+    assert_eq!(
+        database.get_key(TX, "a", &1.into()).unwrap(),
+        Some(1.into())
+    );
+    assert_eq!(database.count(TX, "a").unwrap(), 1);
+    assert!(matches!(
+        database.get(TX, "a", &1.into(), u64::MAX),
+        Err(IndexedDbError::Corruption(_))
+    ));
+    database.abort(TX).unwrap();
+}
+
+#[test]
+fn clearing_records_reclaims_chunks_slots_and_indexes_without_resetting_generator() {
+    let mut database = database();
+    add_unique_index(&mut database);
+    let base_usage = database.committed().usage;
+    begin(&mut database, TransactionMode::ReadWrite);
+    let request = database.prepare_write(TX, "a", None, false).unwrap();
+    database
+        .finish_write(
+            TX,
+            request.id,
+            IndexedDbValue::new(
+                vec![1],
+                vec![External::Blob {
+                    bytes: vec![2; CHUNK_BYTES + 1],
+                    mime_type: "x".into(),
+                }],
+            ),
+            index_keys("saved"),
+            None,
+        )
+        .unwrap();
+    database.clear(TX, "a").unwrap();
+    for table in [
+        "records",
+        "record_chunks",
+        "index_entries",
+        "record_objects",
+        "external_objects",
+        "blob_chunks",
+        "handle_components",
+    ] {
+        assert_eq!(
+            database
+                .connection
+                .query_row(&format!("SELECT count(*) FROM {table}"), [], |row| row
+                    .get::<_, i64>(0))
+                .unwrap(),
+            0,
+            "{table}"
+        );
+    }
+    assert_eq!(database.candidate_metadata(TX).unwrap().usage, base_usage);
+    assert_eq!(
+        put(&mut database, None, b"two", index_keys("new"), None).unwrap(),
+        Key::from(2)
+    );
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn index_backfill_checks_uniqueness_and_upgrade_abort_removes_partial_index() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    put(&mut database, None, b"one", vec![], None).unwrap();
+    put(&mut database, None, b"two", vec![], None).unwrap();
+    database.commit(TX).unwrap();
+    let original = database.committed().clone();
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(2))
+        .unwrap();
+    database
+        .create_index(
+            TX,
+            "a",
+            "unique",
+            IndexOptions {
+                key_path: "value".into(),
+                unique: true,
+                multi_entry: false,
+            },
+        )
+        .unwrap();
+    database
+        .backfill_index_record(TX, "a", "unique", &1.into(), vec!["same".into()])
+        .unwrap();
+    assert!(matches!(
+        database.backfill_index_record(TX, "a", "unique", &2.into(), vec!["same".into()]),
+        Err(IndexedDbError::Constraint(_))
+    ));
+    database.abort(TX).unwrap();
+    assert_eq!(database.committed(), &original);
+    assert_eq!(
+        database
+            .connection
+            .query_row("SELECT count(*) FROM indexes", [], |row| row
+                .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+    assert_eq!(
+        database
+            .connection
+            .query_row("SELECT count(*) FROM index_entries", [], |row| row
+                .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+}
+
+#[test]
+fn integer_overflow_and_stale_generation_are_rejected_without_publishing_data() {
+    let mut database = empty();
+    assert!(matches!(
+        database.begin(
+            TX,
+            TransactionMode::VersionChange,
+            BTreeSet::new(),
+            Some(u64::MAX)
+        ),
+        Err(IndexedDbError::Data(_))
+    ));
+    assert_eq!(database.committed().version, 0);
+    let connection = database.connection;
+    let mut wrong = identity();
+    wrong.generation[0] ^= 1;
+    assert!(matches!(
+        Database::open(connection, wrong),
+        Err(IndexedDbError::Corruption(_))
+    ));
+}
+
+#[test]
+fn multi_entry_deduplication_uses_key_equality_and_caller_cannot_disable_uniqueness() {
+    let mut database = database();
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(2))
+        .unwrap();
+    database
+        .create_index(
+            TX,
+            "a",
+            "multi",
+            IndexOptions {
+                key_path: "values".into(),
+                unique: true,
+                multi_entry: true,
+            },
+        )
+        .unwrap();
+    database.commit(TX).unwrap();
+    begin(&mut database, TransactionMode::ReadWrite);
+    let keys = || {
+        vec![IndexKeys {
+            name: "multi".into(),
+            keys: vec![
+                Key::number(-0.0).unwrap(),
+                Key::number(0.0).unwrap(),
+                "another".into(),
+                "another".into(),
+            ],
+        }]
+    };
+    put(&mut database, None, b"one", keys(), None).unwrap();
+    assert_eq!(
+        database
+            .connection
+            .query_row("SELECT count(*) FROM index_entries", [], |row| row
+                .get::<_, i64>(0))
+            .unwrap(),
+        2
+    );
+    assert!(matches!(
+        put(&mut database, None, b"two", keys(), None),
+        Err(IndexedDbError::Constraint(_))
+    ));
+    assert_eq!(database.count(TX, "a").unwrap(), 1);
+    database.commit(TX).unwrap();
+}
+
+#[test]
+fn missing_external_slots_and_noncontiguous_chunks_are_reported_as_corruption() {
+    let mut database = database();
+    begin(&mut database, TransactionMode::ReadWrite);
+    let request = database.prepare_write(TX, "a", None, false).unwrap();
+    database
+        .finish_write(
+            TX,
+            request.id,
+            IndexedDbValue::new(
+                vec![1; CHUNK_BYTES + 1],
+                vec![External::Blob {
+                    bytes: vec![2],
+                    mime_type: "x".into(),
+                }],
+            ),
+            vec![],
+            None,
+        )
+        .unwrap();
+    database.commit(TX).unwrap();
+    for damage in [
+        "DELETE FROM external_objects",
+        "UPDATE record_objects SET ordinal=1",
+        "DELETE FROM record_chunks WHERE chunk_no=0",
+        "UPDATE record_chunks SET bytes=x'00' WHERE chunk_no=0",
+    ] {
+        begin(&mut database, TransactionMode::ReadWrite);
+        database.connection.execute_batch(damage).unwrap();
+        assert!(
+            matches!(
+                database.get(TX, "a", &1.into(), u64::MAX),
+                Err(IndexedDbError::Corruption(_))
+            ),
+            "{damage}"
+        );
+        database.abort(TX).unwrap();
+    }
+}
+
+#[test]
+fn disk_owner_recovers_committed_schema_values_and_high_water_mark() {
+    struct Directory(std::path::PathBuf);
+    impl Drop for Directory {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+    let mut token = [0; 16];
+    moli_crypto::fill_secure_random(&mut token).unwrap();
+    let directory = Directory(
+        std::env::temp_dir().join(format!("moli-idb-sql-{}", moli_crypto::sha256_hex(token))),
+    );
+    std::fs::create_dir(&directory.0).unwrap();
+    let path = directory.0.join("db.sqlite");
+    let connection =
+        moli_sqlite::open_disk(&path, moli_sqlite::OpenMode::Create, Default::default()).unwrap();
+    let mut database = Database::create(connection, identity()).unwrap();
+    database
+        .begin(TX, TransactionMode::VersionChange, BTreeSet::new(), Some(1))
+        .unwrap();
+    database
+        .create_object_store(
+            TX,
+            "a",
+            ObjectStoreOptions {
+                auto_increment: true,
+                ..Default::default()
+            },
+        )
+        .unwrap();
+    put(&mut database, None, b"persisted", vec![], None).unwrap();
+    database.commit(TX).unwrap();
+    let original = database.committed().clone();
+    begin(&mut database, TransactionMode::ReadWrite);
+    put(&mut database, None, b"abandoned", vec![], None).unwrap();
+    database.close().unwrap();
+    let connection =
+        moli_sqlite::open_disk(&path, moli_sqlite::OpenMode::Existing, Default::default()).unwrap();
+    let mut database = Database::open(connection, identity()).unwrap();
+    assert_eq!(database.committed(), &original);
+    begin(&mut database, TransactionMode::ReadWrite);
+    assert_eq!(database.count(TX, "a").unwrap(), 1);
+    assert_eq!(
+        database
+            .get(TX, "a", &1.into(), u64::MAX)
+            .unwrap()
+            .unwrap()
+            .wire_bytes(),
+        b"persisted"
+    );
+    assert_eq!(
+        put(&mut database, None, b"next", vec![], None).unwrap(),
+        Key::from(2)
+    );
+    database.commit(TX).unwrap();
+    database.close().unwrap();
+    let connection =
+        moli_sqlite::open_disk(&path, moli_sqlite::OpenMode::Existing, Default::default()).unwrap();
+    let mut wrong = identity();
+    wrong.owner_key.push(0);
+    assert!(matches!(
+        Database::open(connection, wrong),
+        Err(IndexedDbError::Corruption(_))
+    ));
+    let connection =
+        moli_sqlite::open_disk(&path, moli_sqlite::OpenMode::Existing, Default::default()).unwrap();
+    let database = Database::open(connection, identity()).unwrap();
+    assert_eq!(database.committed().usage, original.usage + 8 + 4);
+    database.close().unwrap();
+}

moli-indexeddb/src/sqlite/values.rs

@@ -0,0 +1,269 @@
+use rusqlite::types::ValueRef;
+
+use super::*;
+use crate::{
+    IndexedDbExternalObject as External, IndexedDbFileSystemHandleBucket as Bucket,
+    IndexedDbFileSystemHandleKind as HandleKind, IndexedDbValue,
+};
+
+struct ReadBudget(u64);
+
+impl ReadBudget {
+    fn charge(&mut self, bytes: u64) -> Result<(), IndexedDbError> {
+        self.0 = self.0.checked_sub(bytes).ok_or_else(|| {
+            IndexedDbError::Io("IndexedDB result exceeds its reserved byte budget".into())
+        })?;
+        Ok(())
+    }
+}
+
+impl Database {
+    pub(super) fn write_value(
+        &self,
+        record: i64,
+        value: &IndexedDbValue,
+    ) -> Result<(), IndexedDbError> {
+        self.write_chunks("record_chunks", "record_id", record, value.wire_bytes())?;
+        for (ordinal, object) in value.external_objects().iter().enumerate() {
+            let id = self.next_id("next_object_id")?;
+            match object {
+                External::Blob { bytes, mime_type } => {
+                    self.connection.execute("INSERT INTO external_objects(object_id, kind, size, mime_type) VALUES(?1, 'blob', ?2, ?3)", rusqlite::params![id, sql_integer(bytes.len() as u64)?, mime_type])?;
+                    self.write_chunks("blob_chunks", "object_id", id, bytes)?;
+                }
+                External::File {
+                    bytes,
+                    mime_type,
+                    name,
+                    last_modified,
+                } => {
+                    self.connection.execute("INSERT INTO external_objects(object_id, kind, size, mime_type, file_name, last_modified_f64) VALUES(?1, 'file', ?2, ?3, ?4, ?5)", rusqlite::params![id, sql_integer(bytes.len() as u64)?, mime_type, name, last_modified.to_bits().to_be_bytes().as_slice()])?;
+                    self.write_chunks("blob_chunks", "object_id", id, bytes)?;
+                }
+                External::FileSystemHandle { kind, bucket, path } => {
+                    let kind = match kind {
+                        HandleKind::File => "file",
+                        HandleKind::Directory => "directory",
+                    };
+                    let (bucket_kind, bucket_id) = match bucket {
+                        Bucket::Default => ("default", None),
+                        Bucket::Named { bucket_id } => ("named", Some(sql_integer(*bucket_id)?)),
+                    };
+                    if bucket_id == Some(0) {
+                        return Err(IndexedDbError::Data(
+                            "named bucket identity must be positive".into(),
+                        ));
+                    }
+                    self.connection.execute("INSERT INTO external_objects(object_id, kind, size, handle_kind, handle_bucket_kind, handle_bucket_id) VALUES(?1, 'file-system-handle', 0, ?2, ?3, ?4)", rusqlite::params![id, kind, bucket_kind, bucket_id])?;
+                    let mut insert = self
+                        .connection
+                        .prepare_cached("INSERT INTO handle_components VALUES(?1, ?2, ?3)")?;
+                    for (ordinal, component) in path.iter().enumerate() {
+                        insert.execute(rusqlite::params![
+                            id,
+                            sql_integer(ordinal as u64)?,
+                            component
+                        ])?;
+                    }
+                }
+            }
+            self.connection.execute(
+                "INSERT INTO record_objects VALUES(?1, ?2, ?3)",
+                rusqlite::params![record, sql_integer(ordinal as u64)?, id],
+            )?;
+        }
+        Ok(())
+    }
+
+    fn write_chunks(
+        &self,
+        table: &'static str,
+        column: &'static str,
+        id: i64,
+        bytes: &[u8],
+    ) -> Result<(), IndexedDbError> {
+        let mut insert = self.connection.prepare_cached(&format!(
+            "INSERT INTO {table}({column}, chunk_no, bytes) VALUES(?1, ?2, ?3)"
+        ))?;
+        for (ordinal, chunk) in bytes.chunks(CHUNK_BYTES).enumerate() {
+            insert.execute(rusqlite::params![id, sql_integer(ordinal as u64)?, chunk])?;
+        }
+        Ok(())
+    }
+
+    /// The delivered value owns its final bytes; it does not borrow the SQLite
+    /// connection or retain a record after deletion/rollback/database close.
+    pub(super) fn read_value(
+        &self,
+        record: i64,
+        wire_size: u64,
+        max_bytes: u64,
+    ) -> Result<IndexedDbValue, IndexedDbError> {
+        let mut budget = ReadBudget(max_bytes);
+        let wire =
+            self.read_chunks("record_chunks", "record_id", record, wire_size, &mut budget)?;
+        let mut objects = Vec::new();
+        let mut mapping = self.connection.prepare_cached(
+            "SELECT ordinal, object_id FROM record_objects WHERE record_id=?1 ORDER BY ordinal",
+        )?;
+        let mut rows = mapping.query([record])?;
+        while let Some(row) = rows.next()? {
+            let ordinal: i64 = row.get(0)?;
+            let id: i64 = row.get(1)?;
+            if ordinal != objects.len() as i64 || id <= 0 {
+                return Err(corruption("invalid IndexedDB external object slot"));
+            }
+            budget.charge(std::mem::size_of::<External>() as u64)?;
+            let object = self.read_object(id, &mut budget)?;
+            objects.try_reserve(1).map_err(|_| {
+                IndexedDbError::Io("unable to allocate IndexedDB external objects".into())
+            })?;
+            objects.push(object);
+        }
+        Ok(IndexedDbValue::new(wire, objects))
+    }
+
+    fn read_object(&self, id: i64, budget: &mut ReadBudget) -> Result<External, IndexedDbError> {
+        let mut statement = self.connection.prepare_cached("SELECT kind, size, mime_type, file_name, last_modified_f64, handle_kind, handle_bucket_kind, handle_bucket_id FROM external_objects WHERE object_id=?1")?;
+        let mut rows = statement.query([id])?;
+        let row = rows
+            .next()?
+            .ok_or_else(|| corruption("missing IndexedDB external object"))?;
+        let kind: String = row.get(0)?;
+        let size: i64 = row.get(1)?;
+        if size < 0 {
+            return Err(corruption("negative external object size"));
+        }
+        match kind.as_str() {
+            "blob" | "file" => {
+                let mime_type: String = row.get(2)?;
+                budget.charge(mime_type.len() as u64)?;
+                let bytes =
+                    self.read_chunks("blob_chunks", "object_id", id, size as u64, budget)?;
+                if kind == "blob" {
+                    return Ok(External::Blob { bytes, mime_type });
+                }
+                let name: String = row.get(3)?;
+                budget.charge(name.len() as u64 + 8)?;
+                let bits: Vec<u8> = row.get(4)?;
+                let bits: [u8; 8] = bits
+                    .try_into()
+                    .map_err(|_| corruption("invalid File lastModified bits"))?;
+                Ok(External::File {
+                    bytes,
+                    mime_type,
+                    name,
+                    last_modified: f64::from_bits(u64::from_be_bytes(bits)),
+                })
+            }
+            "file-system-handle" => {
+                if size != 0 {
+                    return Err(corruption("filesystem handle contains a body"));
+                }
+                let kind = match row.get::<_, String>(5)?.as_str() {
+                    "file" => HandleKind::File,
+                    "directory" => HandleKind::Directory,
+                    _ => return Err(corruption("invalid filesystem handle kind")),
+                };
+                let bucket = match (
+                    row.get::<_, String>(6)?.as_str(),
+                    row.get::<_, Option<i64>>(7)?,
+                ) {
+                    ("default", None) => Bucket::Default,
+                    ("named", Some(id)) if id > 0 => {
+                        budget.charge(8)?;
+                        Bucket::Named {
+                            bucket_id: id as u64,
+                        }
+                    }
+                    _ => return Err(corruption("invalid filesystem handle bucket")),
+                };
+                let mut path = Vec::new();
+                let mut statement = self.connection.prepare_cached("SELECT ordinal, name FROM handle_components WHERE object_id=?1 ORDER BY ordinal")?;
+                let mut rows = statement.query([id])?;
+                while let Some(row) = rows.next()? {
+                    if row.get::<_, i64>(0)? != path.len() as i64 {
+                        return Err(corruption("filesystem handle path is not contiguous"));
+                    }
+                    let component: String = row.get(1)?;
+                    budget.charge(component.len() as u64 + std::mem::size_of::<String>() as u64)?;
+                    path.push(component);
+                }
+                Ok(External::FileSystemHandle { kind, bucket, path })
+            }
+            _ => Err(corruption("unknown IndexedDB external object kind")),
+        }
+    }
+
+    fn read_chunks(
+        &self,
+        table: &'static str,
+        column: &'static str,
+        id: i64,
+        expected: u64,
+        budget: &mut ReadBudget,
+    ) -> Result<Vec<u8>, IndexedDbError> {
+        budget.charge(expected)?;
+        let expected = usize::try_from(expected)
+            .map_err(|_| corruption("IndexedDB body length exceeds address space"))?;
+        let mut bytes = Vec::new();
+        bytes
+            .try_reserve_exact(expected)
+            .map_err(|_| IndexedDbError::Io("unable to allocate IndexedDB result".into()))?;
+        let mut statement = self.connection.prepare_cached(&format!(
+            "SELECT chunk_no, bytes FROM {table} WHERE {column}=?1 ORDER BY chunk_no"
+        ))?;
+        let mut rows = statement.query([id])?;
+        let mut ordinal = 0_i64;
+        while let Some(row) = rows.next()? {
+            if row.get::<_, i64>(0)? != ordinal {
+                return Err(corruption("IndexedDB chunks are not contiguous"));
+            }
+            let ValueRef::Blob(chunk) = row.get_ref(1)? else {
+                return Err(corruption("IndexedDB chunk is not a BLOB"));
+            };
+            if chunk.is_empty()
+                || chunk.len() > CHUNK_BYTES
+                || chunk.len() > expected.saturating_sub(bytes.len())
+            {
+                return Err(corruption("IndexedDB chunk length mismatch"));
+            }
+            bytes.extend_from_slice(chunk);
+            ordinal += 1;
+        }
+        if bytes.len() != expected {
+            return Err(corruption("IndexedDB body is truncated"));
+        }
+        Ok(bytes)
+    }
+
+    pub(super) fn remove_record_contents(&self, record: i64) -> Result<(), IndexedDbError> {
+        for table in ["blob_chunks", "handle_components", "external_objects"] {
+            self.connection.execute(&format!("DELETE FROM {table} WHERE object_id IN (SELECT object_id FROM record_objects WHERE record_id=?1)"), [record])?;
+        }
+        self.connection
+            .execute("DELETE FROM record_objects WHERE record_id=?1", [record])?;
+        self.connection
+            .execute("DELETE FROM record_chunks WHERE record_id=?1", [record])?;
+        self.connection
+            .execute("DELETE FROM index_entries WHERE record_id=?1", [record])?;
+        Ok(())
+    }
+
+    pub(super) fn remove_store_records(&self, store: i64) -> Result<(), IndexedDbError> {
+        let usage: i64 = self.connection.query_row(
+            "SELECT coalesce(sum(usage), 0) FROM records WHERE store_id=?1",
+            [store],
+            |row| row.get(0),
+        )?;
+        for table in ["blob_chunks", "handle_components", "external_objects"] {
+            self.connection.execute(&format!("DELETE FROM {table} WHERE object_id IN (SELECT object_id FROM record_objects WHERE record_id IN (SELECT record_id FROM records WHERE store_id=?1))"), [store])?;
+        }
+        for table in ["record_objects", "record_chunks", "index_entries"] {
+            self.connection.execute(&format!("DELETE FROM {table} WHERE record_id IN (SELECT record_id FROM records WHERE store_id=?1)"), [store])?;
+        }
+        self.connection
+            .execute("DELETE FROM records WHERE store_id=?1", [store])?;
+        self.adjust_usage(-i128::from(usage))
+    }
+}

moli-indexeddb/src/usage.rs

@@ -47,7 +47,7 @@ fn object_store_usage_bytes(name: &str, store: &ObjectStoreData) -> u64 {
         })))
 }
 
-fn indexed_db_value_usage_bytes(value: &IndexedDbValue) -> u64 {
+pub(crate) fn indexed_db_value_usage_bytes(value: &IndexedDbValue) -> u64 {
     value
         .wire_bytes
         .len()
@@ -96,7 +96,7 @@ fn optional_key_path_usage_bytes(key_path: &Option<KeyPath>) -> u64 {
     key_path.as_ref().map(key_path_usage_bytes).unwrap_or(0)
 }
 
-fn key_path_usage_bytes(key_path: &KeyPath) -> u64 {
+pub(crate) fn key_path_usage_bytes(key_path: &KeyPath) -> u64 {
     match key_path {
         KeyPath::String(value) => string_usage_bytes(value),
         KeyPath::Sequence(values) => {
@@ -105,7 +105,7 @@ fn key_path_usage_bytes(key_path: &KeyPath) -> u64 {
     }
 }
 
-fn key_usage_bytes(key: &Key) -> u64 {
+pub(crate) fn key_usage_bytes(key: &Key) -> u64 {
     match key {
         Key::String(value) => (value.len() as u64).saturating_mul(2),
         Key::Binary(value) => value.len() as u64,