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

error returned from database: current transaction is aborted, commands ignored until end of transaction block at line 1750

p.delay_ms.store(300, Ordering::SeqCst);
let r = tokio::time::timeout(Duration::from_millis(timeout_ms), pool.begin()).await;
println!("timeout({timeout_ms}ms, pool.begin()) = {:?}", r.map(|r| r.map(|_| ())));
p.delay_ms.store(0, Ordering::SeqCst);
tokio::time::sleep(Duration::from_millis(500)).await;

sqlx::query("insert into pit2.items values (1)").execute(&pool).await?;
// 另一条连接查 pg_stat_activity.state 和 count(*)
let e = sqlx::query("insert into pit2.items values (1)").execute(&pool).await;
let e = sqlx::query_scalar::<_, i32>("select 1").fetch_one(&pool).await;
pool.close().await;

--- 默认配置（test_before_acquire 默认开） ---
（慢之前）select 1 = Ok
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Query("BEGIN")
timeout(450ms, pool.begin()) = Err(Elapsed(()))
  [conn 1] C->S Sync
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] S->C ReadyForQuery(T)
insert 1 返回 Ok；池里的连接 pid 1022 -> 1022
  服务端 state = Some("idle in transaction")，别的连接看到 0 行
再 insert 1（主键冲突）= error returned from database: duplicate key value violates unique constraint "items_pkey" at line 666
然后 select 1 = error returned from database: current transaction is aborted, commands ignored until end of transaction block at line 1750
pool.close() 之后别的连接看到 0 行

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

let mut conn = pool.acquire().await?;
let r = tokio::time::timeout(Duration::from_millis(100), conn.begin()).await;
let timed_out = r.is_err();
drop(r);
if timed_out {
  conn.close().await?;
}

--- 先 acquire，超时只套在 begin 上，超时就 close ---
  [conn 6] C->S Query("BEGIN")
timeout(100ms, conn.begin()) 超时 = true
  [conn 6] C->S Terminate
  [conn 6] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
insert 1 返回 Ok；池里的连接 pid 1027 -> 1028
  别的连接看到 1 行

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

let pool = PgPoolOptions::new()
  .max_connections(1)
  .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)
    })
  })
  .connect_with(opts.clone())
  .await?;

--- after_release 探测 ---
（慢之前）select 1 = Ok
  [conn 2] C->S Sync
  [conn 2] S->C ReadyForQuery(I)
  [conn 2] C->S Query("BEGIN")
timeout(450ms, pool.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 1 返回 Ok；池里的连接 pid 1023 -> 1024
  服务端 state = Some("idle")，别的连接看到 1 行
再 insert 1（主键冲突）= error returned from database: duplicate key value violates unique constraint "items_pkey" at line 666
然后 select 1 = Ok(1)
pool.close() 之后别的连接看到 1 行

--- before_acquire 探测 ---
（慢之前）select 1 = Ok
  [conn 4] C->S Sync
  [conn 4] S->C 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 Query("BEGIN")
timeout(750ms, pool.begin()) = Err(Elapsed(()))
  [conn 4] C->S Sync
  [conn 4] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 4] S->C ReadyForQuery(T)
  before_acquire: 连接还在事务里，换一条
insert 1 返回 Ok；池里的连接 pid 1025 -> 1026
  服务端 state = Some("idle")，别的连接看到 1 行
再 insert 1（主键冲突）= error returned from database: duplicate key value violates unique constraint "items_pkey" at line 666
然后 select 1 = Ok(1)
pool.close() 之后别的连接看到 1 行