第15章 Transaction：Drop 保护、savepoint、嵌套

Transaction 只有两个字段：一条连接，和一个 open 标志。事务的 RAII 保护为什么只能做到这个程度——答案是 Drop 不能 await。

15.1 问题引入：Rust async 里的事务挑战

事务是 SQL 世界的原子性保证——一组操作要么全做要么全不做。经典用法：

let mut tx = pool.begin().await?;
sqlx::query("UPDATE accounts SET balance = balance - $1 WHERE id = $2")
    .bind(100).bind(from_id).execute(&mut *tx).await?;
sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
    .bind(100).bind(to_id).execute(&mut *tx).await?;
tx.commit().await?;

两条 UPDATE 在同一事务里——中间任何一步失败（第二条 UPDATE 抛错、? 返回 Err），整个事务要回滚，钱不会凭空消失。

问题是：? 返回 Err 后 tx 变量离开作用域被 drop——Rust 期望 drop 里发 ROLLBACK 给数据库。但 Rust 的 Drop::drop 是同步函数——不能 .await。而"发 ROLLBACK 消息给数据库"是异步 I/O（写 socket + 等响应）——drop 里做不完。

这个矛盾是 sqlx Transaction 设计的根本挑战。sqlx 的对策是两层 API：

1. 显式 commit/rollback 是 async——用户 .await 它们保证消息发出、响应接收。
2. Drop 调同步 start_rollback——Postgres / MySQL 驱动只把 ROLLBACK 查询排进 write buffer、不发送，真正发送要等下次这条连接被使用时（SQLite 驱动的做法不同，见 §15.6.2）。

这是一个"prefer explicit, fall back to best-effort"的设计——用户如果记得 commit/rollback 就一切正常；如果忘了、Drop 也尽可能把事务关掉（但不保证立即）。

本章拆开这条设计的每一处细节——TransactionManager trait、三条 ANSI SQL helper、Transaction struct、Drop 实现、嵌套 savepoint——看 sqlx 如何在 Rust async 限制下把事务 API 做得"尽可能安全"。

15.2 TransactionManager trait：驱动层的抽象

sqlx-core/src/transaction.rs:15-50 的 trait：

#[doc(hidden)]
pub trait TransactionManager {
    type Database: Database;

    fn begin<'conn>(
        conn: &'conn mut <Self::Database as Database>::Connection,
        statement: Option<Cow<'static, str>>,
    ) -> BoxFuture<'conn, Result<(), Error>>;

    fn commit(conn: &mut <Self::Database as Database>::Connection) -> BoxFuture<'_, Result<(), Error>>;
    fn rollback(conn: &mut <Self::Database as Database>::Connection) -> BoxFuture<'_, Result<(), Error>>;

    fn start_rollback(conn: &mut <Self::Database as Database>::Connection);

    fn get_transaction_depth(conn: &<Self::Database as Database>::Connection) -> usize;
}

五个方法：

• begin(conn, statement) async——开事务或进 savepoint。
• commit(conn) async——提交事务或 release savepoint。
• rollback(conn) async——回滚事务或 rollback to savepoint。
• start_rollback(conn) sync——只排队 ROLLBACK 查询，不等执行。
• get_transaction_depth(conn)——查当前 nesting 层级（0 = 无事务、1 = 事务中、>1 = savepoint 嵌套）。

这是 0.8.6 的形状。0.8.3 及以前 trait 只有四个方法：begin(conn) 不带 statement 参数，也没有 get_transaction_depth；两者都来自 0.8.4 的 #3765（合并了 #3427 的 get_transaction_depth 和 #3614 的 begin_with，CHANGELOG 0.8.4 条目）。

#[doc(hidden)] 标注——普通用户不调这些方法，通过 Transaction 类型间接用。trait 是驱动层内部抽象。

没有方法接收 self——所有方法都是 type-associated，通过 &mut Connection 操作。TransactionManager 本身是零大小 marker 类型（PgTransactionManager 是 pub struct PgTransactionManager; 空 struct）——不持状态，只提供方法集。

15.2.1 begin 的 statement 参数

begin(conn, statement: Option<Cow<'static, str>>) 的 statement 允许用户自定义 BEGIN 语句：

• None：用默认 ANSI BEGIN。
• Some("BEGIN ISOLATION LEVEL SERIALIZABLE")：自定义（只对 depth = 0 有效）。这是 Postgres 写法；MySQL 的 BEGIN 不接受隔离级别子句，要事先 SET TRANSACTION ISOLATION LEVEL ...。

第 12 章 §12.5 讨论过的 Connection::begin_with（以及 Pool::begin_with，同样 0.8.4 起才有）就走这条——传 Some(statement) 给 TransactionManager::begin。驱动执行完语句后会检查连接是否真的进了事务，没有就返回 Error::BeginFailed。嵌套事务（depth > 0）不能自定义 statement——sqlx 强制用 SAVEPOINT，传 Some 时返回 Error::InvalidSavePointStatement。

15.3 Transaction struct 的简洁性

sqlx-core/src/transaction.rs:86-92：

pub struct Transaction<'c, DB>
where DB: Database,
{
    connection: MaybePoolConnection<'c, DB>,
    open: bool,
}

只有两个字段：

• connection: MaybePoolConnection<'c, DB>——要么是 &mut Connection（从 Connection::begin 来）、要么是 owned PoolConnection（从 Pool::begin 来）。统一抽象让 Transaction 对两种来源都工作。
• open: bool——标记事务是否还活着。true 表示事务中、false 表示已 commit / rollback。

MaybePoolConnection<'c, DB>（sqlx-core/src/pool/maybe.rs）简化：

pub enum MaybePoolConnection<'c, DB: Database> {
    Connection(&'c mut DB::Connection),  // 来自 Connection::begin
    PoolConnection(PoolConnection<DB>),   // 来自 Pool::begin
}

两种来源通过枚举统一——Transaction 不关心具体来源、只管事务语义。这让 Transaction<'c, DB> 的 'c 生命周期根据来源不同而不同：

• Connection-originated：'c = 被借用的 Connection 的生命周期。
• Pool-originated：'c = 'static——PoolConnection owned 不借用外部。

15.3.1 open: bool 的作用

open 字段跟踪事务状态。只在两处被修改：

• Self::commit 成功后：self.open = false（sqlx-core/src/transaction.rs:118）。
• Self::rollback 成功后：self.open = false（sqlx-core/src/transaction.rs:126）。

Drop 里检查 if self.open——只有还在开着的事务才调 start_rollback：

impl<'c, DB> Drop for Transaction<'c, DB> {
    fn drop(&mut self) {
        if self.open {
            DB::TransactionManager::start_rollback(&mut self.connection);
        }
    }
}

这条简单的标志让 "已 commit/rollback 的事务 drop 不做任何事" 成立——避免重复回滚。

15.4 begin 的流程：ANSI BEGIN 或 SAVEPOINT

Transaction::begin（sqlx-core/src/transaction.rs:98-113）：

pub fn begin(
    conn: impl Into<MaybePoolConnection<'c, DB>>,
    statement: Option<Cow<'static, str>>,
) -> BoxFuture<'c, Result<Self, Error>> {
    let mut conn = conn.into();
    Box::pin(async move {
        DB::TransactionManager::begin(&mut conn, statement).await?;
        Ok(Self { connection: conn, open: true })
    })
}

就是调 TransactionManager::begin + 构造 Self。具体的 BEGIN 语句在 PgTransactionManager::begin（sqlx-postgres/src/transaction.rs:18-43）里决定：

fn begin<'conn>(conn: &'conn mut PgConnection, statement: Option<Cow<'static, str>>) -> BoxFuture<...> {
    Box::pin(async move {
        let depth = conn.inner.transaction_depth;
        let statement = match statement {
            Some(_) if depth > 0 => return Err(Error::InvalidSavePointStatement),
            Some(statement) => statement,
            None => begin_ansi_transaction_sql(depth),
        };

        let rollback = Rollback::new(conn);
        rollback.conn.queue_simple_query(&statement)?;
        rollback.conn.wait_until_ready().await?;
        if !rollback.conn.in_transaction() {
            return Err(Error::BeginFailed);
        }
        rollback.conn.inner.transaction_depth += 1;
        rollback.defuse();
        Ok(())
    })
}

核心流程：

1. 读当前 depth——决定是 BEGIN 还是 SAVEPOINT。
2. 按 depth 选 SQL：
   • depth = 0 + 无自定义 statement：BEGIN。
   • depth = 0 + 有自定义：用户 statement（BEGIN ISOLATION LEVEL SERIALIZABLE 等）。
   • depth > 0 + 自定义：拒绝（InvalidSavePointStatement）。
   • depth > 0：SAVEPOINT _sqlx_savepoint_N。
3. 新建 Rollback guard——一个内部 RAII，如果 begin 过程中异常 drop、自动 start_rollback。
4. 发 simple query——把 BEGIN/SAVEPOINT 发给服务端。
5. 等 ReadyForQuery——wait_until_ready 先 flush，再读到 ReadyForQuery 为止，其中带回服务端的事务状态。
6. 检查 in_transaction()——事务状态不是"事务中"（比如自定义语句根本不是 BEGIN）就返回 Error::BeginFailed。
7. transaction_depth += 1。
8. defuse Rollback guard——成功了不需要 guard 的兜底。

15.4.1 begin_ansi_transaction_sql helper

sqlx-core/src/transaction.rs:277-283：

pub fn begin_ansi_transaction_sql(depth: usize) -> Cow<'static, str> {
    if depth == 0 {
        Cow::Borrowed("BEGIN")
    } else {
        Cow::Owned(format!("SAVEPOINT _sqlx_savepoint_{depth}"))
    }
}

depth = 0：BEGIN（顶层事务）。 depth = 1：SAVEPOINT _sqlx_savepoint_1（第一个嵌套 savepoint）。 depth = 2：SAVEPOINT _sqlx_savepoint_2。

savepoint 命名用连续数字——第 N 层 savepoint 叫 _sqlx_savepoint_N。这样 commit / rollback 对应的 RELEASE / ROLLBACK TO 也能按 depth 反算名字。

返回 Cow<'static, str> 的巧妙：depth = 0 时是 "BEGIN" 字符串字面量（借用）、depth > 0 时是动态格式化 String（owned）。Cow 让两种情况共用签名——常见字面量零分配、少见情况按需分配。

15.4.2 Rollback guard

PgTransactionManager::begin 里的 Rollback 是个内部 RAII：

struct Rollback<'c> {
    conn: &'c mut PgConnection,
    defuse: bool,
}

impl Drop for Rollback<'_> {
    fn drop(&mut self) {
        if !self.defuse {
            PgTransactionManager::start_rollback(self.conn)
        }
    }
}

作用：begin 的 future 在 .await? 处出错返回、中途被取消（future 被 drop），或者因 BeginFailed 提前返回时，guard 没被 defuse，drop 时就调 start_rollback。

但要注意 0.8.6 里的顺序：transaction_depth += 1 排在 wait_until_ready().await 和 in_transaction() 检查之后，guard 触发时 depth 还是旧值。于是顶层 BEGIN（depth = 0）失败时 start_rollback 什么都不做；嵌套 SAVEPOINT 失败时，它回滚的是外面那一层（depth = 1 时发 ROLLBACK，更深时发 ROLLBACK TO SAVEPOINT _sqlx_savepoint_{depth-1}），并把 depth 减一。0.8.3 的写法是先 += 1 再等待，guard 回滚的正是刚开的这一层；#3765 把自增挪到了等待和检查之后。

rollback.defuse() 只在 begin 完全成功后调——cancel 掉自动 rollback、因为 Transaction 本身接管了事务管理责任。

这是一个经典 RAII 模式——"部分成功的错误路径"用 guard 兜底。第 14 章的 DecrementSizeGuard 也是同一种设计。

15.5 commit / rollback：async 完整路径

sqlx-core/src/transaction.rs:115-129：

pub async fn commit(mut self) -> Result<(), Error> {
    DB::TransactionManager::commit(&mut self.connection).await?;
    self.open = false;
    Ok(())
}

pub async fn rollback(mut self) -> Result<(), Error> {
    DB::TransactionManager::rollback(&mut self.connection).await?;
    self.open = false;
    Ok(())
}

两个方法的结构一样——调 TransactionManager 的 commit/rollback + 设置 open = false。

关键是吃 self——commit/rollback 消费整个 Transaction 对象。.await? 之后 self 要么被消费（Ok）要么错误冒泡（Err）——两条路径之后 Transaction 都不可再用。

注意 如果 commit 的 .await? 失败——open 保持 true、Drop 会触发 start_rollback。commit 失败不意味着"事务还活着"——它意味着"客户端不确定事务状态"（网络可能在 COMMIT 消息发出后断了）——rollback 是保守兜底。

15.5.1 PgTransactionManager::commit 的实现

sqlx-postgres/src/transaction.rs:45-56：

fn commit(conn: &mut PgConnection) -> BoxFuture<'_, Result<(), Error>> {
    Box::pin(async move {
        if conn.inner.transaction_depth > 0 {
            conn.execute(&*commit_ansi_transaction_sql(conn.inner.transaction_depth))
                .await?;
            conn.inner.transaction_depth -= 1;
        }
        Ok(())
    })
}

三步：

1. 检查 depth：只在 depth > 0 时执行（否则无事务可 commit）。
2. 发 commit SQL：COMMIT 或 RELEASE SAVEPOINT _sqlx_savepoint_N。
3. depth -= 1。

commit_ansi_transaction_sql（sqlx-core/src/transaction.rs:285-291）：

pub fn commit_ansi_transaction_sql(depth: usize) -> Cow<'static, str> {
    if depth == 1 {
        Cow::Borrowed("COMMIT")
    } else {
        Cow::Owned(format!("RELEASE SAVEPOINT _sqlx_savepoint_{}", depth - 1))
    }
}

• depth = 1：COMMIT（关闭顶层事务）。
• depth > 1：RELEASE SAVEPOINT _sqlx_savepoint_{depth-1}（关闭嵌套的 savepoint）。

注意 depth - 1——命名用 savepoint 建立时的 depth。举例：

• depth = 0 → begin BEGIN（depth 变 1）。
• depth = 1 → begin SAVEPOINT _sqlx_savepoint_1（depth 变 2）。
• commit at depth = 2 → RELEASE SAVEPOINT _sqlx_savepoint_1（depth 变 1）。

这条命名和 release 的反向对称是 savepoint 语义的一部分——让 commit/rollback 能按 depth 反推到正确的 savepoint 名字。

15.6 start_rollback：同步路径

PgTransactionManager::start_rollback（sqlx-postgres/src/transaction.rs:73-80）：

fn start_rollback(conn: &mut PgConnection) {
    if conn.inner.transaction_depth > 0 {
        conn.queue_simple_query(&rollback_ansi_transaction_sql(conn.inner.transaction_depth))
            .expect("BUG: Rollback query somehow too large for protocol");
        conn.inner.transaction_depth -= 1;
    }
}

同步函数——不 .await。做的事：

1. 检查 depth。
2. queue_simple_query——把 ROLLBACK SQL 写进连接的 write buffer（不 flush、不发送）。
3. depth -= 1——逻辑上事务已经"准备" rollback。

expect("BUG: ...")——queue_simple_query 只有在编码消息时出错才返回 Err：缓冲区 try_reserve 分配失败，或整条消息长度放不进协议的 i32 长度前缀（sqlx-postgres/src/io/buf_mut.rs:29-31 的 i32::try_from(size)）。一条 ROLLBACK 不可能触发、panic 是合理的。

15.6.1 为什么"只排队不发送"？

queue_simple_query 往 write buffer 写字节——不调 flush。什么时候真正发送？

1. 下次用这条连接执行 query 时——新 query 写进 buffer 后、flush 会把之前的 ROLLBACK 一起发。
2. 连接归还 Pool 时——Transaction 析构后紧接着析构它持有的 PoolConnection，后者的 Drop（sqlx-core/src/pool/connection.rs:199-210）spawn 一个 return_to_pool 任务；return_to_pool（sqlx-core/src/pool/connection.rs:275-328）在放回空闲队列前无条件 self.raw.ping().await，源码注释写得很直白："test the connection on-release to ensure it is still viable,"，下一行接着说这一 ping 还要 "flush anything time-sensitive like transaction rollbacks"。Postgres 的 ping 先 write_sync 再 wait_until_ready，后者会把 write buffer 里排着的 ROLLBACK 一并 flush 并等到回应。也就是说 Pool 场景下 ROLLBACK 在归还任务里就被发出去了——spawn 发生在 PoolConnection 这一层，不在 start_rollback 里。
3. 连接直接被 drop——驱动没有为 PgConnection 实现 Drop（grep 三个驱动都没有 impl Drop for *Connection），排队的 ROLLBACK 就随 socket 一起丢弃；不过 TCP 断开时服务端本来就会回滚这条 session 的事务。

延迟发送意味着：从 drop(tx) 到 ROLLBACK 真正抵达服务端之间有一段窗口。在这段间隔里，服务端视角下事务还开着——持有锁、持有行版本。

实际影响：

• Pool 场景：窗口很短——drop(tx) 之后，被 spawn 的归还任务一跑就做 ping/flush。
• 裸 Connection 场景：窗口取决于你下次用这条连接的时间；如果这条连接从此不再使用，就要等 socket 关闭或服务端的 idle-in-transaction 超时清理。

这条"尽力"语义是 Rust async Drop 限制的直接结果。Transaction 的文档注释也写明事务"should end with a call to commit or rollback"——Drop 路径只是兜底。

15.6.2 MySQL 与 SQLite 的 start_rollback

MySQL（sqlx-mysql/src/transaction.rs:68-81）和 Postgres 同一思路：往 waiting 队列压一个 Waiting::Result、把 sequence_id 置 0，再用 write_packet 把 COM_QUERY 形式的 ROLLBACK 写进 write buffer，最后 depth 减一。下一次 wait_until_ready（每次执行查询和 ping 都会先调它）先 flush、再把这条 ROLLBACK 的结果读掉。

SQLite（sqlx-sqlite/src/transaction.rs:41-43 → sqlx-sqlite/src/connection/worker.rs:399-403）不一样。SQLite 连接的所有操作都由一个专属 worker 线程执行，start_rollback 只是用 flume 通道的同步 send 投递一条 Command::Rollback { tx: None }（错误被 .ok() 吞掉）。worker 线程按顺序处理命令，轮到它就直接执行 ROLLBACK 并把 depth 减一——不需要等 async 一侧再碰这条连接。depth 由 worker 线程维护（WorkerSharedState 里的 AtomicUsize），所以 drop(tx) 之后立刻读 is_in_transaction() 可能还是 true。实测（sqlx 0.8.6，SQLite 内存库）：drop 之后立即读为 true，不碰连接、等 50ms 再读为 false，插入的行已被回滚。

另有两个细节：通道是有界的（command_channel_size 默认 50），满了时这个同步 send 会阻塞当前线程；worker 还用 ignore_next_start_rollback 标志处理"COMMIT/ROLLBACK 已执行、但应答没送达"的情况，跳过随后由 Drop 发来的那次多余回滚（sqlx-sqlite/src/connection/worker.rs:139-142、sqlx-sqlite/src/connection/worker.rs:268-279）。

15.7 Drop 的"尽力 rollback"

sqlx-core/src/transaction.rs:260-275 的 Drop 实现：

impl<'c, DB> Drop for Transaction<'c, DB>
where DB: Database,
{
    fn drop(&mut self) {
        if self.open {
            // starts a rollback operation
            // what this does depends on the database but generally this means we queue a rollback
            // operation that will happen on the next asynchronous invocation of the underlying
            // connection (including if the connection is returned to a pool)
            DB::TransactionManager::start_rollback(&mut self.connection);
        }
    }
}

注释精炼地说清楚了语义——"queue a rollback operation that will happen on the next asynchronous invocation"。用户看到这段注释就明白"drop 不立即 rollback"。

这条设计的工程权衡：

如果 Drop 试图当场发出 ROLLBACK 并等回应：

• 需要在 Drop 里 block_on——在 Tokio 运行时的工作线程里这样做会直接 panic（Cannot start a runtime from within a runtime），而且会阻塞工作线程。
• 或者 spawn 一个任务去发——但任务必须拥有连接（'static），而 Transaction<'c, DB> 可能只借着一个 &'c mut Connection（MaybePoolConnection::Connection），借来的连接没法交给任务。

如果 Drop 什么都不做：

• ROLLBACK 永远不发——Pool 场景下带着未结束事务的连接会被放回池里，下一个借到它的人接手一个半截事务。

start_rollback 是中间路径：排队不等待——Postgres / MySQL 靠下次 connection I/O 把它发出去，SQLite 靠 worker 线程自行执行。

作为对照，tokio-postgres 0.7.9 的 Transaction::drop（tokio-postgres/src/transaction.rs:36-56）也不 await：它把编码好的 ROLLBACK 通过通道交给独立运行的 Connection 任务，由那个任务去写 socket——所以不必等用户下次使用。这与 sqlx SQLite 驱动的 worker 线程是同一种思路：I/O 由另一个执行单元驱动时，Drop 只需投递消息。

sqlx 的选择是 "尽力而为 + 源码注释写明"——不假装能做到同步 rollback、也不放弃尝试。

15.8 嵌套事务：savepoint 的工作机制

sqlx 支持任意深度的嵌套事务——通过 SAVEPOINT 实现：

let mut tx = pool.begin().await?;                   // depth 1: BEGIN

sqlx::query("INSERT ...").execute(&mut *tx).await?;

let mut inner_tx = (&mut *tx).begin().await?;       // depth 2: SAVEPOINT _sqlx_savepoint_1
sqlx::query("INSERT ...").execute(&mut *inner_tx).await?;

// 嵌套 rollback 只回滚到 savepoint
inner_tx.rollback().await?;                          // ROLLBACK TO SAVEPOINT _sqlx_savepoint_1

// 外层继续
sqlx::query("INSERT ...").execute(&mut *tx).await?;
tx.commit().await?;                                  // COMMIT

(&mut *tx).begin() 调的是 Connection::begin 或 Acquire::begin，二者导入一个即可；若 sqlx::Connection 和 sqlx::Acquire 同时导入，这一行会报 E0034（multiple applicable items in scope），需要写成 sqlx::Connection::begin(&mut *tx)。

语义：内层 rollback 不影响外层——外层继续跑到 commit。内层如果 commit 了、外层再 rollback，内层的改动也会被外层 rollback（因为 COMMIT 嵌套 savepoint 实际是 RELEASE——在外层 COMMIT 前还没真正提交到 DB）。

这条语义对应 Postgres / MySQL / SQLite 的 SAVEPOINT 语义——sqlx 直接发 SQL 的 SAVEPOINT 语句、不做客户端模拟。

15.8.1 嵌套的 SQL 对应

完整映射表：

depth 变化	begin SQL	commit SQL	rollback SQL
0 → 1	BEGIN	COMMIT	ROLLBACK
1 → 2	SAVEPOINT _sqlx_savepoint_1	RELEASE SAVEPOINT _sqlx_savepoint_1	ROLLBACK TO SAVEPOINT _sqlx_savepoint_1
2 → 3	SAVEPOINT _sqlx_savepoint_2	RELEASE SAVEPOINT _sqlx_savepoint_2	ROLLBACK TO SAVEPOINT _sqlx_savepoint_2
3 → 4	SAVEPOINT _sqlx_savepoint_3	...	...

depth 在 begin 时递增、commit/rollback 时递减——计数存在各驱动的连接状态里，由 TransactionManager 的 begin / commit / rollback / start_rollback 维护，get_transaction_depth 只负责读。

15.9 Transaction 的 DerefMut

sqlx-core/src/transaction.rs:212-232：

impl<'c, DB> Deref for Transaction<'c, DB>
where DB: Database,
{
    type Target = DB::Connection;
    fn deref(&self) -> &Self::Target { &self.connection }
}

impl<'c, DB> DerefMut for Transaction<'c, DB>
where DB: Database,
{
    fn deref_mut(&mut self) -> &mut Self::Target { &mut self.connection }
}

Transaction 可以被 deref 成 &mut DB::Connection——这是 sqlx 让 &mut *tx 成为 Executor 的基础。

第 4 章 §4.6 详细讨论过 &mut Transaction 不直接实现 Executor——因为 lazy normalization 限制。直接写 &mut tx 会编译失败（sqlx 0.8.6 + SQLite 实测）：

error[E0277]: the trait bound `&mut Transaction<'_, Sqlite>: Executor<'_>` is not satisfied
   |
 5 |     sqlx::query("SELECT 1").execute(&mut tx).await?;
   |                             ------- ^^^^^^^ the trait `Executor<'_>` is not implemented for `&mut Transaction<'_, Sqlite>`

trait 匹配时编译器不会替你走 Deref。用户写 &mut *tx，*tx 经 DerefMut::deref_mut 得到 DB::Connection，再取 &mut，类型就是 &mut SqliteConnection 这样的具体类型，正好命中驱动提供的 impl Executor for &mut SqliteConnection——绕过这条限制。

这条设计让事务内的 query 写法统一：

sqlx::query(...).execute(&mut *tx).await?;           // &mut Connection via DerefMut
sqlx::query(...).execute(&mut conn).await?;          // 直接 &mut Connection
sqlx::query(...).execute(&pool).await?;              // &Pool

三种 Executor 混用——Transaction 通过 DerefMut 融入统一体系。

15.10 Connection::transaction 便捷方法

第 12 章 §12.5 讨论过 Connection::transaction（sqlx-core/src/connection.rs:97-124）——一个便捷闭包包装：

fn transaction<'a, F, R, E>(&'a mut self, callback: F) -> BoxFuture<'a, Result<R, E>>
where
    for<'c> F: FnOnce(&'c mut Transaction<'_, Self::Database>) -> BoxFuture<'c, Result<R, E>>
        + 'a + Send + Sync,
    Self: Sized,
    R: Send,
    E: From<Error> + Send,
{
    Box::pin(async move {
        let mut transaction = self.begin().await?;
        let ret = callback(&mut transaction).await;
        match ret {
            Ok(ret) => { transaction.commit().await?; Ok(ret) }
            Err(err) => { transaction.rollback().await?; Err(err) }
        }
    })
}

显式 commit 或 rollback——不依赖 Drop。这让 transaction 包装器比手写 begin/commit/rollback 更安全——错误路径显式 rollback、不用担心 Drop 的"尽力"语义。

生产代码推荐优先用 transaction 闭包：

conn.transaction(|tx| Box::pin(async move {
    sqlx::query("UPDATE ...").execute(&mut **tx).await?;
    sqlx::query("INSERT ...").execute(&mut **tx).await?;
    Ok::<_, sqlx::Error>(())
})).await?;

相比手写 begin/commit，少两行代码、错误路径自动显式 rollback。主要限制在借用：闭包要满足 Send + Sync + 'a，返回的 future 要活在 'c 里，想在闭包里借用外部变量时经常过不了编译——这时候 fallback 到手写 begin。另外，若 callback 的 future 被取消或 panic，走的仍是 Drop 兜底。

15.11 跨 DB 的事务差异

三家 DB 的事务行为共性 + 差异：

共性：

• 都用 BEGIN / COMMIT / ROLLBACK 顶层事务。
• 都用 SAVEPOINT / RELEASE / ROLLBACK TO SAVEPOINT 嵌套。
• depth 跟踪机制类似。

Postgres 的特殊：

• 出错即 aborted：Postgres 事务内一条 query 失败后、整个事务进入 "aborted" 状态、后续 query 全 current transaction is aborted——只能 rollback。sqlx 不处理这个、让错误自然冒泡。
• ISOLATION LEVEL：BEGIN ISOLATION LEVEL SERIALIZABLE 等通过 begin_with（0.8.4 起）传入。
• READ ONLY / DEFERRABLE：同样通过 begin_with。

MySQL 的特殊：

• DDL 不事务性：MySQL 的 DDL（CREATE TABLE 等）会自动提交当前事务——不像 Postgres 里 DDL 也事务性。sqlx 不能防止这个（协议层无法拒绝）。
• 隔离级别：BEGIN / START TRANSACTION 不接受隔离级别子句，begin_with 传 BEGIN ISOLATION LEVEL ... 会被服务端报语法错误；要先执行 SET TRANSACTION ISOLATION LEVEL ...（只作用于下一个事务），再 begin。
• Lock 行为：InnoDB 默认 REPEATABLE READ，此级别下搜索和索引扫描加的是 next-key lock（行锁 + 间隙锁），同样的语句锁住的范围可能比 Postgres 大。

SQLite 的特殊：

• 三种 BEGIN：BEGIN DEFERRED（SQLite 默认）、BEGIN IMMEDIATE、BEGIN EXCLUSIVE——获取写锁的时机不同。sqlx 默认发裸 BEGIN（等同 DEFERRED、见 sqlx-core/src/transaction.rs:277-283）——（rollback journal 模式下）第一条 SELECT 拿 shared lock、第一条 UPDATE/INSERT/DELETE 才升级为 reserved lock。如果业务在事务里先读后写且多个连接并发，两个连接都想从读锁升级时会互相卡住，SQLite 判断出这种死锁后直接返回 SQLITE_BUSY（database is locked），busy_timeout 也不起作用——这时应该用 conn.begin_with("BEGIN IMMEDIATE")（0.8.4 起）显式指定。
• 单写者：SQLite 一个数据库文件只能有一个 writing transaction——其他 writer 要等。Pool 的 max_connections 不能解决这个并发限制。

这些差异在业务代码几乎看不见——sqlx 的 Transaction API 屏蔽了大部分方言。但生产事故时需要知道这些——比如"为什么 MySQL 里 DDL 跑完我的事务没了"就是差异 1 导致。

15.12 实战模式

几个常见的事务实战模式：

15.12.1 金钱转账

// AppError 是业务自己的错误类型（`impl From<sqlx::Error>`），不是 sqlx::Error 的变体
pub async fn transfer(pool: &PgPool, from: i32, to: i32, amount: i64) -> Result<(), AppError> {
    let mut tx = pool.begin().await?;

    // 扣款（带余额校验）
    let affected = sqlx::query(
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1"
    ).bind(amount).bind(from).execute(&mut *tx).await?.rows_affected();

    if affected != 1 {
        return Err(AppError::InsufficientFunds);
        // drop tx → start_rollback 排队
    }

    // 入账
    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(amount).bind(to).execute(&mut *tx).await?;

    // 审计记录
    sqlx::query("INSERT INTO audit_log (from_id, to_id, amount) VALUES ($1, $2, $3)")
        .bind(from).bind(to).bind(amount).execute(&mut *tx).await?;

    tx.commit().await?;  // 显式 commit
    Ok(())
}

三条 SQL 原子性——任何一条失败（扣款校验不过、网络断），整个回滚。

15.12.2 带 SAVEPOINT 的错误恢复

pub async fn bulk_import(pool: &PgPool, rows: Vec<Row>) -> Result<u64, sqlx::Error> {
    let mut tx = pool.begin().await?;
    let mut inserted = 0u64;

    for row in rows {
        // 每行一个 savepoint
        let mut sp = (&mut *tx).begin().await?;

        let result = sqlx::query("INSERT INTO target ...").bind(...).execute(&mut *sp).await;

        match result {
            Ok(_) => {
                sp.commit().await?;  // RELEASE SAVEPOINT
                inserted += 1;
            }
            Err(_) => {
                sp.rollback().await?;  // ROLLBACK TO SAVEPOINT
                // 继续下一行
            }
        }
    }

    tx.commit().await?;
    Ok(inserted)
}

逐行 savepoint让 "一行失败不影响其他行" 成为可能——批量导入的常见模式。

15.12.3 重试（乐观并发）

pub async fn update_with_retry(pool: &PgPool, id: i32, f: impl Fn(&Entity) -> Entity) -> Result<(), AppError> {
    for attempt in 0..3 {
        let mut tx = pool.begin().await?;
        // 不加 FOR UPDATE：乐观并发靠下面的 version 条件发现冲突
        let entity: Entity = sqlx::query_as("SELECT ... WHERE id = $1")
            .bind(id).fetch_one(&mut *tx).await?;

        let updated = f(&entity);

        // SET 子句按 updated 写新值，并把 version 加一
        let res = sqlx::query("UPDATE ... WHERE id = $1 AND version = $2")
            .bind(id).bind(entity.version)
            .execute(&mut *tx).await?;

        if res.rows_affected() == 1 {
            tx.commit().await?;
            return Ok(());
        } else {
            tx.rollback().await?;
            // 版本冲突，重试
        }
    }
    Err(AppError::ConflictRetryExceeded)
}

乐观并发 + 版本冲突重试——显式 rollback + loop retry。用 Transaction 保证每次尝试隔离。

15.12.4 事务状态机可视化

把 Transaction 对象的完整生命周期画出来：

stateDiagram-v2
    [*] --> Created: Transaction begin
    Created --> Running: BEGIN or SAVEPOINT 发送成功
    Running --> Running: query execute 多次
    Running --> Committed: commit await
    Running --> RolledBack: rollback await
    Running --> DropRollback: Drop open true
    DropRollback --> Closed: start_rollback 排队 ROLLBACK
    Committed --> [*]: open false
    RolledBack --> [*]: open false
    Closed --> [*]: ROLLBACK 真正送达并执行

三条主要路径：

1. Running → Committed：正常 commit 关闭。
2. Running → RolledBack：显式 rollback 关闭。
3. Running → DropRollback → Closed：没显式关就 drop——start_rollback 排队（Postgres / MySQL 等下次 I/O 发送，SQLite 由 worker 线程执行）。

其中 Closed 不是完整状态——严格说是"已排队 rollback 但 server 可能还没收到"的临界态。

三条路径都最终到 [*]——但第三条的"到 [*]"延迟不定、server 侧最终一致。生产代码应当只走 1 或 2 路径。

15.12.5 Transaction 的三种典型错误场景

场景 1：commit 失败

let mut tx = pool.begin().await?;
sqlx::query(...).execute(&mut *tx).await?;
tx.commit().await?;  // ← 这里可能失败

commit 失败的原因：

• 网络断（ACK 没回来）——客户端不知道事务是否提交成功。
• 延迟约束（DEFERRABLE INITIALLY DEFERRED）在 COMMIT 时才检查，检查失败——事务没提交。
• 并发冲突（SERIALIZABLE 隔离级别下）——事务没提交。

错误处理：tx.commit().await? 返回 Err——tx 已被 commit 消费、但 open 还是 true（commit 没成功设 false）——Drop 会触发 start_rollback——如果事务在 server 端已经结束（提交了或已被回滚），这次 rollback 是 no-op：Postgres 对事务块外的 ROLLBACK 只回一条 WARNING（there is no transaction in progress），不做任何事。

业务含义：commit 返回 Err 时你不能假设事务已生效——可能成功也可能失败。幂等设计是唯一安全做法——重试时先 SELECT 验证状态。

场景 2：begin 失败

let mut tx = pool.begin().await?;  // ← 这里失败

begin 失败的原因：

• pool 满（PoolTimedOut）。
• 连接建立失败。
• BEGIN SQL 被 server 拒绝（罕见）。

错误处理：tx 根本没构造出——无事务状态泄漏。

场景 3：query 失败

let mut tx = pool.begin().await?;
sqlx::query(...).execute(&mut *tx).await?;  // ← 这里失败
sqlx::query(...).execute(&mut *tx).await?;  // 不到这
tx.commit().await?;                          // 不到这

? 冒泡错误——tx 离开作用域 drop——start_rollback 排队 ROLLBACK。

Postgres 特殊：第一条 query 失败后事务进入 "aborted" 状态、后续 query 全拒绝 current transaction is aborted。rollback 是唯一出路——drop 的 start_rollback 正好是期望行为。

三种错误场景的共性：? 传播错误时，Drop 兜底会把未完成的事务排队回滚，不会让事务一直挂着。但它只是兜底——需要确认回滚已完成（比如之后还要接着用这条裸连接，或者想拿到回滚本身的错误）时，在错误分支里显式 tx.rollback().await，或者直接用 §15.10 的 transaction 闭包。

15.13 阶段小结

到这里，sqlx Transaction 的每一处设计都拆开了：

1. Rust async Drop 的根本限制（§15.1）—— Drop 同步、事务回滚需要 async I/O、两者冲突。sqlx 的对策是显式 commit/rollback + start_rollback 同步排队。
2. TransactionManager trait（§15.2）—— 五个方法 begin/commit/rollback（async）+ start_rollback（sync）+ get_transaction_depth。驱动层抽象。
3. Transaction 两字段 struct（§15.3）—— connection + open。MaybePoolConnection 统一 Connection-originated 和 Pool-originated。
4. begin 的 depth-based SQL 选择（§15.4）—— depth = 0 → BEGIN；depth > 0 → SAVEPOINT _sqlx_savepoint_N。Rollback guard 兜底 begin 过程的错误。
5. commit / rollback async（§15.5）—— 执行 SQL + 更新 depth + 设置 open = false。失败时 open 保持 true 让 Drop 兜底。
6. start_rollback 同步排队（§15.6）—— Postgres 用 queue_simple_query 写 buffer 不发送，真正发送要等下次 I/O；MySQL 同理；SQLite 把命令投给 worker 线程执行。
7. Drop 的"尽力 rollback"（§15.7）—— 中间路径：不在 Drop 里 block_on 或 spawn，也不完全放弃（排队等后续 I/O 发送）。
8. SAVEPOINT 嵌套（§15.8）—— 连续数字命名、commit/rollback 按 depth 反推 savepoint 名。
9. DerefMut 到 Connection（§15.9）—— &mut *tx 变成 &mut Connection、满足 Executor bound。
10. Connection::transaction 闭包（§15.10）—— 显式 commit/rollback 包装、错误路径显式 rollback、比手写 begin 更安全。
11. 跨 DB 差异（§15.11）—— Postgres 的 aborted 状态、MySQL 的 DDL 隐式提交、SQLite 的单写者。
12. 三个实战模式（§15.12）—— 转账、带 savepoint 的批量导入、乐观并发重试。

15.14 Transaction 的三种"不要做"

生产里关于 Transaction 使用的三条反模式，值得单独列出：

不要做 1：在事务里调 tokio::spawn

let mut tx = pool.begin().await?;
tokio::spawn(async move {
    sqlx::query("...").execute(&mut *tx).await  // Pool 来的 tx 是 'static，能编译
});

Pool-originated 的 Transaction<'static, DB> 可以 move 进 tokio::spawn（sqlx 0.8.6 + SQLite 实测能编译运行）；Connection-originated 的持有 &'c mut Connection，不满足 'static，编译不过。能编译也不推荐：tx 被 move 进任务后，外面就没法再 commit；而 drop JoinHandle 只是分离任务、不会取消它，任务什么时候结束、事务最后是 commit 还是走 Drop 回滚，外层都控制不了。

修复：所有事务操作在一个 task 内完成。

不要做 2：让事务持续数秒或更久

let mut tx = pool.begin().await?;
long_external_api_call().await?;  // 几秒 HTTP 请求
sqlx::query(...).execute(&mut *tx).await?;
tx.commit().await?;

事务持有时间长 = 持锁时间长 = 其他事务被阻塞更久 = scalability 瓶颈。Postgres 的 row lock 在事务中一直持有。

修复：事务只包围必要的原子操作；外部 API 调用在事务外做。

不要做 3：假设 Drop 会立即 rollback

if condition {
    // 想"丢弃这个事务，不 rollback"——drop 让它 rollback 就够了
} else {
    tx.commit().await?;
}

错——drop 会 start_rollback、排队 ROLLBACK SQL——不是"什么都不做"。如果你真的想"不操作、让 server timeout 清理"——也做不到（start_rollback 已经执行）。

修复：想明确 rollback 就调 tx.rollback().await?；不想 rollback 就 commit 或保持路径一致。Drop 是兜底、不是你该依赖的"行为"。

15.15 Transaction API 的设计哲学

读完本章、回看整个 Transaction 设计——能看出 sqlx 团队的几条核心取舍：

1. 类型系统优于运行时检查。Transaction 吃 self commit/rollback——一次用完就消费掉、不能重复 commit。open 字段只在内部维护——用户看不到。用 Rust 类型一次性表达"事务是线性资源"。

2. 让异常路径尽可能正确。Rollback guard（begin 过程）、start_rollback（Drop）、open 字段（防重复）——全部都是"异常时也能兜底"的设计——完全靠 RAII 和 flags。

3. 诚实暴露 Rust 限制。sqlx 没假装能同步 rollback、也没在 Drop 里 spawn 回滚任务或 block_on——Drop 的源码注释直接写明只是"queue a rollback operation"。

4. 跨 DB 差异压到最低。同一套 Transaction 对外 API——底下 Postgres / MySQL / SQLite 的具体 SQL 和行为不同——但用户代码几乎不需要改。差异只在业务要用 DB 方言特性（如 ISOLATION LEVEL）时才显露。

5. RAII 自动化 + 显式调用鼓励。Drop 保底 + transaction() 闭包促显式。用户按场景选——简单场景让 RAII 管、需要精细控制时手写 begin + explicit commit。

这五条哲学共同构成 sqlx Transaction 的精神——"类型驱动安全 + RAII 保底 + 诚实文档 + 一套 API 跨 DB + 灵活性保留"。这是一份好的Rust 库设计模板——你设计自己的资源管理 API 时都可以套用。

15.16 第四部分回顾

本专栏第四部分"连接与事务"到此结束——四章讲完了 sqlx 最核心的资源管理抽象：

• 第 12 章 Connection——单条连接的生命周期、协议、礼仪。
• 第 13 章 Pool 外部 API——用户如何配置和使用 Pool。
• 第 14 章 Pool 内部实现——idle queue、semaphore、maintenance task。
• 第 15 章 Transaction（本章）——事务的 RAII guard、SAVEPOINT、尽力 rollback。

四章合起来构成了 sqlx 的"资源管理栈"——从最底层的 Connection 到最上层的 Transaction、每一层都有清晰职责、层层叠加。

下一部分进入驱动实现——Postgres / MySQL / SQLite / Any 四章。这部分偏"协议级深入"——读者可以按兴趣选。日常业务用 sqlx 不需要读；但想贡献 sqlx 代码、写自定义驱动、深度优化性能——这部分必读。

15.17 Transaction 常见问题 FAQ

读完本章，读者可能常有的几个问题：

Q1：嵌套事务（SAVEPOINT）在生产里真的用得着吗？

用得着，但场景较窄。典型用例：

• 批量导入——每行一个 savepoint 隔离失败（§15.12.2）。
• 子流程——某个独立操作可能失败但不影响主事务。

大多数业务代码不用嵌套——一个顶层事务包围所有操作就够。

Q2：事务里可以 await 其他非数据库操作吗？

可以但不建议：

let mut tx = pool.begin().await?;
sqlx::query(...).execute(&mut *tx).await?;
some_http_call().await?;  // 危险
sqlx::query(...).execute(&mut *tx).await?;
tx.commit().await?;

http_call 可能慢几秒——在这段时间里事务持有锁、占 Pool 连接、占 server 资源。高并发场景会让 Pool 饱和、其他事务等死。

规范：事务里只做数据库操作；外部调用在事务外。

Q3：query_as::<_, User>(...).fetch_all(&mut *tx) 和 .fetch_all(&pool) 行为有什么不同？

• &mut *tx：走事务连接、看到事务内未提交的改动、读一致性受隔离级别控制。
• &pool：借一个独立连接、看不到你自己事务里未提交的改动、和事务隔离。

在事务里想查"自己刚改的数据"——必须用 &mut *tx。

Q4：为什么 Transaction 不直接用 Drop::drop 等待 rollback 完成？

Rust Drop::drop 是同步、不能 .await。解决方案只有两个：

• spawn task——任务要拥有连接，而借着 &mut Connection 的 Transaction 交不出去；何况 Drop 也等不到任务结束。
• block_on——在 Tokio 运行时的工作线程里直接 panic（Cannot start a runtime from within a runtime）。

两者都不可行。sqlx 选了"start_rollback 同步排队"：Postgres / MySQL 等下次 I/O 触发发送，SQLite 交给 worker 线程执行（§15.6.2）。

Q5：我的代码里 commit 返回错误是不是应该重试？

看情况。

• 网络错：可以重试、但先查 DB 状态确认事务到底提交了没——commit 错误不代表事务失败、可能 ACK 丢了。
• SERIALIZABLE 冲突：重试是正常流程（和 §15.12.3 的版本冲突重试同理）。
• 逻辑错：不要重试、修代码。

规则：重试必须配合幂等设计——单次操作的重复执行不改变结果。

15.18 Transaction 源码阅读推荐

如果你想精读 sqlx Transaction 的源码，推荐顺序：

1. sqlx-core/src/transaction.rs——core trait + Transaction struct + 3 条 ANSI SQL helper（300 行）。本章主要源码。
2. sqlx-postgres/src/transaction.rs——PgTransactionManager 具体实现（110 行）。
3. sqlx-mysql/src/transaction.rs——MySqlTransactionManager（类似 PgTransactionManager）。
4. sqlx-sqlite/src/transaction.rs——SQLite 版本、通过 worker command 发 BEGIN（同样默认 DEFERRED 语义）；配合读 sqlx-sqlite/src/connection/worker.rs 里 Command::Begin / Commit / Rollback 的处理。
5. sqlx-core/src/pool/maybe.rs——MaybePoolConnection 枚举（43 行）——Transaction 用来统一 Connection/PoolConnection 来源。

前五个文件合计不到 600 行（302 + 110 + 86 + 48 + 43），读完后你对"Rust 里实现 RAII 风格事务"这件事有完整理解。

特别留意：

• sqlx-core/src/transaction.rs:260-275 的 Drop 实现和注释——解释了"尽力 rollback"的哲学。
• PgTransactionManager::begin 里的 Rollback guard——部分成功的错误路径兜底。
• 三条 ANSI SQL helper 的 Cow<'static, str> 返回类型——零拷贝常见情况 + 按需分配动态情况。

15.19 事务的往返次数

事务的开销主要看网络往返（RTT）次数。按 0.8.6 源码数一下 Postgres 上各操作的往返（不含服务端执行时间）：

操作	往返次数
pool.begin().await	从空闲队列取连接时默认先 ping 一次（test_before_acquire 默认 true），再发 BEGIN：约 2 次
&mut *tx 一次 query.execute()	1 次（预处理语句缓存命中时）
tx.commit().await	1 次
Drop 的 start_rollback	0 次（同步排队，不碰网络）
嵌套 savepoint begin	1 次

观察：事务的每个操作都要等数据库往返——跨机房累加很快。按 10ms RTT 估算，一个"begin + 3 条 query + commit"事务光往返就要约 60ms（6 次往返），相当明显。

优化方向：

1. 减少事务内 query 数量——把多次查合并成一条 SQL（JOIN / CTE / RETURNING）。
2. batch 操作——一条 INSERT 带 1000 行胜过 1000 次 INSERT。
3. 保持事务短——事务外的只读查询分离出去。
4. 用 sqlx::raw_sql 批量执行 migration 脚本——一条 simple query 一次性发。

这些优化都是"减少 round-trip"——对事务性能影响最大的因素。

15.20 三条 ANSI SQL 生成函数

sqlx 的 SAVEPOINT 嵌套靠三个小函数生成 SQL（sqlx-core/src/transaction.rs:277-302）：

pub fn begin_ansi_transaction_sql(depth: usize) -> Cow<'static, str> {
    if depth == 0 {
        Cow::Borrowed("BEGIN")
    } else {
        Cow::Owned(format!("SAVEPOINT _sqlx_savepoint_{depth}"))
    }
}

pub fn commit_ansi_transaction_sql(depth: usize) -> Cow<'static, str> {
    if depth == 1 {
        Cow::Borrowed("COMMIT")
    } else {
        Cow::Owned(format!("RELEASE SAVEPOINT _sqlx_savepoint_{}", depth - 1))
    }
}

pub fn rollback_ansi_transaction_sql(depth: usize) -> Cow<'static, str> {
    if depth == 1 {
        Cow::Borrowed("ROLLBACK")
    } else {
        Cow::Owned(format!(
            "ROLLBACK TO SAVEPOINT _sqlx_savepoint_{}",
            depth - 1
        ))
    }
}

三个函数一眼看完嵌套语义：

• depth=0 → BEGIN（真事务启动）。
• depth≥1 → SAVEPOINT _sqlx_savepoint_<depth>（嵌套点）。
• commit depth=1 → COMMIT；commit depth>1 → RELEASE SAVEPOINT ...。
• rollback depth=1 → ROLLBACK；rollback depth>1 → ROLLBACK TO SAVEPOINT ...。

SAVEPOINT 命名前缀 _sqlx_savepoint_—— 和用户 SAVEPOINT 不易冲突（用户一般不用下划线开头）—— 和 _sqlx_migrations 表的命名策略一致——下划线前缀 = sqlx 内部空间。

depth 计数由 TransactionManager 维护——每家驱动实现 begin/commit/rollback 时维护自己的 depth：Postgres 用 PgConnectionInner::transaction_depth、MySQL 用 MySqlConnectionInner::transaction_depth（都经 conn.inner 访问）、SQLite 用 WorkerSharedState::transaction_depth（AtomicUsize，由 worker 线程改写）。同一套 ANSI SQL 生成函数 + 三种 depth 存储—— 共享代码最大化。

Cow<'static, str> 的选择—— depth=0/1 的常见路径返回 Borrowed("BEGIN") / Borrowed("COMMIT")—— 零分配；嵌套路径才分配 owned String—— 常见路径 fast、边界路径慢一点但仍正确。

这种"常见路径零开销、少见路径接受开销"是 sqlx 全专栏反复出现的设计风格。

15.21 Transaction::drop 的 start_rollback 含义

源码见 §15.7（sqlx-core/src/transaction.rs:260-275）。

关键词 start_rollback（注意是 start）——不是 "执行 rollback"—— 是 "排队一条 rollback"。因为 Drop 不能 async——不能直接发 ROLLBACK 并等响应——只能把回滚排队。

每家驱动实现 start_rollback 不同：

• Postgres—— queue_simple_query 把 ROLLBACK 写进 write buffer、depth 减一；下次 wait_until_ready 时 flush 出去（§15.6）。
• MySQL—— 同样把 ROLLBACK 包写进 write buffer，并往 waiting 队列登记一个待读结果（§15.6.2）。
• SQLite—— 向 worker 线程投递 Command::Rollback { tx: None }，worker 轮到它就执行，不等 async 一侧再用连接（§15.6.2）。

self.open: bool—— 标志位—— commit/rollback 显式调用时置 false—— Drop 看到 false 就什么都不做。

结果—— 用户忘记显式 commit/rollback、业务代码早退，Drop 自动排队回滚——未提交的改动不会被意外提交，Pool 场景下连接归还时回滚也会被发出去。

Drop 的源码注释明确告诉读者它的语义和限制——"rollback 会在这条连接下次异步使用时发生"（SQLite 驱动更早一些）——用户不需要假设 Drop 立刻 rollback。

补一句对比—— Java JDBC 的 Connection.close() 文档把"关闭时仍有活动事务"的结果定为 implementation-defined，并强烈建议关闭前显式 commit 或 rollback——提交还是回滚取决于驱动。sqlx 把兜底行为固定为 rollback，默认 rollback 比默认 commit 安全。

15.22 实战示例：带 retry 的幂等转账

把本章所有内容用一个最完整的例子收尾——一个生产级的带重试的转账函数：

use sqlx::{PgPool, Error};
use std::time::Duration;

// TransferError 是业务自己的错误类型，含 Conflict / MaybeCommitted / InsufficientFunds /
// RetriesExceeded / Other(sqlx::Error) 几个变体；它的 From<sqlx::Error> 把 SQLSTATE
// 40001（serialization_failure）和 40P01（deadlock_detected）映射成 Conflict，其余映射成 Other
pub async fn transfer_with_retry(
    pool: &PgPool,
    from: i32,
    to: i32,
    amount: i64,
    idempotency_key: &str,
) -> Result<(), TransferError> {
    const MAX_ATTEMPTS: u32 = 3;

    for attempt in 0..MAX_ATTEMPTS {
        match transfer_once(pool, from, to, amount, idempotency_key).await {
            Ok(()) => return Ok(()),
            Err(TransferError::Conflict) if attempt < MAX_ATTEMPTS - 1 => {
                // 序列化失败或死锁，重试
                let backoff = Duration::from_millis(10 * 2u64.pow(attempt));
                tokio::time::sleep(backoff).await;
            }
            Err(TransferError::MaybeCommitted) if attempt < MAX_ATTEMPTS - 1 => {
                // commit 失败但可能已生效——幂等 key 让重试安全
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            Err(e) => return Err(e),
        }
    }
    Err(TransferError::RetriesExceeded)
}

async fn transfer_once(
    pool: &PgPool,
    from: i32, to: i32, amount: i64,
    idempotency_key: &str,
) -> Result<(), TransferError> {
    let mut tx = pool.begin().await?;

    // 幂等 key 防重复
    let existing: Option<i64> = sqlx::query_scalar(
        "SELECT amount FROM transfers WHERE idempotency_key = $1"
    ).bind(idempotency_key).fetch_optional(&mut *tx).await?;

    if existing.is_some() {
        return Ok(());  // 已经转过了
    }

    // 扣款（带余额校验）
    let affected = sqlx::query(
        "UPDATE accounts SET balance = balance - $1
         WHERE id = $2 AND balance >= $1"
    ).bind(amount).bind(from).execute(&mut *tx).await?.rows_affected();

    if affected != 1 { return Err(TransferError::InsufficientFunds); }

    // 入账
    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(amount).bind(to).execute(&mut *tx).await?;

    // 记录
    sqlx::query("INSERT INTO transfers (idempotency_key, from_id, to_id, amount) VALUES ($1, $2, $3, $4)")
        .bind(idempotency_key).bind(from).bind(to).bind(amount)
        .execute(&mut *tx).await?;

    // commit
    match tx.commit().await {
        Ok(()) => Ok(()),
        // I/O 错误：COMMIT 可能已经生效，只是没收到回应
        Err(Error::Io(_)) => Err(TransferError::MaybeCommitted),
        Err(e) => Err(e.into()),
    }
}

这个例子整合了本章所有要点：

• 显式 begin + commit——不依赖 Drop。
• 幂等 key——commit 错误后重试不会重复扣款。
• 冲突重试——序列化失败 / 死锁（40001 / 40P01）映射成 Conflict 后重试。
• 重试逻辑——指数退避 + 有限次数。
• commit 失败分类——I/O 错误 vs 其他错误——区分重试与否。

第 16 章开始进入驱动层——Postgres Extended Query 协议的 Rust async 实现。