From cbaa025b412effcb25ff595c86a99cded15f6ef7 Mon Sep 17 00:00:00 2001 From: Sree Narayanan Date: Wed, 28 Oct 2026 01:03:23 +0400 Subject: [PATCH 1/2] fix(sqlite-embedded): close fts5 connections without double finalize --- engine/packages/sqlite-embedded/Cargo.toml | 3 + engine/packages/sqlite-embedded/src/query.rs | 363 ++++++++---------- .../sqlite-embedded/src/vfs/transport.rs | 36 +- .../sqlite-embedded/tests/fts5_close.rs | 132 +++++++ 4 files changed, 297 insertions(+), 237 deletions(-) create mode 100644 engine/packages/sqlite-embedded/tests/fts5_close.rs diff --git a/engine/packages/sqlite-embedded/Cargo.toml b/engine/packages/sqlite-embedded/Cargo.toml index ad93860c98..843c1444f7 100644 --- a/engine/packages/sqlite-embedded/Cargo.toml +++ b/engine/packages/sqlite-embedded/Cargo.toml @@ -32,3 +32,6 @@ sqlite_types.workspace = true lru = "0.18.5" parking_lot.workspace = true + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/engine/packages/sqlite-embedded/src/query.rs b/engine/packages/sqlite-embedded/src/query.rs index 849e867e0b..2568ae71f7 100644 --- a/engine/packages/sqlite-embedded/src/query.rs +++ b/engine/packages/sqlite-embedded/src/query.rs @@ -163,6 +163,7 @@ pub fn execute_statement( let c_sql = CString::new(sql).map_err(|err| anyhow!(err.to_string()))?; let mut stmt = ptr::null_mut(); let rc = unsafe { sqlite3_prepare_v2(db, c_sql.as_ptr(), -1, &mut stmt, ptr::null_mut()) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error(db, 0, "failed to prepare sqlite statement")); } @@ -170,32 +171,24 @@ pub fn execute_statement( return Ok(ExecResult { changes: 0 }); } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - if let Some(params) = params { - bind_params(db, stmt, params)?; - } + refuse_oversize_vacuum(db, stmt)?; + if let Some(params) = params { + bind_params(db, stmt, params)?; + } - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error(db, 0, "failed to execute sqlite statement")); - } + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error(db, 0, "failed to execute sqlite statement")); } - - Ok(ExecResult { - changes: unsafe { sqlite3_changes(db) as i64 }, - }) - })(); - - unsafe { - sqlite3_finalize(stmt); } - result + Ok(ExecResult { + changes: unsafe { sqlite3_changes(db) as i64 }, + }) } pub fn query_statement( @@ -206,6 +199,7 @@ pub fn query_statement( let c_sql = CString::new(sql).map_err(|err| anyhow!(err.to_string()))?; let mut stmt = ptr::null_mut(); let rc = unsafe { sqlite3_prepare_v2(db, c_sql.as_ptr(), -1, &mut stmt, ptr::null_mut()) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error(db, 0, "failed to prepare sqlite query")); } @@ -216,39 +210,31 @@ pub fn query_statement( }); } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - if let Some(params) = params { - bind_params(db, stmt, params)?; - } - - let columns = collect_columns(stmt); - let mut rows = Vec::new(); + refuse_oversize_vacuum(db, stmt)?; + if let Some(params) = params { + bind_params(db, stmt, params)?; + } - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error(db, 0, "failed to step sqlite query")); - } + let columns = collect_columns(stmt); + let mut rows = Vec::new(); - let mut row = Vec::with_capacity(columns.len()); - for index in 0..columns.len() { - row.push(column_value(stmt, index as i32)); - } - rows.push(row); + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error(db, 0, "failed to step sqlite query")); } - Ok(QueryResult { columns, rows }) - })(); - - unsafe { - sqlite3_finalize(stmt); + let mut row = Vec::with_capacity(columns.len()); + for index in 0..columns.len() { + row.push(column_value(stmt, index as i32)); + } + rows.push(row); } - result + Ok(QueryResult { columns, rows }) } /// Runs one statement and collects its rows. With `max_result_bytes`, the statement stops with @@ -263,6 +249,7 @@ pub fn execute_single_statement( let mut stmt = ptr::null_mut(); let mut tail = ptr::null(); let rc = unsafe { sqlite3_prepare_v2(db, c_sql.as_ptr(), -1, &mut stmt, &mut tail) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error( db, @@ -271,11 +258,6 @@ pub fn execute_single_statement( )); } if has_non_whitespace_tail(tail) { - if !stmt.is_null() { - unsafe { - sqlite3_finalize(stmt); - } - } return Err(SqliteStatementError { code: -1, statement_index: 0, @@ -292,45 +274,37 @@ pub fn execute_single_statement( }); } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - if let Some(params) = params { - bind_params(db, stmt, params)?; - } - - let columns = collect_columns(stmt); - let mut rows = Vec::new(); - let mut budget = ResultBudget::new(max_result_bytes); - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error( - db, - 0, - "failed to step sqlite execute statement", - )); - } + refuse_oversize_vacuum(db, stmt)?; + if let Some(params) = params { + bind_params(db, stmt, params)?; + } - rows.push(read_row(stmt, columns.len(), &mut budget)?); + let columns = collect_columns(stmt); + let mut rows = Vec::new(); + let mut budget = ResultBudget::new(max_result_bytes); + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error( + db, + 0, + "failed to step sqlite execute statement", + )); } - let changes = unsafe { sqlite3_changes(db) as i64 }; - Ok(ExecuteResult { - columns, - rows, - changes, - last_insert_row_id: (changes > 0).then(|| unsafe { sqlite3_last_insert_rowid(db) }), - }) - })(); - - unsafe { - sqlite3_finalize(stmt); + rows.push(read_row(stmt, columns.len(), &mut budget)?); } - result + let changes = unsafe { sqlite3_changes(db) as i64 }; + Ok(ExecuteResult { + columns, + rows, + changes, + last_insert_row_id: (changes > 0).then(|| unsafe { sqlite3_last_insert_rowid(db) }), + }) } /// Like [`execute_single_statement`], also returning the statement's profile. @@ -344,6 +318,7 @@ pub fn execute_single_statement_profiled( let mut stmt = ptr::null_mut(); let mut tail = ptr::null(); let rc = unsafe { sqlite3_prepare_v2(db, c_sql.as_ptr(), -1, &mut stmt, &mut tail) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error( db, @@ -352,11 +327,6 @@ pub fn execute_single_statement_profiled( )); } if has_non_whitespace_tail(tail) { - if !stmt.is_null() { - unsafe { - sqlite3_finalize(stmt); - } - } return Err(SqliteStatementError { code: -1, statement_index: 0, @@ -376,56 +346,48 @@ pub fn execute_single_statement_profiled( )); } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - let mut profile = SqliteQueryProfile::default(); - if let Some(params) = params { - bind_params_profiled(db, stmt, params, &mut profile)?; - } - - let columns = collect_columns(stmt); - profile.result_columns = columns.len() as u64; - let mut rows = Vec::new(); - let mut budget = ResultBudget::new(max_result_bytes); - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error( - db, - 0, - "failed to step sqlite execute statement", - )); - } + refuse_oversize_vacuum(db, stmt)?; + let mut profile = SqliteQueryProfile::default(); + if let Some(params) = params { + bind_params_profiled(db, stmt, params, &mut profile)?; + } - rows.push(read_row_profiled( - stmt, - columns.len(), - &mut budget, - &mut profile, - )?); - profile.result_rows = profile.result_rows.saturating_add(1); + let columns = collect_columns(stmt); + profile.result_columns = columns.len() as u64; + let mut rows = Vec::new(); + let mut budget = ResultBudget::new(max_result_bytes); + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error( + db, + 0, + "failed to step sqlite execute statement", + )); } - let changes = unsafe { sqlite3_changes(db) as i64 }; - Ok(( - ExecuteResult { - columns, - rows, - changes, - last_insert_row_id: (changes > 0).then(|| unsafe { sqlite3_last_insert_rowid(db) }), - }, - profile, - )) - })(); - - unsafe { - sqlite3_finalize(stmt); + rows.push(read_row_profiled( + stmt, + columns.len(), + &mut budget, + &mut profile, + )?); + profile.result_rows = profile.result_rows.saturating_add(1); } - result + let changes = unsafe { sqlite3_changes(db) as i64 }; + Ok(( + ExecuteResult { + columns, + rows, + changes, + last_insert_row_id: (changes > 0).then(|| unsafe { sqlite3_last_insert_rowid(db) }), + }, + profile, + )) } /// Runs every statement in `sql` and returns the rows of the last one that has columns or @@ -447,6 +409,7 @@ pub fn exec_statements( let mut stmt = ptr::null_mut(); let mut tail = ptr::null(); let rc = unsafe { sqlite3_prepare_v2(db, remaining, -1, &mut stmt, &mut tail) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error( db, @@ -472,35 +435,26 @@ pub fn exec_statements( }; } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - let columns = collect_columns(stmt); - let mut rows = Vec::new(); - let mut budget = ResultBudget::new(max_result_bytes); - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error( - db, - statement_index, - "failed to step sqlite exec statement", - )); - } - - rows.push(read_row(stmt, columns.len(), &mut budget)?); + refuse_oversize_vacuum(db, stmt)?; + let columns = collect_columns(stmt); + let mut rows = Vec::new(); + let mut budget = ResultBudget::new(max_result_bytes); + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error( + db, + statement_index, + "failed to step sqlite exec statement", + )); } - Ok((columns, rows)) - })(); - - unsafe { - sqlite3_finalize(stmt); + rows.push(read_row(stmt, columns.len(), &mut budget)?); } - let (columns, rows) = result?; if !columns.is_empty() || !rows.is_empty() { final_result = QueryResult { columns, rows }; } @@ -534,6 +488,7 @@ pub fn exec_statements_profiled( let mut stmt = ptr::null_mut(); let mut tail = ptr::null(); let rc = unsafe { sqlite3_prepare_v2(db, remaining, -1, &mut stmt, &mut tail) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK { return Err(sqlite_error( db, @@ -559,45 +514,36 @@ pub fn exec_statements_profiled( }; } - let result = (|| { - refuse_oversize_vacuum(db, stmt)?; - let columns = collect_columns(stmt); - let mut profile = SqliteQueryProfile { - result_columns: columns.len() as u64, - ..Default::default() - }; - let mut rows = Vec::new(); - let mut budget = ResultBudget::new(max_result_bytes); - loop { - let step_rc = unsafe { sqlite3_step(stmt) }; - if step_rc == SQLITE_DONE { - break; - } - if step_rc != SQLITE_ROW { - return Err(sqlite_error( - db, - statement_index, - "failed to step sqlite exec statement", - )); - } - - rows.push(read_row_profiled( - stmt, - columns.len(), - &mut budget, - &mut profile, - )?); - profile.result_rows = profile.result_rows.saturating_add(1); + refuse_oversize_vacuum(db, stmt)?; + let columns = collect_columns(stmt); + let mut profile = SqliteQueryProfile { + result_columns: columns.len() as u64, + ..Default::default() + }; + let mut rows = Vec::new(); + let mut budget = ResultBudget::new(max_result_bytes); + loop { + let step_rc = unsafe { sqlite3_step(stmt) }; + if step_rc == SQLITE_DONE { + break; + } + if step_rc != SQLITE_ROW { + return Err(sqlite_error( + db, + statement_index, + "failed to step sqlite exec statement", + )); } - Ok((columns, rows, profile)) - })(); - - unsafe { - sqlite3_finalize(stmt); + rows.push(read_row_profiled( + stmt, + columns.len(), + &mut budget, + &mut profile, + )?); + profile.result_rows = profile.result_rows.saturating_add(1); } - let (columns, rows, profile) = result?; if !columns.is_empty() || !rows.is_empty() { final_result = QueryResult { columns, rows }; final_profile = profile; @@ -897,18 +843,29 @@ fn sql_words(sql: &str) -> Vec<&str> { fn pragma_i64(db: *mut sqlite3, sql: &CStr) -> Option { let mut stmt = ptr::null_mut(); let rc = unsafe { sqlite3_prepare_v2(db, sql.as_ptr(), -1, &mut stmt, ptr::null_mut()) }; + let _stmt = FinalizeOnDrop(stmt); if rc != SQLITE_OK || stmt.is_null() { return None; } - let value = if unsafe { sqlite3_step(stmt) } == SQLITE_ROW { + if unsafe { sqlite3_step(stmt) } == SQLITE_ROW { Some(unsafe { sqlite3_column_int64(stmt, 0) }) } else { None - }; - unsafe { - sqlite3_finalize(stmt); } - value +} + +/// Finalizes a prepared statement when its owner leaves scope, including by an error return or +/// a panic, so no statement outlives the call that prepared it. +struct FinalizeOnDrop(*mut sqlite3_stmt); + +impl Drop for FinalizeOnDrop { + fn drop(&mut self) { + // SAFETY: the statement was prepared by the call that owns this guard and is used by no + // one else. Finalizing a null statement is a no-op. + unsafe { + sqlite3_finalize(self.0); + } + } } fn has_non_whitespace_tail(tail: *const c_char) -> bool { diff --git a/engine/packages/sqlite-embedded/src/vfs/transport.rs b/engine/packages/sqlite-embedded/src/vfs/transport.rs index e6a06bf1e8..a62b024a2d 100644 --- a/engine/packages/sqlite-embedded/src/vfs/transport.rs +++ b/engine/packages/sqlite-embedded/src/vfs/transport.rs @@ -4056,18 +4056,9 @@ impl Drop for NativeDatabase { // Every committed transaction is now in Depot or discarded. SQLite's rollback of an // open transaction during close only restores that state, so nothing it writes is - // sent. Statements still prepared are finalized first, because `sqlite3_close_v2` - // would otherwise defer the close and that rollback until they are finalized, which - // never happens once the connection's owner is gone. + // sent. Every statement the connection's callers prepare is finalized before their + // call returns, so the close is never deferred by a statement nobody owns. ctx.state.write().closing = true; - let finalized_statements = finalize_outstanding_statements(self.db); - if finalized_statements > 0 { - tracing::warn!( - actor_id = %ctx.actor_id, - finalized_statements, - "finalized statements left prepared on a closing sqlite database" - ); - } let rc = unsafe { sqlite3_close_v2(self.db) }; if rc != SQLITE_OK { tracing::warn!( @@ -4081,29 +4072,6 @@ impl Drop for NativeDatabase { } } -/// Finalizes every statement still prepared on a connection whose owner is dropping it, and -/// returns how many there were. Every statement on the connection belongs to a caller that -/// borrowed the connection, so none can be used once the connection is gone. A statement -/// left prepared, for example by a panic between prepare and finalize, would otherwise keep -/// `sqlite3_close_v2` from closing the connection until it is finalized, deferring the -/// rollback of an open transaction and the release of the connection indefinitely. -fn finalize_outstanding_statements(db: *mut sqlite3) -> usize { - let mut finalized = 0; - loop { - // SAFETY: `db` is a live connection owned by the caller, and the statement it returns - // has no other user because the connection's owner is dropping it. - let stmt = unsafe { sqlite3_next_stmt(db, ptr::null_mut()) }; - if stmt.is_null() { - break; - } - unsafe { - sqlite3_finalize(stmt); - } - finalized += 1; - } - finalized -} - /// Whether the connection holds a write transaction SQLite has not committed: an explicit /// transaction after `BEGIN`, or a statement whose implicit transaction has not finished. fn write_transaction_open(db: *mut sqlite3) -> bool { diff --git a/engine/packages/sqlite-embedded/tests/fts5_close.rs b/engine/packages/sqlite-embedded/tests/fts5_close.rs new file mode 100644 index 0000000000..abe1aef830 --- /dev/null +++ b/engine/packages/sqlite-embedded/tests/fts5_close.rs @@ -0,0 +1,132 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use anyhow::{Result, bail}; +use async_trait::async_trait; +use parking_lot::Mutex; +use rivet_envoy_protocol as protocol; +use sqlite_embedded::database::open_database_from_transport; +use sqlite_embedded::query::ColumnValue; +use sqlite_embedded::vfs::SqliteTransport; +use tokio::runtime::Handle; + +/// Keeps committed pages in memory so a closed database can be reopened from the same storage. +#[derive(Default)] +struct MemoryTransport { + state: Mutex, +} + +#[derive(Default)] +struct MemoryState { + pages: BTreeMap, + head_txid: u64, +} + +#[async_trait] +impl SqliteTransport for MemoryTransport { + async fn get_pages( + &self, + request: protocol::SqliteGetPagesRequest, + ) -> Result { + let state = self.state.lock(); + Ok(protocol::SqliteGetPagesResponse::SqliteGetPagesOk( + protocol::SqliteGetPagesOk { + pages: request + .pgnos + .into_iter() + .map(|pgno| protocol::SqliteFetchedPage { + pgno, + bytes: state.pages.get(&pgno).cloned(), + }) + .collect(), + head_txid: Some(state.head_txid), + }, + )) + } + + async fn commit( + &self, + request: protocol::SqliteCommitRequest, + ) -> Result { + let mut state = self.state.lock(); + for page in request.dirty_pages { + state.pages.insert(page.pgno, page.bytes); + } + state.pages.retain(|pgno, _| *pgno <= request.db_size_pages); + state.head_txid += 1; + Ok(protocol::SqliteCommitResponse::SqliteCommitOk( + protocol::SqliteCommitOk { + head_txid: Some(state.head_txid), + }, + )) + } + + async fn commit_stage_begin( + &self, + _request: protocol::SqliteCommitStageBeginRequest, + ) -> Result { + bail!("memory transport does not stage commits") + } + + async fn commit_stage_segment( + &self, + _request: protocol::SqliteCommitStageSegmentRequest, + ) -> Result { + bail!("memory transport does not stage commits") + } + + async fn commit_finalize( + &self, + _request: protocol::SqliteCommitFinalizeRequest, + ) -> Result { + bail!("memory transport does not stage commits") + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn database_with_fts5_table_closes_and_reopens_with_its_data() -> Result<()> { + let transport = Arc::new(MemoryTransport::default()); + let actor_id = "fts5-close-actor".to_string(); + + let db = open_database_from_transport( + transport.clone(), + actor_id.clone(), + 1, + Handle::current(), + None, + ) + .await?; + db.exec( + "CREATE TABLE messages (sequence INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT NOT NULL); + CREATE VIRTUAL TABLE search USING fts5(text, content='messages', content_rowid='sequence'); + CREATE TRIGGER search_insert AFTER INSERT ON messages BEGIN + INSERT INTO search(rowid, text) VALUES (new.sequence, new.text); + END;" + .to_string(), + ) + .await?; + db.exec( + "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 3000) + INSERT INTO messages (text) SELECT 'message number ' || i FROM n;" + .to_string(), + ) + .await?; + db.query( + "SELECT count(*) FROM search WHERE search MATCH 'message'".to_string(), + None, + ) + .await?; + db.close_and_wait().await?; + drop(db); + + let db = open_database_from_transport(transport, actor_id, 2, Handle::current(), None).await?; + let result = db + .query( + "SELECT count(*) FROM search WHERE search MATCH 'message'".to_string(), + None, + ) + .await?; + assert_eq!(result.rows, vec![vec![ColumnValue::Integer(3000)]]); + db.close_and_wait().await?; + Ok(()) +} From 2d98fddcae52e289b2a830c05590b59044149205 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Wed, 28 Oct 2026 03:28:07 -0700 Subject: [PATCH 2/2] update --- engine/packages/sqlite-embedded/Cargo.toml | 3 - .../sqlite-embedded/tests/fts5_close.rs | 132 ------------------ 2 files changed, 135 deletions(-) delete mode 100644 engine/packages/sqlite-embedded/tests/fts5_close.rs diff --git a/engine/packages/sqlite-embedded/Cargo.toml b/engine/packages/sqlite-embedded/Cargo.toml index 843c1444f7..ad93860c98 100644 --- a/engine/packages/sqlite-embedded/Cargo.toml +++ b/engine/packages/sqlite-embedded/Cargo.toml @@ -32,6 +32,3 @@ sqlite_types.workspace = true lru = "0.18.5" parking_lot.workspace = true - -[dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/engine/packages/sqlite-embedded/tests/fts5_close.rs b/engine/packages/sqlite-embedded/tests/fts5_close.rs deleted file mode 100644 index abe1aef830..0000000000 --- a/engine/packages/sqlite-embedded/tests/fts5_close.rs +++ /dev/null @@ -1,132 +0,0 @@ -use std::collections::BTreeMap; -use std::sync::Arc; - -use anyhow::{Result, bail}; -use async_trait::async_trait; -use parking_lot::Mutex; -use rivet_envoy_protocol as protocol; -use sqlite_embedded::database::open_database_from_transport; -use sqlite_embedded::query::ColumnValue; -use sqlite_embedded::vfs::SqliteTransport; -use tokio::runtime::Handle; - -/// Keeps committed pages in memory so a closed database can be reopened from the same storage. -#[derive(Default)] -struct MemoryTransport { - state: Mutex, -} - -#[derive(Default)] -struct MemoryState { - pages: BTreeMap, - head_txid: u64, -} - -#[async_trait] -impl SqliteTransport for MemoryTransport { - async fn get_pages( - &self, - request: protocol::SqliteGetPagesRequest, - ) -> Result { - let state = self.state.lock(); - Ok(protocol::SqliteGetPagesResponse::SqliteGetPagesOk( - protocol::SqliteGetPagesOk { - pages: request - .pgnos - .into_iter() - .map(|pgno| protocol::SqliteFetchedPage { - pgno, - bytes: state.pages.get(&pgno).cloned(), - }) - .collect(), - head_txid: Some(state.head_txid), - }, - )) - } - - async fn commit( - &self, - request: protocol::SqliteCommitRequest, - ) -> Result { - let mut state = self.state.lock(); - for page in request.dirty_pages { - state.pages.insert(page.pgno, page.bytes); - } - state.pages.retain(|pgno, _| *pgno <= request.db_size_pages); - state.head_txid += 1; - Ok(protocol::SqliteCommitResponse::SqliteCommitOk( - protocol::SqliteCommitOk { - head_txid: Some(state.head_txid), - }, - )) - } - - async fn commit_stage_begin( - &self, - _request: protocol::SqliteCommitStageBeginRequest, - ) -> Result { - bail!("memory transport does not stage commits") - } - - async fn commit_stage_segment( - &self, - _request: protocol::SqliteCommitStageSegmentRequest, - ) -> Result { - bail!("memory transport does not stage commits") - } - - async fn commit_finalize( - &self, - _request: protocol::SqliteCommitFinalizeRequest, - ) -> Result { - bail!("memory transport does not stage commits") - } -} - -#[tokio::test(flavor = "multi_thread")] -async fn database_with_fts5_table_closes_and_reopens_with_its_data() -> Result<()> { - let transport = Arc::new(MemoryTransport::default()); - let actor_id = "fts5-close-actor".to_string(); - - let db = open_database_from_transport( - transport.clone(), - actor_id.clone(), - 1, - Handle::current(), - None, - ) - .await?; - db.exec( - "CREATE TABLE messages (sequence INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT NOT NULL); - CREATE VIRTUAL TABLE search USING fts5(text, content='messages', content_rowid='sequence'); - CREATE TRIGGER search_insert AFTER INSERT ON messages BEGIN - INSERT INTO search(rowid, text) VALUES (new.sequence, new.text); - END;" - .to_string(), - ) - .await?; - db.exec( - "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 3000) - INSERT INTO messages (text) SELECT 'message number ' || i FROM n;" - .to_string(), - ) - .await?; - db.query( - "SELECT count(*) FROM search WHERE search MATCH 'message'".to_string(), - None, - ) - .await?; - db.close_and_wait().await?; - drop(db); - - let db = open_database_from_transport(transport, actor_id, 2, Handle::current(), None).await?; - let result = db - .query( - "SELECT count(*) FROM search WHERE search MATCH 'message'".to_string(), - None, - ) - .await?; - assert_eq!(result.rows, vec![vec![ColumnValue::Integer(3000)]]); - db.close_and_wait().await?; - Ok(()) -}