From ec75a0aeda15baa356226445577c41ba920ccce9 Mon Sep 17 00:00:00 2001 From: Arda Nakisci Date: Mon, 28 Sep 2026 13:08:59 +0200 Subject: [PATCH 1/3] Add bounded atomic SQL dump import --- docs/sql-dump-import.md | 55 ++ packages/sqlite-web-core/src/coordination.rs | 205 ++++- packages/sqlite-web-core/src/database.rs | 720 +++++++++++++++++- packages/sqlite-web-core/src/messages.rs | 46 ++ packages/sqlite-web/src/db.rs | 137 +++- .../benchmarks/sql-dump-import.benchmark.ts | 46 ++ .../tests/integration/sql-dump-import.test.ts | 112 +++ svelte-test/vitest.benchmark.config.js | 10 + 8 files changed, 1319 insertions(+), 12 deletions(-) create mode 100644 docs/sql-dump-import.md create mode 100644 svelte-test/benchmarks/sql-dump-import.benchmark.ts create mode 100644 svelte-test/tests/integration/sql-dump-import.test.ts create mode 100644 svelte-test/vitest.benchmark.config.js diff --git a/docs/sql-dump-import.md b/docs/sql-dump-import.md new file mode 100644 index 0000000..e63c16f --- /dev/null +++ b/docs/sql-dump-import.md @@ -0,0 +1,55 @@ +# 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. An opening marker requires a matching closing marker. + +```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, cancellation, or worker termination rolls back the whole import. +Row-returning SQL such as `SELECT` or `INSERT ... RETURNING` is rejected because +an import does not consume query results. +An import idle for two minutes is rolled back by the worker. The caller should +retry from a clean session after any failure. Await each append before sending +the next one; concurrent import operations are rejected to keep queued SQL +bounded. 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. + +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..3cb2ded 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,10 +995,28 @@ impl DbWorkerState { } => { self.enqueue_job(request_id, DbJobPayload::Batch { statements }); } + WorkerMessage::ImportSqlDump { request_id, action } => { + self.enqueue_job(request_id, DbJobPayload::Import { action }); + } } } fn enqueue_job(self: &Rc, request_id: u32, payload: DbJobPayload) { + if matches!(&payload, DbJobPayload::Import { .. }) + && self + .db_queue + .borrow() + .iter() + .any(|job| matches!(&job.payload, DbJobPayload::Import { .. })) + { + if let Ok(response) = make_query_result_message( + request_id, + Err("Wait for the previous SQL dump import request to finish".to_string()), + ) { + (self.hooks.deliver)(&response); + } + return; + } self.db_queue.borrow_mut().push_back(DbJob { request_id, payload, @@ -891,7 +1040,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 +1202,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 +1222,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 +1232,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 +1241,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 +1250,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 +1341,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..2b9dd65 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,187 @@ 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())); + if state.statement_count == 0 + || state.saw_begin != state.saw_commit + || !complete_tail + { + self.rollback_if_in_transaction().await; + self.refresh_transaction_state(); + return Err( + "SQL dump ends with an incomplete statement or contains no statements." + .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() + ); + } + 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(); + } + } + fn refresh_transaction_state(&mut self) { self.in_transaction = unsafe { sqlite3_get_autocommit(self.db) } == 0; } @@ -600,6 +881,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 +911,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 +983,7 @@ impl SQLiteDatabase { if i == start { None } else { - Some(&sql[start..i]) + Some((&sql[start..i], &sql[i..])) } } @@ -753,6 +1080,7 @@ impl SQLiteDatabase { Ok(SQLiteDatabase { db, in_transaction: false, + import_state: None, }) } @@ -831,6 +1159,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 +1252,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 +1272,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 +2779,389 @@ mod tests { assert_eq!(array[0]["msg"].as_str().unwrap(), "insert; happened"); assert_eq!(array[1]["msg"].as_str().unwrap(), "second; line"); } + + #[wasm_bindgen_test] + async fn sql_dump_import_accepts_split_statements_and_persists() { + let Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut reopened) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 2 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_rolls_back_on_invalid_statement() { + let Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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'); INSERT INTO dump_trigger_log VALUES (NEW.value); END; INSERT INTO dump_trigger_source VALUES ('split'".into(), + }).await.unwrap(); + db.import_sql_action(SqlImportAction::Chunk { + id: id.clone(), + sql: ");".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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut db) = get_test_db().await else { + return; + }; + 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 Some(mut reopened) = get_test_db().await else { + return; + }; + 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); + } } 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..3e5e2ea 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; @@ -25,6 +25,7 @@ pub struct SQLiteWasmDatabase { db_name: String, pending_queries: Rc>>, next_request_id: Rc>, + import_request_in_flight: Cell, ready_signal: ReadySignal, } @@ -70,6 +71,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 +318,139 @@ 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. A failure or cancellation rolls the entire import back. + #[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. + /// 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", + ))); + } + if js_sys::JsString::from(sql.clone()).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 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(); 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..c0f116a --- /dev/null +++ b/svelte-test/benchmarks/sql-dump-import.benchmark.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import { createTestDatabase } from "../tests/fixtures/test-helpers.js"; + +// Manual comparison of equal rows using the old statement-array path and the +// bounded grouped-SQL path. Run alone to avoid concurrent OPFS/CPU contention. +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; + await db.query("CREATE TABLE old_import (id INTEGER PRIMARY KEY, value TEXT)"); + await db.query("CREATE TABLE new_import (id INTEGER PRIMARY KEY, value TEXT)"); + + const oldStart = performance.now(); + const statements = Array.from({ length: rowCount }, (_, i) => ({ + sql: `INSERT INTO old_import VALUES (${i}, 'value ${i}')`, + })); + const oldResult = await db.transaction(statements); + const oldMs = performance.now() - oldStart; + expect(oldResult.error).toBeUndefined(); + + const newStart = performance.now(); + const id = (await db.beginSqlDumpImport()).value!; + for (let offset = 0; offset < rowCount; offset += 256) { + const tuples = Array.from( + { length: Math.min(256, rowCount - offset) }, + (_, i) => `(${offset + i}, 'value ${offset + i}')`, + ); + const sql = `INSERT INTO new_import VALUES ${tuples.join(",")};\n`; + const result = await db.appendSqlDumpChunk(id, sql); + expect(result.error).toBeUndefined(); + } + const finish = await db.finishSqlDumpImport(id); + const newMs = performance.now() - newStart; + expect(finish.error).toBeUndefined(); + + const oldData = await db.query( + "SELECT COUNT(*) AS count, SUM(id) AS ids, SUM(LENGTH(value)) AS bytes FROM old_import", + ); + const newData = await db.query( + "SELECT COUNT(*) AS count, SUM(id) AS ids, SUM(LENGTH(value)) AS bytes FROM new_import", + ); + expect(JSON.parse(oldData.value || "[]")).toEqual(JSON.parse(newData.value || "[]")); + expect(JSON.parse(newData.value || "[]")[0].count).toBe(rowCount); + console.info(`SQL import benchmark: ${rowCount} rows; statement array ${oldMs.toFixed(1)} ms; grouped chunks ${newMs.toFixed(1)} ms`); + }, 120_000); +}); 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..5cb69b0 --- /dev/null +++ b/svelte-test/tests/integration/sql-dump-import.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it } from "vitest"; +import { createTestDatabase } from "../fixtures/test-helpers.js"; + +describe("SQL dump import", () => { + const dbName = `sql-dump-import-${Date.now()}`; + + 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 begin = await follower.beginSqlDumpImport(); + expect(begin.error).toBeUndefined(); + const id = begin.value!; + 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 db.beginSqlDumpImport()).value!; + 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 db.beginSqlDumpImport()).value!; + + 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 db.beginSqlDumpImport()).value!; + expect((await db.appendSqlDumpChunk(id, "INSERT INTO unicode_dump_test VALUES (1);")).error).toBeUndefined(); + + const oversized = await db.appendSqlDumpChunk(id, "界".repeat(200_000)); + expect(oversized.error).toBeDefined(); + 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 db.beginSqlDumpImport()).value!; + 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"); + }); +}); 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, + }, +}; From 4cbc50a8deb380d63e0f3c3bebe3fe9f5ee0d3a9 Mon Sep 17 00:00:00 2001 From: Arda Nakisci Date: Wed, 30 Sep 2026 10:06:38 +0200 Subject: [PATCH 2/3] Fix SQL dump import review findings --- docs/sql-dump-import.md | 57 +++- packages/sqlite-web-core/src/database.rs | 316 +++++++++++++++--- packages/sqlite-web/src/db.rs | 134 +++++++- .../tests/integration/sql-dump-import.test.ts | 158 +++++++-- 4 files changed, 577 insertions(+), 88 deletions(-) diff --git a/docs/sql-dump-import.md b/docs/sql-dump-import.md index e63c16f..d2c8cae 100644 --- a/docs/sql-dump-import.md +++ b/docs/sql-dump-import.md @@ -6,7 +6,25 @@ 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. An opening marker requires a matching closing marker. +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. + +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. + +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(); @@ -18,7 +36,9 @@ try { 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 text = decoder.decode(bytes.subarray(offset, offset + 256 * 1024), { + stream: true, + }); const append = await db.appendSqlDumpChunk(id, text); if (append.error) throw append.error; } @@ -36,20 +56,25 @@ try { } ``` -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, cancellation, or worker termination rolls back the whole import. +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 and the +finish response never arrives, the commit outcome is unknown: check or reset the +database before retrying, even when the coordinator leader did not change. Row-returning SQL such as `SELECT` or `INSERT ... RETURNING` is rejected because -an import does not consume query results. -An import idle for two minutes is rolled back by the worker. The caller should -retry from a clean session after any failure. Await each append before sending -the next one; concurrent import operations are rejected to keep queued SQL -bounded. 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. +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 are rejected to keep queued SQL bounded. +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. -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 +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/database.rs b/packages/sqlite-web-core/src/database.rs index 2b9dd65..4b3b269 100644 --- a/packages/sqlite-web-core/src/database.rs +++ b/packages/sqlite-web-core/src/database.rs @@ -335,12 +335,32 @@ impl SQLiteDatabase { "SQL dump contains an unexpected transaction statement.".to_string() ); } - self.exec_import_statement(statement).await.map_err(|err| { - format!( - "SQL dump statement {} failed: {err}", - state.statement_count + 1 - ) - })?; + // 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; } } @@ -349,6 +369,19 @@ impl SQLiteDatabase { } } + // 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; } @@ -2780,11 +2813,32 @@ mod tests { 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 Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2815,9 +2869,8 @@ mod tests { ); drop(db); - let Some(mut reopened) = get_test_db().await else { - return; - }; + 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 @@ -2829,9 +2882,9 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_accepts_cli_transaction_markers_and_empty_statements() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2871,9 +2924,7 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_rolls_back_on_invalid_statement() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2896,9 +2947,8 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_rejects_other_clients_and_cancel_rolls_back() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2931,9 +2981,7 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_rejects_incomplete_tail_and_expires() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2975,9 +3023,7 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_handles_trigger_body_and_split_quote() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -2993,11 +3039,17 @@ mod tests { 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'); INSERT INTO dump_trigger_log VALUES (NEW.value); END; INSERT INTO dump_trigger_source VALUES ('split'".into(), + 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: ");".into(), + 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(); @@ -3015,9 +3067,7 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_rejects_unterminated_final_comment() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -3045,9 +3095,7 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_requires_closing_transaction_marker() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -3075,9 +3123,8 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_rejects_row_returning_sql_without_leaking_rows() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -3114,9 +3161,8 @@ mod tests { #[wasm_bindgen_test] async fn sql_dump_import_bounds_chunks_and_rolls_back_on_connection_close() { - let Some(mut db) = get_test_db().await else { - return; - }; + 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(); @@ -3154,9 +3200,8 @@ mod tests { .unwrap(); drop(db); - let Some(mut reopened) = get_test_db().await else { - return; - }; + 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 @@ -3164,4 +3209,187 @@ mod tests { 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/src/db.rs b/packages/sqlite-web/src/db.rs index 3e5e2ea..e4ad281 100644 --- a/packages/sqlite-web/src/db.rs +++ b/packages/sqlite-web/src/db.rs @@ -19,6 +19,32 @@ 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>, @@ -325,7 +351,8 @@ impl SQLiteWasmDatabase { self.send_import_action("begin", None, None).await } - /// Append up to 512 KiB of SQL text. Chunks may split a SQL statement. + /// 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( @@ -344,12 +371,19 @@ impl SQLiteWasmDatabase { "SQL dump chunk must be a string", ))); } - if js_sys::JsString::from(sql.clone()).length() > 512 * 1024 { + 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( @@ -536,6 +570,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"); @@ -732,6 +774,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(); @@ -753,6 +882,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/tests/integration/sql-dump-import.test.ts b/svelte-test/tests/integration/sql-dump-import.test.ts index 5cb69b0..257a817 100644 --- a/svelte-test/tests/integration/sql-dump-import.test.ts +++ b/svelte-test/tests/integration/sql-dump-import.test.ts @@ -1,28 +1,50 @@ -import { describe, expect, it } from "vitest"; +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)"); + await leader.query( + "CREATE TABLE streamed_dump_test (id INTEGER PRIMARY KEY, value TEXT)", + ); - const begin = await follower.beginSqlDumpImport(); - expect(begin.error).toBeUndefined(); - const id = begin.value!; + 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 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, @@ -33,11 +55,14 @@ describe("SQL dump import", () => { 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", - ]); + 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"); }); @@ -46,13 +71,15 @@ describe("SQL dump import", () => { 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 db.beginSqlDumpImport()).value!; + 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"); + 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"); }); @@ -61,7 +88,7 @@ describe("SQL dump import", () => { 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 db.beginSqlDumpImport()).value!; + const id = await beginImport(db); const results = await Promise.all([ db.appendSqlDumpChunk(id, "INSERT INTO serial_dump_test VALUES (1);"), @@ -69,7 +96,9 @@ describe("SQL dump import", () => { ]); 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"); + 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"); }); @@ -78,12 +107,27 @@ describe("SQL dump import", () => { 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 db.beginSqlDumpImport()).value!; - expect((await db.appendSqlDumpChunk(id, "INSERT INTO unicode_dump_test VALUES (1);")).error).toBeUndefined(); + 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(); - const count = await db.query("SELECT COUNT(*) AS count FROM unicode_dump_test"); + 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"); }); @@ -91,8 +135,10 @@ describe("SQL dump import", () => { 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 db.beginSqlDumpImport()).value!; + 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)}*/`, @@ -100,13 +146,73 @@ describe("SQL dump import", () => { 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( + 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"); + 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"); + }); }); From 9c5b64660566b705c8bf1207eef1259213d5c977 Mon Sep 17 00:00:00 2001 From: Arda Nakisci Date: Wed, 30 Sep 2026 14:29:11 +0200 Subject: [PATCH 3/3] Address SQL dump import follow-up review --- docs/sql-dump-import.md | 44 +++++++--- packages/sqlite-web-core/src/coordination.rs | 15 ---- packages/sqlite-web-core/src/database.rs | 83 ++++++++++++++++--- packages/sqlite-web/src/db.rs | 4 +- .../benchmarks/sql-dump-import.benchmark.ts | 69 ++++++++++----- svelte-test/package.json | 1 + 6 files changed, 155 insertions(+), 61 deletions(-) diff --git a/docs/sql-dump-import.md b/docs/sql-dump-import.md index d2c8cae..a01d962 100644 --- a/docs/sql-dump-import.md +++ b/docs/sql-dump-import.md @@ -13,6 +13,11 @@ 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 @@ -22,6 +27,10 @@ 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. @@ -60,20 +69,31 @@ 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 and the -finish response never arrives, the commit outcome is unknown: check or reset the -database before retrying, even when the coordinator leader did not change. -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 +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 are rejected to keep queued SQL bounded. -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. +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 diff --git a/packages/sqlite-web-core/src/coordination.rs b/packages/sqlite-web-core/src/coordination.rs index 3cb2ded..8f33d81 100644 --- a/packages/sqlite-web-core/src/coordination.rs +++ b/packages/sqlite-web-core/src/coordination.rs @@ -1002,21 +1002,6 @@ impl DbWorkerState { } fn enqueue_job(self: &Rc, request_id: u32, payload: DbJobPayload) { - if matches!(&payload, DbJobPayload::Import { .. }) - && self - .db_queue - .borrow() - .iter() - .any(|job| matches!(&job.payload, DbJobPayload::Import { .. })) - { - if let Ok(response) = make_query_result_message( - request_id, - Err("Wait for the previous SQL dump import request to finish".to_string()), - ) { - (self.hooks.deliver)(&response); - } - return; - } self.db_queue.borrow_mut().push_back(DbJob { request_id, payload, diff --git a/packages/sqlite-web-core/src/database.rs b/packages/sqlite-web-core/src/database.rs index 4b3b269..4dacca1 100644 --- a/packages/sqlite-web-core/src/database.rs +++ b/packages/sqlite-web-core/src/database.rs @@ -256,16 +256,19 @@ impl SQLiteDatabase { } let complete_tail = CString::new(state.pending.as_str()) .is_ok_and(|tail| Self::is_trivia_tail_only(tail.as_ptr())); - if state.statement_count == 0 - || state.saw_begin != state.saw_commit - || !complete_tail - { + 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( - "SQL dump ends with an incomplete statement or contains no statements." - .to_string(), - ); + return Err(error.to_string()); } if let Err(err) = self.exec_single_statement("COMMIT").await { self.rollback_if_in_transaction().await; @@ -360,8 +363,8 @@ impl SQLiteDatabase { state.statement_count + 1 ) })?; + state.statement_count += 1; } - state.statement_count += 1; } } consumed += end; @@ -2903,7 +2906,7 @@ mod tests { db.import_sql_action(SqlImportAction::Finish { id }) .await .unwrap(), - "Imported 2 SQL statements." + "Imported 1 SQL statements." ); let rows = db @@ -2922,6 +2925,66 @@ mod tests { )); } + #[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"); diff --git a/packages/sqlite-web/src/db.rs b/packages/sqlite-web/src/db.rs index e4ad281..068042e 100644 --- a/packages/sqlite-web/src/db.rs +++ b/packages/sqlite-web/src/db.rs @@ -345,7 +345,9 @@ impl SQLiteWasmDatabase { } /// Begin an atomic SQL dump import. Append bounded text chunks, then finish - /// to commit. A failure or cancellation rolls the entire import back. + /// 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 diff --git a/svelte-test/benchmarks/sql-dump-import.benchmark.ts b/svelte-test/benchmarks/sql-dump-import.benchmark.ts index c0f116a..f4912a1 100644 --- a/svelte-test/benchmarks/sql-dump-import.benchmark.ts +++ b/svelte-test/benchmarks/sql-dump-import.benchmark.ts @@ -1,46 +1,69 @@ import { describe, expect, it } from "vitest"; import { createTestDatabase } from "../tests/fixtures/test-helpers.js"; -// Manual comparison of equal rows using the old statement-array path and the -// bounded grouped-SQL path. Run alone to avoid concurrent OPFS/CPU contention. +// 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; - await db.query("CREATE TABLE old_import (id INTEGER PRIMARY KEY, value TEXT)"); - await db.query("CREATE TABLE new_import (id INTEGER PRIMARY KEY, value TEXT)"); + expect( + (await db.query("CREATE TABLE benchmark_import (id INTEGER PRIMARY KEY, value TEXT)")).error, + ).toBeUndefined(); - const oldStart = performance.now(); + const singleStart = performance.now(); const statements = Array.from({ length: rowCount }, (_, i) => ({ - sql: `INSERT INTO old_import VALUES (${i}, 'value ${i}')`, + sql: `INSERT INTO benchmark_import VALUES (${i}, 'value ${i}')`, })); - const oldResult = await db.transaction(statements); - const oldMs = performance.now() - oldStart; - expect(oldResult.error).toBeUndefined(); + const singleResult = await db.transaction(statements); + const singleMs = performance.now() - singleStart; + expect(singleResult.error).toBeUndefined(); - const newStart = performance.now(); - const id = (await db.beginSqlDumpImport()).value!; + 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}')`, ); - const sql = `INSERT INTO new_import VALUES ${tuples.join(",")};\n`; - const result = await db.appendSqlDumpChunk(id, sql); - expect(result.error).toBeUndefined(); + 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 newMs = performance.now() - newStart; + const chunkMs = performance.now() - chunkStart; expect(finish.error).toBeUndefined(); + expect(await readData()).toEqual(reference); - const oldData = await db.query( - "SELECT COUNT(*) AS count, SUM(id) AS ids, SUM(LENGTH(value)) AS bytes FROM old_import", - ); - const newData = await db.query( - "SELECT COUNT(*) AS count, SUM(id) AS ids, SUM(LENGTH(value)) AS bytes FROM new_import", + 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`, ); - expect(JSON.parse(oldData.value || "[]")).toEqual(JSON.parse(newData.value || "[]")); - expect(JSON.parse(newData.value || "[]")[0].count).toBe(rowCount); - console.info(`SQL import benchmark: ${rowCount} rows; statement array ${oldMs.toFixed(1)} ms; grouped chunks ${newMs.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",