超时取消 db.begin()：连接带着事务回到池里

p.slow_begin.store(true, Ordering::SeqCst);
let r = tokio::time::timeout(Duration::from_millis(100), db.begin()).await;
println!("timeout(100ms, db.begin()) = {:?}", r.map(|x| x.map(|_| ())));
// 关掉延迟，等归还完成
item::ActiveModel { n: Set(1), ..Default::default() }.insert(&db).await?;
println!("insert(&db) 返回 Ok；pid {again} -> {after}；从另一条连接看 item.n: {:?}", rows().await?);
db.close().await?;

--- L: db.begin() 在 BEGIN 来回途中被超时取消 ---
正常归还两次：pid 5753 -> 5753
  [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)
insert(&db) 返回 Ok；pid 5753 -> 5753；从另一条连接看 item.n: []
db.close() 之后 item.n: []

                    InnerConnection::Postgres(c) => {
                        <sqlx::Postgres as sqlx::Database>::TransactionManager::begin(c, None)
                            .await
                            .map_err(sqlx_error_to_query_err)?;

const PROBE: &str = "select transaction_timestamp() = statement_timestamp()";

opt.map_sqlx_postgres_pool_opts(|o| {
  o.after_release(|conn, _meta| {
    Box::pin(async move {
      let clean: bool = sqlx::raw_sql(PROBE).fetch_one(&mut *conn).await?.try_get(0)?;
      if !clean {
        println!("  after_release: 连接还在事务里，关掉");
      }
      Ok(clean)
    })
  })
});

--- P: 同上，map_sqlx_postgres_pool_opts 挂 after_release 探测 ---
正常归还两次：pid 5758 -> 5758
  [conn 2] C->S Sync
  [conn 2] S->C ReadyForQuery(I)
  [conn 2] C->S Query("BEGIN")
timeout(100ms, db.begin()) = Err(Elapsed(()))
  [conn 2] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 2] C->S Query("select transaction_timestamp() = statement_timestamp()")
  [conn 2] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(T)
  after_release: 连接还在事务里，关掉
  [conn 2] C->S Terminate
insert(&db) 返回 Ok；pid 5758 -> 5759；从另一条连接看 item.n: [1]
db.close() 之后 item.n: [1]

--- Q: 带探测时，一条 &db 语句 ---
  [conn 4] C->S Sync
  [conn 4] S->C ReadyForQuery(I)
  [conn 4] C->S Parse(sqlx_s_2, "select pg_backend_pid() as p"), Describe(S, sqlx_s_2), Sync
  [conn 4] S->C ParseComplete, ParameterDescription(OID ), RowDescription, ReadyForQuery(I)
  [conn 4] C->S Bind(portal unnamed, stmt sqlx_s_2, ), Execute, Close(P, unnamed), Sync
  [conn 4] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 4] C->S Query("select transaction_timestamp() = statement_timestamp()")
  [conn 4] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
  [conn 4] C->S Sync
  [conn 4] S->C ReadyForQuery(I)

let url = format!("{}&options=-c%20search_path%3D{SCHEMA}", proxied_url(port));
let pool = sqlx::postgres::PgPoolOptions::new()
  .max_connections(1)
  .after_release(|conn, _meta| {
    Box::pin(async move {
      let clean: bool = sqlx::Executor::fetch_one(&mut *conn, PROBE).await?.try_get(0)?;
      Ok(clean)
    })
  })
  .connect(&url)
  .await
  .map_err(|e| DbErr::Custom(e.to_string()))?;
let db = SqlxPostgresConnector::from_sqlx_postgres_pool(pool);

--- M: txn.begin()（SAVEPOINT）在来回途中被取消，带探测 ---
  [conn 4] C->S Query("SAVEPOINT _sqlx_savepoint_1")
timeout(100ms, txn.begin()) = Err(Elapsed(()))
  [conn 4] S->C CommandComplete("SAVEPOINT"), ReadyForQuery(T)
  [conn 4] C->S Query("ROLLBACK")
  [conn 4] S->C CommandComplete("ROLLBACK"), ReadyForQuery(I)
  [conn 4] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000006), Execute, Close(P, unnamed), Sync
  [conn 4] S->C BindComplete, DataRow(列长 4/4), CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(I)
  [conn 4] C->S Query("select transaction_timestamp() = statement_timestamp()")
  [conn 4] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
  [conn 4] C->S Sync
  [conn 4] S->C ReadyForQuery(I)
外层 commit = Ok(())
--- 最终 item.n: [6]