Help investigating connection closed error

[ufoscout]

Hi @sfackler,
I am migrating a web server from rust-postgres 0.15 to tokio-postgres 0.17 and I am completely stuck with a weird issue. In fact, in the tokio-postgres based application, I receive tons of connection-closed errors and I have no idea why.

Sorry if I don't provide any explicit example but I was not able to isolate the issue; if you could provide me with some hints on how to investigate or on what could be a possible cause that would really help.

The application never calls connection.close(), there are no network issues, I tried with both bb8_postgres and deadpool_postgres, it is not a problem of timeouts as the error appears immediately as the application starts and the errors are completely random.
Every time I get a new connection I check client.is_closed() and it always returns false.

The original version rust-postgres 0.15 with r2d2 keeps working like a charm.

[sfackler]

What do you mean by connection.close()?

You could check postgres logs to see if there's anything there. Otherwise it may be an issue of the pool failing to validate that a connection is still active before giving it out.

[ufoscout]

@sfackler

What do you mean by connection.close()?

I mean that I never explicitly call any method on the pool or on the client that closes the connection.

The postgres logs are full of this:

2020-04-03 08:31:40.608 UTC [248] LOG: unexpected EOF on client connection with an open transaction

[sfackler]

Is there anything interesting on in the logs on the tokio-postgres side?

[ufoscout]

@sfackler
I was able to create a reproducer. You can find it here:
https://github.com/ufoscout/tokio_postgres_err

You need a local postgres db (I tested it with docker image of postgres 11).

There are three tests that should all succeed, but often the test test_1_good_tx fails when it tries to commit the transaction with this error:

thread 'test_1_good_tx' panicked at 'called Result::unwrap() on an Err value: Error { kind: Closed, cause: None }'

In the reproducer I used deadpool_postgres, but I can reproduce the very same error with bb8_postgres so I suppose the issue is not in the pool.

Sorry if the reproducer is quite verbose, I wasn't able to find anything smaller than that.

[sfackler]

You are sharing a global connection pool between the two tests, but the runtimes associated with the connections are being shut down when the test that caused them to be created completes.

[ufoscout]

@sfackler
thanks for the explanation, I was going mad with it.
Do you know if there is a way to use a global executor to avoid this issue?

[sfackler]

That would be something that you'd need to configure in the connection pool. I would generally recommend against using global state though.

[dunnock]

Had the same issue, thanks that suggestion helped! It's not trivial to figure out that connection is actually tied to runtime, nevertheless you still own the handle. Connection is spawned in the runtime while handle is a channel to that.

[seanlinsley]

Adding more to this thread so it's a bit more clear. Be careful where you start Tokio runtimes!

I happen to be using sidekiq-rs to implement a Sidekiq server in Rust. It uses the threadpool crate instead of Tokio, so workers don't have access to the Tokio runtime. I initially wrapped the worker function in #[tokio::main] so it would be able to create database connections if needed, but that lead the connection to be dropped after every job even though it was still in the connection pool.

In order to maintain the connections for the lifetime of the program, I'm now booting every DB connection inside the program's main function. An improved version would spawn a Tokio worker that refills the connection pool if any are dropped.

static DB_POOL: Lazy<Arc<deadpool::Pool>> = Lazy::new(|| {
    let url = env::var("DATABASE_URL").expect("DATABASE_URL is missing");
    let pg_config = tokio_postgres::Config::from_str(&url).unwrap();
    let mgr_config = deadpool::ManagerConfig {
        recycling_method: deadpool::RecyclingMethod::Fast
    };
    let mgr = deadpool::Manager::from_config(pg_config, NoTls, mgr_config);
    let pool = deadpool::Pool::builder(mgr).max_size(*POOL_SIZE).build().unwrap();
    Arc::new(pool)
});

#[tokio::main]
async fn main() {
    // Start all connections within the Tokio runtime so they last for the lifetime of the program
    // TODO: run a loop in a Tokio worker that makes sure the pool is always at full size
    futures::future::try_join_all((0..*POOL_SIZE).map(|_| DB_POOL.get())).await.unwrap();

    // Sidekiq server boots here
}