Skip to main content

keelson_exec/
transaction.rs

1use std::fmt;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
4
5use keelson_core::Value;
6use tokio::sync::Mutex;
7
8use crate::error::ExecError;
9use crate::executor::{ExecFuture, ExecResult, Executor, Family, Statement};
10use crate::row::Row;
11
12/// One raw connection, exclusively held. The seam a backend implements.
13///
14/// keelson-sqlx implements this once per driver; a future backend implements
15/// it once. Everything else — [`Transaction`], the transaction SQL, the verb
16/// layer — is written once in this crate against it, which is what keeps
17/// transaction semantics identical across backends.
18pub trait RawConnection: Send + fmt::Debug {
19    /// Which engine family this connection talks to.
20    fn family(&self) -> Family;
21
22    /// Run a statement and collect every row.
23    fn fetch<'a>(
24        &'a mut self,
25        sql: &'a str,
26        args: Vec<Value>,
27    ) -> ExecFuture<'a, Result<Vec<Row>, ExecError>>;
28
29    /// Run a statement for its side effect.
30    fn execute<'a>(
31        &'a mut self,
32        sql: &'a str,
33        args: Vec<Value>,
34    ) -> ExecFuture<'a, Result<ExecResult, ExecError>>;
35
36    /// Dispose of a connection whose server-side state is unknown — an
37    /// abandoned transaction. Must **not** return the connection to a pool as
38    /// reusable: close it (the server then discards the open transaction).
39    fn abandon(self: Box<Self>);
40}
41
42/// An open transaction. Owned, lifetime-free, and an [`Executor`] — any
43/// function written as `fn f(db: &dyn Executor)` accepts it, which is what
44/// lets model hooks run inside the caller's transaction without knowing one
45/// exists.
46///
47/// `commit` and `rollback` consume `self`: using a finished transaction is a
48/// compile error, not a runtime one. Dropping without either **abandons the
49/// connection** — it is closed rather than returned to the pool, and the
50/// server rolls the transaction back. The lazy path is therefore always
51/// *safe* and merely expensive; `commit` is the only way to keep work. Prefer
52/// [`BeginExt::within`], where neither a drop nor a forgotten commit can
53/// happen at all.
54///
55/// This crate itself issues the transaction vocabulary — `BEGIN`, `COMMIT`,
56/// `ROLLBACK`, `SAVEPOINT n` / `RELEASE SAVEPOINT n` / `ROLLBACK TO SAVEPOINT
57/// n` — which is identical across PostgreSQL, MySQL and SQLite, so no
58/// backend re-implements (or drifts on) transaction semantics. Where the
59/// engines stop agreeing — isolation levels, access modes — the vocabulary
60/// is still written here, once, but per family and with the disagreements
61/// spelled out: see [`TxOptions`] and [`BeginWith`].
62pub struct Transaction {
63    conn: Mutex<Option<Box<dyn RawConnection>>>,
64    family: Family,
65    opts: TxOptions,
66    finished: AtomicBool,
67    depth: AtomicU32,
68}
69
70impl fmt::Debug for Transaction {
71    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
72        f.debug_struct("Transaction")
73            .field("family", &self.family)
74            .field("options", &self.opts)
75            .field("finished", &self.finished.load(Ordering::Relaxed))
76            .finish_non_exhaustive()
77    }
78}
79
80impl Transaction {
81    /// Open a transaction on an exclusively-owned connection. Backend-facing:
82    /// a backend's [`Begin`] impl checks a connection out and hands it here.
83    pub async fn begin_on(conn: Box<dyn RawConnection>) -> Result<Self, ExecError> {
84        Transaction::begin_on_with(conn, TxOptions::new()).await
85    }
86
87    /// Open a transaction with explicit [`TxOptions`]. Backend-facing, the
88    /// counterpart of [`BeginWith::begin_with`].
89    ///
90    /// The options are turned into statements by [`TxOptions::plan`] *before*
91    /// anything is sent, so an option this engine cannot honour is refused
92    /// with the connection untouched — it goes back to its pool clean rather
93    /// than being abandoned. Every statement of the plan runs on this one
94    /// connection, ahead of any statement the caller issues; if one of them
95    /// fails the connection is abandoned rather than returned, because a
96    /// half-applied plan (MySQL's `SET TRANSACTION` having landed without its
97    /// `START TRANSACTION`) would otherwise leak into whatever transaction
98    /// this pooled connection served next.
99    pub async fn begin_on_with(
100        mut conn: Box<dyn RawConnection>,
101        opts: TxOptions,
102    ) -> Result<Self, ExecError> {
103        let family = conn.family();
104        let plan = opts.plan(family)?;
105        for sql in &plan {
106            if let Err(e) = conn.execute(sql, Vec::new()).await {
107                conn.abandon();
108                return Err(e);
109            }
110        }
111        #[cfg(feature = "tracing")]
112        tracing::debug!(
113            target: "keelson",
114            family = family.as_str(),
115            isolation = opts.get_isolation().map(Isolation::as_sql),
116            "transaction begun"
117        );
118        Ok(Transaction {
119            conn: Mutex::new(Some(conn)),
120            family,
121            opts,
122            finished: AtomicBool::new(false),
123            depth: AtomicU32::new(0),
124        })
125    }
126
127    /// The options this transaction was opened with — [`TxOptions::new`]'s
128    /// defaults for one begun by [`Begin::begin`].
129    pub fn options(&self) -> TxOptions {
130        self.opts
131    }
132
133    /// Commit. Consumes the transaction; on success the connection goes back
134    /// to wherever it came from.
135    pub async fn commit(self) -> Result<(), ExecError> {
136        self.end("COMMIT").await
137    }
138
139    /// Roll back explicitly. Consumes the transaction; cheaper than dropping,
140    /// because the connection is returned cleanly instead of closed.
141    pub async fn rollback(self) -> Result<(), ExecError> {
142        self.end("ROLLBACK").await
143    }
144
145    async fn end(&self, sql: &str) -> Result<(), ExecError> {
146        self.finished.store(true, Ordering::Relaxed);
147        let mut guard = self.conn.lock().await;
148        let mut conn = guard
149            .take()
150            .ok_or_else(|| ExecError::other("transaction connection missing"))?;
151        let res = conn.execute(sql, Vec::new()).await.map(|_| ());
152        #[cfg(feature = "tracing")]
153        tracing::debug!(
154            target: "keelson",
155            family = self.family.as_str(),
156            outcome = if sql == "COMMIT" { "commit" } else { "rollback" },
157            "transaction finished"
158        );
159        if res.is_err() {
160            // The server-side state is unknown; the connection must not be
161            // reused.
162            conn.abandon();
163        }
164        res
165    }
166
167    /// A nested transaction, via `SAVEPOINT`.
168    ///
169    /// The savepoint has no handle to leak: `Ok(_)` releases it, `Err(_)`
170    /// rolls back to it, and the outer transaction lives on either way. The
171    /// closure receives this same transaction, so every `&dyn Executor`-taking
172    /// helper works unchanged inside. Nesting is unbounded; depth-numbered
173    /// names never collide.
174    pub async fn savepoint<T, E, F>(&self, f: F) -> Result<T, E>
175    where
176        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
177        E: From<ExecError>,
178    {
179        let level = self.depth.fetch_add(1, Ordering::Relaxed) + 1;
180        let name = format!("keelson_sp_{level}");
181        if let Err(e) = self.raw(&format!("SAVEPOINT {name}")).await {
182            self.depth.fetch_sub(1, Ordering::Relaxed);
183            return Err(E::from(e));
184        }
185        let out = f(self).await;
186        let cleanup = match &out {
187            Ok(_) => self.raw(&format!("RELEASE SAVEPOINT {name}")).await,
188            // ROLLBACK TO leaves the savepoint in place on every engine we
189            // target, so it is released afterwards to keep names reusable.
190            Err(_) => match self.raw(&format!("ROLLBACK TO SAVEPOINT {name}")).await {
191                Ok(()) => self.raw(&format!("RELEASE SAVEPOINT {name}")).await,
192                Err(e) => Err(e),
193            },
194        };
195        self.depth.fetch_sub(1, Ordering::Relaxed);
196        match (out, cleanup) {
197            (Ok(v), Ok(())) => Ok(v),
198            (Ok(_), Err(e)) => Err(E::from(e)),
199            // The closure's own error wins; if cleanup also failed the
200            // transaction is suspect and the eventual COMMIT will refuse.
201            (Err(e), _) => Err(e),
202        }
203    }
204
205    /// Run transaction-control SQL on the held connection.
206    async fn raw(&self, sql: &str) -> Result<(), ExecError> {
207        let mut guard = self.conn.lock().await;
208        let conn = guard
209            .as_mut()
210            .ok_or_else(|| ExecError::other("transaction already finished"))?;
211        conn.execute(sql, Vec::new()).await.map(|_| ())
212    }
213}
214
215impl Executor for Transaction {
216    fn family(&self) -> Family {
217        self.family
218    }
219
220    fn fetch(&self, stmt: Statement) -> ExecFuture<'_, Result<Vec<Row>, ExecError>> {
221        Box::pin(async move {
222            let Statement { sql, args, .. } = stmt;
223            let mut guard = self.conn.lock().await;
224            let conn = guard
225                .as_mut()
226                .ok_or_else(|| ExecError::other("transaction already finished"))?;
227            conn.fetch(&sql, args).await
228        })
229    }
230
231    fn execute(&self, stmt: Statement) -> ExecFuture<'_, Result<ExecResult, ExecError>> {
232        Box::pin(async move {
233            let Statement { sql, args, .. } = stmt;
234            let mut guard = self.conn.lock().await;
235            let conn = guard
236                .as_mut()
237                .ok_or_else(|| ExecError::other("transaction already finished"))?;
238            conn.execute(&sql, args).await
239        })
240    }
241}
242
243impl Drop for Transaction {
244    fn drop(&mut self) {
245        if self.finished.load(Ordering::Relaxed) {
246            return;
247        }
248        // No runtime is assumed here, so nothing async can run: the
249        // connection is abandoned (closed), and the server rolls back. Safe,
250        // merely expensive — which is the right way round.
251        if let Ok(mut guard) = self.conn.try_lock()
252            && let Some(conn) = guard.take()
253        {
254            conn.abandon();
255            #[cfg(feature = "tracing")]
256            tracing::debug!(
257                target: "keelson",
258                family = self.family.as_str(),
259                outcome = "abandoned",
260                "transaction dropped without commit; connection abandoned"
261            );
262        }
263    }
264}
265
266/// A SQL-standard isolation level, as *asked for*.
267///
268/// keelson accepts a level on an engine only when that engine actually runs
269/// the transaction at it. It never substitutes a neighbouring level and calls
270/// it success — see [`TxOptions::plan`] for the per-engine table and the
271/// refusals.
272#[non_exhaustive]
273#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
274pub enum Isolation {
275    /// Dirty reads permitted. **MySQL only** here: PostgreSQL accepts the
276    /// syntax and runs READ COMMITTED, SQLite has nothing weaker than
277    /// serializable, and both of those are refused rather than faked.
278    ReadUncommitted,
279    /// Each statement sees rows committed before *it* began. PostgreSQL's
280    /// default; MySQL supports it; SQLite cannot offer it.
281    ReadCommitted,
282    /// A transaction-wide snapshot. MySQL/InnoDB's default. **Same name, two
283    /// semantics**: PostgreSQL raises a serialization failure when a
284    /// transaction writes a row changed since its snapshot, while InnoDB's
285    /// consistent read silently coexists with locking reads and `UPDATE`
286    /// taking the *current* row — the classic lost-update shape.
287    RepeatableRead,
288    /// PostgreSQL: true serializability (predicate locks, `40001` on
289    /// conflict). MySQL: `REPEATABLE READ` with every plain `SELECT` promoted
290    /// to a locking read. SQLite: what you always get.
291    Serializable,
292}
293
294impl Isolation {
295    /// The SQL spelling, per each engine's `SET TRANSACTION` grammar (they
296    /// agree on the four names).
297    pub fn as_sql(self) -> &'static str {
298        match self {
299            Isolation::ReadUncommitted => "READ UNCOMMITTED",
300            Isolation::ReadCommitted => "READ COMMITTED",
301            Isolation::RepeatableRead => "REPEATABLE READ",
302            Isolation::Serializable => "SERIALIZABLE",
303        }
304    }
305}
306
307impl fmt::Display for Isolation {
308    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
309        f.write_str(self.as_sql())
310    }
311}
312
313/// A transaction's access mode.
314#[non_exhaustive]
315#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
316pub enum Access {
317    /// The default everywhere; stating it is allowed and explicit.
318    ReadWrite,
319    /// PostgreSQL `BEGIN … READ ONLY`, MySQL `START TRANSACTION READ ONLY`.
320    /// Refused on SQLite, which has no per-transaction read-only mode.
321    ReadOnly,
322}
323
324impl Access {
325    /// The SQL spelling (identical on PostgreSQL and MySQL).
326    pub fn as_sql(self) -> &'static str {
327        match self {
328            Access::ReadWrite => "READ WRITE",
329            Access::ReadOnly => "READ ONLY",
330        }
331    }
332}
333
334/// SQLite's begin modes — **not** isolation levels, and named so nobody can
335/// mistake them for a portable knob.
336///
337/// SQLite has one isolation level (serializable); what it lets you choose is
338/// *when* the transaction takes its locks. Asking for one of these on
339/// PostgreSQL or MySQL is an error, not a no-op.
340#[non_exhaustive]
341#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
342pub enum SqliteBegin {
343    /// Locks are taken at the first statement that needs them. The default.
344    Deferred,
345    /// The write lock is taken at `BEGIN`, so a write-write conflict surfaces
346    /// as [`TxConflict::Busy`] immediately instead of mid-transaction.
347    Immediate,
348    /// As `IMMEDIATE`, and other connections cannot read either (outside WAL).
349    Exclusive,
350}
351
352impl SqliteBegin {
353    /// The keyword this mode contributes to `BEGIN`.
354    pub fn as_sql(self) -> &'static str {
355        match self {
356            SqliteBegin::Deferred => "DEFERRED",
357            SqliteBegin::Immediate => "IMMEDIATE",
358            SqliteBegin::Exclusive => "EXCLUSIVE",
359        }
360    }
361}
362
363/// What a transaction is opened *with*: the parts of transaction control the
364/// three engines do not agree on.
365///
366/// Defaults to "whatever the engine's own default is" on every axis, so
367/// [`Transaction::begin_on_with`] with a default `TxOptions` sends exactly
368/// what [`Transaction::begin_on`] sends.
369#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
370pub struct TxOptions {
371    isolation: Option<Isolation>,
372    access: Option<Access>,
373    sqlite_begin: Option<SqliteBegin>,
374}
375
376impl TxOptions {
377    /// No option set: the engine's own defaults.
378    pub const fn new() -> Self {
379        TxOptions {
380            isolation: None,
381            access: None,
382            sqlite_begin: None,
383        }
384    }
385
386    /// Ask for an isolation level.
387    pub const fn isolation(mut self, level: Isolation) -> Self {
388        self.isolation = Some(level);
389        self
390    }
391
392    /// Ask for an access mode.
393    pub const fn access(mut self, mode: Access) -> Self {
394        self.access = Some(mode);
395        self
396    }
397
398    /// Shorthand for [`Access::ReadOnly`].
399    pub const fn read_only(self) -> Self {
400        self.access(Access::ReadOnly)
401    }
402
403    /// SQLite's begin mode. An error on any other family.
404    pub const fn sqlite_begin(mut self, mode: SqliteBegin) -> Self {
405        self.sqlite_begin = Some(mode);
406        self
407    }
408
409    /// The isolation level asked for, if any.
410    pub const fn get_isolation(self) -> Option<Isolation> {
411        self.isolation
412    }
413
414    /// The access mode asked for, if any.
415    pub const fn get_access(self) -> Option<Access> {
416        self.access
417    }
418
419    /// The SQLite begin mode asked for, if any.
420    pub const fn get_sqlite_begin(self) -> Option<SqliteBegin> {
421        self.sqlite_begin
422    }
423
424    /// Can this engine honour these options? Answers without opening a
425    /// transaction — backends call it before checking a connection out, and
426    /// callers can pre-flight a configuration with it.
427    pub fn check(&self, family: Family) -> Result<(), ExecError> {
428        self.plan(family).map(|_| ())
429    }
430
431    /// The exact statements [`Transaction::begin_on_with`] will run, in
432    /// order, on the transaction's own connection — or the error explaining
433    /// why this engine cannot be asked for this.
434    ///
435    /// Nothing here is inferred at run time from a server version; it is the
436    /// documented grammar of each engine, and it is public so that "what does
437    /// keelson actually send?" is answerable without a packet capture.
438    ///
439    /// | | PostgreSQL | MySQL / InnoDB | SQLite |
440    /// |---|---|---|---|
441    /// | READ UNCOMMITTED | **refused** (accepted by the server, run as READ COMMITTED) | yes | **refused** |
442    /// | READ COMMITTED | yes (default) | yes | **refused** |
443    /// | REPEATABLE READ | yes | yes (default) | **refused** |
444    /// | SERIALIZABLE | yes | yes | yes — SQLite's only level |
445    /// | READ ONLY | yes | yes | **refused** (`PRAGMA query_only` is connection state) |
446    /// | [`SqliteBegin`] | **refused** | **refused** | yes |
447    ///
448    /// The rule behind every refusal: a level is accepted only when the
449    /// engine runs the transaction at *that* level. Substituting a
450    /// neighbouring level would satisfy the SQL standard (it permits running
451    /// stricter than asked) and would still be a lie to the caller, who asked
452    /// in order to get particular behaviour.
453    ///
454    /// The match on [`Family`] is exhaustive on purpose: a new family cannot
455    /// be added to this crate without deciding what it does here.
456    pub fn plan(&self, family: Family) -> Result<Vec<String>, ExecError> {
457        match family {
458            Family::Postgres => self.plan_postgres(),
459            Family::MySql => self.plan_mysql(),
460            Family::Sqlite => self.plan_sqlite(),
461        }
462    }
463
464    /// PostgreSQL puts everything in the `BEGIN` itself:
465    /// `BEGIN [ ISOLATION LEVEL … ] [ READ WRITE | READ ONLY ]`.
466    fn plan_postgres(&self) -> Result<Vec<String>, ExecError> {
467        self.reject_sqlite_begin(Family::Postgres)?;
468        let mut sql = String::from("BEGIN");
469        if let Some(level) = self.isolation {
470            if level == Isolation::ReadUncommitted {
471                return Err(ExecError::other(
472                    "PostgreSQL accepts READ UNCOMMITTED and then runs the transaction as \
473                     READ COMMITTED — it has no weaker level. keelson refuses the request \
474                     rather than hand back an isolation level the engine does not \
475                     implement; ask for Isolation::ReadCommitted if that is the behaviour \
476                     you want.",
477                ));
478            }
479            sql.push_str(" ISOLATION LEVEL ");
480            sql.push_str(level.as_sql());
481        }
482        if let Some(mode) = self.access {
483            sql.push(' ');
484            sql.push_str(mode.as_sql());
485        }
486        Ok(vec![sql])
487    }
488
489    /// MySQL cannot carry an isolation level on `START TRANSACTION`, and
490    /// refuses `SET TRANSACTION` once a transaction is open
491    /// (`ER_CANT_CHANGE_TX_CHARACTERISTICS`). Unqualified — no `SESSION`, no
492    /// `GLOBAL` — the `SET` applies to the *next* transaction on this
493    /// connection and to nothing else, which is exactly the scope wanted: it
494    /// expires with the transaction instead of riding the connection back
495    /// into the pool.
496    fn plan_mysql(&self) -> Result<Vec<String>, ExecError> {
497        self.reject_sqlite_begin(Family::MySql)?;
498        let mut out = Vec::with_capacity(2);
499        if let Some(level) = self.isolation {
500            out.push(format!(
501                "SET TRANSACTION ISOLATION LEVEL {}",
502                level.as_sql()
503            ));
504        }
505        out.push(match self.access {
506            // `BEGIN` when nothing is asked, so the default path is
507            // byte-identical to `Transaction::begin_on`'s.
508            None => "BEGIN".to_owned(),
509            Some(mode) => format!("START TRANSACTION {}", mode.as_sql()),
510        });
511        Ok(out)
512    }
513
514    /// SQLite: `BEGIN [ DEFERRED | IMMEDIATE | EXCLUSIVE ]`, and no standard
515    /// levels at all.
516    fn plan_sqlite(&self) -> Result<Vec<String>, ExecError> {
517        if let Some(level) = self.isolation
518            && level != Isolation::Serializable
519        {
520            return Err(ExecError::other(format!(
521                "SQLite has exactly one isolation level — serializable — and cannot be \
522                 weakened to {level}: a transaction asking for it would silently run \
523                 serializable, which is a different set of permitted anomalies from the \
524                 one you asked for. Ask for Isolation::Serializable (SQLite's only level, \
525                 and what a plain BEGIN already gives), or use TxOptions::sqlite_begin for \
526                 SQLite's own DEFERRED / IMMEDIATE / EXCLUSIVE begin modes."
527            )));
528        }
529        if self.access == Some(Access::ReadOnly) {
530            return Err(ExecError::other(
531                "SQLite has no per-transaction read-only mode. `PRAGMA query_only` is \
532                 connection-level state, and keelson will not set connection state behind \
533                 a pooled connection's back; open the database read-only instead \
534                 (`sqlite://file?mode=ro`).",
535            ));
536        }
537        Ok(vec![match self.sqlite_begin {
538            None => "BEGIN".to_owned(),
539            Some(mode) => format!("BEGIN {}", mode.as_sql()),
540        }])
541    }
542
543    fn reject_sqlite_begin(&self, family: Family) -> Result<(), ExecError> {
544        match self.sqlite_begin {
545            None => Ok(()),
546            Some(mode) => Err(ExecError::other(format!(
547                "SqliteBegin::{mode:?} is SQLite's own begin-mode vocabulary and has no \
548                 meaning on {family}; it is refused rather than ignored. Use \
549                 TxOptions::isolation and TxOptions::access there."
550            ))),
551        }
552    }
553}
554
555impl From<Isolation> for TxOptions {
556    fn from(level: Isolation) -> Self {
557        TxOptions::new().isolation(level)
558    }
559}
560
561impl From<Access> for TxOptions {
562    fn from(mode: Access) -> Self {
563        TxOptions::new().access(mode)
564    }
565}
566
567impl From<SqliteBegin> for TxOptions {
568    fn from(mode: SqliteBegin) -> Self {
569        TxOptions::new().sqlite_begin(mode)
570    }
571}
572
573/// A concurrency conflict the engine raised: this transaction lost, and the
574/// only correct response is to run the whole thing again.
575///
576/// The point of the type is that it is *matchable*. Serialization failures
577/// arrive as engine-specific codes on engine-specific error types; without a
578/// classification a caller ends up matching on message text, which is a bug
579/// waiting for a locale or a version bump. A backend that reports one of
580/// these constructs a [`TxConflictError`]; a caller asks [`TxConflict::of`].
581#[non_exhaustive]
582#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
583pub enum TxConflict {
584    /// The engine could not serialize this transaction against a concurrent
585    /// one — PostgreSQL `40001` (`could not serialize access …`).
586    Serialization,
587    /// A deadlock was detected and this transaction was picked as the victim
588    /// — PostgreSQL `40P01`, MySQL `1213` (`ER_LOCK_DEADLOCK`, SQLSTATE
589    /// `40001`: InnoDB reports its serialization failures this way).
590    Deadlock,
591    /// A lock wait timed out — PostgreSQL `55P03`, MySQL `1205`
592    /// (`ER_LOCK_WAIT_TIMEOUT`).
593    LockTimeout,
594    /// SQLite could not take the lock it needed — `SQLITE_BUSY` /
595    /// `SQLITE_LOCKED`, which is how SQLite says "serialize elsewhere".
596    Busy,
597}
598
599impl TxConflict {
600    /// A short stable name, for logs and assertions.
601    pub fn as_str(self) -> &'static str {
602        match self {
603            TxConflict::Serialization => "serialization failure",
604            TxConflict::Deadlock => "deadlock",
605            TxConflict::LockTimeout => "lock timeout",
606            TxConflict::Busy => "database busy",
607        }
608    }
609
610    /// Classify an error: `Some` when a backend reported a concurrency
611    /// conflict, `None` otherwise. Every variant means the same thing to a
612    /// caller — retry the transaction from the top.
613    pub fn of(e: &ExecError) -> Option<TxConflict> {
614        match e {
615            ExecError::Driver(d) => d.downcast_ref::<TxConflictError>().map(|c| c.kind),
616            _ => None,
617        }
618    }
619}
620
621impl fmt::Display for TxConflict {
622    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
623        f.write_str(self.as_str())
624    }
625}
626
627/// The error a backend reports for a [`TxConflict`], carrying the engine's own
628/// code and message and (usually) the driver error as its `source`.
629///
630/// Backend-facing: construct one, then [`TxConflictError::into_exec_error`].
631#[derive(Debug)]
632pub struct TxConflictError {
633    kind: TxConflict,
634    code: String,
635    message: String,
636    source: Option<Box<dyn std::error::Error + Send + Sync>>,
637}
638
639impl TxConflictError {
640    /// A conflict of `kind`, as the engine reported it (`code` is the
641    /// engine's own: a SQLSTATE, an error number, a SQLite result code).
642    pub fn new(kind: TxConflict, code: impl Into<String>, message: impl Into<String>) -> Self {
643        TxConflictError {
644            kind,
645            code: code.into(),
646            message: message.into(),
647            source: None,
648        }
649    }
650
651    /// Keep the driver error underneath, so nothing is lost by classifying.
652    pub fn with_source(mut self, e: impl std::error::Error + Send + Sync + 'static) -> Self {
653        self.source = Some(Box::new(e));
654        self
655    }
656
657    /// Which conflict this is.
658    pub fn kind(&self) -> TxConflict {
659        self.kind
660    }
661
662    /// The engine's own code.
663    pub fn code(&self) -> &str {
664        &self.code
665    }
666
667    /// Wrap into the error the executor traits speak. This is the one
668    /// construction [`TxConflict::of`] recognises.
669    pub fn into_exec_error(self) -> ExecError {
670        ExecError::Driver(Box::new(self))
671    }
672}
673
674impl fmt::Display for TxConflictError {
675    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
676        write!(f, "{} [{}]: {}", self.kind, self.code, self.message)
677    }
678}
679
680impl std::error::Error for TxConflictError {
681    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
682        self.source
683            .as_ref()
684            .map(|e| &**e as &(dyn std::error::Error + 'static))
685    }
686}
687
688/// Something a transaction can be begun on: a pool or a connection.
689///
690/// Deliberately **not** implemented by [`Transaction`] — nesting is spelled
691/// [`Transaction::savepoint`], so "did I open a transaction or a savepoint?"
692/// cannot be confused at a call site.
693pub trait Begin: Executor {
694    /// Open a transaction on a connection this executor owns or checks out.
695    fn begin(&self) -> ExecFuture<'_, Result<Transaction, ExecError>>;
696}
697
698impl<B: Begin + ?Sized> Begin for &B {
699    fn begin(&self) -> ExecFuture<'_, Result<Transaction, ExecError>> {
700        (**self).begin()
701    }
702}
703
704impl<B: Begin + ?Sized> Begin for Arc<B> {
705    fn begin(&self) -> ExecFuture<'_, Result<Transaction, ExecError>> {
706        (**self).begin()
707    }
708}
709
710/// Opt-in transaction options: [`begin_with`](BeginWith::begin_with) beside
711/// [`begin`](Begin::begin).
712///
713/// A separate trait rather than a second method on [`Begin`], following the
714/// [`StreamExecutor`](crate::StreamExecutor) template: the core traits never
715/// grow methods, so a backend that has no answer for isolation levels stays
716/// compiling and stays honest by *not* implementing this.
717///
718/// ```ignore
719/// let tx = db.begin_with(Isolation::Serializable.into()).await?;
720/// let tx = db.begin_with(TxOptions::new().isolation(Isolation::RepeatableRead).read_only()).await?;
721/// ```
722///
723/// Implementing it is three lines: refuse the options for this family
724/// ([`TxOptions::check`]) *before* taking a connection, then hand the
725/// connection to [`Transaction::begin_on_with`], which owns the SQL.
726pub trait BeginWith: Begin {
727    /// Open a transaction with explicit options.
728    ///
729    /// Options this engine cannot honour are an error — never a silent
730    /// downgrade, never a no-op. [`TxOptions::plan`] is the table of what
731    /// each engine accepts and why.
732    fn begin_with(&self, opts: TxOptions) -> ExecFuture<'_, Result<Transaction, ExecError>>;
733}
734
735impl<B: BeginWith + ?Sized> BeginWith for &B {
736    fn begin_with(&self, opts: TxOptions) -> ExecFuture<'_, Result<Transaction, ExecError>> {
737        (**self).begin_with(opts)
738    }
739}
740
741impl<B: BeginWith + ?Sized> BeginWith for Arc<B> {
742    fn begin_with(&self, opts: TxOptions) -> ExecFuture<'_, Result<Transaction, ExecError>> {
743        (**self).begin_with(opts)
744    }
745}
746
747/// [`BeginExt::within`] with options.
748pub trait BeginWithExt: BeginWith {
749    /// Run `f` inside a fresh transaction opened with `opts`.
750    fn within_with<T, E, F>(
751        &self,
752        opts: TxOptions,
753        f: F,
754    ) -> impl std::future::Future<Output = Result<T, E>>
755    where
756        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
757        E: From<ExecError>,
758    {
759        async move {
760            let tx = self.begin_with(opts).await.map_err(E::from)?;
761            match f(&tx).await {
762                Ok(v) => {
763                    tx.commit().await.map_err(E::from)?;
764                    Ok(v)
765                }
766                Err(e) => {
767                    let _ = tx.rollback().await;
768                    Err(e)
769                }
770            }
771        }
772    }
773}
774
775impl<B: BeginWith + ?Sized> BeginWithExt for B {}
776
777/// The closure form of a transaction: commit on `Ok`, roll back on `Err`.
778///
779/// The closure receives `&Transaction` and so *cannot* commit or consume it —
780/// `within` owns the outcome. Neither a dropped transaction nor a forgotten
781/// commit is expressible here, which is why this is the recommended shape.
782///
783/// # `within` or [`Atomic::atomic`]
784///
785/// On a pool they are the same call: `atomic` for a [`Begin`] *is* `within`.
786/// What differs is what the receiver may be, which is to say what the code
787/// claims:
788///
789/// - `within` takes a [`Begin`] and nothing else — **a transaction begins
790///   here**. Handing it a [`Transaction`] does not compile.
791/// - `atomic` takes either — *I do not care which of the two this is, I care
792///   that it is atomic.*
793///
794/// The weaker claim is not the safer default. Anything whose correctness
795/// depends on where the transaction *ends* needs `within`:
796///
797/// - **A retry loop.** Re-running a savepoint cannot clear a serialization
798///   failure — the snapshot is unchanged, so the conflict recurs. A retry
799///   written against `atomic` would spin when its caller happened to be
800///   inside a transaction; written against `within`, that call does not
801///   compile.
802/// - **Isolation and access mode**, which are properties of the outermost
803///   transaction and so are only reachable through [`BeginWith::begin_with`]
804///   and [`BeginWithExt::within_with`].
805///
806/// Rule of thumb: `within` when the extent of the transaction is part of what
807/// the code is saying, `atomic` for a reusable unit of work that only needs
808/// its own block to be all-or-nothing.
809pub trait BeginExt: Begin {
810    /// Run `f` inside a fresh transaction.
811    ///
812    /// ```ignore
813    /// let order = db.within(async |tx| {
814    ///     let order: Order = insert_order.fetch_one(tx).await?;
815    ///     reserve_stock(tx, &order).await?;
816    ///     Ok(order)
817    /// }).await?;
818    /// ```
819    fn within<T, E, F>(&self, f: F) -> impl std::future::Future<Output = Result<T, E>>
820    where
821        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
822        E: From<ExecError>,
823    {
824        async move {
825            let tx = self.begin().await.map_err(E::from)?;
826            match f(&tx).await {
827                Ok(v) => {
828                    tx.commit().await.map_err(E::from)?;
829                    Ok(v)
830                }
831                Err(e) => {
832                    // Best-effort: the closure's error is what the caller
833                    // needs to see even if the rollback also failed.
834                    let _ = tx.rollback().await;
835                    Err(e)
836                }
837            }
838        }
839    }
840}
841
842impl<B: Begin + ?Sized> BeginExt for B {}
843
844/// **All-or-nothing here, wherever "here" turns out to be**: a transaction
845/// when nothing is open, a savepoint when a transaction already is.
846///
847/// This is the trait a *reusable* unit of work is written against. Without it
848/// a helper has to pick, and both choices are wrong somewhere:
849///
850/// ```text
851/// async fn transfer(db: &dyn Executor) -> …   // composes, but cannot be atomic
852/// async fn transfer(tx: &Transaction) -> …    // atomic, but demands its caller open one
853/// async fn transfer(db: impl Atomic) -> …     // both
854/// ```
855///
856/// The third accepts a pool, a connection, a `&dyn Begin` and a
857/// [`Transaction`], and is atomic in all of them:
858///
859/// ```ignore
860/// async fn transfer(db: impl Atomic, from: i64, to: i64) -> Result<(), ExecError> {
861///     db.atomic(async |tx| {
862///         debit(tx, from).await?;
863///         credit(tx, to).await
864///     })
865///     .await
866/// }
867///
868/// transfer(&pool, a, b).await?;                       // BEGIN … COMMIT
869/// pool.within(async |tx| {                            // BEGIN …
870///     transfer(tx, a, b).await?;                      //   SAVEPOINT … RELEASE
871///     audit(tx).await                                 // … COMMIT
872/// })
873/// .await?;
874/// ```
875///
876/// # What "atomic" promises, and what it does not
877///
878/// The *block* is all-or-nothing. What an `Err` discards is not:
879///
880/// - at the top, the block is the transaction, so an error rolls back
881///   everything;
882/// - nested, an error rolls back to the savepoint and **the caller's
883///   transaction survives** — the caller decides whether its own work still
884///   makes sense.
885///
886/// That is the useful reading of a nested unit of work, and it is why this is
887/// not merely a shorthand for "open a transaction if you can".
888///
889/// A **retry loop belongs at the transaction boundary, not here.** Rolling
890/// back to a savepoint does recover a transaction from an error, but a
891/// serialization failure ([`TxConflict`]) will recur against the same
892/// snapshot; only re-running the whole transaction can win. Retry where the
893/// transaction begins.
894///
895/// # How to spell the parameter
896///
897/// `db: impl Atomic`, by value and with no bound beyond the trait. Every
898/// receiver you would want is accepted, because `Atomic` follows [`Executor`]
899/// in being implemented for handles as well as values:
900///
901/// ```text
902/// f(&pool)      f(pool)      f(Arc::clone(&pool))      f(&dyn Begin)      f(tx)
903/// ```
904///
905/// The last is the `&Transaction` a scope closure hands you, which is what
906/// makes these functions nest into each other. Writing `&impl Atomic` instead
907/// would *reject* `&dyn Begin` (a trait object is not `Sized`), so the bare
908/// form is the wider one as well as the shorter one.
909///
910/// # The cost, stated
911///
912/// The method is generic, so `Atomic` is **not object-safe**: `&dyn Atomic`
913/// does not exist and the erased currency stays `&dyn Executor`, which a
914/// scope parameter is passed on as for everything that is not itself a
915/// scope.
916///
917/// [`Begin`] is still deliberately *not* implemented by [`Transaction`], so
918/// `begin`/`within` and [`savepoint`](Transaction::savepoint) stay
919/// distinguishable wherever the distinction matters. `atomic` is how a call
920/// site says the other thing on purpose: *I do not care which of the two this
921/// is; I care that it is atomic.*
922///
923/// There is no `atomic_with`. Isolation level and access mode are properties
924/// of the outermost transaction and cannot be changed by a nested scope, so a
925/// nested `atomic_with` could only ignore them — and keelson does not accept
926/// options it would have to ignore. Ask for them where the transaction is
927/// opened: [`BeginWith::begin_with`] or [`BeginWithExt::within_with`].
928pub trait Atomic: Executor {
929    /// Run `f` as one all-or-nothing block, nesting if the receiver is
930    /// already a transaction.
931    fn atomic<T, E, F>(&self, f: F) -> impl std::future::Future<Output = Result<T, E>>
932    where
933        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
934        E: From<ExecError>;
935}
936
937/// Anything a transaction can be begun on: `atomic` *is* [`BeginExt::within`].
938impl<B: Begin + ?Sized> Atomic for B {
939    fn atomic<T, E, F>(&self, f: F) -> impl std::future::Future<Output = Result<T, E>>
940    where
941        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
942        E: From<ExecError>,
943    {
944        self.within(f)
945    }
946}
947
948/// Inside a transaction: `atomic` *is* [`Transaction::savepoint`].
949///
950/// Not an overlapping impl, and not by luck: [`Transaction`] does not
951/// implement [`Begin`], which is the design decision above holding the two
952/// impls apart.
953impl Atomic for Transaction {
954    fn atomic<T, E, F>(&self, f: F) -> impl std::future::Future<Output = Result<T, E>>
955    where
956        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
957        E: From<ExecError>,
958    {
959        self.savepoint(f)
960    }
961}
962
963/// A *reference* to a transaction — what makes the bare `impl Atomic`
964/// parameter form work, since that is the type a scope closure hands you.
965///
966/// [`Executor`] is implemented for `&E` and `Arc<E>` for the same reason: a
967/// caller should not have to know whether what it holds is the value or a
968/// handle to it. Coherent with the blanket impl for the same reason
969/// [`Transaction`]'s own impl is — `&Transaction` is not [`Begin`] either.
970impl Atomic for &Transaction {
971    fn atomic<T, E, F>(&self, f: F) -> impl std::future::Future<Output = Result<T, E>>
972    where
973        F: AsyncFnOnce(&Transaction) -> Result<T, E>,
974        E: From<ExecError>,
975    {
976        (**self).savepoint(f)
977    }
978}
979
980/// The hook payload the execution layer fixes
981/// [`QueryExtensions`](keelson_core::QueryExtensions)' `Hook` parameter to:
982/// an async function of `&dyn Executor`.
983///
984/// A hook receives exactly the executor the caller passed in — so it runs
985/// inside the caller's transaction when there is one — and it receives it as
986/// `&dyn Executor`, not `&Transaction`, so a hook *cannot* end a transaction
987/// it did not open.
988pub type ExecHook =
989    Arc<dyn for<'a> Fn(&'a dyn Executor) -> ExecFuture<'a, Result<(), ExecError>> + Send + Sync>;
990
991/// The loader payload: like [`ExecHook`], plus the rows the query produced.
992pub type ExecLoader = Arc<
993    dyn for<'a> Fn(&'a dyn Executor, &'a [Row]) -> ExecFuture<'a, Result<(), ExecError>>
994        + Send
995        + Sync,
996>;
997
998#[cfg(test)]
999mod tests {
1000    use std::sync::Mutex as StdMutex;
1001
1002    use super::*;
1003
1004    /// A RawConnection that records every statement and whether it was
1005    /// abandoned — transaction semantics are testable without a database.
1006    #[derive(Debug)]
1007    struct Script {
1008        family: Family,
1009        log: Arc<StdMutex<Vec<String>>>,
1010        abandoned: Arc<StdMutex<bool>>,
1011        fail_from: Option<usize>,
1012    }
1013
1014    impl RawConnection for Script {
1015        fn family(&self) -> Family {
1016            self.family
1017        }
1018
1019        fn fetch<'a>(
1020            &'a mut self,
1021            sql: &'a str,
1022            _args: Vec<Value>,
1023        ) -> ExecFuture<'a, Result<Vec<Row>, ExecError>> {
1024            self.log.lock().unwrap().push(sql.to_owned());
1025            Box::pin(async { Ok(Vec::new()) })
1026        }
1027
1028        fn execute<'a>(
1029            &'a mut self,
1030            sql: &'a str,
1031            _args: Vec<Value>,
1032        ) -> ExecFuture<'a, Result<ExecResult, ExecError>> {
1033            let n = {
1034                let mut log = self.log.lock().unwrap();
1035                log.push(sql.to_owned());
1036                log.len()
1037            };
1038            let fails = self.fail_from.is_some_and(|from| n > from);
1039            Box::pin(async move {
1040                if fails {
1041                    Err(ExecError::other("statement refused"))
1042                } else {
1043                    Ok(ExecResult::default())
1044                }
1045            })
1046        }
1047
1048        fn abandon(self: Box<Self>) {
1049            *self.abandoned.lock().unwrap() = true;
1050        }
1051    }
1052
1053    type Log = Arc<StdMutex<Vec<String>>>;
1054    type Abandoned = Arc<StdMutex<bool>>;
1055
1056    fn script_of(family: Family, fail_from: Option<usize>) -> (Script, Log, Abandoned) {
1057        let log: Log = Arc::default();
1058        let abandoned: Abandoned = Arc::default();
1059        (
1060            Script {
1061                family,
1062                log: log.clone(),
1063                abandoned: abandoned.clone(),
1064                fail_from,
1065            },
1066            log,
1067            abandoned,
1068        )
1069    }
1070
1071    fn script() -> (Script, Log, Abandoned) {
1072        script_of(Family::Sqlite, None)
1073    }
1074
1075    #[tokio::test]
1076    async fn commit_speaks_begin_then_commit_and_keeps_the_connection() {
1077        let (conn, log, abandoned) = script();
1078        let tx = Transaction::begin_on(Box::new(conn)).await.unwrap();
1079        tx.execute(Statement::new("INSERT 1", vec![]))
1080            .await
1081            .unwrap();
1082        tx.commit().await.unwrap();
1083        assert_eq!(*log.lock().unwrap(), vec!["BEGIN", "INSERT 1", "COMMIT"]);
1084        assert!(!*abandoned.lock().unwrap());
1085    }
1086
1087    #[tokio::test]
1088    async fn drop_without_commit_abandons_the_connection() {
1089        let (conn, log, abandoned) = script();
1090        let tx = Transaction::begin_on(Box::new(conn)).await.unwrap();
1091        drop(tx);
1092        assert_eq!(*log.lock().unwrap(), vec!["BEGIN"]);
1093        assert!(
1094            *abandoned.lock().unwrap(),
1095            "the connection must not be reused"
1096        );
1097    }
1098
1099    /// A pool that hands out its one scripted connection: enough to have a
1100    /// `Begin` on this side of the driver seam.
1101    #[derive(Debug)]
1102    struct Handle(StdMutex<Option<Box<dyn RawConnection>>>);
1103
1104    impl Executor for Handle {
1105        fn family(&self) -> Family {
1106            Family::Sqlite
1107        }
1108
1109        fn fetch(&self, _: Statement) -> ExecFuture<'_, Result<Vec<Row>, ExecError>> {
1110            Box::pin(async { Err(ExecError::other("not the point of this fixture")) })
1111        }
1112
1113        fn execute(&self, _: Statement) -> ExecFuture<'_, Result<ExecResult, ExecError>> {
1114            Box::pin(async { Err(ExecError::other("not the point of this fixture")) })
1115        }
1116    }
1117
1118    impl Begin for Handle {
1119        fn begin(&self) -> ExecFuture<'_, Result<Transaction, ExecError>> {
1120            let conn = self.0.lock().unwrap().take();
1121            Box::pin(async move {
1122                Transaction::begin_on(conn.ok_or_else(|| ExecError::other("checked out twice"))?)
1123                    .await
1124            })
1125        }
1126    }
1127
1128    /// One helper, written once, atomic at both levels — the whole point of
1129    /// [`Atomic`].
1130    async fn unit_of_work(db: impl Atomic, fail: bool) -> Result<(), ExecError> {
1131        db.atomic(async |tx| {
1132            tx.execute(Statement::new("WORK", vec![])).await?;
1133            if fail {
1134                return Err(ExecError::other("no"));
1135            }
1136            Ok(())
1137        })
1138        .await
1139    }
1140
1141    #[tokio::test]
1142    async fn one_helper_is_a_transaction_at_the_top_and_a_savepoint_inside_one() {
1143        let (conn, log, _) = script();
1144        let pool = Handle(StdMutex::new(Some(Box::new(conn))));
1145        unit_of_work(&pool, false).await.unwrap();
1146        assert_eq!(*log.lock().unwrap(), vec!["BEGIN", "WORK", "COMMIT"]);
1147
1148        let (conn, log, _) = script();
1149        let tx = Transaction::begin_on(Box::new(conn)).await.unwrap();
1150        unit_of_work(&tx, false).await.unwrap();
1151        tx.commit().await.unwrap();
1152        assert_eq!(
1153            *log.lock().unwrap(),
1154            vec![
1155                "BEGIN",
1156                "SAVEPOINT keelson_sp_1",
1157                "WORK",
1158                "RELEASE SAVEPOINT keelson_sp_1",
1159                "COMMIT",
1160            ]
1161        );
1162    }
1163
1164    #[tokio::test]
1165    async fn a_nested_failure_costs_the_block_and_not_the_callers_transaction() {
1166        let (conn, log, _) = script();
1167        let tx = Transaction::begin_on(Box::new(conn)).await.unwrap();
1168        let err = unit_of_work(&tx, true).await.unwrap_err();
1169        assert_eq!(err.to_string(), "no");
1170        // The caller's transaction is still usable, and still commits.
1171        tx.execute(Statement::new("AFTER", vec![])).await.unwrap();
1172        tx.commit().await.unwrap();
1173        assert_eq!(
1174            *log.lock().unwrap(),
1175            vec![
1176                "BEGIN",
1177                "SAVEPOINT keelson_sp_1",
1178                "WORK",
1179                "ROLLBACK TO SAVEPOINT keelson_sp_1",
1180                "RELEASE SAVEPOINT keelson_sp_1",
1181                "AFTER",
1182                "COMMIT",
1183            ]
1184        );
1185    }
1186
1187    #[tokio::test]
1188    async fn savepoints_release_on_ok_and_roll_back_on_err() {
1189        let (conn, log, _) = script();
1190        let tx = Transaction::begin_on(Box::new(conn)).await.unwrap();
1191
1192        tx.savepoint(async |sp| {
1193            sp.execute(Statement::new("GOOD", vec![])).await?;
1194            Ok::<_, ExecError>(())
1195        })
1196        .await
1197        .unwrap();
1198
1199        let err = tx
1200            .savepoint(async |sp| {
1201                sp.execute(Statement::new("BAD", vec![])).await?;
1202                Err::<(), _>(ExecError::other("boom"))
1203            })
1204            .await
1205            .unwrap_err();
1206        assert_eq!(err.to_string(), "boom");
1207
1208        // Nested: the inner savepoint numbers deeper.
1209        tx.savepoint(async |sp| {
1210            sp.savepoint(async |sp2| {
1211                sp2.execute(Statement::new("DEEP", vec![])).await?;
1212                Ok::<_, ExecError>(())
1213            })
1214            .await
1215        })
1216        .await
1217        .unwrap();
1218
1219        tx.commit().await.unwrap();
1220        assert_eq!(
1221            *log.lock().unwrap(),
1222            vec![
1223                "BEGIN",
1224                "SAVEPOINT keelson_sp_1",
1225                "GOOD",
1226                "RELEASE SAVEPOINT keelson_sp_1",
1227                "SAVEPOINT keelson_sp_1",
1228                "BAD",
1229                "ROLLBACK TO SAVEPOINT keelson_sp_1",
1230                "RELEASE SAVEPOINT keelson_sp_1",
1231                "SAVEPOINT keelson_sp_1",
1232                "SAVEPOINT keelson_sp_2",
1233                "DEEP",
1234                "RELEASE SAVEPOINT keelson_sp_2",
1235                "RELEASE SAVEPOINT keelson_sp_1",
1236                "COMMIT",
1237            ]
1238        );
1239    }
1240
1241    // ---- transaction options -------------------------------------------
1242    //
1243    // The statement text is derived from each engine's own grammar:
1244    // PostgreSQL `BEGIN [ transaction_mode … ]`, MySQL `SET TRANSACTION
1245    // transaction_characteristic` + `START TRANSACTION [ transaction_
1246    // characteristic ]`, SQLite `BEGIN [ DEFERRED | IMMEDIATE | EXCLUSIVE ]`.
1247    // What these tests pin is the *shape and order*; that the engines accept
1248    // it, and that it changes their behaviour, is proved against real servers
1249    // in keelson-sqlx's tests/transactions.rs.
1250
1251    #[test]
1252    fn default_options_send_exactly_what_a_plain_begin_sends() {
1253        for family in [Family::Postgres, Family::MySql, Family::Sqlite] {
1254            assert_eq!(TxOptions::new().plan(family).unwrap(), vec!["BEGIN"]);
1255        }
1256    }
1257
1258    #[test]
1259    fn postgres_puts_the_modes_on_begin_itself() {
1260        assert_eq!(
1261            TxOptions::from(Isolation::Serializable)
1262                .plan(Family::Postgres)
1263                .unwrap(),
1264            vec!["BEGIN ISOLATION LEVEL SERIALIZABLE"]
1265        );
1266        assert_eq!(
1267            TxOptions::new()
1268                .isolation(Isolation::RepeatableRead)
1269                .read_only()
1270                .plan(Family::Postgres)
1271                .unwrap(),
1272            vec!["BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY"]
1273        );
1274    }
1275
1276    #[test]
1277    fn mysql_sets_the_level_first_then_starts() {
1278        assert_eq!(
1279            TxOptions::from(Isolation::ReadCommitted)
1280                .plan(Family::MySql)
1281                .unwrap(),
1282            vec!["SET TRANSACTION ISOLATION LEVEL READ COMMITTED", "BEGIN"]
1283        );
1284        assert_eq!(
1285            TxOptions::new()
1286                .isolation(Isolation::Serializable)
1287                .read_only()
1288                .plan(Family::MySql)
1289                .unwrap(),
1290            vec![
1291                "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE",
1292                "START TRANSACTION READ ONLY",
1293            ]
1294        );
1295    }
1296
1297    #[test]
1298    fn sqlite_gets_begin_modes_and_nothing_pretending_to_be_a_level() {
1299        assert_eq!(
1300            TxOptions::from(SqliteBegin::Immediate)
1301                .plan(Family::Sqlite)
1302                .unwrap(),
1303            vec!["BEGIN IMMEDIATE"]
1304        );
1305        // Serializable is accepted because it is literally what SQLite runs.
1306        assert_eq!(
1307            TxOptions::from(Isolation::Serializable)
1308                .plan(Family::Sqlite)
1309                .unwrap(),
1310            vec!["BEGIN"]
1311        );
1312    }
1313
1314    #[test]
1315    fn every_level_an_engine_would_only_pretend_to_honour_is_refused() {
1316        // PostgreSQL accepts READ UNCOMMITTED and runs READ COMMITTED.
1317        let e = TxOptions::from(Isolation::ReadUncommitted)
1318            .check(Family::Postgres)
1319            .unwrap_err()
1320            .to_string();
1321        assert!(e.contains("READ UNCOMMITTED"), "{e}");
1322        assert!(e.contains("READ COMMITTED"), "{e}");
1323
1324        // SQLite cannot weaken below serializable.
1325        for level in [
1326            Isolation::ReadUncommitted,
1327            Isolation::ReadCommitted,
1328            Isolation::RepeatableRead,
1329        ] {
1330            let e = TxOptions::from(level)
1331                .check(Family::Sqlite)
1332                .unwrap_err()
1333                .to_string();
1334            assert!(e.contains(level.as_sql()), "{e}");
1335            assert!(e.contains("sqlite_begin"), "{e}");
1336        }
1337
1338        // SQLite has no per-transaction read-only mode.
1339        let e = TxOptions::new()
1340            .read_only()
1341            .check(Family::Sqlite)
1342            .unwrap_err()
1343            .to_string();
1344        assert!(e.contains("query_only"), "{e}");
1345
1346        // SQLite's begin modes are not portable vocabulary.
1347        for family in [Family::Postgres, Family::MySql] {
1348            let e = TxOptions::from(SqliteBegin::Exclusive)
1349                .check(family)
1350                .unwrap_err()
1351                .to_string();
1352            assert!(e.contains("Exclusive"), "{e}");
1353            assert!(e.contains(family.as_str()), "{e}");
1354        }
1355
1356        // MySQL is the one engine that implements all four.
1357        for level in [
1358            Isolation::ReadUncommitted,
1359            Isolation::ReadCommitted,
1360            Isolation::RepeatableRead,
1361            Isolation::Serializable,
1362        ] {
1363            TxOptions::from(level).check(Family::MySql).unwrap();
1364        }
1365    }
1366
1367    #[tokio::test]
1368    async fn a_refused_option_never_reaches_the_wire_and_keeps_the_connection() {
1369        let (conn, log, abandoned) = script_of(Family::Sqlite, None);
1370        let err = Transaction::begin_on_with(Box::new(conn), Isolation::ReadCommitted.into())
1371            .await
1372            .unwrap_err();
1373        assert!(err.to_string().contains("one isolation level"));
1374        assert!(log.lock().unwrap().is_empty(), "nothing may be sent");
1375        assert!(
1376            !*abandoned.lock().unwrap(),
1377            "an untouched connection goes back to its pool"
1378        );
1379    }
1380
1381    #[tokio::test]
1382    async fn the_whole_plan_runs_on_the_transactions_own_connection_before_any_statement() {
1383        let (conn, log, _) = script_of(Family::MySql, None);
1384        let tx = Transaction::begin_on_with(
1385            Box::new(conn),
1386            TxOptions::new()
1387                .isolation(Isolation::Serializable)
1388                .read_only(),
1389        )
1390        .await
1391        .unwrap();
1392        assert_eq!(
1393            tx.options(),
1394            TxOptions::new()
1395                .isolation(Isolation::Serializable)
1396                .access(Access::ReadOnly)
1397        );
1398        tx.execute(Statement::new("SELECT 1", vec![]))
1399            .await
1400            .unwrap();
1401        tx.commit().await.unwrap();
1402        // One connection, one log: the level is set on it, ahead of the
1403        // caller's first statement, and inside the same checkout.
1404        assert_eq!(
1405            *log.lock().unwrap(),
1406            vec![
1407                "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE",
1408                "START TRANSACTION READ ONLY",
1409                "SELECT 1",
1410                "COMMIT",
1411            ]
1412        );
1413    }
1414
1415    #[tokio::test]
1416    async fn a_half_applied_plan_abandons_the_connection() {
1417        // MySQL's SET lands, START TRANSACTION fails: the pending
1418        // next-transaction characteristic would otherwise ride this
1419        // connection back into the pool.
1420        let (conn, log, abandoned) = script_of(Family::MySql, Some(1));
1421        let err = Transaction::begin_on_with(Box::new(conn), Isolation::Serializable.into())
1422            .await
1423            .unwrap_err();
1424        assert_eq!(err.to_string(), "statement refused");
1425        assert_eq!(
1426            *log.lock().unwrap(),
1427            vec!["SET TRANSACTION ISOLATION LEVEL SERIALIZABLE", "BEGIN"]
1428        );
1429        assert!(
1430            *abandoned.lock().unwrap(),
1431            "the connection carries state nobody can see; it must not be reused"
1432        );
1433    }
1434
1435    #[test]
1436    fn conflicts_are_matchable_rather_than_stringly_typed() {
1437        let e = TxConflictError::new(
1438            TxConflict::Serialization,
1439            "40001",
1440            "could not serialize access due to concurrent update",
1441        )
1442        .with_source(ExecError::other("driver"))
1443        .into_exec_error();
1444        assert_eq!(TxConflict::of(&e), Some(TxConflict::Serialization));
1445        assert!(e.to_string().contains("40001"), "{e}");
1446        assert!(std::error::Error::source(&e).is_some());
1447        assert_eq!(TxConflict::of(&ExecError::RowNotFound), None);
1448    }
1449
1450    #[test]
1451    fn every_way_of_holding_a_scope_is_atomic() {
1452        // Checked at compile time: this is what lets a unit of work say
1453        // `db: impl Atomic` and mean "a pool or a transaction, however you
1454        // happen to be holding it".
1455        fn takes(_: impl Atomic) {}
1456        fn prove(pool: Handle, shared: Arc<Handle>, erased: &dyn Begin, tx: &Transaction) {
1457            takes(&pool);
1458            takes(shared);
1459            takes(erased);
1460            // The one a scope closure hands you, and the reason `&Transaction`
1461            // has an impl of its own.
1462            takes(tx);
1463            takes(pool);
1464        }
1465        let _ = prove;
1466    }
1467
1468    #[test]
1469    fn a_transaction_is_a_dyn_executor() {
1470        // The design's center of gravity, checked at compile time: a hook
1471        // signature accepts a transaction as a plain &dyn Executor.
1472        fn takes(_: &dyn Executor) {}
1473        fn prove(tx: &Transaction) {
1474            takes(tx);
1475        }
1476        let _ = prove;
1477    }
1478}