Repository: jOOQ/jOOQ (language: java)

Issue:
R2DBC Connection leak when cancelling pending connection with ConnectionPool
### Expected behavior

When using DSLContext with an R2DBC ConnectionPool, connections should be properly managed throughout the reactive stream lifecycle. Specifically:

- If a subscription is canceled before the connection is fully established, the connection acquisition should be stopped.
- No connections should remain open or leak after the subscription is canceled.

### Actual behavior

Currently, when using DSLContext with an R2DBC ConnectionPool:

- Connections may still be acquired even after a subscription is canceled, leading to open connections that are not returned to the pool.
- Over time, this accumulation of unused connections depletes the pool.

### Steps to reproduce the problem

```
class ConnectionPoolLeakTest {
    companion object {
        private lateinit var connectionPool: ConnectionPool
        private lateinit var dslContext: DSLContext
        private lateinit var db: PostgreSQLContainer<*>

        @BeforeAll
        @JvmStatic
        fun setUp() {
            db = PostgreSQLContainer("postgres:latest")
                .withUsername("postgres")
                .withDatabaseName("postgres")
                .withPassword("postgres")
                .apply { start() }

            val baseConnectionFactory = ConnectionFactories.get(
                ConnectionFactoryOptions
                    .parse("r2dbc:postgresql://${db.host}:${db.firstMappedPort}/${db.databaseName}")
                    .mutate()
                    .option(ConnectionFactoryOptions.USER, db.username)
                    .option(ConnectionFactoryOptions.PASSWORD, db.password)
                    .build()
            )

            val connectionPoolConfiguration = ConnectionPoolConfiguration
                .builder(baseConnectionFactory)
                .initialSize(1)
                .maxSize(1) // maxSize = 1
                .build()

            connectionPool = ConnectionPool(connectionPoolConfiguration)
            dslContext = DSL.using(connectionPool)
        }

        @AfterAll
        @JvmStatic
        fun tearDown() {
            ResourceReaper.instance().stopAndRemoveContainer(db.containerId, db.dockerImageName)
        }
    }

    @Test
    fun test_ConnectionLeak_When_CancelingPendingRequests() {
        // Execute a query in a thread with a sufficient timeout to complete successfully
        val successThread = executeQueryInThread("SUCCESS-THREAD", Duration.ofMillis(3000L))

        // Wait for 500 milliseconds to ensure the above thread acquires a connection
        Thread.sleep(500L)

        // Execute a query in a thread with a very short timeout,
        // simulating a scenario where the request is cancelled before a connection is acquired. (pending state)
        val cancelThread = executeQueryInThread("CANCEL-THREAD", Duration.ofMillis(10L))

        // Wait for both threads to finish
        successThread.join()
        cancelThread.join()

        // Execute another query in a thread with a reasonable timeout to validate connection handling
        executeQueryInThread("VALIDATION-THREAD", Duration.ofMillis(3000L)).join()

        // Verify that the connection pool has no leaks
        assertEquals(0, connectionPool.metrics.get().acquiredSize())
        assertEquals(0, connectionPool.metrics.get().pendingAcquireSize())
        assertEquals(1, connectionPool.metrics.get().idleSize())
    }

    private fun executeQueryInThread(threadName: String, timeout: Duration): Thread {
        return thread(name = threadName) {
            // Executes a query that takes 1 second to complete.
            Mono.from(
                dslContext
                    .select(DSL.`val`(threadName))
                    .from(DSL.sql("pg_sleep(1)"))
                    .limit(1)
            ).log().timeout(timeout).block()
        }
    }
}
```

### jOOQ Version

jOOQ Community Edition 3.9.11

### Database product and version

PostgreSQL 11.16 (Debian 11.16-1.pgdg90+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 6.3.0-18+deb9u1) 6.3.0 20170516, 64-bit

### Java Version

openjdk version "17.0.1" 2021-10-19 LTS

### JDBC / R2DBC driver name and version (include name if unofficial driver)

org.postgresql:r2dbc-postgresql:1.0.5.RELEASE


Discussion / hints:
I will create a pull request for this.

Fix patch:
diff --git a/jOOQ/src/main/java/org/jooq/impl/R2DBC.java b/jOOQ/src/main/java/org/jooq/impl/R2DBC.java
index 687216a92fa..3284e45870c 100644
--- a/jOOQ/src/main/java/org/jooq/impl/R2DBC.java
+++ b/jOOQ/src/main/java/org/jooq/impl/R2DBC.java
@@ -192,7 +192,7 @@ else if (!completed.get()) {
         }
 
         @Override
-        public final void cancel() {
+        public void cancel() {
             complete(() -> {});
         }
 
@@ -376,14 +376,18 @@ static abstract class ConnectionSubscriber<T> implements Subscriber<Connection>
 
         final AbstractNonBlockingSubscription<T> downstream;
         final AtomicReference<Connection>        connection;
+        final AtomicReference<Subscription>      subscription;
 
         ConnectionSubscriber(AbstractNonBlockingSubscription<T> downstream) {
             this.downstream = downstream;
             this.connection = new AtomicReference<>();
+            this.subscription = new AtomicReference<>();
         }
 
         @Override
         public final void onSubscribe(Subscription s) {
+            // Stores the Subscription that handles the connection establishment.
+            subscription.set(s);
             s.request(1);
         }
 
@@ -402,6 +406,13 @@ public final void onError(Throwable t) {
 
         @Override
         public final void onComplete() {}
+
+        public final void cancel() {
+            subscription.updateAndGet(s -> {
+                if (s != null) { s.cancel(); }
+                return null;
+            });
+        }
     }
 
     static final class QueryExecutionSubscriber<T, Q extends Query> extends ConnectionSubscriber<T> {
@@ -634,6 +645,12 @@ final void request2(Subscription s) {
                 s.request(1);
         }
 
+        @Override
+        public final void cancel() {
+            // Safely cancels the connection acquisition if not yet established.
+            complete(() -> delegate().cancel());
+        }
+
         @Override
         final void cancel0(boolean closeAfterTransaction, Runnable onComplete) {
 


Test patch: