feat(do)!: support sqlx's begin(), commit() and rollback()

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"

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 }
 

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<String> = 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

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<worker::Storage>,
     /// `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<Open>,
 }
 
 // `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<worker::Storage> {
         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<Output = Result<Transaction<'_, Do>, Error>> + Send + '_ {
         Transaction::begin(self, None)
     }

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,

crates/sqlx-cloudflare-do/src/lib.rs

@@ -40,18 +40,20 @@
 //! ```
 //!
 //! 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.
 //!
 //! # What Durable Object storage cannot do
 //!
 //! 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`].