pool timed out while waiting for an open connection

async fn small_pool(max: u32) -> anyhow::Result<PgPool> {
  Ok(
    PgPoolOptions::new()
      .max_connections(max)
      .acquire_timeout(Duration::from_secs(1))
      .connect_with(exp_sqlx_pit1::opts())
      .await?,
  )
}

// A helper that takes the pool, called while the caller holds a transaction.
async fn item_count(pool: &PgPool) -> sqlx::Result<i64> {
  sqlx::query_scalar("select count(*) from item").fetch_one(pool).await
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  // 1. transaction open, helper uses &pool
  let pool = small_pool(1).await?;
  let mut tx = pool.begin().await?;
  sqlx::query("update item set note = 'x' where id = 1").execute(&mut *tx).await?;
  let t = Instant::now();
  let r = item_count(&pool).await.map_err(|e| e.to_string());
  println!("1 tx + &pool:     {:?} after {:?}  size={} idle={}", r, t.elapsed(), pool.size(), pool.num_idle());
  tx.rollback().await?;

  // 2. stream not finished, second query on &pool inside the loop
  let mut rows = sqlx::query("select id from item order by id").fetch(&pool);
  let t = Instant::now();
  while let Some(row) = rows.try_next().await? {
    let id: i64 = row.get("id");
    let r = sqlx::query("update item set note = 'y' where id = $1").bind(id).execute(&pool).await.map_err(|e| e.to_string());
    println!("2 inside fetch:   id={id} {:?} after {:?}", r.map(|r| r.rows_affected()), t.elapsed());
    break;
  }
  drop(rows);

  // 3. max_connections(3), 3 tasks each hold a transaction and call the helper
  let pool = small_pool(3).await?;
  let t = Instant::now();
  let mut tasks = Vec::new();
  for i in 0..3 {
    let pool = pool.clone();
    tasks.push(tokio::spawn(async move {
      let tx = pool.begin().await?;
      tokio::time::sleep(Duration::from_millis(50)).await;
      let r = item_count(&pool).await.map_err(|e| e.to_string());
      drop(tx);
      Ok::<_, sqlx::Error>((i, r))
    }));
  }
  tokio::time::sleep(Duration::from_millis(500)).await;
  println!("3 while waiting:  size={} idle={}", pool.size(), pool.num_idle());
  for task in tasks {
    let (i, r) = task.await??;
    println!("3 task {i}:         {:?} after {:?}", r, t.elapsed());
  }
  Ok(())
}

1 tx + &pool:     Err("pool timed out while waiting for an open connection") after 1.001609459s  size=1 idle=0
2 inside fetch:   id=1 Err("pool timed out while waiting for an open connection") after 1.008053958s
3 while waiting:  size=3 idle=0
3 task 0:         Err("pool timed out while waiting for an open connection") after 1.053998083s
3 task 1:         Ok(3) after 1.060094667s
3 task 2:         Ok(3) after 1.061167208s

    fn fetch_many<'e, 'q: 'e, E>(
        self,
        query: E,
    ) -> BoxStream<'e, Result<Either<DB::QueryResult, DB::Row>, Error>>
    where
        E: 'q + Execute<'q, Self::Database>,
    {
        let pool = self.clone();

        Box::pin(try_stream! {
            let mut conn = pool.acquire().await?;
            let mut s = conn.fetch_many(query);

            while let Some(v) = s.try_next().await? {
                r#yield!(v);
            }

            Ok(())
        })
    }

server down: pool timed out while waiting for an open connection after 1.003259917s  size=0 idle=0

// Takes any executor: the pool, a transaction, or a connection.
async fn item_count(conn: impl PgExecutor<'_>) -> sqlx::Result<i64> {
  sqlx::query_scalar("select count(*) from item").fetch_one(conn).await
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  // 1. reuse the transaction's connection
  let pool = small_pool(1).await?;
  let mut tx = pool.begin().await?;
  sqlx::query("update item set note = 'x' where id = 1").execute(&mut *tx).await?;
  let t = Instant::now();
  let r = item_count(&mut *tx).await;
  println!("1 &mut *tx:       {:?} after {:?}", r, t.elapsed());
  tx.rollback().await?;

  // 2. read everything first, then write
  let t = Instant::now();
  let ids: Vec<i64> = sqlx::query_scalar("select id from item order by id").fetch_all(&pool).await?;
  for id in &ids {
    sqlx::query("update item set note = 'y' where id = $1").bind(id).execute(&pool).await?;
  }
  println!("2 fetch_all:      {} rows updated after {:?}", ids.len(), t.elapsed());

  // 3. three tasks, each passes its own transaction
  let pool = small_pool(3).await?;
  // ...
      let mut tx = pool.begin().await?;
      tokio::time::sleep(Duration::from_millis(50)).await;
      let r = item_count(&mut *tx).await;
      tx.commit().await?;
  // ...
}

1 &mut *tx:       Ok(3) after 1.048667ms
2 fetch_all:      3 rows updated after 8.511458ms
3 task 0:         Ok(3) after 57.47625ms
3 task 1:         Ok(3) after 89.657959ms
3 task 2:         Ok(3) after 89.664667ms