From 1de8ec35ad76f866fe03f4b9edd571b5f672724c Mon Sep 17 00:00:00 2001 From: Daniel Vacic Date: Thu, 30 Jul 2026 02:23:07 +1000 Subject: [PATCH] feat(db): add scoped transaction API --- powersync/src/db/connection.rs | 227 +++++++++++++++++++++++++-- powersync/src/db/connection_tests.rs | 201 ++++++++++++++++++++++++ powersync/src/db/mod.rs | 38 +++++ powersync/src/lib.rs | 2 +- 4 files changed, 455 insertions(+), 13 deletions(-) create mode 100644 powersync/src/db/connection_tests.rs diff --git a/powersync/src/db/connection.rs b/powersync/src/db/connection.rs index e0f286e..7af1025 100644 --- a/powersync/src/db/connection.rs +++ b/powersync/src/db/connection.rs @@ -1,7 +1,9 @@ use crate::error::{PowerSyncError, RawPowerSyncError}; use num_traits::cast::FromPrimitive; use powersync_sqlite_nostd::bindings::sqlite3_open_v2; -use powersync_sqlite_nostd::{Connection, ManagedConnection, ManagedStmt, ResultCode, sqlite3}; +use powersync_sqlite_nostd::{ + Connection, Destructor, ManagedConnection, ManagedStmt, ResultCode, sqlite3, +}; use std::ffi::{CStr, CString, c_int}; use std::mem::MaybeUninit; use std::path::Path; @@ -75,42 +77,239 @@ impl SqliteConnection { } } -/// Utility for running a block in a transaction. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum TransactionMode { + Read, + Write, +} + +/// A prepared statement whose lifetime is confined to a scoped transaction. +/// +/// The raw `ManagedStmt` is deliberately private so a statement or row borrow +/// cannot escape after the transaction and pool lease have ended. +pub struct TransactionStatement { + inner: ManagedStmt, +} + +impl TransactionStatement { + pub fn step(&mut self) -> Result { + self.inner.step().map_err(PowerSyncError::from) + } + + pub fn execute(&mut self) -> Result<(), PowerSyncError> { + while let ResultCode::ROW = self.step()? {} + Ok(()) + } + + pub fn bind_null(&mut self, index: i32) -> Result<(), PowerSyncError> { + self.inner.bind_null(index)?; + Ok(()) + } + + pub fn bind_text(&mut self, index: i32, value: &str) -> Result<(), PowerSyncError> { + self.inner.bind_text(index, value, Destructor::TRANSIENT)?; + Ok(()) + } + + pub fn bind_blob(&mut self, index: i32, value: &[u8]) -> Result<(), PowerSyncError> { + self.inner.bind_blob(index, value, Destructor::TRANSIENT)?; + Ok(()) + } + + pub fn bind_int64(&mut self, index: i32, value: i64) -> Result<(), PowerSyncError> { + self.inner.bind_int64(index, value)?; + Ok(()) + } + + pub fn bind_int(&mut self, index: i32, value: i32) -> Result<(), PowerSyncError> { + self.inner.bind_int(index, value)?; + Ok(()) + } + + pub fn bind_double(&mut self, index: i32, value: f64) -> Result<(), PowerSyncError> { + self.inner.bind_double(index, value)?; + Ok(()) + } + + pub fn column_text(&self, index: i32) -> Result<&str, PowerSyncError> { + self.inner.column_text(index).map_err(PowerSyncError::from) + } + + pub fn column_blob(&self, index: i32) -> Result<&[u8], PowerSyncError> { + self.inner.column_blob(index).map_err(PowerSyncError::from) + } + + pub fn column_int64(&self, index: i32) -> i64 { + self.inner.column_int64(index) + } + + pub fn column_int(&self, index: i32) -> i32 { + self.inner.column_int(index) + } + + pub fn column_double(&self, index: i32) -> f64 { + self.inner.column_double(index) + } +} + +/// A transaction whose commit and rollback lifecycle is owned by the SDK. pub struct TransactionGuard<'a> { - pub inner: &'a mut SqliteConnection, + pub(crate) inner: &'a mut SqliteConnection, active: bool, } impl<'a> TransactionGuard<'a> { - pub fn new(connection: &'a mut SqliteConnection) -> Result { + /// Compatibility constructor for existing SDK internals. New public + /// callers should use `PowerSyncDatabase::{read,write}_transaction`. + pub(crate) fn new(connection: &'a mut SqliteConnection) -> Result { + Self::with_mode(connection, TransactionMode::Read) + } + + pub(crate) fn with_mode( + connection: &'a mut SqliteConnection, + mode: TransactionMode, + ) -> Result { if !unsafe { connection.handle().get_autocommit() } { return Err(PowerSyncError::argument_error( "Connection already in transaction", )); } - connection.exec(c"BEGIN")?; + connection.exec(match mode { + TransactionMode::Read => c"BEGIN", + TransactionMode::Write => c"BEGIN IMMEDIATE", + })?; Ok(TransactionGuard { inner: connection, active: true, }) } - pub fn commit(mut self) -> Result<(), PowerSyncError> { + /// Prepare and use one statement without allowing it to escape the callback. + /// + /// ```compile_fail + /// use powersync::{PowerSyncStatement, PowerSyncTransaction}; + /// use powersync::error::PowerSyncError; + /// + /// fn escape<'a>( + /// transaction: &'a mut PowerSyncTransaction<'a>, + /// ) -> Result<&'a mut PowerSyncStatement, PowerSyncError> { + /// transaction.with_statement("SELECT 1", |statement| Ok(statement)) + /// } + /// ``` + pub fn with_statement( + &mut self, + sql: &str, + operation: impl for<'statement> FnOnce( + &'statement mut TransactionStatement, + ) -> Result, + ) -> Result { + reject_transaction_control(sql)?; + let statement = self.inner.prepare(sql)?; + let mut statement = TransactionStatement { inner: statement }; + operation(&mut statement) + } + + /// Execute one statement inside this transaction. + pub fn execute(&mut self, sql: &str) -> Result<(), PowerSyncError> { + self.with_statement(sql, TransactionStatement::execute) + } + + /// Compatibility exit for existing SDK internals. + pub(crate) fn commit(mut self) -> Result<(), PowerSyncError> { + self.commit_in_place() + } + + fn commit_in_place(&mut self) -> Result<(), PowerSyncError> { + if !self.active { + return Ok(()); + } + self.inner.exec(c"COMMIT")?; self.active = false; - self.inner.exec(c"COMMIT") + Ok(()) } - fn rollback_internal(&mut self) -> Result<(), PowerSyncError> { - self.inner.exec(c"ROLLBACK") + fn rollback(&mut self) -> Result<(), PowerSyncError> { + if !self.active { + return Ok(()); + } + self.inner.exec(c"ROLLBACK")?; + self.active = false; + Ok(()) } } impl Drop for TransactionGuard<'_> { fn drop(&mut self) { - if self.active { - // Rollback if the transaction hasn't explicitly been committed. - let _ = self.rollback_internal(); + if self.active + && let Err(error) = self.rollback() + { + log::error!("Could not roll back scoped PowerSync transaction: {error}"); + } + } +} + +fn reject_transaction_control(sql: &str) -> Result<(), PowerSyncError> { + let Some(keyword) = first_sql_keyword(sql) else { + return Err(PowerSyncError::argument_error( + "Scoped transaction statement is empty", + )); + }; + + if matches!( + keyword.as_str(), + "BEGIN" | "COMMIT" | "END" | "ROLLBACK" | "SAVEPOINT" | "RELEASE" + ) { + return Err(PowerSyncError::argument_error( + "Transaction control is owned by the scoped transaction API", + )); + } + + Ok(()) +} + +fn first_sql_keyword(mut sql: &str) -> Option { + loop { + sql = sql.trim_start_matches(|character: char| { + character.is_ascii_whitespace() || character == '\u{feff}' + }); + if let Some(rest) = sql.strip_prefix("--") { + sql = rest.split_once('\n').map_or("", |(_, remaining)| remaining); + continue; + } + if let Some(rest) = sql.strip_prefix("/*") { + sql = rest.split_once("*/").map_or("", |(_, remaining)| remaining); + continue; + } + break; + } + + let keyword = sql + .bytes() + .take_while(u8::is_ascii_alphabetic) + .collect::>(); + (!keyword.is_empty()).then(|| String::from_utf8_lossy(&keyword).to_ascii_uppercase()) +} + +pub(crate) fn run_transaction( + connection: &mut SqliteConnection, + mode: TransactionMode, + operation: impl FnOnce(&mut TransactionGuard<'_>) -> Result, +) -> Result { + let mut transaction = TransactionGuard::with_mode(connection, mode)?; + match operation(&mut transaction) { + Ok(value) => { + transaction.commit_in_place()?; + Ok(value) + } + Err(original) => { + if let Err(rollback_error) = transaction.rollback() { + log::error!( + "Could not roll back scoped PowerSync transaction after error: \ + {rollback_error}" + ); + } + Err(original) } } } @@ -192,6 +391,10 @@ pub fn exec_stmt(stmt: ManagedStmt) -> Result<(), PowerSyncError> { Ok(()) } +#[cfg(all(test, feature = "rusqlite"))] +#[path = "connection_tests.rs"] +mod tests; + #[cfg(unix)] fn path_to_cstring(p: &Path) -> Result { use std::os::unix::ffi::OsStrExt; diff --git a/powersync/src/db/connection_tests.rs b/powersync/src/db/connection_tests.rs new file mode 100644 index 0000000..9bb27d7 --- /dev/null +++ b/powersync/src/db/connection_tests.rs @@ -0,0 +1,201 @@ +use super::{SqliteConnection, TransactionMode, run_transaction}; +use crate::error::PowerSyncError; +use powersync_sqlite_nostd::Connection; +use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn connection(tag: &str) -> (SqliteConnection, PathBuf) { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock") + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "powersync-transaction-{tag}-{}-{nonce}.db", + std::process::id() + )); + let connection = rusqlite::Connection::open(&path).expect("connection"); + connection + .execute_batch( + "PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0;", + ) + .expect("WAL"); + (SqliteConnection::from(connection), path) +} + +fn autocommit(connection: &SqliteConnection) -> bool { + unsafe { connection.handle().get_autocommit() } +} + +fn assert_wal_reclaimed(connection: &SqliteConnection) { + let result: (i64, i64, i64) = connection + .rusqlite_connection() + .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?)) + }) + .expect("checkpoint"); + assert_eq!( + result, + (0, 0, 0), + "transaction exit must leave no WAL read mark" + ); +} + +fn remove_db(connection: SqliteConnection, path: PathBuf) { + drop(connection); + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(path.with_extension("db-wal")); + let _ = std::fs::remove_file(path.with_extension("db-shm")); +} + +#[test] +fn success_commits_and_error_rolls_back() { + let (mut connection, path) = connection("success-error"); + connection + .exec(c"CREATE TABLE t(id INTEGER)") + .expect("table"); + + run_transaction(&mut connection, TransactionMode::Write, |transaction| { + transaction.execute("INSERT INTO t VALUES (1)") + }) + .expect("commit"); + + let error = run_transaction(&mut connection, TransactionMode::Write, |transaction| { + transaction.execute("INSERT INTO t VALUES (2)")?; + Err::<(), _>(PowerSyncError::argument_error("expected operation error")) + }) + .expect_err("operation error"); + assert!(error.to_string().contains("expected operation error")); + assert!(autocommit(&connection)); + + let count: i64 = connection + .rusqlite_connection() + .query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)) + .expect("count"); + assert_eq!(count, 1); + assert_wal_reclaimed(&connection); + remove_db(connection, path); +} + +#[test] +fn panic_rolls_back_on_unwind() { + let (mut connection, path) = connection("panic"); + connection + .exec(c"CREATE TABLE t(id INTEGER)") + .expect("table"); + + let panic = catch_unwind(AssertUnwindSafe(|| { + let _ = run_transaction( + &mut connection, + TransactionMode::Write, + |transaction| -> Result<(), PowerSyncError> { + transaction.execute("INSERT INTO t VALUES (1)")?; + panic!("fixture panic"); + }, + ); + })); + assert!(panic.is_err()); + assert!(autocommit(&connection)); + + let count: i64 = connection + .rusqlite_connection() + .query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)) + .expect("count"); + assert_eq!(count, 0); + assert_wal_reclaimed(&connection); + remove_db(connection, path); +} + +#[test] +fn commit_failure_rolls_back_and_releases_the_wal() { + let (mut connection, path) = connection("commit-failure"); + connection + .rusqlite_connection() + .execute_batch( + "PRAGMA foreign_keys=ON; + CREATE TABLE parent(id INTEGER PRIMARY KEY); + CREATE TABLE child( + parent_id INTEGER, + FOREIGN KEY(parent_id) REFERENCES parent(id) + DEFERRABLE INITIALLY DEFERRED + );", + ) + .expect("schema"); + + run_transaction(&mut connection, TransactionMode::Write, |transaction| { + transaction.execute("INSERT INTO child(parent_id) VALUES (999)") + }) + .expect_err("deferred foreign-key violation must fail at commit"); + + assert!( + autocommit(&connection), + "failed COMMIT must still roll back" + ); + let count: i64 = connection + .rusqlite_connection() + .query_row("SELECT COUNT(*) FROM child", [], |row| row.get(0)) + .expect("count"); + assert_eq!(count, 0); + assert_wal_reclaimed(&connection); + remove_db(connection, path); +} + +#[test] +fn transaction_control_is_owned_by_the_scope() { + let (mut connection, path) = connection("owned-exit"); + connection + .exec(c"CREATE TABLE t(id INTEGER)") + .expect("table"); + + for sql in [ + "BEGIN", + " commit", + "-- deliberately obscured\nROLLBACK", + "/* deliberately obscured */ SAVEPOINT nested", + "\u{feff}END", + "RELEASE nested", + ] { + let error = run_transaction(&mut connection, TransactionMode::Write, |transaction| { + transaction.execute("INSERT INTO t VALUES (1)")?; + transaction.execute(sql) + }) + .expect_err("transaction control must be rejected"); + assert!( + error.to_string().contains("Transaction control is owned"), + "{sql}: {error}" + ); + } + + let count: i64 = connection + .rusqlite_connection() + .query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)) + .expect("count"); + assert_eq!(count, 0); + assert_wal_reclaimed(&connection); + remove_db(connection, path); +} + +#[test] +fn read_mode_releases_a_real_snapshot() { + let (mut reader, path) = connection("read-snapshot"); + reader.exec(c"CREATE TABLE t(id INTEGER)").expect("table"); + reader.exec(c"INSERT INTO t VALUES (1)").expect("seed"); + let writer = rusqlite::Connection::open(&path).expect("second writer"); + + run_transaction(&mut reader, TransactionMode::Read, |transaction| { + transaction.with_statement("SELECT id FROM t", |statement| { + assert_eq!(statement.step()?, powersync_sqlite_nostd::ResultCode::ROW); + writer + .execute("INSERT INTO t VALUES (2)", []) + .map_err(PowerSyncError::from)?; + Ok(()) + }) + }) + .expect("read transaction"); + + assert!(autocommit(&reader)); + assert_wal_reclaimed(&reader); + drop(writer); + remove_db(reader, path); +} diff --git a/powersync/src/db/mod.rs b/powersync/src/db/mod.rs index bae02a9..3026110 100644 --- a/powersync/src/db/mod.rs +++ b/powersync/src/db/mod.rs @@ -29,6 +29,10 @@ pub mod schema; pub mod streams; pub mod watch; +pub use connection::{ + TransactionGuard as PowerSyncTransaction, TransactionStatement as PowerSyncStatement, +}; + #[derive(Clone)] pub struct PowerSyncDatabase { sync: Arc, @@ -315,6 +319,40 @@ impl PowerSyncDatabase { self.inner.writer().await } + /// Run a read transaction whose connection and statements cannot escape + /// the synchronous callback. + /// + /// Lease acquisition is asynchronous, but the callback cannot suspend + /// while SQLite holds a transaction or WAL read mark. + pub async fn read_transaction( + &self, + operation: impl FnOnce(&mut PowerSyncTransaction<'_>) -> Result, + ) -> Result { + let mut connection = self.inner.reader().await?; + connection::run_transaction( + connection.sqlite_connection_mut(), + connection::TransactionMode::Read, + operation, + ) + } + + /// Run a write transaction whose connection and statements cannot escape + /// the synchronous callback. + /// + /// Write transactions use `BEGIN IMMEDIATE` so the writer reservation is + /// acquired before the callback starts. + pub async fn write_transaction( + &self, + operation: impl FnOnce(&mut PowerSyncTransaction<'_>) -> Result, + ) -> Result { + let mut connection = self.inner.writer().await?; + connection::run_transaction( + connection.sqlite_connection_mut(), + connection::TransactionMode::Write, + operation, + ) + } + /* /// Returns the shared [InnerPowerSyncState] backing this database. /// diff --git a/powersync/src/lib.rs b/powersync/src/lib.rs index 2a68d1b..a76e6d1 100644 --- a/powersync/src/lib.rs +++ b/powersync/src/lib.rs @@ -3,7 +3,6 @@ pub mod env; mod sync; mod util; -pub use db::PowerSyncDatabase; pub use db::crud::{CrudEntry, CrudTransaction, UpdateType}; #[cfg(feature = "ffi")] pub use db::internal::InnerPowerSyncState; @@ -11,6 +10,7 @@ pub use db::pool::{ConnectionPool, LeasedConnection}; pub use db::streams::StreamSubscription; pub use db::streams::StreamSubscriptionOptions; pub use db::streams::SyncStream; +pub use db::{PowerSyncDatabase, PowerSyncStatement, PowerSyncTransaction}; pub use sync::connector::{BackendConnector, PowerSyncCredentials}; pub use sync::options::SyncOptions; pub use sync::status::SyncStatusData;