Cancel pool connection timeout timer when acquisition fails

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java

@@ -248,7 +248,12 @@ public <R> void execute(CommandBase<R> cmd, Completable<R> handler, long timeout
       });
     }, t -> {
       dequeueMetric(metric);
-      return Future.failedFuture(t);
+      if (timerId != -1 && !vertx.cancelTimer(timerId)) {
+        // The timer already failed the handler
+        return Future.failedFuture(POOL_QUERY_TIMEOUT_EXCEPTION);
+      } else {
+        return Future.failedFuture(t);
+      }
     }).onComplete(ar -> {
       if (ar.succeeded()) {
         handler.succeed(ar.result());

vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/spi/backend/DriverBaseTest.java

@@ -3,6 +3,7 @@
 import io.vertx.core.Completable;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
+import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.net.NetClientOptions;
 import io.vertx.core.net.SocketAddress;
@@ -18,22 +19,31 @@
 import io.vertx.sqlclient.desc.ColumnDescriptor;
 import io.vertx.sqlclient.impl.RowBase;
 import io.vertx.sqlclient.spi.connection.Connection;
+import io.vertx.sqlclient.internal.PoolInternal;
 import io.vertx.sqlclient.internal.QueryResultHandler;
 import io.vertx.sqlclient.internal.RowDescriptorBase;
 import io.vertx.sqlclient.spi.connection.ConnectionContext;
 import io.vertx.sqlclient.spi.protocol.CommandBase;
+import io.vertx.sqlclient.spi.protocol.CommandScheduler;
 import io.vertx.sqlclient.spi.protocol.SimpleQueryCommand;
 import io.vertx.sqlclient.spi.connection.ConnectionFactory;
 import io.vertx.sqlclient.spi.DatabaseMetadata;
 import io.vertx.sqlclient.spi.DriverBase;
 import org.junit.Test;
 
 import java.sql.JDBCType;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.stream.Collector;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.Assert.*;
 
 public class DriverBaseTest {
@@ -136,6 +146,13 @@ private <T, A> void scheduleQueryCommand(SimpleQueryCommand<T> simpleQuery, Comp
   }
 
   private static DriverBase<SqlConnectOptions> createDriver(
+    java.util.function.Function<Connection, Future<Void>> afterAcquire,
+    java.util.function.Function<Connection, Future<Void>> beforeRecycle) {
+    return createDriver(context -> Future.succeededFuture(fakeConnection()), afterAcquire, beforeRecycle);
+  }
+
+  private static DriverBase<SqlConnectOptions> createDriver(
+    java.util.function.Function<Context, Future<Connection>> connector,
     java.util.function.Function<Connection, Future<Void>> afterAcquire,
     java.util.function.Function<Connection, Future<Void>> beforeRecycle) {
     return new DriverBase<>("generic", afterAcquire, beforeRecycle) {
@@ -144,7 +161,7 @@ public ConnectionFactory<SqlConnectOptions> createConnectionFactory(Vertx vertx,
         return new ConnectionFactory<>() {
           @Override
           public Future<Connection> connect(Context context, SqlConnectOptions options) {
-            return Future.succeededFuture(fakeConnection());
+            return connector.apply(context);
           }
 
           @Override
@@ -234,4 +251,67 @@ public void testAfterAcquireFailureReleasesConnection() {
       vertx.close().await();
     }
   }
+
+  @Test
+  public void testConnectFailureCancelsPoolConnectionTimeout() throws Exception {
+    RuntimeException connectError = new RuntimeException("connect failed");
+    DriverBase<SqlConnectOptions> driver = createDriver(context -> Future.failedFuture(connectError), null, null);
+
+    Vertx vertx = Vertx.vertx();
+
+    try {
+      PoolInternal pool = driver.newPool(vertx, () -> Future.succeededFuture(new SqlConnectOptions()),
+        new PoolOptions().setMaxSize(1).setConnectionTimeout(1000).setConnectionTimeoutUnit(MILLISECONDS), new NetClientOptions(), null);
+
+      List<Throwable> failures = new CopyOnWriteArrayList<>();
+      CountDownLatch latch = new CountDownLatch(1);
+      // The public query path completes its promise with tryFail, which hides a double completion
+      ((CommandScheduler) pool).schedule(new CommandBase<Void>() {}, (res, err) -> {
+        failures.add(err);
+        latch.countDown();
+      });
+      assertTrue(latch.await(10, SECONDS));
+
+      // Let the connection timeout elapse
+      Thread.sleep(1500);
+      assertEquals(Collections.singletonList(connectError), failures);
+    } finally {
+      vertx.close().await();
+    }
+  }
+
+  @Test
+  public void testConnectFailureAfterPoolConnectionTimeout() throws Exception {
+    RuntimeException connectError = new RuntimeException("connect failed");
+    Promise<Connection> connectPromise = Promise.promise();
+    AtomicReference<Context> connectContext = new AtomicReference<>();
+    DriverBase<SqlConnectOptions> driver = createDriver(context -> {
+      connectContext.set(context);
+      return connectPromise.future();
+    }, null, null);
+
+    Vertx vertx = Vertx.vertx();
+
+    try {
+      PoolInternal pool = driver.newPool(vertx, () -> Future.succeededFuture(new SqlConnectOptions()),
+        new PoolOptions().setMaxSize(1).setConnectionTimeout(100).setConnectionTimeoutUnit(MILLISECONDS), new NetClientOptions(), null);
+
+      List<Throwable> failures = new CopyOnWriteArrayList<>();
+      CountDownLatch latch = new CountDownLatch(1);
+      // The public query path completes its promise with tryFail, which hides a double completion
+      ((CommandScheduler) pool).schedule(new CommandBase<Void>() {}, (res, err) -> {
+        failures.add(err);
+        latch.countDown();
+      });
+      assertTrue(latch.await(10, SECONDS));
+      assertEquals(1, failures.size());
+      assertEquals("Timeout waiting for connection", failures.get(0).getMessage());
+
+      connectContext.get().runOnContext(v -> connectPromise.fail(connectError));
+      Thread.sleep(300);
+      assertEquals(1, failures.size());
+    } finally {
+      vertx.close().await();
+    }
+  }
 }