diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java index 113cec157..63eb5e658 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java @@ -279,8 +279,10 @@ public void complete(Lease lease, Throwable failure) { if (ar2.succeeded()) { handle(lease); } else { - // Should we do some cleanup ? - handler.fail(failure); + // return the lease to the pool, otherwise its slot leaks forever + lease.recycle(); + dequeueMetric(metric); + handler.fail(ar2.cause()); } }); } else { diff --git a/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/SqlConnectionPoolAcquireHookTest.java b/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/SqlConnectionPoolAcquireHookTest.java new file mode 100644 index 000000000..13c670b22 --- /dev/null +++ b/vertx-sql-client/src/test/java/io/vertx/tests/sqlclient/SqlConnectionPoolAcquireHookTest.java @@ -0,0 +1,161 @@ +/* + * Copyright (c) 2011-2026 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0 + * which is available at https://www.apache.org/licenses/LICENSE-2.0. + * + * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0 + */ + +package io.vertx.tests.sqlclient; + +import io.vertx.core.Completable; +import io.vertx.core.Context; +import io.vertx.core.Future; +import io.vertx.core.Vertx; +import io.vertx.core.internal.ContextInternal; +import io.vertx.core.internal.VertxInternal; +import io.vertx.core.net.SocketAddress; +import io.vertx.core.spi.metrics.ClientMetrics; +import io.vertx.core.tracing.TracingPolicy; +import io.vertx.ext.unit.Async; +import io.vertx.ext.unit.TestContext; +import io.vertx.ext.unit.junit.VertxUnitRunner; +import io.vertx.sqlclient.SqlConnectOptions; +import io.vertx.sqlclient.impl.pool.SqlConnectionPool; +import io.vertx.sqlclient.spi.DatabaseMetadata; +import io.vertx.sqlclient.spi.connection.Connection; +import io.vertx.sqlclient.spi.connection.ConnectionContext; +import io.vertx.sqlclient.spi.connection.ConnectionFactory; +import io.vertx.sqlclient.spi.protocol.CommandBase; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * When the {@code afterAcquire} hook fails, the acquisition must fail with the hook failure and + * the lease must be returned to the pool, otherwise the slot leaks and the pool eventually starves. + */ +@RunWith(VertxUnitRunner.class) +public class SqlConnectionPoolAcquireHookTest { + + static class StubConnection implements Connection { + + @Override + public TracingPolicy tracingPolicy() { + return TracingPolicy.IGNORE; + } + + @Override + public ClientMetrics metrics() { + return null; + } + + @Override + public SocketAddress server() { + return SocketAddress.inetSocketAddress(5432, "localhost"); + } + + @Override + public String database() { + return "db"; + } + + @Override + public String user() { + return "user"; + } + + @Override + public void init(ConnectionContext context) { + } + + @Override + public void close(ConnectionContext holder, Completable promise) { + promise.succeed(); + } + + @Override + public boolean isSsl() { + return false; + } + + @Override + public boolean isValid() { + return true; + } + + @Override + public DatabaseMetadata databaseMetadata() { + return null; + } + + @Override + public void schedule(CommandBase cmd, Completable handler) { + handler.fail("not supported"); + } + } + + private Vertx vertx; + private SqlConnectionPool pool; + private final AtomicBoolean hookFails = new AtomicBoolean(true); + private final Exception hookFailure = new Exception("hook failure"); + + @Before + public void setUp() { + vertx = Vertx.vertx(); + ConnectionFactory factory = new ConnectionFactory() { + @Override + public Future connect(Context context, SqlConnectOptions options) { + return Future.succeededFuture(new StubConnection()); + } + @Override + public void close(Completable completion) { + completion.succeed(); + } + }; + pool = new SqlConnectionPool( + () -> Future.succeededFuture(new SqlConnectOptions()), + factory, + null, + null, + conn -> hookFails.get() ? Future.failedFuture(hookFailure) : Future.succeededFuture(), + conn -> Future.succeededFuture(), + (VertxInternal) vertx, + 0, + 0, + 1, + false, + -1, + 0); + } + + @After + public void tearDown(TestContext ctx) { + pool.close().onComplete(ctx.asyncAssertSuccess(v -> { + vertx.close().onComplete(ctx.asyncAssertSuccess()); + })); + } + + @Test + public void testAcquireFailsWithHookFailureAndReleasesTheSlot(TestContext ctx) { + Async async = ctx.async(); + ContextInternal context = (ContextInternal) vertx.getOrCreateContext(); + pool.acquire(context, 0, (conn, err) -> { + // the acquisition must report the hook failure, not null + ctx.assertEquals(hookFailure, err); + hookFails.set(false); + // the pool has a single slot: this acquisition succeeds only if the failed one was released + pool.acquire(context, 0, (conn2, err2) -> { + ctx.assertNull(err2); + ctx.assertNotNull(conn2); + async.complete(); + }); + }); + } +}