事务

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

    #[inline]
    fn is_in_transaction(&self) -> bool {
        <Self::Database as Database>::TransactionManager::get_transaction_depth(self) != 0
    }

pub fn begin_ansi_transaction_sql(depth: usize) -> SqlStr {
    if depth == 0 {
        "BEGIN".into_sql_str()
    } else {
        AssertSqlSafe(format!("SAVEPOINT _sqlx_savepoint_{depth}")).into_sql_str()
    }
}

pub fn commit_ansi_transaction_sql(depth: usize) -> SqlStr {
    if depth == 1 {
        "COMMIT".into_sql_str()
    } else {
        AssertSqlSafe(format!("RELEASE SAVEPOINT _sqlx_savepoint_{}", depth - 1)).into_sql_str()
    }
}

    async fn begin(conn: &mut PgConnection, statement: Option<SqlStr>) -> Result<(), Error> {
        let depth = conn.inner.transaction_depth;

        let statement = match statement {
            // custom `BEGIN` statements are not allowed if we're already in
            // a transaction (we need to issue a `SAVEPOINT` instead)
            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.as_str())?;
        rollback.conn.wait_until_ready().await?;
        if !rollback.conn.in_transaction() {
            return Err(Error::BeginFailed);
        }
        rollback.conn.inner.transaction_depth += 1;
        rollback.defuse();

        Ok(())
    }

let mut tx = pool.begin().await?;
sqlx::query("insert into items values ($1)").bind(1_i64).execute(&mut *tx).await?;
let sp = tx.begin().await?;
sp.commit().await?;
let sp = tx.begin().await?;
sp.rollback().await?;
tx.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)
depth 1 begin 返回
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
depth 2 begin 返回
  [conn 1] C->S Query("RELEASE SAVEPOINT _sqlx_savepoint_1")
  [conn 1] S->C CommandComplete("RELEASE"), ReadyForQuery(T)
depth 2 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)
depth 2 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)
depth 1 commit 返回

    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);
        }
    }

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

            conn.inner.transaction_depth -= 1;
        }
    }

// B: pool connection
let mut tx = pool.begin().await?;
sqlx::query("insert into items values ($1)").bind(2_i64).execute(&mut *tx).await?;
drop(tx);

// C: bare PgConnection
let mut tx = conn.begin().await?;
sqlx::query("insert into items values ($1)").bind(3_i64).execute(&mut *tx).await?;
drop(tx);
println!("is_in_transaction = {}", conn.is_in_transaction());
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
let n: i64 = sqlx::query_scalar("select count(*) from items").fetch_one(&mut conn).await?;

--- B: 池连接上的 Transaction 不提交就 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), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
drop(tx)
  [conn 1] C->S Query("ROLLBACK"), Sync
  [conn 1] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 1] S->C ReadyForQuery(I)
drop 之后 200ms
--- C: 裸 PgConnection 上不提交就 drop ---
  [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), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
drop(tx)
is_in_transaction = false
drop 之后 500ms，下一次使用：
  [conn 2] C->S Query("ROLLBACK")
  [conn 2] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 2] C->S Parse(sqlx_s_3, "select count(*) from items"), Describe(S, sqlx_s_3), Sync
  [conn 2] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_3), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 8), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
count = 1

// D
let mut tx = conn.begin_with("BEGIN ISOLATION LEVEL SERIALIZABLE").await?;
let level: String = sqlx::query_scalar("select current_setting('transaction_isolation')")
  .fetch_one(&mut *tx)
  .await?;
tx.commit().await?;

// E
let r = conn.begin_with("SELECT 1").await;

--- D: begin_with 指定隔离级别 ---
  [conn 2] C->S Query("BEGIN ISOLATION LEVEL SERIALIZABLE")
  [conn 2] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 2] C->S Parse(sqlx_s_4, "select current_setting('transaction_isolation')"), Describe(S, sqlx_s_4), Sync
  [conn 2] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(T)
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_4), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, DataRow(列长 12), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(T)
  [conn 2] C->S Query("COMMIT")
  [conn 2] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
transaction_isolation = serializable
--- E: begin_with 不是 BEGIN ---
  [conn 2] C->S Query("SELECT 1")
  [conn 2] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
结果 Err(BeginFailed)

        Box::pin(async move {
            let mut tx = Self {
                connection: conn,

                // If the call to `begin` fails or doesn't complete we want to attempt a rollback in case the transaction was started.
                open: true,
            };

            DB::TransactionManager::begin(&mut tx.connection, statement).await?;

            Ok(tx)
        })

// H
let polled = conn.begin().now_or_never().map(|r| r.map(|_| ()));
println!("now_or_never = {polled:?}");
println!("is_in_transaction = {}", conn.is_in_transaction());
sqlx::query("insert into items values ($1)").bind(7_i64).execute(&mut conn).await?;
conn.close().await?;

// I
let mut pc = pool.acquire().await?;
let polled = pc.begin().now_or_never().map(|r| r.map(|_| ()));
drop(pc);
sqlx::query("insert into items values ($1)").bind(8_i64).execute(&pool).await?;
pool.close().await;

--- H: begin 只 poll 一次就丢掉 ---
now_or_never = None
  [conn 2] C->S Query("BEGIN")
  [conn 2] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
is_in_transaction = false
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
关闭连接
  [conn 2] C->S Terminate
--- I: 池连接上 begin 只 poll 一次就丢掉，然后归还 ---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
now_or_never = None
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
drop(pc)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
下一次从池里用：
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(T)
pool.close()
  [conn 1] C->S Terminate
--- 最终 tx.items: [1, 5]

    fn transaction<'a, F, R, E>(
        &'a mut self,
        callback: F,
    ) -> impl Future<Output = Result<R, E>> + Send + 'a
    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,
    {
        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)
                }
            }
        }
    }

let r: Result<(), sqlx::Error> = conn
  .transaction(|t| {
    Box::pin(async move {
      sqlx::query("insert into items values ($1)").bind(6_i64).execute(&mut **t).await?;
      Err(sqlx::Error::RowNotFound)
    })
  })
  .await;

--- G: Connection::transaction 闭包，返回 Err ---
  [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), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 2] C->S Query("ROLLBACK")
  [conn 2] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
结果 Err(RowNotFound)

--- F: 事务里再 begin_with ---
  [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), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
结果 Err(InvalidSavePointStatement)
外层 is_in_transaction = false
  [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), Execute, Close(P, unnamed), Sync
  [conn 2] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(I)
外层 commit 结果 Ok(())