diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index 36a06e842..7a036dad6 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -117,8 +117,13 @@ public final class NettyResponseFuture implements ListenableFuture { // volatile where we don't need CAS ops private volatile long touch = unpreciseMillisTime(); private volatile ChannelState channelState = ChannelState.NEW; + // Written and read from the event loop, the caller thread (attachChannel on a pooled hit, cancel()) and + // the HashedWheelTimer thread (terminateAndExit, and TimeoutTimerTask.expire which closes this channel + // on a request timeout). Without volatile those readers can miss the write and skip the close, which + // since the connect path publishes the channel before the TLS handshake would strand a stuck + // connection, and its permit, until handshakeTimeout (issue #2189). + private volatile Channel channel; // state mutated only inside the event loop - private Channel channel; private boolean keepAlive = true; private Request targetRequest; private Request currentRequest; diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java index 580ab0799..06290b735 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java @@ -77,9 +77,10 @@ private static final class PendingOpener { private final AtomicBoolean redundant = new AtomicBoolean(false); private volatile Object partitionKey; // Releases the per-host permit this connection holds in ROUND_ROBIN mode (installed by - // NettyConnectListener, absent in DEFAULT mode). The GOAWAY handler and the channel closeFuture race - // to fire it; getAndSet(null) makes it run at most once, since a double release would push the - // semaphore above maxConnectionsPerHost. + // NettyConnectListener, absent in DEFAULT mode). Fired by the GOAWAY handler so the permit is freed at + // drain start rather than at close; the channel closeFuture releases the same permit directly, and both + // funnel through a single getAndSet, since a double release would push the semaphore above + // maxConnectionsPerHost. private final AtomicReference permitRelease = new AtomicReference<>(); public boolean tryAcquireStream() { @@ -281,8 +282,8 @@ public void setPermitRelease(Runnable release) { /** * Runs the installed permit-release action the first time it is called; later calls do nothing, as do - * calls when no action was installed (DEFAULT mode). Fired at drain start (GOAWAY) and on channel - * close, whichever comes first. + * calls when no action was installed (DEFAULT mode). Fired at drain start (GOAWAY); a channel that + * closes without draining has its permit released by the closeFuture listener instead. */ public void releasePermitOnce() { Runnable release = permitRelease.getAndSet(null); diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java index 9f0f75bd1..049921c13 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java @@ -34,6 +34,7 @@ import java.net.ConnectException; import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicReference; /** * Non Blocking connect. @@ -75,22 +76,34 @@ private void writeRequest(Channel channel) { LOGGER.debug("Using new Channel '{}' for '{}' to '{}'", channel, httpRequest.method(), httpRequest.uri()); } + // Publishing the future on the channel is deliberately deferred to here, unlike attachChannel which + // runs before the handshake: it arms AsyncHttpClientHandler.channelInactive, whose + // handleUnexpectedClosedChannel would race NettyConnectListener.onFailure to retry the same connect + // failure twice if a mid-handshake close could reach it. Channels.setAttribute(channel, future); channelManager.registerOpenChannel(channel); - future.attachChannel(channel, false); requestSender.writeRequest(future, channel); } public void onSuccess(Channel channel, InetSocketAddress remoteAddress) { - // Take the semaphore lock from the future. For HTTP/1.1, we'll transfer it to channel.closeFuture(). - // For HTTP/2, we release it immediately after ALPN negotiation since the connection is multiplexed. - final Object partitionKeyLock = (connectionSemaphore != null) ? future.takePartitionKeyLock() : null; + // The permit must be owned by the channel from the moment it leaves the future: takePartitionKeyLock() + // is a getAndSet(null), so abort() can no longer return it, and everything below can abandon this + // channel without being able to reach the token -- a failed or timed-out TLS handshake, a crashing + // AsyncHandler callback, a connect retry (issue #2189). The HTTP/2 branches release it earlier and + // leave this listener a no-op. The listener captures no enclosing instance, so it does not pin this + // request's object graph for the lifetime of a long-lived (HTTP/2) connection. + final ConnectionSemaphore semaphore = connectionSemaphore; + final Object partitionKeyLock = semaphore != null ? future.takePartitionKeyLock() : null; + final AtomicReference permit = new AtomicReference<>(partitionKeyLock); + if (partitionKeyLock != null) { + channel.closeFuture().addListener(f -> releasePermitOnce(semaphore, permit)); + } Channels.setActiveToken(channel); if (futureIsAlreadyCompleted(channel)) { - releaseSemaphoreImmediately(partitionKeyLock); + releasePermitOnce(semaphore, permit); return; } @@ -102,10 +115,15 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) { // futureIsAlreadyCompleted check above can pass while the holder is already null. // Drop this connection rather than NPE-ing on setResolvedRemoteAddress below. Channels.silentlyCloseChannel(channel); - releaseSemaphoreImmediately(partitionKeyLock); + releasePermitOnce(semaphore, permit); return; } + // Publish the channel before the (possibly long) TLS handshake. Until this runs future.channel() is + // null, and NettyRequestSender.abort only closes a non-null channel -- so a request timeout firing + // mid-handshake could not close the socket, stranding it until handshakeTimeout (issue #2189). + future.attachChannel(channel, false); + Request request = future.getTargetRequest(); Uri uri = request.getUri(); // don't set a null resolved address - if the remoteAddress is null we keep @@ -153,7 +171,6 @@ protected void onSuccess(Channel value) { return; } // After SSL handshake to proxy, continue with normal proxy request - attachSemaphoreToChannelClose(channel, partitionKeyLock); writeRequest(channel); } @@ -215,9 +232,7 @@ protected void onSuccess(Channel value) { } if (http2Negotiated && !uri.isWebSocket()) { channelManager.upgradePipelineToHttp2(channel.pipeline()); - registerHttp2AndManageSemaphore(channel, partitionKeyLock); - } else { - attachSemaphoreToChannelClose(channel, partitionKeyLock); + registerHttp2AndManageSemaphore(channel, semaphore, permit); } writeRequest(channel); } @@ -240,32 +255,22 @@ protected void onFailure(Throwable cause) { // excluded for the same RFC 8441 reason as the TLS path above — it stays on HTTP/1.1. if (!uri.isSecured() && channelManager.isHttp2CleartextEnabled() && !uri.isWebSocket()) { channelManager.upgradePipelineToHttp2(channel.pipeline()); - registerHttp2AndManageSemaphore(channel, partitionKeyLock); - } else { - attachSemaphoreToChannelClose(channel, partitionKeyLock); + registerHttp2AndManageSemaphore(channel, semaphore, permit); } writeRequest(channel); } } /** - * Attaches the semaphore lock to the channel's close future (HTTP/1.1 behavior). - * The semaphore slot is released when the connection closes. - */ - private void attachSemaphoreToChannelClose(Channel channel, Object partitionKeyLock) { - if (connectionSemaphore != null && partitionKeyLock != null) { - channel.closeFuture().addListener(f -> connectionSemaphore.releaseChannelLock(partitionKeyLock)); - } - } - - /** - * Releases the semaphore lock immediately (HTTP/2 behavior). - * HTTP/2 connections are multiplexed, so the semaphore should not be held - * for the lifetime of the connection. + * Returns the per-host permit this connection holds, at most once. The channel closeFuture, the + * early-exit paths and the HTTP/2 immediate release all race to call this; a double release would + * push the semaphore above {@code maxConnectionsPerHost}, and an unmatched one can prematurely + * prune a live entry from the per-host map. */ - private void releaseSemaphoreImmediately(Object partitionKeyLock) { - if (connectionSemaphore != null && partitionKeyLock != null) { - connectionSemaphore.releaseChannelLock(partitionKeyLock); + private static void releasePermitOnce(ConnectionSemaphore semaphore, AtomicReference permit) { + Object partitionKeyLock = permit.getAndSet(null); + if (partitionKeyLock != null && semaphore != null) { + semaphore.releaseChannelLock(partitionKeyLock); } } @@ -281,7 +286,7 @@ private void releaseSemaphoreImmediately(Object partitionKeyLock) { * a further request fails to acquire a permit and multiplexes onto a sibling-IP connection via * {@link ChannelManager#pollHttp2SiblingConnection(Object)} rather than opening another (issue #2214). */ - private void registerHttp2AndManageSemaphore(Channel channel, Object partitionKeyLock) { + private void registerHttp2AndManageSemaphore(Channel channel, ConnectionSemaphore semaphore, AtomicReference permit) { // Register under the future's partition key so the H2 connection is found by the same key the // pool is polled with, including the IP-aware key used by LoadBalance.ROUND_ROBIN. Read the key // once: it is volatile (repinned on IP failover) and both uses below must agree. @@ -289,19 +294,14 @@ private void registerHttp2AndManageSemaphore(Channel channel, Object partitionKe channelManager.registerHttp2Connection(partitionKey, channel); if (partitionKey instanceof RoundRobinPartitionKey) { // The permit must be freed as soon as the connection stops serving new requests: on close, or - // at drain start after GOAWAY (issue #2214). Both paths go through releasePermitOnce(), which - // guarantees a single release; a double release would push the semaphore above the cap. The - // state attribute is always present after upgradePipelineToHttp2, the fallback is just - // belt and braces. + // at drain start after GOAWAY (issue #2214). The closeFuture listener installed in onSuccess + // covers close; this adds the drain hook. Both funnel through releasePermitOnce. Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); - if (state != null && connectionSemaphore != null && partitionKeyLock != null) { - state.setPermitRelease(() -> connectionSemaphore.releaseChannelLock(partitionKeyLock)); - channel.closeFuture().addListener(f -> state.releasePermitOnce()); - } else { - attachSemaphoreToChannelClose(channel, partitionKeyLock); + if (state != null) { + state.setPermitRelease(() -> releasePermitOnce(semaphore, permit)); } } else { - releaseSemaphoreImmediately(partitionKeyLock); + releasePermitOnce(semaphore, permit); } } diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java b/client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java index 3ffd781b5..112720cf5 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java @@ -107,6 +107,16 @@ private void abort(Channel channel, NettyResponseFuture future, WebSocketUpgr public void handleRead(Channel channel, NettyResponseFuture future, Object e) throws Exception { if (e instanceof HttpResponse) { + // Unlike HttpHandler, this check cannot guard the whole method: a successful upgrade completes + // the future (see upgrade() below) and the channel then keeps serving frames, so isDone() is + // true for all normal WebSocket traffic. It belongs on the upgrade response alone, where a 101 + // arriving just as a request timeout aborted the future would otherwise still run onOpen and + // deliver the WebSocket lifecycle after onThrowable had already fired. + if (future.isDone()) { + channelManager.closeChannel(channel); + return; + } + HttpResponse response = (HttpResponse) e; if (logger.isDebugEnabled()) { HttpRequest httpRequest = future.getNettyRequest().getHttpRequest(); diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 03e757ad6..b524e476d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -1071,18 +1071,21 @@ private static void scheduleReadTimeout(NettyResponseFuture nettyResponseFutu } public void abort(Channel channel, NettyResponseFuture future, Throwable t) { - if (channel != null) { - if (channel.isActive()) { - channelManager.closeChannel(channel); - } - } - + // Complete the future before closing, so the caller's cause is the one the user sees. Closing first + // can fail an in-flight TLS handshake, and that failure races back through + // NettyConnectListener.onFailure to abort the same future with a ConnectException instead -- which a + // request timeout on the connect path can now hit, since the channel is published before the + // handshake (issue #2189). The close still uses the channel passed in, which abort() does not clear. if (!future.isDone()) { future.setChannelState(ChannelState.CLOSED); LOGGER.debug("Aborting Future {}\n", future); LOGGER.debug(t.getMessage(), t); future.abort(t); } + + if (channel != null && channel.isActive()) { + channelManager.closeChannel(channel); + } } public void handleUnexpectedClosedChannel(Channel channel, NettyResponseFuture future) { diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java index 53fd0cc5a..6e23b265f 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java @@ -27,6 +27,7 @@ import java.net.InetAddress; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicReference; import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; @@ -81,9 +82,19 @@ private EmbeddedChannel registerRrConnectionHoldingPermit(ConnectionSemaphore se Http2ConnectionState state = new Http2ConnectionState(); channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state); channelManager.registerHttp2Connection(registryKey, channel); - // Mirror NettyConnectListener.registerHttp2AndManageSemaphore's round-robin branch. - state.setPermitRelease(() -> semaphore.releaseChannelLock(baseKey)); - channel.closeFuture().addListener(f -> state.releasePermitOnce()); + // Stand in for NettyConnectListener's round-robin wiring - the drain hook from + // registerHttp2AndManageSemaphore plus the closeFuture release from onSuccess, both funnelling + // through one getAndSet. This fixture only pins ChannelManager's GOAWAY drain contract; that + // NettyConnectListener actually installs this wiring is covered end-to-end by BasicHttp2Test. + AtomicReference permit = new AtomicReference<>(baseKey); + Runnable release = () -> { + Object key = permit.getAndSet(null); + if (key != null) { + semaphore.releaseChannelLock(key); + } + }; + state.setPermitRelease(release); + channel.closeFuture().addListener(f -> release.run()); return channel; } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionSemaphoreLeakTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionSemaphoreLeakTest.java new file mode 100644 index 000000000..8ae9cf6ba --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ConnectionSemaphoreLeakTest.java @@ -0,0 +1,409 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.channel; + +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import org.asynchttpclient.AsyncHttpClient; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; +import org.asynchttpclient.Response; +import org.asynchttpclient.exception.TooManyConnectionsPerHostException; +import org.asynchttpclient.testserver.HttpServer; +import org.eclipse.jetty.server.handler.AbstractHandler; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; +import static org.asynchttpclient.test.TestUtils.createSslEngineFactory; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +/** + * End-to-end regression tests for issue #2189 over real sockets: a connection attempt that fails after TCP + * connect must return its per-host connection permit, otherwise a run of failures against one degraded + * origin permanently pins {@code maxConnectionsPerHost} and every later request to it fails with + * {@link TooManyConnectionsPerHostException} instead of the real error. + * + *

The deterministic interleavings are covered by {@code NettyConnectListenerPermitLeakTest}; these prove + * the user-visible symptom is gone. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class ConnectionSemaphoreLeakTest { + + private HttpServer server; + + @BeforeEach + void start() throws Exception { + server = new HttpServer(); + server.start(); + } + + @AfterEach + void stop() throws IOException { + if (server != null) { + server.close(); + } + } + + /** Counts acquires/releases and signals when every acquired permit has been returned. */ + private static final class CountingSemaphore implements ConnectionSemaphore { + + private final ConnectionSemaphore delegate; + final AtomicInteger acquires = new AtomicInteger(); + final AtomicInteger releases = new AtomicInteger(); + final CountDownLatch released = new CountDownLatch(1); + + CountingSemaphore(ConnectionSemaphore delegate) { + this.delegate = delegate; + } + + @Override + public void acquireChannelLock(Object partitionKey) throws IOException { + delegate.acquireChannelLock(partitionKey); + acquires.incrementAndGet(); + } + + @Override + public void acquireChannelLock(Object partitionKey, boolean nonBlocking) throws IOException { + delegate.acquireChannelLock(partitionKey, nonBlocking); + acquires.incrementAndGet(); + } + + @Override + public void releaseChannelLock(Object partitionKey) { + delegate.releaseChannelLock(partitionKey); + releases.incrementAndGet(); + released.countDown(); + } + } + + /** + * Builds a config with a counting semaphore installed. HTTP/2 is disabled because on permit exhaustion + * the sender otherwise parks the request in an Http2ConnectionWaiter for connectTimeout before failing, + * which would mask the symptom under a timeout rather than surfacing it. + */ + private static DefaultAsyncHttpClientConfig.Builder countingConfig(AtomicReference probe) { + return config() + .setMaxConnectionsPerHost(1) + .setHttp2Enabled(false) + .setConnectionSemaphoreFactory(cfg -> { + CountingSemaphore semaphore = new CountingSemaphore( + new DefaultConnectionSemaphoreFactory().newConnectionSemaphore(cfg)); + probe.set(semaphore); + return semaphore; + }); + } + + /** + * A TLS handshake that fails on an untrusted certificate must return the permit. Before the fix the + * token had already been taken off the future and was never bound to the channel, so the abort released + * nothing and the second request failed on the exhausted cap rather than on TLS. + */ + @Test + @Timeout(60) + void failedTlsHandshakeDoesNotLeakThePerHostPermit() throws Exception { + AtomicReference probe = new AtomicReference<>(); + AsyncHttpClientConfig cfg = countingConfig(probe) + .setMaxRequestRetry(0) + .setRequestTimeout(Duration.ofSeconds(10)) + .setSslEngineFactory(createSslEngineFactory(new AtomicBoolean(false))) + .build(); + + try (AsyncHttpClient client = asyncHttpClient(cfg)) { + String url = server.getHttpsUrl() + "/foo"; + + ExecutionException first = assertThrows(ExecutionException.class, + () -> client.prepareGet(url).execute().get(30, TimeUnit.SECONDS)); + assertFalse(first.getCause() instanceof TooManyConnectionsPerHostException, + "sanity: the first request must fail on TLS, not on the permit"); + + CountingSemaphore semaphore = probe.get(); + assertTrue(semaphore.released.await(10, TimeUnit.SECONDS), + "issue #2189: a failed TLS handshake must return the per-host permit"); + + // With a cap of 1, a leaked permit makes every later request to this origin fail on the cap. + ExecutionException second = assertThrows(ExecutionException.class, + () -> client.prepareGet(url).execute().get(30, TimeUnit.SECONDS)); + assertFalse(second.getCause() instanceof TooManyConnectionsPerHostException, + "issue #2189: the second request must still fail on TLS, not on a leaked permit"); + + assertEquals(semaphore.acquires.get(), semaphore.releases.get(), + "every acquired permit must be returned"); + } + } + + /** + * The reported production scenario: a peer that accepts TCP and never speaks TLS, with the request + * timeout firing while the handshake is still in flight. The abort cannot reclaim the token, so the + * permit must come back through the channel. + */ + @Test + @Timeout(60) + void requestTimeoutDuringTlsHandshakeDoesNotLeakThePerHostPermit() throws Exception { + AtomicReference probe = new AtomicReference<>(); + AsyncHttpClientConfig cfg = countingConfig(probe) + .setMaxRequestRetry(0) + .setRequestTimeout(Duration.ofMillis(300)) + .setHandshakeTimeout(2000) + .setSslEngineFactory(createSslEngineFactory(new AtomicBoolean(true))) + .build(); + + try (BlackHoleServer blackHole = new BlackHoleServer(); + AsyncHttpClient client = asyncHttpClient(cfg)) { + ExecutionException e = assertThrows(ExecutionException.class, + () -> client.prepareGet(blackHole.url()).execute().get(30, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, e.getCause(), + "sanity: the request timeout must win over the handshake timeout"); + + CountingSemaphore semaphore = probe.get(); + assertTrue(semaphore.released.await(20, TimeUnit.SECONDS), + "issue #2189: the permit must come back after a request timeout during the TLS handshake"); + assertEquals(semaphore.acquires.get(), semaphore.releases.get()); + } + } + + /** + * The payoff of publishing the channel before the handshake: a request timeout must close the socket + * itself, at requestTimeout. Without that the connecting channel is invisible to the aborting timeout + * (NettyRequestSender.abort only closes a non-null channel) and the socket, and its permit, survive + * until the far longer handshakeTimeout - so the deadline below is what distinguishes the two. + */ + @Test + @Timeout(60) + void requestTimeoutClosesTheConnectingSocketWithoutWaitingForHandshakeTimeout() throws Exception { + AtomicReference probe = new AtomicReference<>(); + AsyncHttpClientConfig cfg = countingConfig(probe) + .setMaxRequestRetry(0) + .setRequestTimeout(Duration.ofMillis(300)) + .setHandshakeTimeout(10_000) + .setSslEngineFactory(createSslEngineFactory(new AtomicBoolean(true))) + .build(); + + try (BlackHoleServer blackHole = new BlackHoleServer(); + AsyncHttpClient client = asyncHttpClient(cfg)) { + ExecutionException e = assertThrows(ExecutionException.class, + () -> client.prepareGet(blackHole.url()).execute().get(30, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, e.getCause()); + + Socket peer = blackHole.awaitFirstConnection(10, TimeUnit.SECONDS); + assertNotNull(peer, "sanity: the client established the TCP connection"); + // Well inside handshakeTimeout: if the abort could not close the connecting channel, draining + // this socket blocks until the read times out instead of reaching EOF. + peer.setSoTimeout(3000); + try (InputStream in = peer.getInputStream()) { + byte[] buffer = new byte[1024]; + while (in.read(buffer) >= 0) { + // drain the ClientHello until the client closes its end + } + } catch (SocketTimeoutException timeout) { + fail("issue #2189: the request timeout must close the connecting socket, not leave it " + + "until handshakeTimeout"); + } catch (IOException reset) { + // a reset rather than a clean FIN still means the client closed in time + } + + assertTrue(probe.get().released.await(10, TimeUnit.SECONDS), "the permit must come back with it"); + } + } + + /** + * The other half of the invariant: an HTTP/1.1 connection must hold its permit for as long as it is + * serving, so the fix must not release it early. With a cap of 1 and a request held open, a second + * request to the same host must be refused - a permit released at handshake/response time instead of at + * close would let it through and breach maxConnectionsPerHost. + */ + @Test + @Timeout(60) + void permitIsHeldWhileTheConnectionIsStillServing() throws Exception { + AtomicReference probe = new AtomicReference<>(); + AsyncHttpClientConfig cfg = countingConfig(probe) + .setRequestTimeout(Duration.ofSeconds(30)) + .build(); + + CountDownLatch release = new CountDownLatch(1); + CountDownLatch started = new CountDownLatch(1); + server.enqueue(new AbstractHandler() { + @Override + public void handle(String target, org.eclipse.jetty.server.Request baseRequest, + HttpServletRequest request, HttpServletResponse response) throws IOException { + baseRequest.setHandled(true); + started.countDown(); + try { + release.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + response.setStatus(200); + response.getOutputStream().flush(); + } + }); + + try (AsyncHttpClient client = asyncHttpClient(cfg)) { + Future inFlight = client.prepareGet(server.getHttpUrl() + "/foo").execute(); + assertTrue(started.await(20, TimeUnit.SECONDS), "sanity: the first request reached the server"); + + try { + ExecutionException refused = assertThrows(ExecutionException.class, + () -> client.prepareGet(server.getHttpUrl() + "/foo").execute().get(20, TimeUnit.SECONDS), + "a second connection must not be admitted while the first still holds the only permit"); + assertInstanceOf(TooManyConnectionsPerHostException.class, refused.getCause()); + } finally { + release.countDown(); + } + assertEquals(200, inFlight.get(30, TimeUnit.SECONDS).getStatusCode()); + + CountingSemaphore semaphore = probe.get(); + assertEquals(1, semaphore.acquires.get(), "only the served connection ever took a permit"); + } + } + + /** + * A permit must be returned exactly once per connection: with keep-alive off, three sequential requests + * against a cap of 1 each need the previous connection's permit back. + */ + @Test + @Timeout(60) + void successfulPlaintextRequestsReturnTheirPermit() throws Exception { + AtomicReference probe = new AtomicReference<>(); + AsyncHttpClientConfig cfg = countingConfig(probe) + .setKeepAlive(false) + .setRequestTimeout(Duration.ofSeconds(10)) + .build(); + + try (AsyncHttpClient client = asyncHttpClient(cfg)) { + for (int i = 0; i < 3; i++) { + server.enqueueOk(); + Response response = client.prepareGet(server.getHttpUrl() + "/foo").execute().get(30, TimeUnit.SECONDS); + assertEquals(200, response.getStatusCode(), "request " + i + " must succeed"); + } + + CountingSemaphore semaphore = probe.get(); + assertEquals(3, semaphore.acquires.get(), "one new connection per request with keep-alive off"); + assertEquals(semaphore.acquires.get(), semaphore.releases.get(), + "every connection's permit must be returned exactly once"); + } + } + + /** + * Accepts TCP connections on the loopback address and then says nothing at all, so a TLS handshake + * started against it stalls until something else closes the socket. Bound to the loopback address rather + * than the wildcard so the client's connect cannot miss it where localhost resolves to ::1 first. + */ + private static final class BlackHoleServer implements Closeable { + + private final ServerSocket serverSocket; + private final List accepted = Collections.synchronizedList(new ArrayList<>()); + private final BlockingQueue firstAccepted = new ArrayBlockingQueue<>(1); + private final Thread thread; + private volatile boolean closed; + + BlackHoleServer() throws IOException { + serverSocket = new ServerSocket(0, 0, InetAddress.getLoopbackAddress()); + thread = new Thread(() -> { + try { + while (!closed) { + Socket socket = serverSocket.accept(); + if (closed) { + socket.close(); + return; + } + accepted.add(socket); + firstAccepted.offer(socket); + } + } catch (IOException ignored) { + // the server socket was closed: normal shutdown + } + }); + thread.setDaemon(true); + thread.start(); + } + + /** + * Brackets an IPv6 literal, which the loopback address is wherever java.net.preferIPv6Addresses is + * set: an unbracketed "https://::1:443/foo" does not parse as a URI at all. + */ + String url() { + String host = serverSocket.getInetAddress().getHostAddress(); + if (host.indexOf(':') >= 0) { + host = '[' + host + ']'; + } + return "https://" + host + ':' + serverSocket.getLocalPort() + "/foo"; + } + + Socket awaitFirstConnection(long timeout, TimeUnit unit) throws InterruptedException { + return firstAccepted.poll(timeout, unit); + } + + @Override + public void close() { + closed = true; + try { + serverSocket.close(); + } catch (IOException ignored) { + // best effort test cleanup + } + try { + // join before draining, so a connection accepted during close() cannot be added afterwards + thread.join(TimeUnit.SECONDS.toMillis(5)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + synchronized (accepted) { + for (Socket socket : accepted) { + try { + socket.close(); + } catch (IOException ignored) { + // best effort test cleanup + } + } + accepted.clear(); + } + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/NettyConnectListenerPermitLeakTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/NettyConnectListenerPermitLeakTest.java new file mode 100644 index 000000000..afdb994bf --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/NettyConnectListenerPermitLeakTest.java @@ -0,0 +1,393 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.channel; + +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.DecoderException; +import io.netty.handler.ssl.NotSslRecordException; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.timeout.TimeoutsHolder; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Regression tests for issue #2189: the per-host connection permit must be conserved across every exit + * from {@link NettyConnectListener#onSuccess}. + * + *

{@code onSuccess} moves the permit token off the future with a {@code getAndSet(null)}, after which + * {@code future.abort(...)} can no longer release it. Before the fix the token then lived in a bare local + * until the TLS handshake completed, so every failure in between - a failed or timed-out handshake, a + * crashing {@code AsyncHandler} callback - dropped it permanently: the channel closed but no release had + * ever been bound to it. That is the "permits pinned at the cap while zero sockets to the peer exist" + * state the issue reports. + * + *

These drive the listener directly over an {@link EmbeddedChannel}, so the interleavings are exact and + * single-threaded rather than timing-dependent. The end-to-end path over a real socket is covered by + * {@code ConnectionSemaphoreLeakTest}. + */ +class NettyConnectListenerPermitLeakTest { + + private static final String HTTPS_URL = "https://example.com:12345"; + + private AsyncHttpClientConfig config; + private ChannelManager channelManager; + private Timer timer; + + @BeforeEach + void setUp() { + config = config().build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, timer); + } + + @AfterEach + void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + private static AsyncHandler noopHandler() { + return new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }; + } + + private NettyResponseFuture newFuture(String url, ConnectionSemaphore semaphore, AsyncHandler handler) { + Request request = new RequestBuilder().setUrl(url).build(); + // maxRetry = 0 keeps onFailure from reaching requestSender.retry(), which is why every test here + // can pass a null requestSender. + NettyResponseFuture future = new NettyResponseFuture<>(request, handler, null, 0, + ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, semaphore, null); + future.setTimeoutsHolder(new TimeoutsHolder(null, future, null, config, null)); + return future; + } + + /** + * Available per-host permits for {@code key}. Reads the map directly rather than through + * {@code getFreeConnectionsForHost}, which is a {@code computeIfAbsent} and would resurrect an entry + * that was legitimately pruned after its last release. + */ + private static int availablePerHost(PerHostConnectionSemaphore semaphore, Object key) { + Semaphore freeConnections = semaphore.freeChannelsPerHost.get(key); + return freeConnections != null ? freeConnections.availablePermits() : semaphore.maxConnectionsPerHost; + } + + /** + * Counts acquire/release calls. Once the last permit is returned the per-host entry is pruned, so a + * double release is invisible to a permit count - only a call count can prove "exactly once". + */ + private static final class CountingSemaphore implements ConnectionSemaphore { + + private final ConnectionSemaphore delegate; + final AtomicInteger acquires = new AtomicInteger(); + final AtomicInteger releases = new AtomicInteger(); + + CountingSemaphore(ConnectionSemaphore delegate) { + this.delegate = delegate; + } + + @Override + public void acquireChannelLock(Object partitionKey) throws IOException { + delegate.acquireChannelLock(partitionKey); + acquires.incrementAndGet(); + } + + @Override + public void acquireChannelLock(Object partitionKey, boolean nonBlocking) throws IOException { + delegate.acquireChannelLock(partitionKey, nonBlocking); + acquires.incrementAndGet(); + } + + @Override + public void releaseChannelLock(Object partitionKey) { + releases.incrementAndGet(); + delegate.releaseChannelLock(partitionKey); + } + } + + /** + * The peer answers the TLS ClientHello with something that is not a TLS record, so the handshake fails + * and {@code SslHandler} closes the connection. The permit must come back. + */ + @Test + void tlsHandshakeFailureReleasesThePerHostPermit() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + Object key = future.basePartitionKey(); + assertEquals(0, availablePerHost(semaphore, key), "sanity: the future holds the only permit"); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertTrue(channel.isOpen(), "sanity: the TLS handshake is in flight"); + assertEquals(0, availablePerHost(semaphore, key), "the permit is still held during the handshake"); + + DecoderException decoderException = assertThrows(DecoderException.class, + () -> channel.writeInbound(Unpooled.wrappedBuffer("not-a-tls-record!".getBytes(StandardCharsets.US_ASCII))), + "sanity: garbage in place of a ServerHello fails the handshake"); + assertInstanceOf(NotSslRecordException.class, decoderException.getCause()); + + assertEquals(1, availablePerHost(semaphore, key), + "issue #2189: a failed TLS handshake must not strand the per-host permit"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * The permit must be owned by the channel from the moment it leaves the future, so that closing the + * channel alone returns it. Before the fix the binding happened only after the handshake completed, so + * a connection that closed first leaked it. + */ + @Test + void permitIsBoundToTheChannelBeforeTheHandshakeCompletes() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + Object key = future.basePartitionKey(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertEquals(0, availablePerHost(semaphore, key)); + + channel.close().sync(); + + assertEquals(1, availablePerHost(semaphore, key), + "issue #2189: the permit must be bound to the channel's close, not to handshake completion"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * The production trigger: a peer that accepts TCP but never speaks TLS. Netty's own handshake timeout + * eventually closes the socket - which is why the reporter saw zero sockets to the peer - and the permit + * must come back with it. Driven on the EmbeddedChannel's virtual clock, so this costs no wall time. + */ + @Test + void sslHandshakeTimeoutReleasesThePerHostPermit() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + Object key = future.basePartitionKey(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertEquals(0, availablePerHost(semaphore, key)); + + channel.advanceTimeBy(config.getHandshakeTimeout() + 1L, TimeUnit.MILLISECONDS); + channel.runScheduledPendingTasks(); + + assertFalse(channel.isOpen(), "sanity: the handshake timeout closes the connection"); + assertEquals(1, availablePerHost(semaphore, key), + "issue #2189: a handshake timeout must not strand the per-host permit"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * An AsyncHandler crashing inside the TLS window routes through onFailure, which closes the channel and + * aborts the future - but the abort cannot reach a token onSuccess already took. + */ + @Test + void asyncHandlerCrashInTheTlsWindowReleasesThePerHostPermit() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + AsyncHandler crashing = new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + + @Override + public void onTlsHandshakeAttempt() { + throw new IllegalStateException("boom"); + } + }; + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, crashing); + future.acquirePartitionLockLazily(); + Object key = future.basePartitionKey(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + + assertFalse(channel.isOpen(), "sanity: onFailure closes the freshly connected channel"); + assertEquals(1, availablePerHost(semaphore, key), + "issue #2189: an AsyncHandler crash in the TLS window must not strand the permit"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * The interleaving from the issue, sequenced on one thread: onSuccess takes the token, then the request + * timeout aborts the future. The abort provably cannot reclaim the token, so conservation depends + * entirely on the channel owning it by then. + */ + @Test + void requestTimeoutAbortDuringTheHandshakeDoesNotStrandThePermit() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + Object key = future.basePartitionKey(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertEquals(0, availablePerHost(semaphore, key)); + + future.abort(new TimeoutException("Request timeout to example.com:12345 after 100 ms")); + + assertTrue(future.isDone()); + // The premise the whole fix rests on: the token is already off the future, so the abort has + // nothing to give back and conservation depends entirely on the channel owning it. + assertEquals(0, availablePerHost(semaphore, key), + "abort cannot reclaim a token onSuccess already took"); + + // The orphaned socket eventually goes away (peer close, or the handshake timeout). + channel.close().sync(); + + assertEquals(1, availablePerHost(semaphore, key), + "issue #2189: the permit must come back when the orphaned connection closes"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * A request timeout aborts via {@code NettyRequestSender.abort(future.channel(), ...)}, which closes + * nothing when the channel is null. Publishing the channel before the handshake is what lets that abort + * close the socket at timeout time instead of leaving it until handshakeTimeout. + */ + @Test + void connectingChannelIsPublishedBeforeTheTlsHandshake() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + + assertTrue(channel.isOpen(), "sanity: the handshake is still in flight"); + assertSame(channel, future.channel(), + "issue #2189: a request-timeout abort must be able to close the connecting channel"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * Retries compound the leak. Once onSuccess has taken the token, a retry re-enters sendRequest, where + * acquirePartitionLockLazily sees a null partitionKeyLock and takes a *fresh* permit - so each failed + * attempt burned one, and a cap of 1 exhausted itself after the first. Replays the sequence a retry + * performs, in order, against a cap of 1. + */ + @Test + void retryAfterAFailedHandshakeCanReacquireThePermit() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertThrows(DecoderException.class, + () -> channel.writeInbound(Unpooled.wrappedBuffer("not-a-tls-record!".getBytes(StandardCharsets.US_ASCII)))); + + // What NettyRequestSender.sendRequestWithNewChannel does on the retry. + assertDoesNotThrow(() -> future.acquirePartitionLockLazily(false), + "issue #2189: a retry must not be blocked by the permit its own failed attempt leaked"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * Every release path races the channel closeFuture; a double release would push the semaphore above + * maxConnectionsPerHost and can prematurely prune a live per-host entry. + */ + @Test + void thePermitIsReleasedExactlyOnce() throws Exception { + CountingSemaphore semaphore = new CountingSemaphore(new PerHostConnectionSemaphore(1, 0)); + NettyResponseFuture future = newFuture(HTTPS_URL, semaphore, noopHandler()); + future.acquirePartitionLockLazily(); + + NettyConnectListener listener = new NettyConnectListener<>(future, null, channelManager, semaphore); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertThrows(DecoderException.class, + () -> channel.writeInbound(Unpooled.wrappedBuffer("not-a-tls-record!".getBytes(StandardCharsets.US_ASCII)))); + + // Neither a later close nor a late abort may release a second time. + channel.close().sync(); + future.abort(new IOException("late abort")); + + assertEquals(1, semaphore.acquires.get(), "sanity: one connection attempt took one permit"); + assertEquals(1, semaphore.releases.get(), "the permit must be released exactly once"); + } finally { + channel.finishAndReleaseAll(); + } + } + +} diff --git a/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderAbortTest.java b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderAbortTest.java new file mode 100644 index 000000000..bc8d388ea --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderAbortTest.java @@ -0,0 +1,111 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.request; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.channel.ChannelManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.net.ConnectException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Pins the ordering inside {@link NettyRequestSender#abort(io.netty.channel.Channel, NettyResponseFuture, + * Throwable)}: the future is completed with the caller's cause before the channel is closed. + * + *

Closing first is not inert. Once the connect path publishes the channel on the future (issue #2189), a + * request timeout closes a socket whose TLS handshake is still in flight; that failure races back through + * {@code NettyConnectListener.onFailure}, which aborts the same future with a {@link ConnectException}. If + * the close runs first that cause can win, and a request timeout surfaces as a connect error instead of a + * {@link TimeoutException}. An {@link EmbeddedChannel} makes this deterministic - it runs close listeners + * inline, so the induced abort always beats a subsequent one. + */ +class NettyRequestSenderAbortTest { + + private AsyncHttpClientConfig config; + private ChannelManager channelManager; + private NettyRequestSender sender; + private Timer timer; + + @BeforeEach + void setUp() { + config = config().build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, timer); + sender = new NettyRequestSender(config, channelManager, timer, null); + } + + @AfterEach + void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + private NettyResponseFuture newFuture() { + Request request = new RequestBuilder().setUrl("https://example.com:12345").build(); + return new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + } + + @Test + void abortCauseSurvivesAnAbortInducedByTheCloseItTriggers() throws Exception { + NettyResponseFuture future = newFuture(); + EmbeddedChannel channel = new EmbeddedChannel(); + future.attachChannel(channel, false); + // Stand in for the TLS handshake failing because we closed the channel, which NettyConnectListener + // reports by aborting this same future with a ConnectException. + channel.closeFuture().addListener(f -> future.abort(new ConnectException("closed mid-handshake"))); + + TimeoutException requestTimeout = new TimeoutException("Request timeout to example.com:12345 after 300 ms"); + try { + sender.abort(channel, future, requestTimeout); + + ExecutionException thrown = assertThrows(ExecutionException.class, + () -> future.get(5, TimeUnit.SECONDS)); + assertSame(requestTimeout, thrown.getCause(), + "the caller's cause must win over the one induced by the close it triggered"); + assertFalse(channel.isOpen(), "abort must still close the channel"); + } finally { + channel.finishAndReleaseAll(); + } + } +}