From 931b72c9a132dd6f18bad75dbd274079ebf15d88 Mon Sep 17 00:00:00 2001 From: Svetlin Ralchev Date: Sat, 26 Sep 2026 13:08:39 +0400 Subject: [PATCH] feat(do)!: support sqlx's begin(), commit() and rollback() `sql.exec` rejects `BEGIN`, and `Storage::transaction` wraps a callback, so `begin()` parks one open: it opens `Storage::transaction` in a task of its own with a callback that only waits for a commit or rollback signal, while the statements run on the connection through `&mut *tx` as with any sqlx driver. A spike measured that this works: statements outside the callback are part of the transaction, the request can await timers and sub-requests while it is open, and other requests are held off until it ends. - `commit()` / `rollback()` signal the parked transaction and wait for it. - A dropped `Transaction` signals the rollback without waiting (sqlx's `start_rollback` is synchronous), and every query on the connection first waits for it to land -- a write right after the drop was measured to be rolled back with it otherwise. - sqlx drops a guard, calling `start_rollback`, when `begin()` fails; a refused nested `begin()` records a phantom level so that drop cannot roll back the outer transaction. - Nested `begin()` (savepoints), a custom `BEGIN` statement, and batches or `transaction()` while one is open are refused rather than deadlocking. - `DoConnection::transaction(callback)` stays, as the form that cannot be left open; its docs note that sqlx's own `Connection::transaction` wins method resolution on a `&mut DoConnection`, and works too now. Worker scenarios cover commit, rollback, drop-then-write, read-your-writes across an await, the nested refusal keeping the outer's writes, the refusals while open, and a second request held off while a `begin()` transaction is open. BREAKING CHANGE: `DoConnection::into_storage` returns `Rc` instead of `worker::Storage`: the parked transaction needs its own handle to the storage, which cannot be cloned. Method calls still work through auto-deref; only code naming the type changes. --- Cargo.lock | 1 + Cargo.toml | 1 + crates/sqlx-cloudflare-do/Cargo.toml | 5 +- crates/sqlx-cloudflare-do/README.md | 37 ++- crates/sqlx-cloudflare-do/src/connection.rs | 38 ++- crates/sqlx-cloudflare-do/src/executor.rs | 14 +- crates/sqlx-cloudflare-do/src/lib.rs | 12 +- crates/sqlx-cloudflare-do/src/transaction.rs | 305 ++++++++++++++++-- crates/sqlx-cloudflare-do/tests/facade.rs | 10 + .../tests/worker/Cargo.lock | 3 +- .../tests/worker/src/lib.rs | 248 ++++++++++++-- 11 files changed, 583 insertions(+), 91 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index beb9d60..a379dbc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1011,6 +1011,7 @@ dependencies = [ name = "sqlx-cloudflare-do" version = "0.2.0" dependencies = [ + "futures-channel", "futures-core", "futures-util", "js-sys", diff --git a/Cargo.toml b/Cargo.toml index e6b3806..400d1e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,6 +48,7 @@ wasm-bindgen = "0.2" js-sys = "0.3" futures-core = { version = "0.3", default-features = false } futures-util = { version = "0.3", default-features = false } +futures-channel = { version = "0.3", default-features = false } log = { version = "0.4", default-features = false } ureq = { version = "3", default-features = false } serde_json = "1" diff --git a/crates/sqlx-cloudflare-do/Cargo.toml b/crates/sqlx-cloudflare-do/Cargo.toml index 0638a6c..ba6a748 100644 --- a/crates/sqlx-cloudflare-do/Cargo.toml +++ b/crates/sqlx-cloudflare-do/Cargo.toml @@ -29,7 +29,10 @@ worker = { workspace = true } wasm-bindgen = { workspace = true } js-sys = { workspace = true } futures-core = { workspace = true } -futures-util = { workspace = true, features = ["alloc"] } +# `std` for `Shared`: every query waits on a dropped transaction's rollback. +futures-util = { workspace = true, features = ["alloc", "std"] } +# One-shot signals to the transaction `begin()` parks open. +futures-channel = { workspace = true, features = ["alloc"] } # Only for `log::LevelFilter`, which `ConnectOptions` names in its signatures. log = { workspace = true } diff --git a/crates/sqlx-cloudflare-do/README.md b/crates/sqlx-cloudflare-do/README.md index 82955fe..a9b474c 100644 --- a/crates/sqlx-cloudflare-do/README.md +++ b/crates/sqlx-cloudflare-do/README.md @@ -86,9 +86,24 @@ two share one definition. ## Transactions -`sql.exec` rejects `BEGIN`, so a transaction is a callback. It commits when -the callback returns `Ok` and rolls back every write when it returns `Err`, -and inside it you read your own writes: +Real transactions -- they read their own writes and roll back cleanly -- +two ways. sqlx's own `begin()` works as with any driver: + +```rust +use sqlx::Connection; + +let mut tx = conn.begin().await?; +sqlx::query("UPDATE accounts SET balance = balance - 10 WHERE id = 1") + .execute(&mut *tx) + .await?; +sqlx::query("UPDATE accounts SET balance = balance + 10 WHERE id = 2") + .execute(&mut *tx) + .await?; +tx.commit().await?; // dropping `tx` instead rolls both back +``` + +Or as a callback, which commits on `Ok`, rolls back on `Err`, and cannot be +left open by mistake: ```rust let id = conn @@ -106,12 +121,16 @@ let id = conn .await?; ``` +- **A transaction holds the whole object.** No other request reaches the + object until it ends, even while your code awaits something that is not + storage. That keeps other requests' writes out of it -- and is why it + should be short: no slow `fetch()` inside, and never a request to the object + itself. A `begin()` transaction left open stalls the object until it is + committed, rolled back or dropped. - **The callback owns what it uses.** It is `'static` -- move data in, and run queries through the `tx` it is handed, not the connection. -- **It holds the whole object.** No other request reaches the object until - the transaction ends, even while the callback awaits something that is not - storage. That keeps other requests' writes out of it -- and is why it should - be short: no `fetch()` inside, and never a request to the object itself. +- **No nesting.** `begin()` inside an open transaction -- a savepoint, to sqlx + -- fails, and so do batches and `transaction` while one is open. ## Batches @@ -136,8 +155,8 @@ let names: Vec = conn - **No `query!` / `query_as!` macros.** sqlx's macros only know its built-in drivers. The runtime API (`sqlx::query`, `query_as`, `FromRow`) works. -- **No `begin()`.** `sql.exec` rejects `BEGIN` and `SAVEPOINT`, so sqlx's - `begin()` fails. Use a [transaction callback](#transactions). +- **No savepoints.** `sql.exec` rejects `SAVEPOINT`, so a nested `begin()` + fails. See [Transactions](#transactions). - **No `sqlx::Pool`.** Nothing to pool: the connection is a handle to storage the object owns. Keep one in the object's struct. - **Integers are limited to ±(2^53 − 1).** Values cross as JavaScript diff --git a/crates/sqlx-cloudflare-do/src/connection.rs b/crates/sqlx-cloudflare-do/src/connection.rs index 72c9012..28fb817 100644 --- a/crates/sqlx-cloudflare-do/src/connection.rs +++ b/crates/sqlx-cloudflare-do/src/connection.rs @@ -1,5 +1,7 @@ +use std::cell::RefCell; use std::fmt::{self, Debug, Formatter}; use std::future::{self, Future}; +use std::rc::Rc; use std::str::FromStr; use std::time::Duration; @@ -9,6 +11,7 @@ use sqlx_core::error::Error; use sqlx_core::transaction::Transaction; use sqlx_core::Url; +use crate::transaction::Open; use crate::Do; /// A connection to a Durable Object's SQL storage. @@ -21,19 +24,23 @@ use crate::Do; /// connection kept in the object's struct serves every request through /// `&self`. pub struct DoConnection { - /// For [`execute_batch`](Self::execute_batch) and - /// [`fetch_batch`](Self::fetch_batch), which run through - /// `Storage::transaction`. - pub(crate) storage: worker::Storage, + /// For everything that runs through `Storage::transaction`. Shared, + /// because the transaction `begin()` opens is parked in a task of its own + /// that needs the storage too -- and `worker::Storage` cannot be cloned. + pub(crate) storage: Rc, /// `storage.sql()`, kept rather than fetched for every query. pub(crate) sql: worker::SqlStorage, + /// Where a transaction `begin()` opened stands; see + /// [`DoTransactionManager`](crate::DoTransactionManager). + pub(crate) open: RefCell, } // `worker::Storage` is neither, only because it holds a JavaScript handle -- -// and `worker` itself makes `SqlStorage` both on the grounds that follow. A -// Durable Object runs on one thread, so the handle never meets another one. -// The executor's futures rely on this: they hold `&DoConnection`, and sqlx -// requires them to be `Send`. +// and `worker` itself makes `SqlStorage` both on the grounds that follow -- +// and `Rc` and `RefCell` are neither for the same reason: they assume one +// thread. A Durable Object runs on one thread, so none of them ever meets +// another one. The executor's futures rely on this: they hold +// `&DoConnection`, and sqlx requires them to be `Send`. // // SAFETY: a Durable Object is single-threaded; see above. unsafe impl Send for DoConnection {} @@ -45,7 +52,11 @@ impl DoConnection { #[must_use] pub fn new(storage: worker::Storage) -> Self { let sql = storage.sql(); - Self { storage, sql } + Self { + storage: Rc::new(storage), + sql, + open: RefCell::default(), + } } /// The storage this connection wraps, for anything this crate does not @@ -63,8 +74,12 @@ impl DoConnection { /// The storage this connection wraps, giving up the connection. See /// [`storage`](Self::storage) to borrow it instead. + /// + /// Shared, because a transaction `begin()` opened holds the storage too + /// until it has ended -- which, for one dropped without a commit, can be + /// a moment after the drop. #[must_use] - pub fn into_storage(self) -> worker::Storage { + pub fn into_storage(self) -> Rc { self.storage } } @@ -100,7 +115,8 @@ impl Connection for DoConnection { future::ready(Ok(())) } - /// Always fails -- see [`DoTransactionManager`](crate::DoTransactionManager). + /// Parks a transaction open -- see + /// [`DoTransactionManager`](crate::DoTransactionManager). fn begin(&mut self) -> impl Future, Error>> + Send + '_ { Transaction::begin(self, None) } diff --git a/crates/sqlx-cloudflare-do/src/executor.rs b/crates/sqlx-cloudflare-do/src/executor.rs index c6e2858..094b6fa 100644 --- a/crates/sqlx-cloudflare-do/src/executor.rs +++ b/crates/sqlx-cloudflare-do/src/executor.rs @@ -14,11 +14,12 @@ use crate::{ js, Do, DoArguments, DoConnection, DoQueryResult, DoRow, DoStatement, DoTransaction, DoTypeInfo, }; -// `sql.exec` is synchronous, so each future here finishes on its first poll -// and holds nothing across an `await` -- which is why none of them needs -// D1's `SendFuture`: they are `Send` because `DoConnection` and -// `DoTransaction` are `Sync`. They still run the query when polled, not when -// built, as sqlx's futures do. +// `sql.exec` is synchronous, so each future here runs its query on its first +// poll -- after, on a connection, waiting for any dropped transaction's +// rollback to land (see `DoConnection::settle`). None of them needs D1's +// `SendFuture`: what they hold across that wait is `Send`, and they are +// `Send` because `DoConnection` and `DoTransaction` are `Sync`. They still +// run the query when polled, not when built, as sqlx's futures do. // // One implementation for both executors, a connection and an open transaction: // each only needs its `sql` handle, and a macro writes the impls for each. @@ -34,6 +35,7 @@ macro_rules! impl_executor { E: 'q + Execute<'q, Do>, { async move { + self.settle().await; let (sql, arguments) = take(query)?; js::run(&self.sql, sql.as_str(), arguments_of(arguments.as_ref())) } @@ -62,6 +64,7 @@ macro_rules! impl_executor { E: 'q + Execute<'q, Do>, { async move { + self.settle().await; let (sql, arguments) = take(query)?; js::fetch( &self.sql, @@ -85,6 +88,7 @@ macro_rules! impl_executor { E: 'q + Execute<'q, Do>, { async move { + self.settle().await; let (sql, arguments) = take(query)?; let (rows, _) = js::fetch( &self.sql, diff --git a/crates/sqlx-cloudflare-do/src/lib.rs b/crates/sqlx-cloudflare-do/src/lib.rs index 4396fa7..b529566 100644 --- a/crates/sqlx-cloudflare-do/src/lib.rs +++ b/crates/sqlx-cloudflare-do/src/lib.rs @@ -40,7 +40,9 @@ //! ``` //! //! The database lives in the Durable Object: queries never leave the process, -//! and each one runs to completion before its future first returns. The class +//! and each one runs to completion when its future is first polled. +//! Transactions are real: sqlx's `begin()` / `commit()` work as with any +//! driver, and [`DoConnection::transaction`] takes them as a callback. The class //! must be declared with `new_sqlite_classes` in `wrangler.toml` -- the //! key-value-backed kind has no SQL storage. //! @@ -48,10 +50,10 @@ //! //! Each of these fails loudly rather than pretending: //! -//! - **`begin()`.** It returns an error: `sql.exec` rejects `BEGIN` and -//! `SAVEPOINT`. Transactions take a callback instead -- -//! [`DoConnection::transaction`] -- and [`DoConnection::execute_batch`] runs -//! several statements atomically without one. +//! - **Nested transactions.** sqlx's `begin()`, `commit()` and `rollback()` +//! work -- see [`DoTransactionManager`] -- as does +//! [`DoConnection::transaction`], but `begin()` inside an open transaction +//! (a savepoint, to sqlx) fails: `sql.exec` rejects `SAVEPOINT`. //! - **Integers beyond ±(2^53 − 1).** Values cross as JavaScript numbers, so //! binding a wider `i64` is an encode error rather than a rounded value. //! See [`types`]. diff --git a/crates/sqlx-cloudflare-do/src/transaction.rs b/crates/sqlx-cloudflare-do/src/transaction.rs index 8a8772f..9c6e1cc 100644 --- a/crates/sqlx-cloudflare-do/src/transaction.rs +++ b/crates/sqlx-cloudflare-do/src/transaction.rs @@ -1,8 +1,11 @@ use std::cell::RefCell; use std::fmt::{self, Debug, Formatter}; -use std::future::{self, Future}; +use std::future::Future; +use std::mem; use std::rc::Rc; +use futures_channel::oneshot; +use futures_util::future::{self, Either, FutureExt, Shared}; use sqlx_core::error::Error; use sqlx_core::sql_str::SqlStr; use sqlx_core::transaction::TransactionManager; @@ -11,8 +14,12 @@ use worker::send::SendFuture; use crate::error::unsupported; use crate::{Do, DoConnection}; -const NO_TRANSACTIONS: &str = "Durable Object storage cannot begin a transaction here; \ - run the statements inside `DoConnection::transaction`, which takes them as a callback"; +/// Why a parked transaction rejects: a rollback, asked for. Any error would +/// do -- rejecting is what rolls back -- but this one is recognisable. +const ROLLED_BACK: &str = "rolled back"; + +const ALREADY_OPEN: &str = "a transaction begun with `begin()` is open on this connection; \ + run the statements through it rather than in a batch or another transaction"; impl DoConnection { /// Runs `callback` in a transaction: it commits if the callback returns @@ -22,7 +29,9 @@ impl DoConnection { /// A real interactive transaction, unlike a batch: the callback reads its /// own writes, and one statement can use what the one before it returned. /// It runs inside `Storage::transaction`, and the callback's value -- or - /// its error, as it returned it -- is what this resolves to. + /// its error, as it returned it -- is what this resolves to. sqlx's own + /// `begin()` works too (see [`DoTransactionManager`]); this is the form + /// that cannot be left open by mistake. /// /// ```no_run /// # async fn create(conn: &sqlx_cloudflare_do::DoConnection, email: String) -> Result { @@ -48,8 +57,15 @@ impl DoConnection { /// `Storage::transaction` takes a `'static` callback, so it cannot borrow /// the connection or anything else from the caller: move what it needs /// in, and run its queries through the [`DoTransaction`] it is handed. - /// That is also why this is a callback and not sqlx's `begin()`, which - /// borrows the connection -- `begin()` still fails. + /// + /// # Which `transaction` you get + /// + /// sqlx's `Connection` trait has a `transaction` method of its own, taking + /// `&mut self` and a boxed callback over `&mut Transaction`. With that + /// trait in scope, `conn.transaction(…)` on a `&mut DoConnection` resolves + /// to sqlx's -- which works too, through `begin()` and `commit()`. Call + /// this one on a `&DoConnection`, or as `DoConnection::transaction(&conn, + /// …)`. /// /// # It holds the whole object /// @@ -62,7 +78,8 @@ impl DoConnection { /// # Errors /// /// The callback's own error, after rolling back; or, converted into `E`, - /// an error from storage when the transaction could not commit. + /// an error from storage when the transaction could not commit, or when a + /// transaction begun with `begin()` is still open on this connection. pub fn transaction( &self, callback: F, @@ -81,6 +98,14 @@ impl DoConnection { // handles, so neither is `Send`; `SendFuture` asserts it, soundly, // because a Durable Object runs on one thread. SendFuture::new(async move { + self.settle().await; + + // Storage would nest this inside the parked transaction, whose + // callback waits on a commit that cannot come while this runs. + if self.is_open() { + return Err(E::from(unsupported(ALREADY_OPEN))); + } + // The callback reports through this rather than its return // value, which `worker` flattens to a string: the caller gets its // value, or its error as it returned it. @@ -116,6 +141,41 @@ impl DoConnection { } }) } + + /// Whether a transaction `begin()` opened is still open. + pub(crate) fn is_open(&self) -> bool { + matches!(*self.open.borrow(), Open::Parked { .. }) + } + + /// Records a `begin()` refused while a transaction is open, so the + /// `start_rollback` its guard's drop brings is not taken for this one's. + fn refuse_nested(&mut self, error: Error) -> Error { + if let Open::Parked { refused, .. } = self.open.get_mut() { + *refused += 1; + } + error + } + + /// Waits for a dropped transaction's rollback to land. + /// + /// sqlx rolls back a dropped `Transaction` synchronously -- it can only + /// signal the parked transaction, not wait for it -- and a statement run + /// before the rollback lands would be rolled back with it. So every query + /// on the connection waits here first. + pub(crate) async fn settle(&self) { + let rolling_back = match &*self.open.borrow() { + Open::RollingBack(done) => Some(done.clone()), + _ => None, + }; + + if let Some(done) = rolling_back { + let _ = done.await; + let mut open = self.open.borrow_mut(); + if matches!(*open, Open::RollingBack(_)) { + *open = Open::None; + } + } + } } /// An open transaction: the executor [`DoConnection::transaction`] hands its @@ -128,50 +188,231 @@ pub struct DoTransaction { pub(crate) sql: worker::SqlStorage, } +impl DoTransaction { + // The executor waits on this for a connection; a callback's transaction + // has no dropped `begin()` to wait for. + #[allow( + clippy::unused_async, + reason = "the executor awaits it alike for both kinds of executor" + )] + pub(crate) async fn settle(&self) {} +} + impl Debug for DoTransaction { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { f.debug_struct("DoTransaction").finish_non_exhaustive() } } -/// Refuses `begin()`. +/// How a parked transaction ended, as `Storage::transaction` reported it. +type Ended = Result<(), String>; + +/// Where a transaction `begin()` opened stands. +#[derive(Default)] +pub(crate) enum Open { + #[default] + None, + /// Open: `Storage::transaction` is running a callback that waits for + /// `signal` -- `true` to commit, `false` to roll back -- and `ended` + /// reports how the transaction finished. + Parked { + signal: oneshot::Sender, + ended: oneshot::Receiver, + /// `begin()` calls refused while this was open, whose guards have + /// yet to drop. sqlx builds a `Transaction` before calling `begin`, + /// and drops it -- calling `start_rollback` -- when `begin` fails; each + /// such drop is owed to a refusal, not to this transaction. + refused: usize, + }, + /// Dropped without a commit: rollback signalled, not yet landed. Shared, + /// because every query on the connection waits for it. + RollingBack(Shared>), +} + +/// Opens `Storage::transaction` in a task of its own, with a callback that +/// waits for a signal rather than running any statements. +/// +/// The statements are the caller's, run on the connection while the callback +/// waits. They are part of the transaction all the same -- Durable Object +/// storage has one database, and the transaction is on it, not on the +/// callback -- and the object delivers no other request until it ends. Both +/// were measured before this was written: the request that opened it can +/// await timers and sub-requests meanwhile, and another request's write, held +/// off until the end, survives the rollback. /// -/// `sql.exec` rejects `BEGIN` and `SAVEPOINT` outright: the only transactions -/// Durable Object storage offers wrap a callback, and sqlx's `begin` and -/// `commit` are two separate calls with the caller's code in between. -/// Buffering writes until `commit` would look like a transaction and not be -/// one -- reads inside it would see the database as it was before -- which is -/// worse than an error. So `conn.begin()` compiles and fails at runtime, and -/// transactions go through [`DoConnection::transaction`]. +/// Returns the signal, a receiver that resolves once the callback runs, and +/// one that says how the transaction ended. +fn park( + storage: Rc, +) -> ( + oneshot::Sender, + oneshot::Receiver<()>, + oneshot::Receiver, +) { + let (signal, signalled) = oneshot::channel::(); + let (started_tx, started) = oneshot::channel(); + let (ended_tx, ended) = oneshot::channel(); + + worker::wasm_bindgen_futures::spawn_local(async move { + let result = storage + .transaction(move |_| async move { + let _ = started_tx.send(()); + match signalled.await { + Ok(true) => Ok(()), + // `false` -- or the connection dropped mid-transaction, + // taking the signal with it: roll back either way. + _ => Err(worker::Error::RustError(ROLLED_BACK.into())), + } + }) + .await; + + let _ = ended_tx.send(result.map_err(|error| error.to_string())); + }); + + (signal, started, ended) +} + +/// sqlx's `begin()`, `commit()` and `rollback()` for Durable Object storage. +/// +/// `sql.exec` rejects `BEGIN` and `SAVEPOINT`, and `Storage::transaction` +/// wraps a callback -- so `begin()` parks one open: the callback only waits +/// for `commit()` or `rollback()`, while the statements run on the +/// connection, through `&mut *tx`, as with any sqlx driver. They read their +/// own writes, and a rollback undoes them. +/// +/// ```no_run +/// # async fn transfer(conn: &mut sqlx_cloudflare_do::DoConnection) -> Result<(), sqlx::Error> { +/// use sqlx::Connection; +/// +/// let mut tx = conn.begin().await?; +/// sqlx::query("UPDATE accounts SET balance = balance - 10 WHERE id = 1") +/// .execute(&mut *tx) +/// .await?; +/// sqlx::query("UPDATE accounts SET balance = balance + 10 WHERE id = 2") +/// .execute(&mut *tx) +/// .await?; +/// tx.commit().await?; // dropping `tx` instead rolls both back +/// # Ok(()) +/// # } +/// ``` +/// +/// # It holds the whole object +/// +/// While a transaction is open the Durable Object delivers no other request, +/// which keeps other requests' writes out of it -- and means one left open +/// stalls every caller of the object until it is committed, rolled back or +/// dropped. Keep it short, and never await a request to the object itself +/// inside one. [`DoConnection::transaction`] is the form that cannot be left +/// open. +/// +/// # Not supported +/// +/// Nested `begin()` -- savepoints, to sqlx -- and a custom `BEGIN` statement +/// both fail; so do [`DoConnection::execute_batch`], +/// [`DoConnection::fetch_batch`] and [`DoConnection::transaction`] while a +/// transaction is open, since storage would nest them inside it. #[derive(Debug)] pub struct DoTransactionManager; impl TransactionManager for DoTransactionManager { type Database = Do; - fn begin( - _conn: &mut DoConnection, - _statement: Option, - ) -> impl Future> + Send + '_ { - future::ready(Err(unsupported(NO_TRANSACTIONS))) + async fn begin(conn: &mut DoConnection, statement: Option) -> Result<(), Error> { + conn.settle().await; + + if conn.is_open() { + return Err(conn.refuse_nested(unsupported( + "nested transactions (savepoints) are not supported", + ))); + } + + if statement.is_some() { + return Err(unsupported( + "a custom BEGIN statement cannot be run: `sql.exec` rejects them", + )); + } + + let (signal, started, mut ended) = park(Rc::clone(&conn.storage)); + + // The callback runs once storage grants the transaction; if + // storage refuses it instead, `ended` says why. + let refused = match future::select(started, &mut ended).await { + Either::Left((Ok(()), _)) => None, + Either::Left((Err(oneshot::Canceled), _)) => Some((&mut ended).await), + Either::Right((result, _)) => Some(result), + }; + + if let Some(result) = refused { + let why = match result { + Ok(Err(why)) => why, + Ok(Ok(())) => "it ended before it began".to_owned(), + Err(oneshot::Canceled) => "it never reported back".to_owned(), + }; + return Err(Error::Protocol(format!( + "could not begin a Durable Object storage transaction: {why}" + ))); + } + + *conn.open.get_mut() = Open::Parked { + signal, + ended, + refused: 0, + }; + Ok(()) } - // Unreachable in practice: `begin` never succeeds, so there is never a - // transaction to end. They still answer rather than panic. - fn commit(_conn: &mut DoConnection) -> impl Future> + Send + '_ { - future::ready(Err(unsupported(NO_TRANSACTIONS))) + async fn commit(conn: &mut DoConnection) -> Result<(), Error> { + let Open::Parked { signal, ended, .. } = mem::take(conn.open.get_mut()) else { + return Err(unsupported("there is no open transaction to commit")); + }; + + let _ = signal.send(true); + match ended.await { + Ok(Ok(())) => Ok(()), + Ok(Err(why)) => Err(Error::Protocol(format!( + "the Durable Object storage transaction did not commit: {why}" + ))), + Err(oneshot::Canceled) => Err(Error::Protocol( + "the Durable Object storage transaction ended without reporting back".into(), + )), + } } - fn rollback(_conn: &mut DoConnection) -> impl Future> + Send + '_ { - future::ready(Err(unsupported(NO_TRANSACTIONS))) + async fn rollback(conn: &mut DoConnection) -> Result<(), Error> { + match mem::take(conn.open.get_mut()) { + // The callback rejects to roll back, so the transaction + // "failing" is the rollback succeeding. + Open::Parked { signal, ended, .. } => { + let _ = signal.send(false); + let _ = ended.await; + Ok(()) + } + Open::RollingBack(done) => { + let _ = done.await; + Ok(()) + } + Open::None => Ok(()), + } } - // `Transaction` calls this from `Drop` when `begin` failed -- it cannot - // tell a failed begin from an abandoned transaction -- so it must be a - // quiet no-op, not an error. - fn start_rollback(_conn: &mut DoConnection) {} + // `Transaction`'s `Drop`: signal the rollback, which cannot be awaited + // here, and leave the connection to wait for it before its next query -- + // unless the drop is a refused `begin()`'s guard, which owns nothing. + fn start_rollback(conn: &mut DoConnection) { + if let Open::Parked { refused, .. } = conn.open.get_mut() { + if *refused > 0 { + *refused -= 1; + return; + } + } + + if let Open::Parked { signal, ended, .. } = mem::take(conn.open.get_mut()) { + let _ = signal.send(false); + *conn.open.get_mut() = Open::RollingBack(ended.shared()); + } + } - fn get_transaction_depth(_conn: &DoConnection) -> usize { - 0 + fn get_transaction_depth(conn: &DoConnection) -> usize { + usize::from(conn.is_open()) } } diff --git a/crates/sqlx-cloudflare-do/tests/facade.rs b/crates/sqlx-cloudflare-do/tests/facade.rs index 34566d5..9f297c8 100644 --- a/crates/sqlx-cloudflare-do/tests/facade.rs +++ b/crates/sqlx-cloudflare-do/tests/facade.rs @@ -89,6 +89,15 @@ async fn insert_members(conn: &DoConnection, ids: &[UserId]) -> Result Result<(), sqlx::Error> { + use sqlx::Connection; + + let mut tx = conn.begin().await?; + sqlx::query("DELETE FROM users").execute(&mut *tx).await?; + tx.commit().await +} + // Handlers need `Send` futures (axum, for one), even inside a Worker. fn assert_send(_: T) {} @@ -100,6 +109,7 @@ fn futures_are_send(conn: &mut DoConnection) { assert_send(insert_members(conn, &[UserId(1)])); assert_send(conn.execute_batch([sqlx::query("DELETE FROM users")])); assert_send(conn.fetch_batch([sqlx::query("SELECT * FROM users")])); + assert_send(begin_and_commit(conn)); assert_send(conn.transaction(|tx| async move { sqlx::query("DELETE FROM users").execute(&tx).await?; Ok::<_, sqlx::Error>(()) diff --git a/crates/sqlx-cloudflare-do/tests/worker/Cargo.lock b/crates/sqlx-cloudflare-do/tests/worker/Cargo.lock index c01a4db..ac71636 100644 --- a/crates/sqlx-cloudflare-do/tests/worker/Cargo.lock +++ b/crates/sqlx-cloudflare-do/tests/worker/Cargo.lock @@ -980,8 +980,9 @@ dependencies = [ [[package]] name = "sqlx-cloudflare-do" -version = "0.1.0" +version = "0.2.0" dependencies = [ + "futures-channel", "futures-core", "futures-util", "js-sys", diff --git a/crates/sqlx-cloudflare-do/tests/worker/src/lib.rs b/crates/sqlx-cloudflare-do/tests/worker/src/lib.rs index 98166ae..aa3455e 100644 --- a/crates/sqlx-cloudflare-do/tests/worker/src/lib.rs +++ b/crates/sqlx-cloudflare-do/tests/worker/src/lib.rs @@ -43,7 +43,6 @@ scenarios![ constraint_errors, batch_reports_each_statement, batch_is_atomic, - begin_fails, from_row, invalid_sql_is_an_error, wrong_argument_count_is_an_error, @@ -63,6 +62,12 @@ scenarios![ transaction_rolls_back_when_the_callback_fails, transaction_keeps_the_statement_error, transaction_reads_its_own_writes, + begin_commit_keeps_writes, + begin_rollback_undoes_writes, + dropped_transaction_rolls_back_before_the_next_query, + begin_reads_its_own_writes_across_awaits, + nested_begin_is_refused, + batches_and_callbacks_are_refused_while_open, select_reports_no_changes, begin_statement_is_refused, ]; @@ -73,7 +78,10 @@ const SCHEMA: &str = include_str!("../schema.sql"); // share one database, and `reset` clears what each one depends on. // Scenarios the Worker runs itself rather than the object: they need more // than one request to the object at a time. -const WORKER_SCENARIOS: &[&str] = &["transaction_holds_off_other_requests"]; +const WORKER_SCENARIOS: &[&str] = &[ + "transaction_holds_off_other_requests", + "begin_holds_off_other_requests", +]; #[event(fetch)] async fn fetch(req: Request, env: Env, _ctx: Context) -> worker::Result { @@ -88,8 +96,14 @@ async fn fetch(req: Request, env: Env, _ctx: Context) -> worker::Result { - let body = match transaction_holds_off_other_requests(&env).await { + path @ ("/transaction_holds_off_other_requests" | "/begin_holds_off_other_requests") => { + // The same check for both ways to hold a transaction open. + let hold = if path.starts_with("/begin") { + "begin-hold" + } else { + "hold" + }; + let body = match holds_off_other_requests(&env, hold).await { Ok(()) => serde_json::json!({ "ok": true }), Err(error) => serde_json::json!({ "ok": false, "error": error }), }; @@ -120,9 +134,9 @@ impl DurableObject for TestObject { // The steps of the Worker-level scenarios, which are not scenarios // themselves. if let Some(op) = name.strip_prefix("op/") { - let conn = DoConnection::new(self.state.storage()); + let mut conn = DoConnection::new(self.state.storage()); conn.sql().exec(SCHEMA, None)?; - return op_step(&conn, op).await; + return op_step(&mut conn, op).await; } let Some((_, scenario)) = SCENARIOS.iter().find(|(n, _)| *n == name) else { @@ -130,7 +144,7 @@ impl DurableObject for TestObject { }; // A fresh connection per request, from the object's own storage: - // `begin_fails` needs one it can borrow mutably. + // `begin()` needs one it can borrow mutably. let mut conn = DoConnection::new(self.state.storage()); conn.sql().exec(SCHEMA, None)?; @@ -458,10 +472,6 @@ async fn batch_is_atomic(conn: &mut DoConnection) -> Outcome { ) } -async fn begin_fails(conn: &mut DoConnection) -> Outcome { - ensure(conn.begin().await.is_err(), "begin succeeded") -} - #[derive(Debug, sqlx::FromRow)] struct User { id: i64, @@ -1075,7 +1085,7 @@ async fn transaction_reads_its_own_writes(conn: &mut DoConnection) -> Outcome { // The object delivers no other request while a transaction runs, even while // its callback awaits a timer -- which is what keeps a concurrent request's // writes out of it, and out of its rollback. Measured in #7; pinned here. -async fn transaction_holds_off_other_requests(env: &Env) -> Outcome { +async fn holds_off_other_requests(env: &Env, hold: &'static str) -> Outcome { let step = |op: &'static str| async move { step_on(env, op) .await @@ -1087,7 +1097,7 @@ async fn transaction_holds_off_other_requests(env: &Env) -> Outcome { worker::Delay::from(std::time::Duration::from_millis(50)).await; step("insert").await }; - let (held, other) = futures_util::future::try_join(step("hold"), other).await?; + let (held, other) = futures_util::future::try_join(step(hold), other).await?; let after = step("emails").await?; ensure(held["rolled_back"] == true, format!("hold: {held}"))?; @@ -1133,7 +1143,7 @@ impl From for HoldError { } } -async fn op_step(conn: &DoConnection, op: &str) -> worker::Result { +async fn op_step(conn: &mut DoConnection, op: &str) -> worker::Result { let now = || worker::Date::now().as_millis(); let body = match op { @@ -1157,27 +1167,211 @@ async fn op_step(conn: &DoConnection, op: &str) -> worker::Result { serde_json::json!({ "at": at }) } "hold" => { - let result = conn - .transaction(|tx| async move { - sqlx::query("INSERT INTO users (email) VALUES ('held')") - .execute(&tx) - .await?; - worker::Delay::from(std::time::Duration::from_millis(300)).await; - let seen: Vec = - sqlx::query_scalar("SELECT email FROM users ORDER BY id") - .fetch_all(&tx) - .await?; - Err::<(), _>(HoldError::Held(seen)) - }) - .await; + // Named in full: with `sqlx::Connection` in scope, `conn.transaction` + // on a `&mut DoConnection` is sqlx's own method. + let result = DoConnection::transaction(conn, |tx| async move { + sqlx::query("INSERT INTO users (email) VALUES ('held')") + .execute(&tx) + .await?; + worker::Delay::from(std::time::Duration::from_millis(300)).await; + let seen: Vec = sqlx::query_scalar("SELECT email FROM users ORDER BY id") + .fetch_all(&tx) + .await?; + Err::<(), _>(HoldError::Held(seen)) + }) + .await; let seen = match result { Err(HoldError::Held(seen)) => seen, other => return Response::error(format!("{other:?}"), 500), }; serde_json::json!({ "rolled_back": true, "seen_inside": seen, "ended": now() }) } + // The same, held open with sqlx's `begin()` and dropped, not + // committed, at the end. + "begin-hold" => { + let mut tx = conn + .begin() + .await + .map_err(|error| worker::Error::RustError(error.to_string()))?; + sqlx::query("INSERT INTO users (email) VALUES ('held')") + .execute(&mut *tx) + .await + .map_err(|error| worker::Error::RustError(error.to_string()))?; + worker::Delay::from(std::time::Duration::from_millis(300)).await; + let seen: Vec = sqlx::query_scalar("SELECT email FROM users ORDER BY id") + .fetch_all(&mut *tx) + .await + .map_err(|error| worker::Error::RustError(error.to_string()))?; + tx.rollback() + .await + .map_err(|error| worker::Error::RustError(error.to_string()))?; + serde_json::json!({ "rolled_back": true, "seen_inside": seen, "ended": now() }) + } _ => return Response::error(format!("no step `{op}`"), 404), }; Response::from_json(&body) } + +// ---- sqlx's own transactions: `begin()` / `commit()` / `rollback()`. + +async fn count_users(conn: &DoConnection) -> Result { + sqlx::query_scalar("SELECT COUNT(*) FROM users") + .fetch_one(conn) + .await + .map_err(fail) +} + +async fn begin_commit_keeps_writes(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + for email in ["c1@example.com", "c2@example.com"] { + sqlx::query("INSERT INTO users (email) VALUES (?)") + .bind(email) + .execute(&mut *tx) + .await + .map_err(fail)?; + } + tx.commit().await.map_err(fail)?; + + let count = count_users(conn).await?; + ensure(count == 2, format!("{count} users after commit")) +} + +async fn begin_rollback_undoes_writes(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + sqlx::query("INSERT INTO users (email) VALUES ('undone@example.com')") + .execute(&mut *tx) + .await + .map_err(fail)?; + tx.rollback().await.map_err(fail)?; + + let count = count_users(conn).await?; + ensure(count == 0, format!("{count} users after rollback")) +} + +// sqlx rolls a dropped `Transaction` back without waiting; a write made right +// after the drop must not land inside the dying transaction and be undone +// with it. Measured to happen without `DoConnection::settle`. +async fn dropped_transaction_rolls_back_before_the_next_query(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + sqlx::query("INSERT INTO users (email) VALUES ('dropped@example.com')") + .execute(&mut *tx) + .await + .map_err(fail)?; + drop(tx); + + sqlx::query("INSERT INTO users (email) VALUES ('after@example.com')") + .execute(&*conn) + .await + .map_err(fail)?; + + let emails: Vec = sqlx::query_scalar("SELECT email FROM users ORDER BY id") + .fetch_all(&*conn) + .await + .map_err(fail)?; + ensure(emails == ["after@example.com"], format!("left {emails:?}")) +} + +// Statements run outside the parked callback are part of the transaction, +// across a timer: they see their own writes, and a `RETURNING` value feeds +// the next statement. +async fn begin_reads_its_own_writes_across_awaits(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + let id: i64 = + sqlx::query_scalar("INSERT INTO users (email) VALUES ('own@example.com') RETURNING id") + .fetch_one(&mut *tx) + .await + .map_err(fail)?; + worker::Delay::from(std::time::Duration::from_millis(50)).await; + sqlx::query("INSERT INTO posts (user_id, title) VALUES (?, 'mine')") + .bind(id) + .execute(&mut *tx) + .await + .map_err(fail)?; + let inside: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM posts WHERE user_id = ?") + .bind(id) + .fetch_one(&mut *tx) + .await + .map_err(fail)?; + tx.commit().await.map_err(fail)?; + + ensure(inside == 1, format!("saw {inside} posts inside"))?; + let owner: i64 = sqlx::query_scalar("SELECT user_id FROM posts WHERE title = 'mine'") + .fetch_one(&*conn) + .await + .map_err(fail)?; + ensure(owner == id, format!("post owned by {owner}, user {id}")) +} + +// sqlx drops a guard -- and calls `start_rollback` -- when `begin()` fails, +// so a refused nested `begin()` must not take the outer transaction with it: +// what the outer wrote before and after the refusal both commit. +async fn nested_begin_is_refused(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + sqlx::query("INSERT INTO users (email) VALUES ('before@example.com')") + .execute(&mut *tx) + .await + .map_err(fail)?; + + let nested = tx.begin().await.map(|_| ()); + ensure(nested.is_err(), "a nested begin() succeeded")?; + + sqlx::query("INSERT INTO users (email) VALUES ('after@example.com')") + .execute(&mut *tx) + .await + .map_err(fail)?; + tx.commit().await.map_err(fail)?; + + let emails: Vec = sqlx::query_scalar("SELECT email FROM users ORDER BY id") + .fetch_all(&*conn) + .await + .map_err(fail)?; + ensure( + emails == ["before@example.com", "after@example.com"], + format!("left {emails:?}: the refusal rolled the outer transaction back"), + ) +} + +// Storage would nest these inside the parked transaction, whose callback +// waits on a commit that could never come; they fail instead, and the open +// transaction carries on. +async fn batches_and_callbacks_are_refused_while_open(conn: &mut DoConnection) -> Outcome { + reset(conn).await?; + + let mut tx = conn.begin().await.map_err(fail)?; + let batch = tx + .execute_batch([sqlx::query( + "INSERT INTO users (email) VALUES ('batch@example.com')", + )]) + .await; + ensure(batch.is_err(), "a batch ran inside an open transaction")?; + let callback = tx + .transaction(|_| async move { Ok::<_, sqlx::Error>(()) }) + .await; + ensure( + callback.is_err(), + "a callback transaction ran inside an open one", + )?; + + sqlx::query("INSERT INTO users (email) VALUES ('open@example.com')") + .execute(&mut *tx) + .await + .map_err(fail)?; + tx.commit().await.map_err(fail)?; + + let emails: Vec = sqlx::query_scalar("SELECT email FROM users ORDER BY id") + .fetch_all(&*conn) + .await + .map_err(fail)?; + ensure(emails == ["open@example.com"], format!("left {emails:?}")) +}