Skip to main content

cratestack_sqlx/
transaction.rs

1//! `db.transaction(...)` combinator (cratestack#513): compose several
2//! write-builder calls in one Postgres transaction using only CrateStack's
3//! own API — no `sqlx` dependency in the caller's `Cargo.toml`, and no
4//! `sqlx::Transaction` named in the caller's own source.
5//!
6//! **Design (see the PR body for the sub-questions this answers in full):**
7//!
8//! - [`Tx`] is an opaque, crate-owned newtype around
9//!   `sqlx::Transaction<'static, sqlx::Postgres>`. It implements
10//!   [`Deref`]/[`DerefMut`] to that type, which is what lets every existing
11//!   `run_in_tx(&mut sqlx::Transaction<'tx, Postgres>, ctx)` on the write
12//!   builders (`create.rs`, `update.rs`, ...) keep their exact current
13//!   signature: passing `&mut Tx` at a `&mut sqlx::Transaction<'_, _>` call
14//!   site coerces automatically via `DerefMut`, so nothing downstream of
15//!   `run_in_tx` had to change. The caller's closure never has to name
16//!   `sqlx::Transaction` (or even `Tx` — the type is inferred), it's not a
17//!   breaking change to `run_in_tx`, and the transaction still round-trips
18//!   through the real `sqlx` machinery underneath.
19//! - The closure is bound by `AsyncFnOnce(&mut Tx) -> Result<T, CoolError>`
20//!   (the native async-closure traits stabilized in Rust 1.85; this
21//!   workspace pins 1.95) rather than the `FnMut(...) -> Fut` shape
22//!   `run_in_isolated_tx` uses. That older shape requires the body to hand
23//!   the transaction *back* out of the future on every call (see
24//!   `isolation.rs`) because a plain closure returning `async move { .. }`
25//!   can't express "the returned future borrows the argument for its own
26//!   lifetime" — the classic Rust lending-closure problem. `AsyncFnOnce`
27//!   solves exactly that: callers can write
28//!   `db.transaction(async |tx| { ...; Ok(value) }).await` and reuse `tx`
29//!   across as many sequential `.await`s as they like without threading it
30//!   back through the return type. Verified against a standalone
31//!   reproduction before adopting it here — see the PR body.
32//! - No retry loop: unlike [`crate::run_in_isolated_tx`], `transaction`
33//!   doesn't re-run `body` on a serialization failure, since `body` isn't
34//!   guaranteed idempotent (it's arbitrary caller code, not caller code
35//!   already scoped to "safe to retry" the way `@isolation` procedures
36//!   are). Retrying is exactly what `run_in_isolated_tx` is for; the two
37//!   are orthogonal and composable (see the PR body's isolation
38//!   discussion), not alternatives to pick between.
39//!
40//! ## Composing through here does not close the `AuditSink`/outbox gap (cratestack#534)
41//!
42//! It is tempting to assume that because this is the *sanctioned* way to
43//! compose several write-builder calls, it also gets you the fan-out that
44//! `run()` gives you automatically — an installed [`cratestack_core::AuditSink`]
45//! observing every `@@audit` write, and `@@emit` events reaching their
46//! subscribers. **It does not.** `body` still calls each write builder's
47//! `run_in_tx`, which still only writes the in-database `cratestack_audit`
48//! row / outbox row and hands back a `RunInTxOutcome` — it never dispatches
49//! anything itself, for exactly the same reason it doesn't when called
50//! against a transaction obtained directly from `db.pool().begin()`: there
51//! is still no reliable "after commit" point *inside this crate*, because
52//! `transaction` only knows `body` returned `Ok::<T, _>` for an arbitrary,
53//! caller-chosen `T` — it has no way to discover which `RunInTxOutcome`s
54//! (if any) `body` produced along the way unless `body` hands them back as
55//! part of its own return value.
56//!
57//! This was investigated as a candidate host for cratestack#534's option
58//! (b) ("the runtime takes ownership of dispatch") and found not cleanly
59//! achievable here: even setting aside the arbitrary-`T` problem above,
60//! [`SqlxRuntime::pool`] stays public, so a caller can always open a
61//! transaction with `db.pool().begin()` directly and pass it straight to
62//! `run_in_tx`, bypassing this combinator entirely — the same call
63//! `run_in_tx` accepts from here, because [`Tx`] derefs to a plain
64//! `sqlx::Transaction` before `run_in_tx` ever sees it (see above), so
65//! `run_in_tx` cannot even tell which door the transaction came through.
66//! Any auto-dispatch hook attached only to `transaction()` would therefore
67//! be incomplete by construction, reproducing the exact invisible gap
68//! cratestack#534 exists to close, just for a subset of callers instead of
69//! all of them. **The contract is caller-driven, unconditionally**: after
70//! `transaction()` returns `Ok`, dispatch the audit events yourself via
71//! the generated `Cratestack::dispatch_audit_sink` and drain the outbox
72//! yourself via `Cratestack::events().drain()` — see
73//! [`crate::dispatch_audit_sink`]'s doc comment for the full reasoning,
74//! which applies here unchanged.
75
76use std::ops::{Deref, DerefMut};
77
78use cratestack_core::CoolError;
79
80use crate::descriptor::SqlxRuntime;
81use crate::error::cool_error_from_sqlx;
82use crate::sqlx;
83
84/// Opaque handle onto a live Postgres transaction. Obtained only via
85/// [`SqlxRuntime::transaction`]; never constructed directly by consumers.
86///
87/// Derefs to `sqlx::Transaction<'static, sqlx::Postgres>` purely so the
88/// existing write-builder `run_in_tx` methods keep working unchanged (see
89/// the module doc comment) — this is an implementation detail, not an
90/// invitation to import `sqlx` yourself. Nothing about the public
91/// `db.transaction(...)` call site requires it.
92pub struct Tx(sqlx::Transaction<'static, sqlx::Postgres>);
93
94impl Deref for Tx {
95    type Target = sqlx::Transaction<'static, sqlx::Postgres>;
96
97    fn deref(&self) -> &Self::Target {
98        &self.0
99    }
100}
101
102impl DerefMut for Tx {
103    fn deref_mut(&mut self) -> &mut Self::Target {
104        &mut self.0
105    }
106}
107
108impl SqlxRuntime {
109    /// Run `body` inside one Postgres transaction: commit if it returns
110    /// `Ok`, roll back if it returns `Err`. `body` receives an opaque [`Tx`]
111    /// it can pass straight through to any write builder's `run_in_tx` —
112    /// see the module doc comment for why no `sqlx` type ever needs to be
113    /// named to do that.
114    ///
115    /// On the `Err` path this issues an explicit `tx.rollback().await`
116    /// rather than relying on `sqlx::Transaction`'s `Drop` impl. That
117    /// matters: `sqlx-core`'s `Drop for Transaction` only *queues* a
118    /// rollback for the next time the underlying connection is used (see
119    /// `sqlx-core::transaction::Transaction`'s `Drop` impl) — it does not
120    /// synchronously roll back. A caller asserting "neither write is
121    /// visible" immediately after an `Err` return needs that to have
122    /// already happened, not to be pending on some future unrelated query.
123    ///
124    /// Does not retry — see the module doc comment for why that's left to
125    /// [`crate::run_in_isolated_tx`] instead, and how the two compose.
126    ///
127    /// **Does not dispatch to an installed `AuditSink` or drain the
128    /// `@@emit` outbox on its own** — see the module doc comment's
129    /// "Composing through here does not close the `AuditSink`/outbox gap"
130    /// section (cratestack#534) for why that can't be made automatic here.
131    pub async fn transaction<F, T>(&self, body: F) -> Result<T, CoolError>
132    where
133        F: AsyncFnOnce(&mut Tx) -> Result<T, CoolError>,
134    {
135        let inner = self.pool().begin().await.map_err(cool_error_from_sqlx)?;
136        let mut tx = Tx(inner);
137
138        match body(&mut tx).await {
139            Ok(value) => {
140                tx.0.commit().await.map_err(cool_error_from_sqlx)?;
141                Ok(value)
142            }
143            Err(error) => {
144                // Best-effort: if the rollback itself fails (e.g. the
145                // connection already dropped), the original `error` is
146                // still the one that matters to the caller — a failed
147                // rollback attempt shouldn't mask it.
148                let _ = tx.0.rollback().await;
149                Err(error)
150            }
151        }
152    }
153}