Connections leak on subscription cancellation

[stefannegele]

Bug Report

Versions

• Driver: r2dbc-postgresql-0.8.10.RELEASE
• Database: PostgreSQL
• Java: 14
• OS: macOS

Current Behavior

When a subscriber sends a cancellation signal upstream, connections might not get closed. This produces connections leaks which lead to non-responding applications in the long term.

Table schema

Input Code

create extension if not exists "uuid-ossp";

create table if not exists test_entity
(
    id uuid default public.uuid_generate_v4(),
    name text not null,
    primary key (id)
);

Steps to reproduce

Input Code

private val logger = LoggerFactory.getLogger("StressTest")

fun main(args: Array<String>) {
    runStressTest()
}

private fun runStressTest() {
    val pool = connectionPool()
    val client = DatabaseClient.create(pool)

    val threads = (1..5).map { runStressTestRequestThread(it, client) }
    threads.forEach { it.join() }
    Thread.sleep(20_000) // connection should be evicted after 10 seconds

    assert(pool.metrics.get().allocatedSize() == 0) // this fails
}

private fun runStressTestRequestThread(num: Int, databaseClient: DatabaseClient): Thread =
    thread(name = "stress-test-$num") {
        for (i in (1..1_000)) {
            runCatching {
                databaseClient.sql("select id, name from test_entity")
                    .then()
                    .timeout(Duration.ofMillis(15)) // cancel before answer
                    .block()
            }.onFailure {
                logger.debug(it.message, it)
            }
        }

        logger.info("done")
    }

private fun connectionPool(): ConnectionPool = ConnectionPoolConfiguration
    .builder(connectionFactory())
    .maxSize(1)
    .initialSize(0)
    .maxIdleTime(Duration.ofSeconds(1))
    .maxLifeTime(Duration.ofSeconds(1))
    .backgroundEvictionInterval(Duration.ofSeconds(1))
    .build()
    .let { ConnectionPool(it) }

private fun connectionFactory(): ConnectionFactory = ConnectionFactories.get(
    ConnectionFactoryOptions.parse("r2dbc:postgresql://localhost:5432/test?schema=test")
        .mutate()
        .option(ConnectionFactoryOptions.USER, "postgres")
        .option(ConnectionFactoryOptions.PASSWORD, "pass")
        .build()
)

I also created a demo where this occurs in combination with a netty based spring webflux setup here, that's how i found out about this. This example uses spring boot r2dbc.
https://github.com/stefannegele/connection-leak-demo

Expected behavior/code

Connection does not leak.

[mp911de]

We fixed the issue in R2DBC pool just yesterday with #140. There are a few potential leaks in Spring Framework so I'd ask you to file your ticket at https://github.com/spring-projects/spring-framework/issues.

[stefannegele]

Sorry for the duplicate. Did not catch the other issue, because it was raised while i was still investigating.
I will now wait for a new release of the pool and run my tests with the new version again. Thanks a lot!