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}