From dab7ca2d976dba14d134cbcbd496fcefca13dcb2 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Wed, 28 Oct 2026 22:20:12 +0800 Subject: [PATCH] feat(indexeddb): implement SQL transactions and chunked record storage --- Cargo.lock | 1 + moli-indexeddb/Cargo.toml | 5 +- moli-indexeddb/src/lib.rs | 1 + moli-indexeddb/src/sqlite/mod.rs | 446 +++++++++++++++ moli-indexeddb/src/sqlite/records.rs | 410 ++++++++++++++ moli-indexeddb/src/sqlite/schema.rs | 240 +++++++++ moli-indexeddb/src/sqlite/tests.rs | 776 +++++++++++++++++++++++++++ moli-indexeddb/src/sqlite/values.rs | 269 ++++++++++ moli-indexeddb/src/usage.rs | 6 +- 9 files changed, 2148 insertions(+), 6 deletions(-) create mode 100644 moli-indexeddb/src/sqlite/mod.rs create mode 100644 moli-indexeddb/src/sqlite/records.rs create mode 100644 moli-indexeddb/src/sqlite/schema.rs create mode 100644 moli-indexeddb/src/sqlite/tests.rs create mode 100644 moli-indexeddb/src/sqlite/values.rs diff --git a/Cargo.lock b/Cargo.lock index d951a99269..59a13b0c7f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3035,6 +3035,7 @@ name = "moli-indexeddb" version = "0.1.0" dependencies = [ "moli-crypto", + "moli-sqlite", "parking_lot", "rusqlite", "serde", diff --git a/moli-indexeddb/Cargo.toml b/moli-indexeddb/Cargo.toml index 885bf510f6..cd560985b9 100644 --- a/moli-indexeddb/Cargo.toml +++ b/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 diff --git a/moli-indexeddb/src/lib.rs b/moli-indexeddb/src/lib.rs index a5ca3de9d4..6087d98b2d 100644 --- a/moli-indexeddb/src/lib.rs +++ b/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; diff --git a/moli-indexeddb/src/sqlite/mod.rs b/moli-indexeddb/src/sqlite/mod.rs new file mode 100644 index 0000000000..48b287e4a0 --- /dev/null +++ b/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, + 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, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DatabaseMetadata { + pub name: String, + pub version: u64, + pub usage: u64, + pub stores: BTreeMap, +} + +struct ActiveTransaction { + handle: TransactionHandle, + mode: TransactionMode, + stores: BTreeSet, + pending: Option, +} + +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, +} + +pub struct Database { + connection: Connection, + identity: DatabaseIdentity, + committed: DatabaseMetadata, + active: Option, + next_request: u64, +} + +impl Database { + pub fn create( + mut connection: Connection, + identity: DatabaseIdentity, + ) -> Result { + 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 { + moli_sqlite::require_schema_version(&connection, SCHEMA_VERSION) + .map_err(|error| corruption(&error.to_string()))?; + let saved: Option> = 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 = 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, + upgrade_version: Option, + ) -> 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 { + self.transaction(transaction)?; + schema::read_metadata(&self.connection) + } + + pub fn commit( + &mut self, + transaction: TransactionHandle, + ) -> Result { + 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 { + 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( + &mut self, + transaction: TransactionHandle, + operation: impl FnOnce(&mut Self) -> Result, + ) -> Result { + 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( + &mut self, + outcome: Result, + ) -> Result { + 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 { + 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::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 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; diff --git a/moli-indexeddb/src/sqlite/records.rs b/moli-indexeddb/src/sqlite/records.rs new file mode 100644 index 0000000000..7ca1d5a7c7 --- /dev/null +++ b/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, + add_only: bool, + ) -> Result { + 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, + quota: Option, + ) -> Result { + 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, + quota: Option, + ) -> Result { + let sort = pending.key.encode_ordered()?; + let old: Option<(i64, Vec, 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, + ) -> 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, 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::>()?) + } + + fn insert_index_keys( + &self, + index: i64, + unique: bool, + multi_entry: bool, + record: i64, + primary: &[u8], + keys: Vec, + ) -> 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, + ) -> 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, 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, IndexedDbError> { + self.ready(transaction)?; + let store = self.store_id(transaction, store_name)?; + let value: Option> = 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 { + 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)) + } +} diff --git a/moli-indexeddb/src/sqlite/schema.rs b/moli-indexeddb/src/sqlite/schema.rs new file mode 100644 index 0000000000..0238349f70 --- /dev/null +++ b/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 { + 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> = 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>(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::(); + 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 { + 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))) + }) + } +} diff --git a/moli-indexeddb/src/sqlite/tests.rs b/moli-indexeddb/src/sqlite/tests.rs new file mode 100644 index 0000000000..17798d9c22 --- /dev/null +++ b/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, + bytes: &[u8], + indexes: Vec, + quota: Option, +) -> Result { + 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 { + 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(); +} diff --git a/moli-indexeddb/src/sqlite/values.rs b/moli-indexeddb/src/sqlite/values.rs new file mode 100644 index 0000000000..42bb9b7098 --- /dev/null +++ b/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 { + 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::() 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 { + 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 = 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>(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::() 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, 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)) + } +} diff --git a/moli-indexeddb/src/usage.rs b/moli-indexeddb/src/usage.rs index 20be15bb80..74950d964d 100644 --- a/moli-indexeddb/src/usage.rs +++ b/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) -> 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,