A heavy query early stopped may lead getting connection from pool timeout

[lewiszlw]

bb8-postgres: v0.9

    let mut config = bb8_postgres::tokio_postgres::config::Config::new();
    config
        .host(&options.host)
        .port(options.port)
        .user(&options.username)
        .password(&options.password);
    if let Some(database) = &options.database {
        config.dbname(database);
    }
    let manager = PostgresConnectionManager::new(config, NoTls);
    let pool = bb8::Pool::builder()
        .max_size(options.pool_max_size as u32)
        .build(manager)
        .await
        .map_err(|e| {
            DataFusionError::Execution(format!(
                "Failed to create postgres connection pool due to {e}",
            ))
        })?;

For example, I execute a heavy query (full table scan) and only fetch first row then drop the stream, later if I try to get a new connection from pool, I will get timeout error.

[djc]

I think I need a minimal reproduction for this. I suspect the timeout you're seeing is not from bb8 while getting a connection, but is from trying the reacquired connection to do a new thing.

[lewiszlw]

Here https://github.com/systemxlabs/datafusion-remote-table is my project. I'll try to give a minimal reproduction next Monday.

[lewiszlw]

Here is a minimal reproduction @djc

// [dependencies]
// bb8 = { version = "0.9" }
// bb8-postgres = { version = "0.9", features = ["with-chrono-0_4", "with-serde_json-1"] }
// futures = "0.3"
// tokio = { version = "1", features = ["full"] }

use bb8_postgres::tokio_postgres::NoTls;
use bb8_postgres::{PostgresConnectionManager, tokio_postgres};
use futures::StreamExt;

#[tokio::main]
async fn main() {
    let pool = connect().await;
    let conn = pool.get().await;
    fetch_one_row(conn).await;
    let new_conn = pool.get().await; // this line will cause panic: called `Result::unwrap()` on an `Err` value: TimedOut
    println!("Hello, world!");
}

async fn fetch_one_row(conn: PostgresConnection) {
    let mut stream = conn
        .conn
        .query_raw("select * from dltb3440", Vec::<String>::new())
        .await
        .unwrap()
        .chunks(1)
        .boxed();

    let Some(first_chunk) = stream.next().await else {
        panic!()
    };
    println!("First chunk: {:?}", first_chunk);
}

pub async fn connect() -> PostgresPool {
    let mut config = tokio_postgres::config::Config::new();
    config
        .host("xxx")
        .port(5432)
        .user("xxx")
        .password("xxx")
        .dbname("xxx");
    let manager = PostgresConnectionManager::new(config, NoTls);
    let pool = bb8::Pool::builder().build(manager).await.unwrap();

    PostgresPool { pool }
}

pub struct PostgresPool {
    pub pool: bb8::Pool<PostgresConnectionManager<NoTls>>,
}

impl PostgresPool {
    pub async fn get(&self) -> PostgresConnection {
        PostgresConnection {
            conn: self.pool.get_owned().await.unwrap(),
        }
    }
}

pub struct PostgresConnection {
    pub conn: bb8::PooledConnection<'static, PostgresConnectionManager<NoTls>>,
}