Rust in Everything
基于 sea-orm 2.0.4

事务

主线位置：查询主线和写入主线的执行器那一格换了一种实现。主线上的 .one(&db)、.insert(&db) 每条语句都去池里 acquire 一次；换成 &txn 之后，语句在 DatabaseTransaction 的 ConnectionTrait 实现里锁住同一条连接执行，不再经过池。2.0 的 ActiveModelEx 嵌套保存在内部也开事务，走的是同一套代码。

flowchart TB
  begin1["db.begin()"] --> acq["pool.acquire"]
  acq --> wrap["Arc Mutex InnerConnection"]
  wrap --> tm["sqlx begin 发 BEGIN"]
  tm --> cfg{"有隔离级别或只读"}
  cfg -->|"有"| settx["SET TRANSACTION"]
  cfg -->|"无"| txn[DatabaseTransaction]
  settx --> txn
  txn --> nested["txn.begin() 克隆 Arc"]
  nested --> sp["SAVEPOINT _sqlx_savepoint_N"]
  txn --> commitnode["commit 发 COMMIT"]
  txn --> rb["rollback 发 ROLLBACK"]
  txn --> dropnode["drop 没提交"]
  dropnode --> q["ROLLBACK 排进写缓冲"]
  q --> ret["归还时 ping 发出"]

sqlx 的事务怎么按深度发 BEGIN / SAVEPOINT、Drop 为什么只能排队，sqlx 卷的 事务 讲过，本页只讲 sea-orm 在它上面加了什么、改了什么。代码在 src/database/transaction.rs（816 行，大半是每种后端一个 match 分支）和 src/driver/sqlx_postgres.rs 的三个函数。

实验 crate 依赖 sea-orm =2.0.4（sqlx-postgres、runtime-tokio-rustls、macros、with-json）和 sqlx =0.9.0，平台 M2 Max、macOS、rustc 1.99.0、Postgres 16.14。表是 create table so_conn2.item (id serial primary key, n int not null)，实体 item 两列。看线上消息用 一次 find 的全过程 里的协议代理（[conn N] C->S 是客户端发的，S->C 是服务端回的，Bind 后面是参数的字节）；池只有一条连接，每一步之间睡 200 毫秒再打印说明，连接建立时的认证消息省略了。

DatabaseTransaction：一条连接，放在 Arc 和 Mutex 里

DatabaseTransaction：

pub struct DatabaseTransaction {
    conn: Arc<Mutex<InnerConnection>>,
    backend: DbBackend,
    open: bool,
    metric_callback: Option<crate::metric::Callback>,
    record_stmt_in_spans: bool,
}

InnerConnection 是按 feature 打开分支的 enum，Postgres 分支装的是一条 PoolConnection<sqlx::Postgres>。sea-orm 没有用 sqlx 的 Transaction 结构体，它直接拿着池连接，自己调 sqlx 驱动的 TransactionManager::begin / commit / rollback / start_rollback。所以嵌套深度仍然记在 PgConnection 的 transaction_depth 上，DatabaseTransaction 自己只记一个 open：这一层还要不要在 Drop 时回滚。

两层包装各有用处：

  • Mutex（futures_util::lock::Mutex，异步锁）：ConnectionTrait 的方法都收 &self，sqlx 执行要 &mut PgConnection。每条语句先 self.conn.lock().await 拿到可变引用再执行，execute_raw 的 Postgres 分支就是 query.execute(&mut *conn)。同一个事务上的语句因此排队执行，不再 acquire。
  • Arc：嵌套事务和外层共用这条连接。txn.begin() 把 Arc::clone(&self.conn) 交给新的 DatabaseTransaction，不借用 txn。最后一个 Arc 被 drop 时，PoolConnection 才被 drop，连接回池。

begin：先 acquire，再 BEGIN，再 SET TRANSACTION

db.begin() 是 TransactionTrait for DatabaseConnection 的方法，按 inner 分派到 SqlxPostgresPoolConnection::begin：self.pool.acquire() 借一条连接（池满了等到超时，报 ConnectionAcquire），new_postgres 包进 Arc<Mutex<..>>，再进 DatabaseTransaction::begin：

        let res = DatabaseTransaction {
            conn,
            backend,
            open: true,
            metric_callback,
            record_stmt_in_spans,
        };

        let begin_result: Result<(), DbErr> = super::tracing_spans::with_db_span!(
            // ...
            async {
                #[cfg(not(feature = "sync"))]
                let conn = &mut *res.conn.lock().await;
                // ...
                match conn {
                    // ...
                    #[cfg(feature = "sqlx-postgres")]
                    InnerConnection::Postgres(c) => {
                        <sqlx::Postgres as sqlx::Database>::TransactionManager::begin(c, None)
                            .await
                            .map_err(sqlx_error_to_query_err)?;
                        // in PostgreSQL SET TRANSACTION operations must be executed inside transaction
                        crate::driver::sqlx_postgres::set_transaction_config(
                            c,
                            isolation_level,
                            access_mode,
                        )
                        .await
                    }
                    // ...
                }
            }
        );

        begin_result?;
        Ok(res)

两点：

  • res 先造出来，open: true。后面任何一步失败或被取消，res 被 drop，Drop 会去回滚。这和 sqlx 0.9 的 Transaction::begin 是同一个思路。
  • 给 sqlx 的第二个参数永远是 None，sea-orm 从不调 sqlx 的 begin_with。隔离级别和只读是 BEGIN 成功之后，再由 set_transaction_config 发一条语句：
    let mut settings = Vec::new();

    if let Some(isolation_level) = isolation_level {
        settings.push(format!("ISOLATION LEVEL {isolation_level}"));
    }

    if let Some(access_mode) = access_mode {
        settings.push(access_mode.to_string());
    }

    if !settings.is_empty() {
        let sql = format!("SET TRANSACTION {}", settings.join(" "));
        sqlx::query(sqlx::AssertSqlSafe(sql))
            .execute(&mut **conn)
            .await
            .map_err(sqlx_error_to_exec_err)?;
    }

IsolationLevel 和 AccessMode 的 Display 就是 SQL 关键字：

参数拼进去的文本
IsolationLevel::ReadUncommittedISOLATION LEVEL READ UNCOMMITTED
IsolationLevel::ReadCommittedISOLATION LEVEL READ COMMITTED
IsolationLevel::RepeatableReadISOLATION LEVEL REPEATABLE READ
IsolationLevel::SerializableISOLATION LEVEL SERIALIZABLE
AccessMode::ReadOnlyREAD ONLY
AccessMode::ReadWriteREAD WRITE

两个都给就拼成一条，两个都是 None 就什么也不发。这条 SET TRANSACTION 走的是 sqlx::query，也就是扩展协议，会进 sqlx 的语句缓存。

实验：BEGIN、嵌套 begin、begin_with_config

// A
let txn = db.begin().await?;
add(&txn, 1).await?; // item::ActiveModel { n: Set(1), .. }.insert(&txn)
let sp = txn.begin().await?;
sp.commit().await?;
let sp = txn.begin().await?;
sp.rollback().await?;
txn.commit().await?;

// B，跑两次
let txn = db
  .begin_with_config(Some(IsolationLevel::Serializable), Some(AccessMode::ReadOnly))
  .await?;
// select current_setting('transaction_isolation'), current_setting('transaction_read_only')
txn.commit().await?;

--- A: begin / 嵌套 begin / commit / rollback ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
begin 返回
  [conn 1] C->S Parse(sqlx_s_1, "INSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\""), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23), RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000001), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
嵌套 begin 返回
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
嵌套 commit 返回
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Query("ROLLBACK TO SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(T)
嵌套 rollback 返回
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
commit 返回
--- B: begin_with_config(Serializable, ReadOnly)，两次 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_2, "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE READ ONLY"), Describe(S, sqlx_s_2), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID ), NoData, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2, ), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("SET"), CloseComplete, ReadyForQuery(T)
  ...
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
transaction_isolation = serializable, transaction_read_only = on
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2, ), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("SET"), CloseComplete, ReadyForQuery(T)
  ...
transaction_isolation = serializable, transaction_read_only = on
  • 每个事务的第一条消息是 acquire 时的 ping（Sync），commit 之后的 Sync 是 PoolConnection 归还时的 ping。事务中间的语句一个 Sync 都没有：它们不经过池。
  • 嵌套 begin 是 sqlx 按深度生成的 _sqlx_savepoint_1，同一深度先后两次重名，前一个已经 RELEASE 了，不冲突。
  • 带配置的事务是两条语句：BEGIN 走简单查询，SET TRANSACTION ... 走扩展协议。第一次 Parse 进了缓存（sqlx_s_2），第二次只发 Bind。每一种隔离级别和只读的组合各占一条缓存的预备语句。
📝NOTE
冷知识：两项设置以前是两条 SET TRANSACTION

1.1.20 的 set_transaction_config 对隔离级别和访问模式各发一条：先 SET TRANSACTION ISOLATION LEVEL ...，再 SET TRANSACTION READ ONLY。#2287 报告说这样「causes two statements to be submitted to the DB where one would suffice」，远程数据库上「the time delay can be considerable - 27ms per statement due to the RTT」，建议拼成一条。#2893「Postgres: combine SET TRANSACTION ISOLATION LEVEL & ACCESS MODE into a single statement」照做，改成上面的 settings.join(" ")。CHANGELOG 记在 2.0.0-rc.28，git tag --contains 也是 2.0.0-rc.28。BEGIN 本身仍是单独一次来回，所以带配置的事务现在是两次，第一次用某个组合时还要多一次 Parse。

事务里再 begin_with_config

DatabaseTransaction 自己也实现了 TransactionTrait，begin_with_config 和 begin 一样走 DatabaseTransaction::begin：sqlx 在深度 1 上发 SAVEPOINT，sea-orm 接着发 SET TRANSACTION。Postgres 不允许在事务里已经执行过语句之后改隔离级别：

// D
let txn = db.begin().await?;
add(&txn, 3).await?;
let r = txn.begin_with_config(Some(IsolationLevel::Serializable), None).await;
add(&txn, 4).await?;
txn.commit().await?;

--- D: 事务里 begin_with_config ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000003), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_4, "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE"), Describe(S, sqlx_s_4), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID ), NoData, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_4, ), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, ErrorResponse
  [conn 1] S->C ReadyForQuery(E)
嵌套 begin_with_config = Err(Exec(SqlxError(Database(PgDatabaseError { severity: Error, code: "25001", message: "SET TRANSACTION ISOLATION LEVEL must be called before any query", detail: None, hint: None, position: None, where: None, schema: None, table: None, column: None, data_type: None, constraint: None, file: Some("guc.c"), line: Some(6822), routine: Some("call_enum_check_hook") }))))
  [conn 1] C->S Query("ROLLBACK TO SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000004), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  ...
--- 最终 item.n: [1, 3, 4]

报错是 25001，ReadyForQuery(E) 说明外层事务这时处在出错状态。begin 返回 Err 之前，res 被 drop，open 还是 true，start_rollback 按当前深度 2 排进 ROLLBACK TO SAVEPOINT _sqlx_savepoint_1。它跟在下一条 INSERT 前面发出，把外层从出错状态救回来，3 和 4 都提交了。sqlx 0.9.0 的 begin_with 在同样位置会把外层事务回滚掉（事务里调 begin_with），sea-orm 不走 begin_with，没有这个问题，代价是隔离级别只在最外层有意义。

嵌套事务不借用外层

TransactionTrait for DatabaseTransaction 的 begin：

    async fn begin(&self) -> Result<DatabaseTransaction, DbErr> {
        DatabaseTransaction::begin(
            Arc::clone(&self.conn),
            self.backend,
            self.metric_callback.clone(),
            self.record_stmt_in_spans,
            None,
            None,
            None,
        )
        .await
    }

内层和外层是同一种类型，各自拿着一份 Arc。保存点叫什么、提交时 RELEASE 哪一个，全看连接上的深度，不看是哪个 DatabaseTransaction 调的。sqlx 的嵌套 Transaction 借用外层（&mut），编译器保证内层先结束；sea-orm 的内层是一个独立的值，可以活得比外层的 commit 还久。

❗IMPORTANT
动脑筋：内层还活着时先提交外层，表里剩什么？表是空的，池一条连接（M2 Max、macOS、rustc 1.99.0、Postgres 16.14）：
let txn = db.begin().await?;
item::ActiveModel { n: Set(1), ..Default::default() }.insert(&txn).await?;
let inner = txn.begin().await?;
item::ActiveModel { n: Set(2), ..Default::default() }.insert(&inner).await?;
let c = txn.commit().await;
println!("外层 commit = {c:?}");
drop(inner);

这段能编译，txn.commit() 返回 Ok(())。最后表里是 [1, 2]、[1] 还是 []？

commit、rollback 与 Drop

commit 和 rollback 按值收 mut self：锁住连接，调 sqlx 的 TransactionManager::commit / rollback（按深度发 COMMIT / RELEASE SAVEPOINT 或 ROLLBACK / ROLLBACK TO SAVEPOINT），成功后 self.open = false，然后 self 被 drop。

没提交就 drop，Drop 调 start_rollback：

    // the rollback is queued and will be performed on next async operation, like returning the connection to the pool
    #[instrument(level = "trace")]
    fn start_rollback(&mut self) -> Result<(), DbErr> {
        if self.open {
            if let Some(mut conn) = self.conn.try_lock() {
                match &mut *conn {
                    // ...
                    #[cfg(feature = "sqlx-postgres")]
                    InnerConnection::Postgres(c) => {
                        <sqlx::Postgres as sqlx::Database>::TransactionManager::start_rollback(c);
                    }
                    // ...
                }
            } else {
                //this should never happen
                return Err(conn_err("Dropping a locked Transaction"));
            }
        }
        Ok(())
    }
}

impl Drop for DatabaseTransaction {
    fn drop(&mut self) {
        self.start_rollback().expect("Fail to rollback transaction");
    }
}

Drop 里不能 await，所以用 try_lock，拿到锁后交给 sqlx 的 start_rollback：把 ROLLBACK 写进连接的写缓冲、深度减一，不发网络。DatabaseTransaction 的 drop 跑完，字段 conn 跟着 drop；如果这是最后一个 Arc，PoolConnection 被 drop，归还任务里的 ping 把缓冲里的 ROLLBACK 和自己的 Sync 一起写出去：

// C
let txn = db.begin().await?;
add(&txn, 2).await?;
drop(txn);

--- C: 不提交就 drop ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000002), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
drop(txn)
  [conn 1] C->S Query("ROLLBACK"), Sync
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 1] S->C ReadyForQuery(I)
drop 之后 200ms

和 sqlx 的池连接上不提交就 drop 完全一样（不报错、不打日志，写入静默消失，见 事务没 commit）。sea-orm 的事务连接一定来自池，不会出现 sqlx 裸连接上「ROLLBACK 一直等到下一次使用」的情况；但内层事务的 Arc 还在时，连接不归还，排队的回滚要等下一条语句。D 段就是这样：内层的 ROLLBACK TO SAVEPOINT 跟着下一条 INSERT 才发出去。

sqlx 的语句日志也看不到这条回滚。打开 sqlx_logging 跑一遍带配置的事务和一次 drop（RUST_LOG=sqlx=info）：

--- V: sqlx 日志里的事务 ---
 INFO summary="SET TRANSACTION ISOLATION LEVEL …" db.statement="\n\nSET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ WRITE\n" rows_affected=0 rows_returned=0 elapsed=620.167µs elapsed_secs=0.000620167
 INFO summary="INSERT INTO \"item\" (\"n\") …" db.statement="\n\nINSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\"\n" rows_affected=1 rows_returned=1 elapsed=2.517666ms elapsed_secs=0.002517666
 INFO summary="ROLLBACK TO SAVEPOINT _sqlx_savepoint_1" db.statement="" rows_affected=0 rows_returned=0 elapsed=233.917µs elapsed_secs=0.000233917
 INFO summary="COMMIT" db.statement="" rows_affected=0 rows_returned=0 elapsed=819.542µs elapsed_secs=0.000819542
--- 不提交就 drop ---
 INFO summary="INSERT INTO \"item\" (\"n\") …" db.statement="\n\nINSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\"\n" rows_affected=1 rows_returned=1 elapsed=330.916µs elapsed_secs=0.000330916

BEGIN、SAVEPOINT、Drop 排进去的 ROLLBACK 都是 queue_simple_query 发的，不经过 sqlx 的日志；显式的 rollback() / commit() 走 conn.execute，会出现。SET TRANSACTION 是普通查询，也会出现。

try_lock 失败：Drop 里 panic

源码注释说拿不到锁「should never happen」。嵌套事务让它能发生：内层事务可以 move 进另一个任务，那边的语句正锁着连接时，这边 drop 外层：

// W
let txn = db.begin().await?;
let inner = txn.begin().await?;
let task = tokio::spawn(async move {
  inner
    .execute_raw(Statement::from_string(DbBackend::Postgres, "select pg_sleep(1)"))
    .await
    .map(|_| ())
});
tokio::time::sleep(Duration::from_millis(100)).await;
let dropped = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(txn)));

--- W: 内层在别的任务里查询时 drop 外层 ---

thread 'main' (3856345) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/sea-orm-2.0.4/src/database/transaction.rs:404:31:
Fail to rollback transaction: Conn(Internal("Dropping a locked Transaction"))
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
drop(txn) panicked = true
内层任务 = Ok(Ok(()))

start_rollback 返回 Err，Drop 的 expect 把它变成 panic。这里用 catch_unwind 接住了；在正常代码里这是一次 drop 时的 panic，外层的回滚也没排进去。

transaction 闭包与 TransactionError

TransactionTrait::transaction 要的闭包签名是 FnOnce(&'c Self::Transaction) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'c>>：返回的 future 借用事务，所以要装箱，调用处写成 Box::pin(async move { ... })。Postgres 上它先 acquire、new_postgres，然后交给 DatabaseTransaction::run：

        let res = callback(&self).await.map_err(TransactionError::Transaction);
        if res.is_ok() {
            self.commit().await.map_err(TransactionError::Connection)?;
        } else {
            self.rollback()
                .await
                .map_err(TransactionError::Connection)?;
        }
        res

闭包返回 Err 就当场 rollback().await，不走 Drop 排队；闭包 panic 时这两支都不执行，回滚回到 Drop 那条路。返回的 TransactionError 只有两支：Connection(DbErr) 是 BEGIN / COMMIT / ROLLBACK 自己失败，Transaction(E) 是闭包返回的错误。E 只要求 Display + Debug + Send，可以是自己的错误类型（1.1.12 之前要求 std::error::Error，#2602 放宽成现在这样，为的是让 anyhow::Error 也能用）：

// E
let r: Result<(), TransactionError<OutOfStock>> = db
  .transaction(|txn| {
    Box::pin(async move {
      add(txn, 10).await.map_err(|_| OutOfStock)?;
      Err(OutOfStock)
    })
  })
  .await;
println!("闭包返回");

// F：闭包里的语句自己失败
let r: Result<(), TransactionError<DbErr>> = db
  .transaction(|txn| {
    Box::pin(async move {
      txn
        .execute_raw(Statement::from_string(DbBackend::Postgres, "insert into item (n) values (null)"))
        .await?;
      Ok(())
    })
  })
  .await;

--- E: transaction 闭包返回 Err ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_1, "INSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\""), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23), RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x0000000a), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("ROLLBACK")
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
闭包返回
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
结果 Err(Transaction(OutOfStock))
Display: out of stock
--- F: 闭包里的 DbErr 和 BEGIN 失败分不开吗 ---
  ...
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2, ), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, ErrorResponse
  [conn 1] S->C ReadyForQuery(E)
  [conn 1] C->S Query("ROLLBACK")
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
Transaction 分支: Execution Error: error returned from database: null value in column "n" of relation "item" violates not-null constraint at line 2057

ROLLBACK 在 transaction 返回之前就发出了，和 C 段的排队不同。Display 直接转给里面的错误，看不出是哪一支；要区分得 match。闭包里的 ? 把 DbErr 变成 E，所以 F 里数据库报的错落在 Transaction 分支，Connection 只装事务控制语句本身的失败。DbErr 的各个分支和 Display 见 错误。

2.0 的 transaction_async

2.0.0-rc.39（#3074）给 DatabaseConnection 和 DatabaseTransaction 加了固有方法 transaction_async / transaction_with_config_async，闭包是 AsyncFnOnce(&'c DatabaseTransaction) -> Result<T, E>，Rust 的异步闭包可以直接借用参数，不用装箱：

// G
let n = db
  .transaction_async(async |txn| {
    add(txn, 11).await?;
    add(txn, 12).await?;
    Ok::<_, DbErr>(2)
  })
  .await
  .map_err(|e| DbErr::Custom(e.to_string()))?;

--- G: transaction_async，返回 Ok ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x0000000b), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x0000000c), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
transaction_async 返回 2

它是先 self.begin()，再交给 run_async_transaction_callback，提交和回滚的逻辑和 run 一字不差。它是固有方法，不在 TransactionTrait 里，泛型函数 fn f<C: TransactionTrait> 里调不到，只能用装箱的 transaction。

TransactionTrait：一个函数同时收 &db 和 &txn

TransactionTrait：

#[async_trait::async_trait]
pub trait TransactionTrait: Sync {
    /// The concrete type for the transaction
    type Transaction: ConnectionTrait
        + TransactionTrait<Transaction = Self::Transaction>
        + TransactionSession
        + Send;

    /// Execute SQL `BEGIN` transaction.
    /// Returns a Transaction that can be committed or rolled back
    async fn begin(&self) -> Result<Self::Transaction, DbErr>;
    // ...
}

commit / rollback 在 TransactionSession 里。实现它的有 DatabaseConnection、DatabaseTransaction、DatabaseExecutor（两者之一的引用或一个拥有的事务，迁移的 SchemaManager 用）、RBAC 的 RestrictedConnection / RestrictedTransaction。只执行语句的函数收 C: ConnectionTrait 就够了，本页实验里的 add 就是这样，&db、&txn 都能传；函数里还要开事务（或保存点）才需要 TransactionTrait。下面这个函数每层插一行再往下开一层，递归地收 C、C::Transaction：

fn nest<'a, C>(c: &'a C, depth: i32) -> Pin<Box<dyn Future<Output = Result<(), DbErr>> + Send + 'a>>
where
  C: ConnectionTrait + TransactionTrait,
{
  Box::pin(async move {
    item::ActiveModel { n: Set(depth), ..Default::default() }.insert(c).await?;
    if depth == 0 {
      return Ok(());
    }
    let t = c.begin().await?;
    nest(&t, depth - 1).await?;
    t.commit().await
  })
}

// <C::Transaction as TransactionTrait>::Transaction is the same type as C::Transaction
fn same<C: TransactionTrait>(t: <C::Transaction as TransactionTrait>::Transaction) -> C::Transaction {
  t
}

let txn = db.begin().await?;
nest(&txn, 3).await?;
let txn = same::<sea_orm::DatabaseConnection>(txn);
txn.commit().await?;

--- O: 泛型递归 nest(&db, 3) ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000003), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000002), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_2")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000001), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_3")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000000), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_3")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_2")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)

保存点按深度编号，_sqlx_savepoint_1 到 _3，释放时倒着来。same 能编译，说明 <C::Transaction as TransactionTrait>::Transaction 和 C::Transaction 在类型系统里就是同一个类型；nest 递归下去，泛型参数在 C::Transaction 停住，不会越套越长。

📝NOTE
冷知识：「事务的事务」是 2.0.1 才钉住的

2.0.0 的 TransactionTrait 是 pub trait TransactionTrait，关联类型只要求 ConnectionTrait + TransactionTrait + TransactionSession，没有 Sync、Send，也没说 Transaction 的 Transaction 是谁。#3147 报告新实体格式编译不过：#[sea_orm::model] 处报「future returned by cascade_delete is not Send」。#3153 的说明解释了原因：生成的方法「open a transaction and save related models inside it, so action::<C> recurses into action::<C::Transaction>」，没有具体的 C 时证明 future 是 Send 要沿着「the unbounded C::Transaction::Transaction::… chain」走下去，于是溢出（E0275 / clippy::future_not_send）。修法放在 trait 上：要求 Sync，并把 Transaction 钉成一个 Send 的不动点，「collapsing the chain to one type」，也就是上面 TransactionTrait<Transaction = Self::Transaction> 那一行。CHANGELOG 记在 2.0.1 的 Bug Fixes，提醒自己实现 TransactionTrait 的人：Transaction::Transaction = Transaction 必须成立。

事务开着时用了 &db

&db 和 &txn 都实现 ConnectionTrait，传错了编译器不会拦。用 &db 的语句照常去池里 acquire 另一条连接，后果按池的大小分三种：

// I：max_connections(1)，acquire_timeout(1s)
let txn = db.begin().await?;
item::ActiveModel { n: Set(1), ..Default::default() }.insert(&txn).await?;
let r = item::Entity::find().count(&db).await;
txn.commit().await?;

// J：max_connections(2)
let txn = db.begin().await?;
item::ActiveModel { n: Set(2), ..Default::default() }.insert(&txn).await?;
println!("count(&txn) = {}", item::Entity::find().count(&txn).await?);
println!("count(&db)  = {}", item::Entity::find().count(&db).await?);

// K：接着 J，&txn 改了 id = 1 那一行，再用 &db 改同一行
item::Entity::update_many()
  .col_expr(item::Column::N, Expr::value(100))
  .filter(item::Column::Id.eq(1))
  .exec(&txn)
  .await?;
let r = tokio::time::timeout(
  Duration::from_secs(3),
  item::Entity::update_many()
    .col_expr(item::Column::N, Expr::value(200))
    .filter(item::Column::Id.eq(1))
    .exec(&db),
)
.await;
txn.commit().await?;

--- I: max_connections(1)，事务开着时用 &db ---
count(&db) = Err(ConnectionAcquire(Timeout))，用时 1.0s
Display: Failed to acquire connection from pool: Connection pool timed out
--- J: max_connections(2)，同一个任务里 &txn 和 &db ---
count(&txn) = 2
count(&db)  = 1
--- K: &db 去改 &txn 改过的行 ---
update(&db) = Err(Elapsed(()))，用时 3.0s
txn 提交之后 item.n: [200, 2]
  • I：唯一的连接在事务手里，&db 等的是自己占着的那条，acquire_timeout 到了报 ConnectionAcquire(Timeout)。ConnectOptions 不设这一项时用 sqlx 池自己的默认值（见 DatabaseConnection 与 ConnectOptions）。sqlx 里同样的情形见 pool timed out。
  • J：池有空位，&db 拿到另一条连接，看不到事务里还没提交的那一行。
  • K：这次不是池的问题，&db 的 UPDATE 在服务端等 &txn 持有的行锁，而 &txn 要等这个任务往下走才会提交：同一个任务里自己等自己，没有超时就永远等下去。更麻烦的是，timeout 丢掉的只是客户端的 future，服务端那条 UPDATE 还在排队；txn 提交、锁释放之后它在自动提交模式下执行了，最后的值是被「取消」的 200，不是事务写的 100。
📝NOTE
没有蠢问题

问： 函数签名写成 C: ConnectionTrait，调用方传了 &db 而不是 &txn，有没有办法在编译期拦住？

答： 在 ConnectionTrait 这一层没有：它描述的是「能执行语句」，池和事务都能。要拦，得让函数收具体类型 &DatabaseTransaction（这样它也就不能在事务外调用了），或者让需要原子性的函数自己开事务：收 C: TransactionTrait，在里面 c.begin()。调用方传 &txn 时，里面开的是保存点，仍在调用方的事务里；传 &db 时，它自己是一个完整的事务。

取消 begin：继承 sqlx 0.9.0 的问题

sea-orm 的 begin 和 sqlx 的一样先把 open 设成 true，最后靠 sqlx 的 start_rollback 回滚，而 start_rollback 只看连接上的深度。sqlx 0.9.0 里 BEGIN 发出、回应没到时取消，深度还是 0，两层守卫（sqlx 驱动内部的 Rollback 和 sea-orm 的 Drop）都什么也不排，见 超时取消 pool.begin()。实验把代理改成收到含 CommandComplete("BEGIN") 或 ("SAVEPOINT") 的回应后先压 300 毫秒再转发，再用 100 毫秒的 timeout 包住 begin：

// L
let r = tokio::time::timeout(Duration::from_millis(100), db.begin()).await;
// 关掉延迟，200ms × 2 之后
item::ActiveModel { n: Set(1), ..Default::default() }.insert(&db).await?;
println!("insert(&db) 返回 Ok；从另一条连接看 item.n: {:?}", rows().await?);
db.close().await?;

// M：新开一个池，外层事务里取消嵌套 begin
let txn = db.begin().await?;
item::ActiveModel { n: Set(5), ..Default::default() }.insert(&txn).await?;
let r = tokio::time::timeout(Duration::from_millis(100), txn.begin()).await;
item::ActiveModel { n: Set(6), ..Default::default() }.insert(&txn).await?;
let c = txn.commit().await;

--- L: db.begin() 在 BEGIN 来回途中被超时取消 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
timeout(100ms, db.begin()) = Err(Elapsed(()))
  [conn 1] C->S Sync
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] S->C ReadyForQuery(T)
下一次从池里用：
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_1, "INSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\""), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23), RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000001), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
insert(&db) 返回 Ok；从另一条连接看 item.n: []
  [conn 1] C->S Terminate
db.close() 之后 item.n: []
--- M: txn.begin()（SAVEPOINT）在来回途中被取消 ---
  ...
  [conn 2] C->S Query("BEGIN")
  [conn 2] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  ...
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000005), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 2] C->S Query("SAVEPOINT _sqlx_savepoint_1")
timeout(100ms, txn.begin()) = Err(Elapsed(()))
  [conn 2] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 2] C->S Query("ROLLBACK")
  [conn 2] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000006), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(I)
  [conn 2] C->S Sync
  [conn 2] S->C ReadyForQuery(I)
外层 commit = Ok(())
--- 最终 item.n: [6]
  • L：取消之后连接回池，归还时的 ping 收到 ReadyForQuery(T) 照单全收。下一个用 &db 的人以为在自动提交，实际写进了一个没人管的事务：insert 返回 Ok，从别的连接看不到；db.close() 发 Terminate，服务端回滚，这一行彻底没了。和 sqlx 0.9.0 的现象一模一样，上游是 #4393。
  • M：取消的是保存点。sqlx 的 Rollback 守卫在深度 1 上调 start_rollback，排进去的是 ROLLBACK：外层事务的回滚，深度变成 0。内层 DatabaseTransaction 的 Drop 看到深度 0，什么也不做。外层随后的 INSERT 6 在自动提交模式下执行，commit 看到深度 0 什么都不发，返回 Ok(())。结果 5 丢了、6 绕过了事务，这是 sqlx 那边深度大于 0 的 #4445。

sea-orm 2.0.4 锁的是 sqlx 0.9.0，两种取消都原样继承。sqlx 卷给的绕法（先 acquire 拿到 PoolConnection，在它上面带超时 begin，超时就 close）在 sea-orm 里用不上：acquire 和 BEGIN 都在 begin 内部，从一条现成的 PoolConnection 造 DatabaseTransaction 的 new_postgres 是 pub(crate)。第 0 层的那一种可以在池上兜住：sea-orm 自己不设 after_release，用 map_sqlx_postgres_pool_opts 挂一个检查「连接还在事务里」的 after_release，就能把带着事务回来的连接关掉，做法和实测见 常见坑：取消 begin。深度大于 0 的那一种兜不住（连接回来时已经是空闲状态），只能别让 begin 半路被丢掉：不给 db.begin() / txn.begin() 单独套 timeout，select! 里也不要拿它和别的分支赛跑。

动脑筋答案

--- N: 内层还活着，先提交外层 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_1, "INSERT INTO \"item\" (\"n\") VALUES ($1) RETURNING \"id\", \"n\""), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23), RowDescription, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000001), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000002), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
外层 commit = Ok(())
  [conn 1] C->S Query("ROLLBACK"), Sync
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 1] S->C ReadyForQuery(I)
drop(inner) 之后
--- 最终 item.n: []

表是空的，两行都没了。

  • txn.commit() 调 sqlx 的 commit，sqlx 只看连接上的深度：这时是 2，于是发的是 RELEASE SAVEPOINT _sqlx_savepoint_1，提交的是内层的保存点，深度变成 1。txn 的 open 设成 false，commit 返回 Ok(())。
  • txn 被 drop 时 open 已是 false，什么也不做；连接还被 inner 的 Arc 拿着，不归还，服务端仍在事务里。
  • drop(inner)：inner.open 还是 true，start_rollback 按深度 1 排进 ROLLBACK，这是整个事务的回滚。随后最后一个 Arc 消失，归还时的 ping 把它发了出去，1 和 2 一起回滚。

外层的 commit 返回 Ok，提交的却是内层。sea-orm 的嵌套事务不借用外层，编译器不检查先后，内层必须在外层 commit 之前 commit 或 rollback。MockDatabase 里同样的顺序记下的是 BEGIN、SAVEPOINT savepoint_1、RELEASE SAVEPOINT savepoint_1、ROLLBACK，见 Mock 与 Proxy。

小结

  • DatabaseTransaction 是一条 PoolConnection 放在 Arc<Mutex<InnerConnection>> 里加一个 open 标记；事务里的语句锁住这条连接执行，不经过池。深度记在 sqlx 的连接上，BEGIN、SAVEPOINT _sqlx_savepoint_N、COMMIT、RELEASE、ROLLBACK 都由 sqlx 按深度生成。
  • begin_with_config 不用 sqlx 的 begin_with：先 BEGIN，再发一条 SET TRANSACTION ISOLATION LEVEL ... READ ONLY（2.0.0-rc.28 起合成一条，走扩展协议、进语句缓存）。在事务里调它，保存点开了但 SET TRANSACTION 报 25001，Drop 回滚到保存点，外层不受影响。
  • 嵌套事务克隆 Arc，不借用外层；内层还活着时提交外层，commit 返回 Ok 却只释放了保存点，内层 drop 时把整个事务回滚。
  • 没提交就 drop：start_rollback 把 ROLLBACK 排进写缓冲，最后一个 Arc 消失、连接归还时的 ping 把它发出去；sqlx 日志里看不到。连接被别的任务锁着时，Drop 会 panic。
  • transaction(|txn| Box::pin(async move { .. })) 按闭包的结果当场提交或回滚；TransactionError::Transaction 装闭包的错误（包括闭包里语句的 DbErr），Connection 只装事务控制语句的失败。2.0 的 transaction_async 收异步闭包，是固有方法，泛型代码里用不了。
  • 函数收 C: ConnectionTrait 能同时接 &db 和 &txn，传错了编译器不拦：池小时等到 ConnectionAcquire(Timeout)，池大时看不到未提交的写入，碰到同一行时自己等自己的行锁。
  • 取消 begin 继承 sqlx 0.9.0 的 #4393 和 #4445：取消 BEGIN 让连接带着事务回池，取消保存点会回滚外层，外层 commit 还返回 Ok。

💬 关联讨论

发起讨论
加载中...