diff --git a/powersync/src/db/connection.rs b/powersync/src/db/connection.rs index e0f286e..a2d4705 100644 --- a/powersync/src/db/connection.rs +++ b/powersync/src/db/connection.rs @@ -5,7 +5,7 @@ use powersync_sqlite_nostd::{Connection, ManagedConnection, ManagedStmt, ResultC use std::ffi::{CStr, CString, c_int}; use std::mem::MaybeUninit; use std::path::Path; -use std::ptr::null; +use std::ptr::{null, null_mut}; /// The SQLite connection used by the PowerSync Rust SDK. /// @@ -73,6 +73,43 @@ impl SqliteConnection { .into() }) } + + /// Restore SQLite's connection-level invariants before returning a handle + /// to the pool. + /// + /// A stepped statement can own an implicit read transaction while + /// `sqlite3_get_autocommit()` still returns true. Resetting every busy + /// statement is therefore required before the autocommit check. + pub(crate) fn clean_for_pool(&self) -> Result { + use powersync_sqlite_nostd::bindings::{ + sqlite3_next_stmt, sqlite3_reset, sqlite3_stmt_busy, + }; + + let mut reset_statements = 0; + unsafe { + // Safety: pool release has exclusive ownership of the leased + // connection. These APIs inspect or reset statements without + // closing the connection. + let db = self.handle(); + let mut statement = sqlite3_next_stmt(db, null_mut()); + while !statement.is_null() { + let next = sqlite3_next_stmt(db, statement); + if sqlite3_stmt_busy(statement) != 0 { + // sqlite3_reset returns the statement's prior result code, + // not whether the reset itself released the VM. + let _ = sqlite3_reset(statement); + reset_statements += 1; + } + statement = next; + } + + if !db.get_autocommit() { + self.exec(c"ROLLBACK")?; + } + } + + Ok(reset_statements) + } } /// Utility for running a block in a transaction. diff --git a/powersync/src/db/pool.rs b/powersync/src/db/pool.rs index 763c51f..4ab0af1 100644 --- a/powersync/src/db/pool.rs +++ b/powersync/src/db/pool.rs @@ -214,6 +214,17 @@ impl Drop for OwnedConnectionLease { OwnedConnectionLease::Writer { connection, pool } => { // Send update notifications for writes made on this connection while leased. let _ = pool.take_update_notifications(connection); + match connection.clean_for_pool() { + Ok(reset) if reset > 0 => { + log::warn!( + "Reset {reset} busy SQLite statement(s) before releasing writer" + ); + } + Ok(_) => {} + Err(error) => { + log::error!("Could not clean writer before pool release: {error}"); + } + } } OwnedConnectionLease::Reader { connection, pool } => { let connection = std::mem::replace(connection, MaybeUninit::uninit()); @@ -222,6 +233,18 @@ impl Drop for OwnedConnectionLease { connection.assume_init() }; + match connection.clean_for_pool() { + Ok(reset) if reset > 0 => { + log::warn!( + "Reset {reset} busy SQLite statement(s) before releasing reader" + ); + } + Ok(_) => {} + Err(error) => { + log::error!("Could not clean reader before pool release: {error}"); + } + } + pool.state .readers .as_ref() diff --git a/powersync/tests/wal_pool_repro.rs b/powersync/tests/wal_pool_repro.rs new file mode 100644 index 0000000..b9dc64e --- /dev/null +++ b/powersync/tests/wal_pool_repro.rs @@ -0,0 +1,140 @@ +use powersync::{ConnectionPool, env::PowerSyncEnvironment}; +use rusqlite::ffi::{ + SQLITE_OK, SQLITE_ROW, sqlite3_finalize, sqlite3_prepare_v2, sqlite3_step, sqlite3_stmt, + sqlite3_stmt_busy, +}; +use std::{ + ffi::CString, + path::PathBuf, + ptr, + time::{SystemTime, UNIX_EPOCH}, +}; + +fn temp_db() -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock") + .as_nanos(); + std::env::temp_dir().join(format!( + "powersync-pool-wal-repro-{}-{nonce}.db", + std::process::id() + )) +} + +fn checkpoint(connection: &rusqlite::Connection) -> (i64, i64, i64) { + connection + .query_row("PRAGMA wal_checkpoint(PASSIVE)", [], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?)) + }) + .expect("checkpoint") +} + +unsafe fn step_without_reset(connection: &rusqlite::Connection, sql: &str) -> *mut sqlite3_stmt { + let sql = CString::new(sql).expect("SQL has no NUL"); + let mut statement = ptr::null_mut(); + let prepared = unsafe { + sqlite3_prepare_v2( + connection.handle(), + sql.as_ptr(), + -1, + &mut statement, + ptr::null_mut(), + ) + }; + assert_eq!(prepared, SQLITE_OK, "prepare"); + assert_eq!(unsafe { sqlite3_step(statement) }, SQLITE_ROW, "step"); + statement +} + +#[test] +fn returned_reader_must_not_keep_a_busy_statement_or_pin_the_wal() { + PowerSyncEnvironment::powersync_auto_extension().expect("register PowerSync extension"); + let path = temp_db(); + let writer = rusqlite::Connection::open(&path).expect("writer"); + writer + .execute_batch( + "PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0; + CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT); + INSERT INTO items(value) VALUES ('snapshot');", + ) + .expect("fixture"); + let reader = rusqlite::Connection::open(&path).expect("reader"); + let pool = ConnectionPool::wrap_connections(writer, [reader]); + + let reader = pool.reader_sync(); + let statement = unsafe { step_without_reset(&reader, "SELECT * FROM items") }; + assert!( + reader.is_autocommit(), + "a stepped SELECT can pin WAL state while SQLite still reports autocommit" + ); + drop(reader); + + let writer = pool.writer_sync(); + writer + .execute_batch( + "BEGIN IMMEDIATE; + WITH RECURSIVE n(x) AS ( + VALUES(1) UNION ALL SELECT x + 1 FROM n WHERE x < 2000 + ) + INSERT INTO items(value) SELECT printf('value-%04d', x) FROM n; + COMMIT;", + ) + .expect("grow WAL behind snapshot"); + let (_, log_frames, checkpointed_frames) = checkpoint(&writer); + drop(writer); + + let returned_reader = pool.reader_sync(); + let still_busy = unsafe { sqlite3_stmt_busy(statement) } != 0; + unsafe { + sqlite3_finalize(statement); + } + drop(returned_reader); + + let writer = pool.writer_sync(); + let (_, final_log_frames, final_checkpointed_frames) = checkpoint(&writer); + + assert!( + !still_busy, + "pool returned a reader with a stepped statement still busy: \ + log={log_frames}, checkpointed={checkpointed_frames}; \ + after explicit finalize log={final_log_frames}, checkpointed={final_checkpointed_frames}" + ); + assert_eq!( + log_frames, checkpointed_frames, + "a released reader lease must not strand WAL frames" + ); +} + +#[test] +fn released_leases_roll_back_explicit_transactions() { + PowerSyncEnvironment::powersync_auto_extension().expect("register PowerSync extension"); + let path = temp_db(); + let writer = rusqlite::Connection::open(&path).expect("writer"); + writer + .execute_batch( + "PRAGMA journal_mode=WAL; + CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT);", + ) + .expect("fixture"); + let reader = rusqlite::Connection::open(&path).expect("reader"); + let pool = ConnectionPool::wrap_connections(writer, [reader]); + + let reader = pool.reader_sync(); + reader.execute_batch("BEGIN").expect("reader begin"); + assert!(!reader.is_autocommit()); + drop(reader); + assert!( + pool.reader_sync().is_autocommit(), + "reader transaction survived lease release" + ); + + let writer = pool.writer_sync(); + writer.execute_batch("BEGIN").expect("writer begin"); + assert!(!writer.is_autocommit()); + drop(writer); + assert!( + pool.writer_sync().is_autocommit(), + "writer transaction survived lease release" + ); +}