Fix race condition causing IllegalStateException: Promise already completed (#6170)

See #6169

A bug in SimpleConnectionPool allows a pool waiter's promise to be completed twice under concurrent recycle + cancel with a short connect/acquire timeout.

Recycle, ConnectSuccess (secondary waiter loop), and SetConcurrency polled waiters without setting waiter.disposed=true.

Cancel therefore saw disposed=false, reported cancellation success, and failed the promise, racing with the recycle path that had already completed it.

Some portions of this content were created with the assistance of Claude Code.

Signed-off-by: Thomas Segismont <tsegismont@gmail.com>

Author: tsegismont

vertx-core/src/main/java/io/vertx/core/internal/pool/SimpleConnectionPool.java

@@ -323,7 +323,9 @@ public void run() {
           c += m;
           leases = new LeaseImpl[m];
           for (int i = 0;i < m;i++) {
-            leases[i] = new LeaseImpl<>(slot, pool.waiters.poll().handler);
+            PoolWaiter<C> w = pool.waiters.poll();
+            w.disposed = true;
+            leases[i] = new LeaseImpl<>(slot, w.handler);
           }
         } else {
           leases = null;
@@ -465,7 +467,9 @@ public Task execute(SimpleConnectionPool<C> pool) {
           if (m > 0) {
             extra = new LeaseImpl[m];
             for (int i = 0;i < m;i++) {
-              extra[i] = new LeaseImpl<>(slot, pool.waiters.poll().handler);
+              PoolWaiter<C> w = pool.waiters.poll();
+              w.disposed = true;
+              extra[i] = new LeaseImpl<>(slot, w.handler);
             }
             slot.usage += m;
             return new Task() {
@@ -731,6 +735,7 @@ public Task execute(SimpleConnectionPool<C> pool) {
       if (!pool.closed && slot.connection != null) {
         PoolWaiter<C> waiter;
         if (slot.usage <= slot.concurrency && (waiter = pool.waiters.poll()) != null) {
+          waiter.disposed = true;
           LeaseImpl<C> lease = new LeaseImpl<>(slot, waiter.handler);
           return new Task() {
             @Override

vertx-core/src/test/java/io/vertx/tests/pool/ConnectionPoolTest.java

@@ -876,6 +876,46 @@ public void onConnect(PoolWaiter<Connection> waiter) {
     await();
   }
 
+  @Test
+  public void testCancelWaiterDispatchedByRecycle() throws Exception {
+    waitFor(1);
+    ContextInternal context = vertx.createEventLoopContext();
+    ConnectionManager mgr = new ConnectionManager();
+    ConnectionPool<Connection> pool = ConnectionPool.pool(mgr, new int[]{1});
+
+    // Step 1: Acquire and establish a connection
+    CountDownLatch connectedLatch = new CountDownLatch(1);
+    AtomicReference<Lease<Connection>> leaseRef = new AtomicReference<>();
+    pool.acquire(context, 0, TestUtils.onSuccess2(lease -> {
+      leaseRef.set(lease);
+      connectedLatch.countDown();
+    }));
+    mgr.assertRequest().connect(new Connection(), 0);
+    TestUtils.awaitLatch(connectedLatch);
+
+    // Step 2: Queue a second waiter (pool is at capacity)
+    CompletableFuture<PoolWaiter<Connection>> waiterFut = new CompletableFuture<>();
+    pool.acquire(context, new PoolWaiter.Listener<>() {
+      @Override
+      public void onEnqueue(PoolWaiter<Connection> waiter) {
+        waiterFut.complete(waiter);
+      }
+    }, 0, (res, err) -> {
+    });
+    PoolWaiter<Connection> waiter = waiterFut.get(10, TimeUnit.SECONDS);
+    Assert.assertEquals(1, pool.waiters());
+
+    // Step 3: Recycle the first lease - dispatches connection to the queued waiter
+    leaseRef.get().recycle();
+
+    // Step 4: Cancel the waiter after recycle dispatched it - should return false
+    pool.cancel(waiter, TestUtils.onSuccess2(cancelled -> {
+      Assert.assertFalse(cancelled);
+      testComplete();
+    }));
+    await();
+  }
+
   @Test
   public void testConnectionSelector() throws Exception {
     waitFor(1);