diff --git a/docs/sql-dump-import.md b/docs/sql-dump-import.md new file mode 100644 index 0000000..a01d962 --- /dev/null +++ b/docs/sql-dump-import.md @@ -0,0 +1,100 @@ +# Bounded SQL dump import + +Use the import session API for a streamed SQL dump. The caller sends decoded +UTF-8 text chunks one at a time; it does not need to split at SQL statement +boundaries. Keep each chunk at or below 512 KiB in UTF-8 bytes. A single SQL +statement may span chunks, up to 16 MiB. The producer's outer transaction +markers, including `BEGIN TRANSACTION;` and `COMMIT;`, are accepted. The worker +owns the actual transaction and commits only after `finishSqlDumpImport` +succeeds. Markers are optional, but when present there must be one opening +`BEGIN [DEFERRED|IMMEDIATE|EXCLUSIVE] [TRANSACTION];` and one closing +`COMMIT [TRANSACTION];` or `END [TRANSACTION];` marker. A closing marker without +an opening marker, SQL after the closing marker, and other transaction controls +(`SAVEPOINT`, `RELEASE`, `ROLLBACK`, or additional `BEGIN`/`COMMIT`) are +rejected. + +At least one executable statement is required. Empty dumps are rejected, even +when they contain transaction markers or the skipped CLI PRAGMA header. The +reported statement count excludes those markers, the skipped header, comments, +and empty statements. + +PRAGMA settings are rejected before SQLite prepares them because connection +settings are not restored by transaction rollback. Only the standard SQLite CLI +header `PRAGMA foreign_keys=OFF;` (or `=0`, with optional whitespace and +case-insensitive keywords) is accepted and skipped. It does not disable the +connection's existing foreign-key enforcement. Arrange imported data to satisfy +any enabled foreign-key constraints. `EXPLAIN` is also rejected before +preparation because it can still apply PRAGMA settings. `ATTACH` and `DETACH` +are rejected because attachment changes cannot be rolled back. + +This supports the raindex producer's SQL dumps and CLI dumps that satisfy these +restrictions, not arbitrary SQLite CLI `.dump` output. Virtual-table dumps such +as FTS5 can include `PRAGMA writable_schema=ON` and are rejected. + +Chunks must be valid UTF-16 strings. Do not use JavaScript `slice` to split a +surrogate pair; invalid chunks are rejected and cancel the session. The decoder +in the example below preserves whole characters across byte chunks. + +```ts +const begin = await db.beginSqlDumpImport(); +if (begin.error) throw begin.error; +const id = begin.value!; + +try { + const decoder = new TextDecoder(); + for await (const bytes of decompressedByteChunks) { + // TextDecoder preserves UTF-8 characters split between byte chunks. + for (let offset = 0; offset < bytes.length; offset += 256 * 1024) { + const text = decoder.decode(bytes.subarray(offset, offset + 256 * 1024), { + stream: true, + }); + const append = await db.appendSqlDumpChunk(id, text); + if (append.error) throw append.error; + } + } + const tail = decoder.decode(); + if (tail) { + const append = await db.appendSqlDumpChunk(id, tail); + if (append.error) throw append.error; + } + const finish = await db.finishSqlDumpImport(id); + if (finish.error) throw finish.error; +} catch (error) { + await db.cancelSqlDumpImport(id); // harmless if a failed chunk already rolled back + throw error; +} +``` + +The worker serializes imports with ordinary queries and transactions. While an +import is active, those operations return an error on every client sharing the +database. A malformed statement, incomplete final statement, oversized +statement, or cancellation rolls back the whole import. Worker termination +before commit also rolls back. If a worker is lost while finishing, the commit +outcome is unknown whether the finish response never arrives or the coordinator +returns a worker-failure error. Such an error does not confirm rollback: check +or reset the database before retrying, even when the coordinator leader did not +change. A normal SQL validation or execution error from a live worker rolls back +the import. Row-returning SQL such as `SELECT` or `INSERT ... RETURNING` is +rejected because an import does not consume query results. An import idle for at +least 120 seconds is rolled back on the next database request or watchdog tick. +The watchdog checks every 30 seconds, so expiry normally occurs within 120–150 +seconds; browser scheduling can delay it further. After a confirmed rollback, +the caller can retry from a clean session. Await each append before sending the +next one; concurrent import operations on the same client instance are rejected +to keep queued SQL bounded per instance. If a concurrent call is rejected, await +the active call and then cancel the session before retrying. If the leader +changes before a cross-tab import response arrives, the commit outcome is +unknown. Check or reset the database before retrying that import. + +Follower import requests intentionally have no timeout: a large append or index +build can take longer than ordinary queries. While an import request is in +flight, that client instance also rejects cancellation. If the leader remains +alive but stops responding, the promise can remain pending until a leader +change; there is no same-instance cancellation escape during that wait. If an +importing follower tab closes, the shared import can block other clients until +the idle watchdog rolls it back, normally after 120–150 seconds and potentially +longer if the browser delays scheduling. These are current liveness limitations. + +The chunk limit bounds each worker message and intermediate SQL statement. The +full dump remains with the caller's stream and is never materialized as a +statement array by this API. This API does not download or decompress the dump. diff --git a/packages/sqlite-web-core/src/coordination.rs b/packages/sqlite-web-core/src/coordination.rs index 0756ec4..8f33d81 100644 --- a/packages/sqlite-web-core/src/coordination.rs +++ b/packages/sqlite-web-core/src/coordination.rs @@ -1,6 +1,6 @@ use js_sys::{Function, Object, Promise, Reflect}; use std::cell::{Cell, RefCell}; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::pin::Pin; use std::rc::Rc; @@ -16,8 +16,8 @@ use web_sys::{ use crate::database::SQLiteDatabase; use crate::messages::{ - ChannelMessage, MainThreadMessage, SqlBatchStatement, WorkerErrorPayload, WorkerMessage, - WORKER_ERROR_TYPE_INITIALIZATION_PENDING, + ChannelMessage, MainThreadMessage, SqlBatchStatement, SqlImportAction, WorkerErrorPayload, + WorkerMessage, WORKER_ERROR_TYPE_INITIALIZATION_PENDING, }; use crate::util::{js_value_to_string, sanitize_identifier, set_js_property}; @@ -99,6 +99,9 @@ pub(crate) enum DbJobPayload { Batch { statements: Vec, }, + Import { + action: SqlImportAction, + }, } type DbExecFuture = Pin> + 'static>>; @@ -141,6 +144,7 @@ pub struct CoordinatorState { pub db_name: String, db_pending: Rc>>, pub follower_pending: Rc>>, + follower_pending_imports: RefCell>, pub next_db_request_id: Rc>, db_worker_restart_attempts: Rc>, } @@ -150,6 +154,7 @@ pub struct DbWorkerState { pub db_name: String, db_queue: Rc>>, db_processing: Rc>, + import_watchdog_running: Cell, hooks: DbWorkerHooks, } @@ -174,6 +179,7 @@ impl CoordinatorState { db_name: config.db_name, db_pending: Rc::new(RefCell::new(HashMap::new())), follower_pending: Rc::new(RefCell::new(HashMap::new())), + follower_pending_imports: RefCell::new(HashSet::new()), next_db_request_id: Rc::new(RefCell::new(1)), db_worker_restart_attempts: Rc::new(Cell::new(0)), })) @@ -203,6 +209,7 @@ impl CoordinatorState { fn handle_follower_forward_failure(&self, query_id: &str, request_id: u32, err: &str) { self.follower_pending.borrow_mut().remove(query_id); + self.follower_pending_imports.borrow_mut().remove(query_id); let _ = send_query_result_to_main(request_id, Err(err.to_string())); } @@ -534,6 +541,46 @@ impl CoordinatorState { } } }, + WorkerMessage::ImportSqlDump { request_id, action } => match *self.role.borrow() { + LeadershipRole::Leader => { + if !*self.db_worker_ready.borrow() { + let _ = send_query_result_to_main( + request_id, + Err(WORKER_ERROR_TYPE_INITIALIZATION_PENDING.to_string()), + ); + return; + } + self.forward_import_to_db(DbRequestOrigin::Local { request_id }, action); + } + LeadershipRole::Follower => { + if !*self.leader_ready.borrow() { + let _ = send_query_result_to_main( + request_id, + Err(WORKER_ERROR_TYPE_INITIALIZATION_PENDING.to_string()), + ); + return; + } + let query_id = Uuid::new_v4().to_string(); + self.follower_pending + .borrow_mut() + .insert(query_id.clone(), request_id); + self.follower_pending_imports + .borrow_mut() + .insert(query_id.clone()); + // An import request must resolve from the DB worker. A follower-side + // timeout could report failure while a delayed finish still commits. + if let Err(err) = send_channel_message( + &self.channel, + &ChannelMessage::ImportRequest { + query_id: query_id.clone(), + action, + }, + ) { + self.handle_follower_forward_failure(&query_id, request_id, &err); + let _ = send_worker_error_message(&err); + } + } + }, } } @@ -610,12 +657,29 @@ impl CoordinatorState { self.forward_batch_to_db(DbRequestOrigin::Forwarded { query_id }, statements); } } + ChannelMessage::ImportRequest { query_id, action } => { + if matches!(*self.role.borrow(), LeadershipRole::Leader) { + if !*self.db_worker_ready.borrow() { + let _ = send_channel_message( + &self.channel, + &ChannelMessage::QueryResponse { + query_id, + result: None, + error: Some(WORKER_ERROR_TYPE_INITIALIZATION_PENDING.to_string()), + }, + ); + return; + } + self.forward_import_to_db(DbRequestOrigin::Forwarded { query_id }, action); + } + } ChannelMessage::QueryResponse { query_id, result, error, } => { if let Some(request_id) = self.follower_pending.borrow_mut().remove(&query_id) { + self.follower_pending_imports.borrow_mut().remove(&query_id); let outcome = match (result, error) { (Some(res), _) => Ok(res), (_, Some(err)) => Err(err), @@ -721,6 +785,32 @@ impl CoordinatorState { self.post_db_worker_message(worker, db_request_id, msg); } + fn forward_import_to_db(self: &Rc, origin: DbRequestOrigin, action: SqlImportAction) { + let worker = { + let borrow = self.db_worker.borrow(); + let Some(worker) = borrow.as_ref() else { + self.fail_origin(origin, WORKER_ERROR_TYPE_INITIALIZATION_PENDING.to_string()); + return; + }; + worker.clone() + }; + let db_request_id = { + let mut next = self.next_db_request_id.borrow_mut(); + let id = *next; + *next = next.wrapping_add(1).max(1); + id + }; + self.db_pending.borrow_mut().insert(db_request_id, origin); + self.post_db_worker_message( + worker, + db_request_id, + WorkerMessage::ImportSqlDump { + request_id: db_request_id, + action, + }, + ); + } + fn post_db_worker_message(&self, worker: Worker, db_request_id: u32, msg: WorkerMessage) { match serde_wasm_bindgen::to_value(&msg) { Ok(val) => { @@ -805,6 +895,24 @@ impl CoordinatorState { } fn mark_leader_known(&self, leader_id: String) { + if self + .leader_id + .borrow() + .as_ref() + .is_some_and(|current| current != &leader_id) + { + for query_id in self.follower_pending_imports.borrow_mut().drain() { + if let Some(request_id) = self.follower_pending.borrow_mut().remove(&query_id) { + let _ = send_query_result_to_main( + request_id, + Err( + "Leader changed before SQL dump import response; outcome unknown." + .to_string(), + ), + ); + } + } + } *self.leader_id.borrow_mut() = Some(leader_id); } @@ -830,6 +938,7 @@ impl DbWorkerState { db_name: config.db_name, db_queue: Rc::new(RefCell::new(VecDeque::new())), db_processing: Rc::new(Cell::new(false)), + import_watchdog_running: Cell::new(false), hooks, }) } @@ -849,6 +958,28 @@ impl DbWorkerState { }); } + fn start_import_watchdog(self: &Rc) { + if self.import_watchdog_running.replace(true) { + return; + } + let state = Rc::clone(self); + spawn_local(async move { + loop { + sleep_ms(30_000).await; + let db = state.db.borrow_mut().take(); + if let Some(mut database) = db { + database.expire_import_if_idle().await; + let active = database.has_active_import(); + *state.db.borrow_mut() = Some(database); + if !active { + state.import_watchdog_running.set(false); + break; + } + } + } + }); + } + pub fn handle_message(self: &Rc, msg: WorkerMessage) { match msg { WorkerMessage::ExecuteQuery { @@ -864,6 +995,9 @@ impl DbWorkerState { } => { self.enqueue_job(request_id, DbJobPayload::Batch { statements }); } + WorkerMessage::ImportSqlDump { request_id, action } => { + self.enqueue_job(request_id, DbJobPayload::Import { action }); + } } } @@ -891,7 +1025,16 @@ impl DbWorkerState { let db = Rc::clone(&state.db); let exec = Rc::clone(&hooks.exec); let deliver = Rc::clone(&hooks.deliver); + let starts_import = matches!( + &job.payload, + DbJobPayload::Import { + action: SqlImportAction::Begin + } + ); let result = exec.as_ref()(db, job.payload).await; + if starts_import && result.is_ok() { + state.start_import_watchdog(); + } match make_query_result_message(job.request_id, result) { Ok(resp) => deliver.as_ref()(&resp), Err(err) => { @@ -1044,12 +1187,14 @@ async fn exec_on_db( let db_opt = db.borrow_mut().take(); let result = match db_opt { Some(mut database) => { + database.expire_import_if_idle().await; let result = match payload { DbJobPayload::Query { sql, params } => match params { Some(p) => database.exec_with_params(&sql, p).await, None => database.exec(&sql).await, }, DbJobPayload::Batch { statements } => database.exec_batch(statements).await, + DbJobPayload::Import { action } => database.import_sql_action(action).await, }; *db.borrow_mut() = Some(database); result @@ -1062,7 +1207,7 @@ async fn exec_on_db( pub async fn sleep_ms(ms: i32) { let promise = js_sys::Promise::new(&mut |resolve, _| { let resolve_for_timeout = resolve.clone(); - let closure = Closure::once(move || { + let callback = Closure::once_into_js(move || { let _ = resolve_for_timeout.call0(&JsValue::NULL); }); @@ -1072,7 +1217,7 @@ pub async fn sleep_ms(ms: i32) { .and_then(|scope| { scope .set_timeout_with_callback_and_timeout_and_arguments_0( - closure.as_ref().unchecked_ref(), + callback.unchecked_ref(), ms, ) .map(|_| ()) @@ -1081,7 +1226,7 @@ pub async fn sleep_ms(ms: i32) { .or_else(|_| { web_sys::window().ok_or(()).and_then(|win| { win.set_timeout_with_callback_and_timeout_and_arguments_0( - closure.as_ref().unchecked_ref(), + callback.unchecked_ref(), ms, ) .map(|_| ()) @@ -1090,10 +1235,8 @@ pub async fn sleep_ms(ms: i32) { }); if timeout_result.is_err() { - let _ = resolve.call0(&JsValue::NULL); + let _ = callback.unchecked_ref::().call0(&JsValue::NULL); } - - closure.forget(); }); let _ = JsFuture::from(promise).await; } @@ -1183,6 +1326,35 @@ mod tests { ); } + #[wasm_bindgen_test] + fn leader_change_clears_pending_imports_only() { + let state = CoordinatorState::new(WorkerConfig { + db_name: "testdb-import-leader-change".to_string(), + follower_timeout_ms: 10.0, + query_timeout_ms: 50.0, + }) + .expect("state"); + state.mark_leader_known("old-leader".to_string()); + state + .follower_pending + .borrow_mut() + .insert("import-1".to_string(), 42); + state + .follower_pending_imports + .borrow_mut() + .insert("import-1".to_string()); + state + .follower_pending + .borrow_mut() + .insert("query-1".to_string(), 43); + + state.mark_leader_known("new-leader".to_string()); + + assert!(!state.follower_pending.borrow().contains_key("import-1")); + assert!(state.follower_pending_imports.borrow().is_empty()); + assert!(state.follower_pending.borrow().contains_key("query-1")); + } + #[wasm_bindgen_test(async)] async fn coordinator_broadcasts_leader_and_ready() { set_global_str("__SQLITE_DB_NAME", "testdb-coordinator"); diff --git a/packages/sqlite-web-core/src/database.rs b/packages/sqlite-web-core/src/database.rs index 32a13c7..4dacca1 100644 --- a/packages/sqlite-web-core/src/database.rs +++ b/packages/sqlite-web-core/src/database.rs @@ -1,16 +1,116 @@ use crate::database_functions::register_custom_functions; -use crate::messages::SqlBatchStatement; +use crate::messages::{SqlBatchStatement, SqlImportAction}; use crate::util::sanitize_db_filename; use base64::Engine; use sqlite_wasm_rs::export::{install_opfs_sahpool, *}; use std::ffi::{CStr, CString}; use std::os::raw::c_void; +use uuid::Uuid; use wasm_bindgen::prelude::*; +const MAX_IMPORT_CHUNK_BYTES: usize = 512 * 1024; +const MAX_IMPORT_STATEMENT_BYTES: usize = 16 * 1024 * 1024; +const IMPORT_IDLE_TIMEOUT_MS: f64 = 120_000.0; + +struct SqlImportState { + id: String, + pending: String, + scanner: SqlStatementScanner, + statement_count: usize, + saw_begin: bool, + saw_commit: bool, + last_activity_ms: f64, +} + +#[derive(Default)] +struct SqlStatementScanner { + offset: usize, + mode: SqlScanMode, +} + +#[derive(Clone, Copy, Default)] +enum SqlScanMode { + #[default] + Sql, + Quoted(u8), + Bracket, + LineComment, + BlockComment, +} + +/// Find a statement boundary without treating semicolons in SQL literals or +/// comments as delimiters. A trailing partial statement stays in `pending`. +fn next_sql_statement_end(sql: &str, scanner: &mut SqlStatementScanner) -> Option { + let bytes = sql.as_bytes(); + let mut i = scanner.offset; + while i < bytes.len() { + let byte = bytes[i]; + let next = bytes.get(i + 1).copied(); + match scanner.mode { + SqlScanMode::Sql => match (byte, next) { + (b'\'', _) | (b'"', _) | (b'`', _) => scanner.mode = SqlScanMode::Quoted(byte), + (b'[', _) => scanner.mode = SqlScanMode::Bracket, + (b'-', Some(b'-')) => { + scanner.mode = SqlScanMode::LineComment; + i += 1; + } + (b'/', Some(b'*')) => { + scanner.mode = SqlScanMode::BlockComment; + i += 1; + } + (b'-' | b'/', None) => break, + (b';', _) => { + // SQLite knows whether a trigger body is complete. Ordinary + // INSERTs skip this extra parse in the import hot path. + if SQLiteDatabase::first_sql_keyword(sql) + .is_some_and(|keyword| keyword.eq_ignore_ascii_case("CREATE")) + { + let Ok(candidate) = CString::new(&sql[..i + 1]) else { + scanner.offset = i + 1; + return Some(i + 1); + }; + if unsafe { sqlite3_complete(candidate.as_ptr()) } == 0 { + i += 1; + continue; + } + } + scanner.offset = i + 1; + return Some(i + 1); + } + _ => {} + }, + SqlScanMode::Quoted(quote) if byte == quote => { + if next == Some(quote) { + i += 1; + } else if next.is_none() { + break; + } else { + scanner.mode = SqlScanMode::Sql; + } + } + SqlScanMode::Quoted(_) => {} + SqlScanMode::Bracket if byte == b']' => scanner.mode = SqlScanMode::Sql, + SqlScanMode::Bracket => {} + SqlScanMode::LineComment if byte == b'\n' => scanner.mode = SqlScanMode::Sql, + SqlScanMode::LineComment => {} + SqlScanMode::BlockComment if byte == b'*' && next == Some(b'/') => { + scanner.mode = SqlScanMode::Sql; + i += 1; + } + SqlScanMode::BlockComment if byte == b'*' && next.is_none() => break, + SqlScanMode::BlockComment => {} + } + i += 1; + } + scanner.offset = i; + None +} + // Real SQLite database using sqlite-wasm-rs FFI pub struct SQLiteDatabase { db: *mut sqlite3, in_transaction: bool, + import_state: Option, } unsafe impl Send for SQLiteDatabase {} @@ -68,6 +168,223 @@ enum ParamKind { } impl SQLiteDatabase { + fn reject_during_import(&self) -> Result<(), String> { + if self.import_state.is_some() { + Err("SQL dump import is in progress.".to_string()) + } else { + Ok(()) + } + } + + pub fn has_active_import(&self) -> bool { + self.import_state.is_some() + } + + pub async fn expire_import_if_idle(&mut self) { + let expired = self.import_state.as_ref().is_some_and(|state| { + js_sys::Date::now() - state.last_activity_ms >= IMPORT_IDLE_TIMEOUT_MS + }); + if expired { + self.import_state = None; + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + } + } + + pub async fn import_sql_action(&mut self, action: SqlImportAction) -> Result { + self.expire_import_if_idle().await; + match action { + SqlImportAction::Begin => { + self.reject_during_import()?; + self.refresh_transaction_state(); + if self.in_transaction { + return Err("Cannot import while a transaction is already open.".to_string()); + } + self.exec_single_statement("BEGIN IMMEDIATE").await?; + self.refresh_transaction_state(); + let id = Uuid::new_v4().to_string(); + self.import_state = Some(SqlImportState { + id: id.clone(), + pending: String::new(), + scanner: SqlStatementScanner::default(), + statement_count: 0, + saw_begin: false, + saw_commit: false, + last_activity_ms: js_sys::Date::now(), + }); + Ok(id) + } + SqlImportAction::Chunk { id, sql } => { + let mut state = self + .import_state + .take() + .ok_or("No SQL dump import is active.")?; + if state.id != id { + self.import_state = Some(state); + return Err("SQL dump import belongs to another session.".to_string()); + } + if sql.len() > MAX_IMPORT_CHUNK_BYTES { + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + return Err(format!( + "SQL dump chunk exceeds {MAX_IMPORT_CHUNK_BYTES} bytes." + )); + } + state.pending.push_str(&sql); + let result = self.execute_import_chunk(&mut state).await; + match result { + Ok(()) => { + state.last_activity_ms = js_sys::Date::now(); + self.import_state = Some(state); + Ok("SQL dump chunk imported.".to_string()) + } + Err(err) => { + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + Err(err) + } + } + } + SqlImportAction::Finish { id } => { + let state = self + .import_state + .take() + .ok_or("No SQL dump import is active.")?; + if state.id != id { + self.import_state = Some(state); + return Err("SQL dump import belongs to another session.".to_string()); + } + let complete_tail = CString::new(state.pending.as_str()) + .is_ok_and(|tail| Self::is_trivia_tail_only(tail.as_ptr())); + let validation_error = if state.saw_begin != state.saw_commit { + Some("SQL dump contains unmatched transaction markers.") + } else if !complete_tail { + Some("SQL dump ends with an incomplete statement.") + } else if state.statement_count == 0 { + Some("SQL dump contains no executable statements.") + } else { + None + }; + if let Some(error) = validation_error { + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + return Err(error.to_string()); + } + if let Err(err) = self.exec_single_statement("COMMIT").await { + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + return Err(format!("SQL dump commit failed: {err}")); + } + self.refresh_transaction_state(); + Ok(format!( + "Imported {} SQL statements.", + state.statement_count + )) + } + SqlImportAction::Cancel { id } => { + let state = self + .import_state + .take() + .ok_or("No SQL dump import is active.")?; + if state.id != id { + self.import_state = Some(state); + return Err("SQL dump import belongs to another session.".to_string()); + } + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + Ok("SQL dump import cancelled.".to_string()) + } + } + } + + async fn execute_import_chunk(&mut self, state: &mut SqlImportState) -> Result<(), String> { + let mut consumed = 0; + loop { + let remaining = &state.pending[consumed..]; + let Some(end) = next_sql_statement_end(remaining, &mut state.scanner) else { + if remaining.len() > MAX_IMPORT_STATEMENT_BYTES { + return Err(format!( + "SQL dump statement exceeds {MAX_IMPORT_STATEMENT_BYTES} bytes." + )); + } + state.pending.drain(..consumed); + return Ok(()); + }; + if end > MAX_IMPORT_STATEMENT_BYTES { + return Err(format!( + "SQL dump statement exceeds {MAX_IMPORT_STATEMENT_BYTES} bytes." + )); + } + let statement = &remaining[..end]; + let body = statement.strip_suffix(';').unwrap_or(statement); + if !Self::is_sql_trivia_only(body) { + let (keyword, tail) = Self::first_sql_keyword_and_tail(statement) + .ok_or("SQL dump contains an invalid statement.")?; + if keyword.eq_ignore_ascii_case("BEGIN") + && Self::is_outer_transaction_marker(tail, true) + && !state.saw_begin + && !state.saw_commit + { + state.saw_begin = true; + } else if (keyword.eq_ignore_ascii_case("COMMIT") + || keyword.eq_ignore_ascii_case("END")) + && Self::is_outer_transaction_marker(tail, false) + && !state.saw_commit + { + state.saw_commit = true; + } else { + if state.saw_commit || Self::is_transaction_control_statement(statement) { + return Err( + "SQL dump contains an unexpected transaction statement.".to_string() + ); + } + // EXPLAIN can still apply PRAGMA flags during preparation, + // before its result columns let us reject it. Attachments + // likewise change connection state outside rollback. + if keyword.eq_ignore_ascii_case("EXPLAIN") { + return Err("SQL dump cannot contain row-returning statements.".to_string()); + } + if keyword.eq_ignore_ascii_case("ATTACH") + || keyword.eq_ignore_ascii_case("DETACH") + { + return Err("SQL dump cannot change database attachments.".to_string()); + } + if keyword.eq_ignore_ascii_case("PRAGMA") { + // SQLite CLI dumps include this header. It is a no-op + // inside our transaction; never apply connection settings + // that ROLLBACK cannot restore (including at prepare time). + if !Self::is_dump_foreign_keys_header(tail) { + return Err("SQL dump cannot change PRAGMA settings.".to_string()); + } + } else { + self.exec_import_statement(statement).await.map_err(|err| { + format!( + "SQL dump statement {} failed: {err}", + state.statement_count + 1 + ) + })?; + state.statement_count += 1; + } + } + } + consumed += end; + state.scanner = SqlStatementScanner::default(); + } + } + + // Accept only the harmless CLI header, without executing it. The caller's + // existing foreign-key enforcement remains in effect during the import. + fn is_dump_foreign_keys_header(tail: &str) -> bool { + let Some(body) = tail.trim().strip_suffix(';') else { + return false; + }; + let Some((name, value)) = body.split_once('=') else { + return false; + }; + name.trim().eq_ignore_ascii_case("foreign_keys") + && (value.trim().eq_ignore_ascii_case("OFF") || value.trim() == "0") + } + fn refresh_transaction_state(&mut self) { self.in_transaction = unsafe { sqlite3_get_autocommit(self.db) } == 0; } @@ -600,6 +917,25 @@ impl SQLiteDatabase { self.exec_prepared_statement(stmt_guard.take()) } + /// Execute one import statement without allowing result sets to allocate + /// unbounded row arrays (including INSERT ... RETURNING). + async fn exec_import_statement(&self, sql: &str) -> Result<(), String> { + let sql_cstr = CString::new(sql).map_err(|e| format!("Invalid SQL string: {e}"))?; + let ptr = sql_cstr.as_ptr(); + let (stmt_opt, tail) = self.prepare_one(ptr)?; + let Some(stmt) = stmt_opt else { + return Err("SQL dump statement is empty.".to_string()); + }; + let mut guard = StmtGuard::new(stmt); + if !Self::is_trivia_tail_only(tail) { + return Err("SQL dump statement contains multiple statements.".to_string()); + } + if unsafe { sqlite3_column_count(stmt) } > 0 { + return Err("SQL dump cannot contain row-returning statements.".to_string()); + } + self.exec_prepared_statement(guard.take()).map(|_| ()) + } + fn is_transaction_control_statement(sql: &str) -> bool { let Some(keyword) = Self::first_sql_keyword(sql) else { return false; @@ -611,7 +947,34 @@ impl SQLiteDatabase { ) } + fn is_outer_transaction_marker(tail: &str, allow_mode: bool) -> bool { + let Some(body) = tail.trim().strip_suffix(';') else { + return false; + }; + let mut words = body.split_ascii_whitespace().peekable(); + if allow_mode + && words.peek().is_some_and(|word| { + ["DEFERRED", "IMMEDIATE", "EXCLUSIVE"] + .iter() + .any(|mode| word.eq_ignore_ascii_case(mode)) + }) + { + words.next(); + } + if words + .peek() + .is_some_and(|word| word.eq_ignore_ascii_case("TRANSACTION")) + { + words.next(); + } + words.next().is_none() + } + fn first_sql_keyword(sql: &str) -> Option<&str> { + Self::first_sql_keyword_and_tail(sql).map(|(keyword, _)| keyword) + } + + fn first_sql_keyword_and_tail(sql: &str) -> Option<(&str, &str)> { let bytes = sql.as_bytes(); let mut i = 0; @@ -656,7 +1019,7 @@ impl SQLiteDatabase { if i == start { None } else { - Some(&sql[start..i]) + Some((&sql[start..i], &sql[i..])) } } @@ -753,6 +1116,7 @@ impl SQLiteDatabase { Ok(SQLiteDatabase { db, in_transaction: false, + import_state: None, }) } @@ -831,6 +1195,7 @@ impl SQLiteDatabase { /// Execute potentially multiple SQL statements pub async fn exec(&mut self, sql: &str) -> Result { + self.reject_during_import()?; let trimmed = sql.trim(); // Single-statement mode: execute only the first statement, ignore tail @@ -923,6 +1288,7 @@ impl SQLiteDatabase { sql: &str, params: Vec, ) -> Result { + self.reject_during_import()?; let (results, affected) = self.exec_single_statement_with_params(sql, params).await?; self.refresh_transaction_state(); @@ -942,6 +1308,7 @@ impl SQLiteDatabase { &mut self, statements: Vec, ) -> Result { + self.reject_during_import()?; if statements.is_empty() { return Err("Batch must contain at least one statement.".to_string()); } @@ -2448,4 +2815,644 @@ mod tests { assert_eq!(array[0]["msg"].as_str().unwrap(), "insert; happened"); assert_eq!(array[1]["msg"].as_str().unwrap(), "second; line"); } + + // Core import tests exercise real SQLite through its memory VFS. OPFS is + // available only in dedicated workers and is covered by the UI integration + // suite; an open failure here must fail the test, never skip assertions. + fn get_import_test_db(name: &str) -> SQLiteDatabase { + let filename = CString::new(name).unwrap(); + let vfs = CString::new("memvfs").unwrap(); + let mut db = std::ptr::null_mut(); + let ret = unsafe { + sqlite3_open_v2( + filename.as_ptr(), + &mut db, + SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, + vfs.as_ptr(), + ) + }; + assert_eq!(ret, SQLITE_OK, "failed to open import test database"); + SQLiteDatabase { + db, + in_transaction: false, + import_state: None, + } + } + + #[wasm_bindgen_test] + async fn sql_dump_import_accepts_split_statements_and_persists() { + let mut db = get_import_test_db("sql_dump_import_accepts_split_statements_and_persists"); + db.exec("DROP TABLE IF EXISTS dump_chunk_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_chunk_test (id INTEGER PRIMARY KEY, value TEXT)") + .await + .unwrap(); + + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "-- dump header\nBEGIN;\nINSERT INTO dump_chunk_test VALUES (1, 'first; value');\nINS".into(), + }) + .await + .unwrap(); + assert!(db.exec("SELECT * FROM dump_chunk_test").await.is_err()); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "ERT INTO dump_chunk_test VALUES (2, 'second\nvalue');\n/* footer */ COMMIT;\n" + .into(), + }) + .await + .unwrap(); + assert_eq!( + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap(), + "Imported 2 SQL statements." + ); + drop(db); + + let mut reopened = + get_import_test_db("sql_dump_import_accepts_split_statements_and_persists"); + let result = reopened + .exec("SELECT value FROM dump_chunk_test ORDER BY id") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["value"], "first; value"); + assert_eq!(rows[1]["value"], "second\nvalue"); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_accepts_cli_transaction_markers_and_empty_statements() { + let mut db = get_import_test_db( + "sql_dump_import_accepts_cli_transaction_markers_and_empty_statements", + ); + db.exec("DROP TABLE IF EXISTS dump_cli_marker_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_cli_marker_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "PRAGMA foreign_keys=OFF;\nBEGIN TRANSACTION;\n; /* empty */ ;\nINSERT INTO dump_cli_marker_test VALUES (1);\nCOMMIT;\n".into(), + }) + .await + .unwrap(); + assert_eq!( + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap(), + "Imported 1 SQL statements." + ); + + let rows = db + .exec("SELECT COUNT(*) AS count FROM dump_cli_marker_test") + .await + .unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&rows).unwrap(); + assert_eq!(parsed[0]["count"], 1); + assert!(SQLiteDatabase::is_outer_transaction_marker( + " IMMEDIATE TRANSACTION;", + true + )); + assert!(SQLiteDatabase::is_outer_transaction_marker( + " TRANSACTION;", + false + )); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_empty_dumps_and_unmatched_markers() { + let mut db = + get_import_test_db("sql_dump_import_rejects_empty_dumps_and_unmatched_markers"); + for sql in [ + "", + "; /* empty */ ;", + "BEGIN; COMMIT;", + "PRAGMA foreign_keys=OFF; BEGIN TRANSACTION; COMMIT;", + ] { + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: sql.into(), + }) + .await + .unwrap(); + assert_eq!( + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap_err(), + "SQL dump contains no executable statements.", + "{sql}" + ); + assert!(!db.has_active_import()); + assert_eq!(unsafe { sqlite3_get_autocommit(db.db) }, 1); + } + + db.exec("DROP TABLE IF EXISTS dump_unmatched_marker_test") + .await + .unwrap(); + for sql in [ + "COMMIT;", + "BEGIN; CREATE TABLE dump_unmatched_marker_test (id INTEGER);", + ] { + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: sql.into(), + }) + .await + .unwrap(); + assert_eq!( + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap_err(), + "SQL dump contains unmatched transaction markers.", + "{sql}" + ); + assert!(!db.has_active_import()); + assert_eq!(unsafe { sqlite3_get_autocommit(db.db) }, 1); + assert_eq!( + db.exec("SELECT name FROM sqlite_master WHERE name = 'dump_unmatched_marker_test'") + .await + .unwrap(), + "[]" + ); + } + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rolls_back_on_invalid_statement() { + let mut db = get_import_test_db("sql_dump_import_rolls_back_on_invalid_statement"); + db.exec("DROP TABLE IF EXISTS dump_rollback_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_rollback_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let error = db.import_sql_action(SqlImportAction::Chunk { + id, + sql: "INSERT INTO dump_rollback_test VALUES (1); INSERT INTO dump_rollback_test (missing) VALUES (2);".into(), + }).await.unwrap_err(); + assert!(error.contains("statement 2 failed"), "{error}"); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_rollback_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_other_clients_and_cancel_rolls_back() { + let mut db = + get_import_test_db("sql_dump_import_rejects_other_clients_and_cancel_rolls_back"); + db.exec("DROP TABLE IF EXISTS dump_cancel_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_cancel_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "INSERT INTO dump_cancel_test VALUES (1);".into(), + }) + .await + .unwrap(); + assert!(db.import_sql_action(SqlImportAction::Begin).await.is_err()); + assert!(db + .import_sql_action(SqlImportAction::Cancel { id: "wrong".into() }) + .await + .is_err()); + assert!(db.exec("SELECT * FROM dump_cancel_test").await.is_err()); + db.import_sql_action(SqlImportAction::Cancel { id }) + .await + .unwrap(); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_cancel_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_incomplete_tail_and_expires() { + let mut db = get_import_test_db("sql_dump_import_rejects_incomplete_tail_and_expires"); + db.exec("DROP TABLE IF EXISTS dump_tail_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_tail_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "INSERT INTO dump_tail_test VALUES (1); INSERT INTO dump_tail_test VALUES (" + .into(), + }) + .await + .unwrap(); + assert!(db + .import_sql_action(SqlImportAction::Finish { id }) + .await + .is_err()); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "INSERT INTO dump_tail_test VALUES (2);".into(), + }) + .await + .unwrap(); + db.import_state.as_mut().unwrap().last_activity_ms -= IMPORT_IDLE_TIMEOUT_MS + 1.0; + db.expire_import_if_idle().await; + assert!(db + .import_sql_action(SqlImportAction::Finish { id }) + .await + .is_err()); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_tail_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_handles_trigger_body_and_split_quote() { + let mut db = get_import_test_db("sql_dump_import_handles_trigger_body_and_split_quote"); + db.exec("DROP TABLE IF EXISTS dump_trigger_source") + .await + .unwrap(); + db.exec("DROP TABLE IF EXISTS dump_trigger_log") + .await + .unwrap(); + db.exec("CREATE TABLE dump_trigger_source (value TEXT)") + .await + .unwrap(); + db.exec("CREATE TABLE dump_trigger_log (value TEXT)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "CREATE TRIGGER dump_trigger AFTER INSERT ON dump_trigger_source BEGIN INSERT INTO dump_trigger_log VALUES ('first; log');".into(), + }).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: " INSERT INTO dump_trigger_log VALUES (NEW.value); END; INSERT INTO dump_trigger_source VALUES ('spl".into(), + }) + .await + .unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "it');".into(), + }) + .await + .unwrap(); + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap(); + let result = db + .exec("SELECT value FROM dump_trigger_log ORDER BY rowid") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["value"], "first; log"); + assert_eq!(rows[1]["value"], "split"); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_unterminated_final_comment() { + let mut db = get_import_test_db("sql_dump_import_rejects_unterminated_final_comment"); + db.exec("DROP TABLE IF EXISTS dump_comment_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_comment_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "INSERT INTO dump_comment_test VALUES (1); /* incomplete".into(), + }) + .await + .unwrap(); + assert!(db + .import_sql_action(SqlImportAction::Finish { id }) + .await + .is_err()); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_comment_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_requires_closing_transaction_marker() { + let mut db = get_import_test_db("sql_dump_import_requires_closing_transaction_marker"); + db.exec("DROP TABLE IF EXISTS dump_marker_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_marker_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "BEGIN; INSERT INTO dump_marker_test VALUES (1);".into(), + }) + .await + .unwrap(); + assert!(db + .import_sql_action(SqlImportAction::Finish { id }) + .await + .is_err()); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_marker_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_row_returning_sql_without_leaking_rows() { + let mut db = + get_import_test_db("sql_dump_import_rejects_row_returning_sql_without_leaking_rows"); + db.exec("DROP TABLE IF EXISTS dump_results_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_results_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let err = db + .import_sql_action(SqlImportAction::Chunk { + id, + sql: "INSERT INTO dump_results_test VALUES (1); SELECT * FROM dump_results_test;" + .into(), + }) + .await + .unwrap_err(); + assert!(err.contains("row-returning"), "{err}"); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_results_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let err = db + .import_sql_action(SqlImportAction::Chunk { + id, + sql: "INSERT INTO dump_results_test VALUES (2) RETURNING id;".into(), + }) + .await + .unwrap_err(); + assert!(err.contains("row-returning"), "{err}"); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_bounds_chunks_and_rolls_back_on_connection_close() { + let mut db = + get_import_test_db("sql_dump_import_bounds_chunks_and_rolls_back_on_connection_close"); + db.exec("DROP TABLE IF EXISTS dump_close_test") + .await + .unwrap(); + db.exec("CREATE TABLE dump_close_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "INSERT INTO dump_close_test VALUES (1);".into(), + }) + .await + .unwrap(); + let oversized = "x".repeat(MAX_IMPORT_CHUNK_BYTES + 1); + assert!(db + .import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: oversized, + }) + .await + .unwrap_err() + .contains("chunk exceeds")); + let result = db + .exec("SELECT COUNT(*) AS count FROM dump_close_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id, + sql: "INSERT INTO dump_close_test VALUES (2);".into(), + }) + .await + .unwrap(); + drop(db); + + let mut reopened = + get_import_test_db("sql_dump_import_bounds_chunks_and_rolls_back_on_connection_close"); + let result = reopened + .exec("SELECT COUNT(*) AS count FROM dump_close_test") + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(rows[0]["count"], 0); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_rejects_connection_pragmas_before_they_take_effect() { + let mut db = get_import_test_db("dump_pragma_test"); + db.exec("CREATE TABLE dump_pragma_test (id INTEGER PRIMARY KEY)") + .await + .unwrap(); + for pragma in [ + "PRAGMA query_only=1;", + "PRAGMA writable_schema=1;", + "PRAGMA ignore_check_constraints=1;", + "/* header */ PRAGMA main.query_only(1);", + "PRAGMA foreign_keys=ON;", + "EXPLAIN PRAGMA query_only=1;", + "EXPLAIN PRAGMA ignore_check_constraints=1;", + "EXPLAIN PRAGMA writable_schema=1;", + ] { + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let error = db + .import_sql_action(SqlImportAction::Chunk { + id, + sql: format!("INSERT INTO dump_pragma_test VALUES (1); {pragma}"), + }) + .await + .unwrap_err(); + assert!( + error.contains("PRAGMA") || error.contains("row-returning"), + "{error}" + ); + for name in ["query_only", "writable_schema", "ignore_check_constraints"] { + let rows: serde_json::Value = + serde_json::from_str(&db.exec(&format!("PRAGMA {name}")).await.unwrap()) + .unwrap(); + assert_eq!(rows[0][name], 0, "{pragma} changed {name}"); + } + let rows: serde_json::Value = serde_json::from_str( + &db.exec("SELECT COUNT(*) AS count FROM dump_pragma_test") + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(rows[0]["count"], 0); + db.exec("INSERT INTO dump_pragma_test VALUES (2)") + .await + .unwrap(); + db.exec("DELETE FROM dump_pragma_test").await.unwrap(); + } + db.exec("PRAGMA foreign_keys=ON").await.unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: "PRAGMA foreign_keys=OFF; INSERT INTO dump_pragma_test VALUES (3);".into(), + }) + .await + .unwrap(); + db.import_sql_action(SqlImportAction::Cancel { id }) + .await + .unwrap(); + let rows: serde_json::Value = + serde_json::from_str(&db.exec("PRAGMA foreign_keys").await.unwrap()).unwrap(); + assert_eq!(rows[0]["foreign_keys"], 1); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_does_not_change_database_attachments() { + let mut db = get_import_test_db("dump_attachments_test"); + db.exec("CREATE TABLE dump_attachments_test (id INTEGER)") + .await + .unwrap(); + db.exec("ATTACH DATABASE ':memory:' AS existing") + .await + .unwrap(); + let initial = db.exec("PRAGMA database_list").await.unwrap(); + for statement in [ + "ATTACH DATABASE ':memory:' AS added;", + "DETACH DATABASE existing;", + ] { + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let error = db + .import_sql_action(SqlImportAction::Chunk { + id, + sql: format!("INSERT INTO dump_attachments_test VALUES (1); {statement}"), + }) + .await + .unwrap_err(); + assert!(error.contains("attachments"), "{error}"); + assert_eq!(db.exec("PRAGMA database_list").await.unwrap(), initial); + let rows: serde_json::Value = serde_json::from_str( + &db.exec("SELECT COUNT(*) AS count FROM dump_attachments_test") + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(rows[0]["count"], 0); + } + } + + #[wasm_bindgen_test] + async fn sql_dump_import_scanner_resumes_inside_lexical_states() { + let mut db = get_import_test_db("dump_scanner_test"); + db.exec("CREATE TABLE dump_scanner_test (value TEXT)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + for sql in [ + "INSERT INTO dump_scanner_test VALUES ('a;", + "b'); /", + "* comment; *", + "/ INSERT INTO dump_scanner_test VALUES ('it'", + "'s; fine'); -", + "- line; comment", + "\nINSERT INTO dump_scanner_test VALUES (\"double;", + " quote\"); INSERT INTO [dump_scanner_", + "test] VALUES ('bracket'); INSERT INTO `dump_scanner_", + "test` VALUES ('backtick');", + ] { + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: sql.into(), + }) + .await + .unwrap(); + } + db.import_sql_action(SqlImportAction::Finish { id }) + .await + .unwrap(); + let rows: serde_json::Value = serde_json::from_str( + &db.exec("SELECT value FROM dump_scanner_test ORDER BY rowid") + .await + .unwrap(), + ) + .unwrap(); + assert_eq!( + rows, + json!([ + {"value": "a;b"}, {"value": "it's; fine"}, {"value": "double; quote"}, + {"value": "bracket"}, {"value": "backtick"} + ]) + ); + } + + #[wasm_bindgen_test] + async fn sql_dump_import_bounds_unfinished_statements_and_rolls_back() { + let mut db = get_import_test_db("dump_statement_limit_test"); + db.exec("CREATE TABLE dump_statement_limit_test (value TEXT)") + .await + .unwrap(); + let id = db.import_sql_action(SqlImportAction::Begin).await.unwrap(); + let prefix = "INSERT INTO dump_statement_limit_test VALUES ('partial'); INSERT INTO dump_statement_limit_test VALUES ('"; + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: prefix.into(), + }) + .await + .unwrap(); + let chunk = "x".repeat(MAX_IMPORT_CHUNK_BYTES); + // Reach exactly the statement limit across individually legal chunks. + while db.import_state.as_ref().unwrap().pending.len() < MAX_IMPORT_STATEMENT_BYTES { + let remaining = + MAX_IMPORT_STATEMENT_BYTES - db.import_state.as_ref().unwrap().pending.len(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: chunk[..remaining.min(chunk.len())].into(), + }) + .await + .unwrap(); + } + let error = db + .import_sql_action(SqlImportAction::Chunk { + id, + sql: "x".into(), + }) + .await + .unwrap_err(); + assert!(error.contains("statement exceeds"), "{error}"); + let rows: serde_json::Value = serde_json::from_str( + &db.exec("SELECT COUNT(*) AS count FROM dump_statement_limit_test") + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(rows[0]["count"], 0); + } } diff --git a/packages/sqlite-web-core/src/messages.rs b/packages/sqlite-web-core/src/messages.rs index 6fba81b..a579e7b 100644 --- a/packages/sqlite-web-core/src/messages.rs +++ b/packages/sqlite-web-core/src/messages.rs @@ -21,6 +21,15 @@ pub struct SqlBatchStatement { pub params: Option>, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +#[serde(tag = "kind", rename_all = "kebab-case")] +pub enum SqlImportAction { + Begin, + Chunk { id: String, sql: String }, + Finish { id: String }, + Cancel { id: String }, +} + // Message types for BroadcastChannel communication #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] #[serde(tag = "type")] @@ -50,6 +59,12 @@ pub enum ChannelMessage { query_id: String, statements: Vec, }, + #[serde(rename = "import-request")] + ImportRequest { + #[serde(rename = "queryId")] + query_id: String, + action: SqlImportAction, + }, #[serde(rename = "query-response")] QueryResponse { #[serde(rename = "queryId")] @@ -83,6 +98,12 @@ pub enum WorkerMessage { request_id: u32, statements: Vec, }, + #[serde(rename = "import-sql-dump")] + ImportSqlDump { + #[serde(rename = "requestId")] + request_id: u32, + action: SqlImportAction, + }, } // Messages to main thread @@ -242,6 +263,31 @@ mod tests { } } + #[wasm_bindgen_test] + fn test_import_message_wire_format() { + let begin = WorkerMessage::ImportSqlDump { + request_id: 5, + action: SqlImportAction::Begin, + }; + assert_serialization_roundtrip(begin, "import-sql-dump", |json| { + assert!(json.contains("\"requestId\":5")); + assert!(json.contains("\"kind\":\"begin\"")); + }); + + let chunk = ChannelMessage::ImportRequest { + query_id: "query-1".to_string(), + action: SqlImportAction::Chunk { + id: "import-1".to_string(), + sql: "INSERT INTO data VALUES (1);".to_string(), + }, + }; + assert_serialization_roundtrip(chunk, "import-request", |json| { + assert!(json.contains("\"queryId\":\"query-1\"")); + assert!(json.contains("\"kind\":\"chunk\"")); + assert!(json.contains("\"id\":\"import-1\"")); + }); + } + #[wasm_bindgen_test] fn test_main_thread_messages_serialization() { let success_result = MainThreadMessage::QueryResult { diff --git a/packages/sqlite-web/src/db.rs b/packages/sqlite-web/src/db.rs index f3aa0ee..068042e 100644 --- a/packages/sqlite-web/src/db.rs +++ b/packages/sqlite-web/src/db.rs @@ -1,4 +1,4 @@ -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::collections::HashMap; use std::rc::Rc; @@ -19,12 +19,39 @@ use crate::utils::describe_js_value; use crate::worker::{create_worker_from_code, install_onmessage_handler}; use crate::worker_template::generate_self_contained_worker; +// Bind RegExp.test with a JS string argument so invalid UTF-16 is not +// replaced during conversion to the Rust &str required by js_sys::RegExp.test. +#[wasm_bindgen] +extern "C" { + #[wasm_bindgen(extends = js_sys::Object, js_name = RegExp)] + type SqlChunkUtf16Pattern; + + #[wasm_bindgen(constructor, js_class = RegExp)] + fn new(pattern: &str) -> SqlChunkUtf16Pattern; + + #[wasm_bindgen(method, js_name = test)] + fn has_invalid_utf16(this: &SqlChunkUtf16Pattern, text: &js_sys::JsString) -> bool; +} + +thread_local! { + // A cached, non-global JS regex checks UTF-16 code units entirely in JS. + // JsString::is_valid_utf16 instead crosses Wasm once per code unit. + static INVALID_SQL_CHUNK_UTF16: SqlChunkUtf16Pattern = SqlChunkUtf16Pattern::new( + r"[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?:^|[^\uD800-\uDBFF])[\uDC00-\uDFFF]", + ); +} + +fn is_well_formed_sql_chunk(text: &js_sys::JsString) -> bool { + INVALID_SQL_CHUNK_UTF16.with(|pattern| !pattern.has_invalid_utf16(text)) +} + #[wasm_bindgen] pub struct SQLiteWasmDatabase { worker: Rc>, db_name: String, pending_queries: Rc>>, next_request_id: Rc>, + import_request_in_flight: Cell, ready_signal: ReadySignal, } @@ -70,6 +97,7 @@ impl SQLiteWasmDatabase { db_name: db_name.to_string(), pending_queries, next_request_id, + import_request_in_flight: Cell::new(false), ready_signal, }) } @@ -316,6 +344,149 @@ impl SQLiteWasmDatabase { Ok(result.as_string().unwrap_or_else(|| format!("{result:?}"))) } + /// Begin an atomic SQL dump import. Append bounded text chunks, then finish + /// to commit. SQL validation/execution failures and cancellation roll back. + /// Worker loss during finish can leave the commit outcome unknown, including + /// when the coordinator returns an error; check or reset before retrying. + #[wasm_export(js_name = "beginSqlDumpImport", unchecked_return_type = "string")] + pub async fn begin_sql_dump_import(&self) -> Result { + self.send_import_action("begin", None, None).await + } + + /// Append up to 512 KiB of SQL text. Chunks may split a SQL statement, + /// but must contain valid UTF-16 without splitting a surrogate pair. + /// The dump's outer BEGIN/COMMIT statements are accepted but owned by the worker. + #[wasm_export(js_name = "appendSqlDumpChunk", unchecked_return_type = "string")] + pub async fn append_sql_dump_chunk( + &self, + id: &str, + sql: JsValue, + ) -> Result { + if self.import_request_in_flight.get() { + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "Wait for the previous SQL dump import request to finish", + ))); + } + if !sql.is_string() { + let _ = self.cancel_sql_dump_import(id).await; + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "SQL dump chunk must be a string", + ))); + } + let text = js_sys::JsString::from(sql.clone()); + if text.length() > 512 * 1024 { + let _ = self.cancel_sql_dump_import(id).await; + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "SQL dump chunk exceeds 512 KiB", + ))); + } + if !is_well_formed_sql_chunk(&text) { + let _ = self.cancel_sql_dump_import(id).await; + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "SQL dump chunk contains invalid UTF-16; do not split a surrogate pair", + ))); + } + if sql.as_string().is_some_and(|text| text.len() > 512 * 1024) { + let _ = self.cancel_sql_dump_import(id).await; + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "SQL dump chunk exceeds 512 KiB", + ))); + } + self.send_import_action("chunk", Some(id), Some(sql)).await + } + + /// Commit all imported statements. A partial trailing statement rolls back. + #[wasm_export(js_name = "finishSqlDumpImport", unchecked_return_type = "string")] + pub async fn finish_sql_dump_import( + &self, + id: &str, + ) -> Result { + self.send_import_action("finish", Some(id), None).await + } + + /// Roll back the active import. + #[wasm_export(js_name = "cancelSqlDumpImport", unchecked_return_type = "string")] + pub async fn cancel_sql_dump_import( + &self, + id: &str, + ) -> Result { + self.send_import_action("cancel", Some(id), None).await + } + + async fn send_import_action( + &self, + kind: &str, + id: Option<&str>, + sql: Option, + ) -> Result { + if let InitializationState::Failed(reason) = self.ready_signal.current_state() { + return Err(SQLiteWasmDatabaseError::InitializationFailed(reason)); + } + let action = js_sys::Object::new(); + Reflect::set( + &action, + &JsValue::from_str("kind"), + &JsValue::from_str(kind), + ) + .map_err(SQLiteWasmDatabaseError::JsError)?; + if let Some(id) = id { + Reflect::set(&action, &JsValue::from_str("id"), &JsValue::from_str(id)) + .map_err(SQLiteWasmDatabaseError::JsError)?; + } + if let Some(sql) = sql { + Reflect::set(&action, &JsValue::from_str("sql"), &sql) + .map_err(SQLiteWasmDatabaseError::JsError)?; + } + let message = js_sys::Object::new(); + Reflect::set( + &message, + &JsValue::from_str("type"), + &JsValue::from_str("import-sql-dump"), + ) + .map_err(SQLiteWasmDatabaseError::JsError)?; + let request_id = { + let mut next = self.next_request_id.borrow_mut(); + let id = *next; + *next = next.wrapping_add(1).max(1); + id + }; + Reflect::set( + &message, + &JsValue::from_str("requestId"), + &JsValue::from_f64(request_id as f64), + ) + .map_err(SQLiteWasmDatabaseError::JsError)?; + Reflect::set(&message, &JsValue::from_str("action"), &action) + .map_err(SQLiteWasmDatabaseError::JsError)?; + if self.import_request_in_flight.replace(true) { + return Err(SQLiteWasmDatabaseError::JsError(JsValue::from_str( + "Wait for the previous SQL dump import request to finish", + ))); + } + let worker = Rc::clone(&self.worker); + let pending = Rc::clone(&self.pending_queries); + let promise = js_sys::Promise::new(&mut |resolve, reject| match worker + .borrow() + .post_message(&message) + { + Ok(()) => { + pending.borrow_mut().insert(request_id, (resolve, reject)); + } + Err(err) => { + let _ = reject.call1(&JsValue::NULL, &err); + } + }); + let outcome = JsFuture::from(promise).await; + self.import_request_in_flight.set(false); + match outcome { + Ok(value) => Ok(value.as_string().unwrap_or_else(|| format!("{value:?}"))), + Err(err) if is_initialization_pending_error(&err) => { + Err(SQLiteWasmDatabaseError::InitializationPending) + } + Err(err) => Err(SQLiteWasmDatabaseError::JsError(err)), + } + } + #[wasm_export(js_name = "wipeAndRecreate", unchecked_return_type = "void")] pub async fn wipe_and_recreate(&self) -> Result<(), SQLiteWasmDatabaseError> { self.worker.borrow().terminate(); @@ -401,6 +572,14 @@ mod tests { wasm_bindgen_test_configure!(run_in_browser); + struct TestWorkerCleanup(Rc>); + + impl Drop for TestWorkerCleanup { + fn drop(&mut self) { + self.0.borrow().terminate(); + } + } + #[wasm_bindgen_test] fn normalize_params_handles_none_and_empty_arrays() { let empty = SQLiteWasmDatabase::normalize_params(None).expect("None => empty array"); @@ -597,6 +776,93 @@ mod tests { } } + #[wasm_bindgen_test] + fn sql_dump_chunk_utf16_validation_preserves_pairs_and_replacement_characters() { + let cases: &[(&[u16], bool)] = &[ + (&[], true), + (&[0x61], true), + (&[0xFFFD], true), + (&[0xD800, 0xDC00], true), + (&[0xDBFF, 0xDFFF], true), + (&[0xD83D, 0xDE00, 0xD83D, 0xDE03], true), + (&[0xD800], false), + (&[0xDBFF], false), + (&[0xDC00], false), + (&[0xDFFF], false), + (&[0xD83D, 0xD83D], false), + (&[0xDE00, 0xD83D], false), + (&[0x61, 0xDE00], false), + (&[0xD83D, 0x61], false), + (&[0xD83D, 0xDE00, 0xDE03], false), + ]; + for (units, expected) in cases { + let text = js_sys::JsString::from_char_code(units); + assert_eq!(is_well_formed_sql_chunk(&text), *expected, "{units:?}"); + } + } + + #[wasm_bindgen_test] + async fn follower_import_reports_unknown_outcome_after_leader_termination() { + // A controlled DB worker deliberately never answers chunks. The real + // coordinator and public follower must settle that lost response when + // the leader terminates, independent of SQL/browser scheduling speed. + let name = "test_import_leader_termination"; + let stalled_db_worker = r#" + self.postMessage({ type: "worker-ready" }); + self.onmessage = event => { + const message = event.data; + if (message.type === "import-sql-dump" && message.action.kind === "begin") { + self.postMessage({ type: "query-result", requestId: message.requestId, + result: "test-session", error: null }); + } + }; + "#; + let code = format!( + "{}\nself.__SQLITE_EMBEDDED_WORKER = {};\n", + generate_self_contained_worker(name), + serde_json::to_string(stalled_db_worker).unwrap(), + ); + let leader_worker = create_worker_from_code(&code).unwrap(); + let _leader_cleanup = TestWorkerCleanup(Rc::new(RefCell::new(leader_worker.clone()))); + let ready = ReadySignal::new(); + let ready_promise = ready.wait_promise().unwrap(); + install_onmessage_handler(&leader_worker, Rc::new(RefCell::new(HashMap::new())), ready); + JsFuture::from(ready_promise).await.unwrap(); + let follower = SQLiteWasmDatabase::new(name).await.unwrap(); + let _follower_cleanup = TestWorkerCleanup(Rc::clone(&follower.worker)); + let id = follower.begin_sql_dump_import().await.unwrap(); + let channel = web_sys::BroadcastChannel::new(&format!("sqlite-queries-{name}")).unwrap(); + let listener = Closure::wrap(Box::new(move |event: web_sys::MessageEvent| { + let message = event.data(); + if Reflect::get(&message, &JsValue::from_str("type")) + .unwrap() + .as_string() + .as_deref() + == Some("import-request") + { + leader_worker.terminate(); + } + }) as Box); + channel.set_onmessage(Some(listener.as_ref().unchecked_ref())); + let result = follower + .append_sql_dump_chunk( + &id, + JsValue::from_str("CREATE TABLE lost_response_test (id INTEGER);"), + ) + .await; + channel.set_onmessage(None); + channel.close(); + let error = result.unwrap_err(); + match error { + SQLiteWasmDatabaseError::JsError(value) => { + let message = describe_js_value(&value); + assert!(message.contains("outcome unknown"), "{message}"); + } + other => panic!("unexpected error: {other:?}"), + } + assert!(!follower.import_request_in_flight.get()); + } + #[wasm_bindgen_test] fn detects_structured_initialization_pending_errors() { let err = Object::new(); @@ -618,6 +884,7 @@ mod tests { #[wasm_bindgen_test(async)] async fn wipe_and_recreate_tests() { let db = SQLiteWasmDatabase::new("test_wipe").await.unwrap(); + let _cleanup = TestWorkerCleanup(Rc::clone(&db.worker)); db.wipe_and_recreate().await.unwrap(); db.query( diff --git a/svelte-test/benchmarks/sql-dump-import.benchmark.ts b/svelte-test/benchmarks/sql-dump-import.benchmark.ts new file mode 100644 index 0000000..f4912a1 --- /dev/null +++ b/svelte-test/benchmarks/sql-dump-import.benchmark.ts @@ -0,0 +1,69 @@ +import { describe, expect, it } from "vitest"; +import { createTestDatabase } from "../tests/fixtures/test-helpers.js"; + +// Run alone to avoid concurrent OPFS/CPU contention. The two grouped cases use +// identical SQL so their difference isolates the API rather than row grouping. +// This fixed-order, single-run comparison uses warm storage and small chunks; +// treat it as a smoke measurement, not a production speedup estimate. +describe("SQL dump import benchmark", () => { + it("records sequential equal-data timings", async () => { + const db = await createTestDatabase(`sql-import-benchmark-${Date.now()}`); + const rowCount = 10_000; + expect( + (await db.query("CREATE TABLE benchmark_import (id INTEGER PRIMARY KEY, value TEXT)")).error, + ).toBeUndefined(); + + const singleStart = performance.now(); + const statements = Array.from({ length: rowCount }, (_, i) => ({ + sql: `INSERT INTO benchmark_import VALUES (${i}, 'value ${i}')`, + })); + const singleResult = await db.transaction(statements); + const singleMs = performance.now() - singleStart; + expect(singleResult.error).toBeUndefined(); + + const readData = async () => { + const result = await db.query( + "SELECT COUNT(*) AS count, SUM(id) AS ids, SUM(LENGTH(value)) AS bytes FROM benchmark_import", + ); + expect(result.error).toBeUndefined(); + return JSON.parse(result.value || "[]"); + }; + const reference = await readData(); + expect(reference[0].count).toBe(rowCount); + expect((await db.query("DELETE FROM benchmark_import")).error).toBeUndefined(); + + // Build once outside both grouped timings; each timed path receives the same + // prebuilt strings. The single-row timing above includes its array creation. + const groupedSql: string[] = []; + for (let offset = 0; offset < rowCount; offset += 256) { + const tuples = Array.from( + { length: Math.min(256, rowCount - offset) }, + (_, i) => `(${offset + i}, 'value ${offset + i}')`, + ); + groupedSql.push(`INSERT INTO benchmark_import VALUES ${tuples.join(",")};\n`); + } + + const groupedStart = performance.now(); + const groupedResult = await db.transaction(groupedSql.map((sql) => ({ sql }))); + const groupedMs = performance.now() - groupedStart; + expect(groupedResult.error).toBeUndefined(); + expect(await readData()).toEqual(reference); + expect((await db.query("DELETE FROM benchmark_import")).error).toBeUndefined(); + + const chunkStart = performance.now(); + const begin = await db.beginSqlDumpImport(); + expect(begin.error).toBeUndefined(); + const id = begin.value!; + for (const sql of groupedSql) { + expect((await db.appendSqlDumpChunk(id, sql)).error).toBeUndefined(); + } + const finish = await db.finishSqlDumpImport(id); + const chunkMs = performance.now() - chunkStart; + expect(finish.error).toBeUndefined(); + expect(await readData()).toEqual(reference); + + console.info( + `SQL import benchmark: ${rowCount} rows; single-row transaction (including construction) ${singleMs.toFixed(1)} ms; grouped transaction ${groupedMs.toFixed(1)} ms; identical grouped chunks ${chunkMs.toFixed(1)} ms`, + ); + }, 120_000); +}); diff --git a/svelte-test/package.json b/svelte-test/package.json index 1955d31..138630f 100644 --- a/svelte-test/package.json +++ b/svelte-test/package.json @@ -32,6 +32,7 @@ "check": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json", "check:watch": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json --watch", "test": "vitest run", + "test:benchmark": "vitest run --config vitest.benchmark.config.js", "test:browser": "vitest --browser", "test:ui": "vitest --ui --browser", "lint": "eslint --fix src", diff --git a/svelte-test/tests/integration/sql-dump-import.test.ts b/svelte-test/tests/integration/sql-dump-import.test.ts new file mode 100644 index 0000000..257a817 --- /dev/null +++ b/svelte-test/tests/integration/sql-dump-import.test.ts @@ -0,0 +1,218 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { SQLiteWasmDatabase } from "@rainlanguage/sqlite-web"; +import { createTestDatabase } from "../fixtures/test-helpers.js"; + +describe("SQL dump import", () => { + const dbName = `sql-dump-import-${Date.now()}`; + const cancellations: Array<() => Promise> = []; + + async function beginImport(db: SQLiteWasmDatabase): Promise { + const begin = await db.beginSqlDumpImport(); + expect(begin.error).toBeUndefined(); + const id = begin.value!; + cancellations.push(() => db.cancelSqlDumpImport(id)); + return id; + } + + afterEach(async () => { + try { + // Cancel even when an assertion fails. Already finished/failed sessions + // return an error without disturbing a later import. + for (const cancel of cancellations.splice(0)) await cancel(); + } finally { + vi.restoreAllMocks(); + } + }); + + it("keeps a multi-chunk import private across clients until commit", async () => { + const leader = await createTestDatabase(dbName); + const follower = await createTestDatabase(dbName); + await leader.query("DROP TABLE IF EXISTS streamed_dump_test"); + await leader.query( + "CREATE TABLE streamed_dump_test (id INTEGER PRIMARY KEY, value TEXT)", + ); + + const id = await beginImport(follower); + const first = await follower.appendSqlDumpChunk( + id, + "BEGIN; INSERT INTO streamed_dump_test VALUES (1, 'one; value'); INS", + ); + expect(first.error).toBeUndefined(); + + const competingRead = await leader.query( + "SELECT COUNT(*) AS count FROM streamed_dump_test", + ); + expect( + competingRead.error?.readableMsg ?? competingRead.error?.msg ?? "", + ).toContain("SQL dump import is in progress"); + + const second = await follower.appendSqlDumpChunk( + id, + "ERT INTO streamed_dump_test VALUES (2, 'two'); COMMIT;", + ); + expect(second.error).toBeUndefined(); + const finish = await follower.finishSqlDumpImport(id); + expect(finish.error).toBeUndefined(); + expect(finish.value).toContain("Imported 2 SQL statements"); + + const finalRead = await leader.query( + "SELECT value FROM streamed_dump_test ORDER BY id", + ); + expect( + JSON.parse(finalRead.value || "[]").map( + (row: { value: string }) => row.value, + ), + ).toEqual(["one; value", "two"]); + await leader.query("DROP TABLE streamed_dump_test"); + }); + + it("rolls back a malformed chunk", async () => { + const db = await createTestDatabase(dbName); + await db.query("DROP TABLE IF EXISTS failed_dump_test"); + await db.query("CREATE TABLE failed_dump_test (id INTEGER PRIMARY KEY)"); + + const id = await beginImport(db); + const failed = await db.appendSqlDumpChunk( + id, + "INSERT INTO failed_dump_test VALUES (1); INSERT INTO failed_dump_test (missing) VALUES (2);", + ); + expect(failed.error).toBeDefined(); + const result = await db.query( + "SELECT COUNT(*) AS count FROM failed_dump_test", + ); + expect(JSON.parse(result.value || "[]")[0].count).toBe(0); + await db.query("DROP TABLE failed_dump_test"); + }); + + it("requires the caller to await each chunk", async () => { + const db = await createTestDatabase(dbName); + await db.query("DROP TABLE IF EXISTS serial_dump_test"); + await db.query("CREATE TABLE serial_dump_test (id INTEGER PRIMARY KEY)"); + const id = await beginImport(db); + + const results = await Promise.all([ + db.appendSqlDumpChunk(id, "INSERT INTO serial_dump_test VALUES (1);"), + db.appendSqlDumpChunk(id, "INSERT INTO serial_dump_test VALUES (2);"), + ]); + expect(results.filter((result) => result.error).length).toBe(1); + expect((await db.finishSqlDumpImport(id)).error).toBeUndefined(); + const count = await db.query( + "SELECT COUNT(*) AS count FROM serial_dump_test", + ); + expect(JSON.parse(count.value || "[]")[0].count).toBe(1); + await db.query("DROP TABLE serial_dump_test"); + }); + + it("rejects oversized UTF-8 chunks before dispatch and cancels the import", async () => { + const db = await createTestDatabase(dbName); + await db.query("DROP TABLE IF EXISTS unicode_dump_test"); + await db.query("CREATE TABLE unicode_dump_test (id INTEGER PRIMARY KEY)"); + const id = await beginImport(db); + expect( + ( + await db.appendSqlDumpChunk( + id, + "INSERT INTO unicode_dump_test VALUES (1);", + ) + ).error, + ).toBeUndefined(); + + const dispatch = vi.spyOn(Worker.prototype, "postMessage"); + const oversized = await db.appendSqlDumpChunk(id, "界".repeat(200_000)); + expect(oversized.error).toBeDefined(); + expect( + dispatch.mock.calls.filter( + ([message]) => message?.action?.kind === "chunk", + ), + ).toHaveLength(0); + const count = await db.query( + "SELECT COUNT(*) AS count FROM unicode_dump_test", + ); + expect(JSON.parse(count.value || "[]")[0].count).toBe(0); + await db.query("DROP TABLE unicode_dump_test"); + }); + + it("does not race an invalid chunk's cancellation with an active append", async () => { + const db = await createTestDatabase(dbName); + await db.query("DROP TABLE IF EXISTS concurrent_invalid_dump_test"); + await db.query( + "CREATE TABLE concurrent_invalid_dump_test (id INTEGER PRIMARY KEY)", + ); + const id = await beginImport(db); + const first = db.appendSqlDumpChunk( + id, + `INSERT INTO concurrent_invalid_dump_test VALUES (1); /*${"x".repeat(400_000)}*/`, + ); + const invalid = db.appendSqlDumpChunk(id, "界".repeat(200_000)); + const [firstResult, invalidResult] = await Promise.all([first, invalid]); + expect(firstResult.error).toBeUndefined(); + expect( + invalidResult.error?.readableMsg ?? invalidResult.error?.msg ?? "", + ).toContain("Wait for the previous SQL dump import request"); + + expect((await db.cancelSqlDumpImport(id)).error).toBeUndefined(); + const count = await db.query( + "SELECT COUNT(*) AS count FROM concurrent_invalid_dump_test", + ); + expect(JSON.parse(count.value || "[]")[0].count).toBe(0); + await db.query("DROP TABLE concurrent_invalid_dump_test"); + }); + + it.each(["\uD83D", "\uDE00"])( + "rejects an unpaired surrogate before dispatch and rolls back", + async (surrogate) => { + const db = await createTestDatabase(dbName); + await db.query( + "CREATE TABLE IF NOT EXISTS surrogate_dump_test (value TEXT)", + ); + const id = await beginImport(db); + expect( + ( + await db.appendSqlDumpChunk( + id, + "INSERT INTO surrogate_dump_test VALUES ('before');", + ) + ).error, + ).toBeUndefined(); + const dispatch = vi.spyOn(Worker.prototype, "postMessage"); + const invalid = await db.appendSqlDumpChunk( + id, + `INSERT INTO surrogate_dump_test VALUES ('${surrogate}`, + ); + expect(invalid.error?.readableMsg ?? invalid.error?.msg ?? "").toContain( + "invalid UTF-16", + ); + expect( + dispatch.mock.calls.filter( + ([message]) => message?.action?.kind === "chunk", + ), + ).toHaveLength(0); + const rows = await db.query( + "SELECT COUNT(*) AS count FROM surrogate_dump_test", + ); + expect(rows.error).toBeUndefined(); + expect(JSON.parse(rows.value || "[]")[0].count).toBe(0); + await db.query("DROP TABLE surrogate_dump_test"); + }, + ); + + it("preserves complete surrogate pairs across SQL chunks", async () => { + const db = await createTestDatabase(dbName); + await db.query("CREATE TABLE emoji_dump_test (value TEXT)"); + const id = await beginImport(db); + expect( + ( + await db.appendSqlDumpChunk( + id, + "INSERT INTO emoji_dump_test VALUES ('😀", + ) + ).error, + ).toBeUndefined(); + expect((await db.appendSqlDumpChunk(id, "😃');")).error).toBeUndefined(); + expect((await db.finishSqlDumpImport(id)).error).toBeUndefined(); + const rows = await db.query("SELECT value FROM emoji_dump_test"); + expect(rows.error).toBeUndefined(); + expect(JSON.parse(rows.value || "[]")[0].value).toBe("😀😃"); + await db.query("DROP TABLE emoji_dump_test"); + }); +}); diff --git a/svelte-test/vitest.benchmark.config.js b/svelte-test/vitest.benchmark.config.js new file mode 100644 index 0000000..43cd2c9 --- /dev/null +++ b/svelte-test/vitest.benchmark.config.js @@ -0,0 +1,10 @@ +import testConfig from "./vitest.config.js"; + +export default { + ...testConfig, + test: { + ...testConfig.test, + include: ["benchmarks/**/*.benchmark.ts"], + fileParallelism: false, + }, +};