Guard the WebSocket read path against a completed future

Author: hyperxpro

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();

client/src/test/java/org/asynchttpclient/netty/channel/ConnectionSemaphoreLeakTest.java

@@ -194,10 +194,10 @@ void requestTimeoutDuringTlsHandshakeDoesNotLeakThePerHostPermit() throws Except
                 .setSslEngineFactory(createSslEngineFactory(new AtomicBoolean(true)))
                 .build();
 
-        try (BlackHoleServer server = new BlackHoleServer();
+        try (BlackHoleServer blackHole = new BlackHoleServer();
              AsyncHttpClient client = asyncHttpClient(cfg)) {
             ExecutionException e = assertThrows(ExecutionException.class,
-                    () -> client.prepareGet(server.url()).execute().get(30, TimeUnit.SECONDS));
+                    () -> client.prepareGet(blackHole.url()).execute().get(30, TimeUnit.SECONDS));
             assertInstanceOf(TimeoutException.class, e.getCause(),
                     "sanity: the request timeout must win over the handshake timeout");
 
@@ -225,13 +225,13 @@ void requestTimeoutClosesTheConnectingSocketWithoutWaitingForHandshakeTimeout()
                 .setSslEngineFactory(createSslEngineFactory(new AtomicBoolean(true)))
                 .build();
 
-        try (BlackHoleServer server = new BlackHoleServer();
+        try (BlackHoleServer blackHole = new BlackHoleServer();
              AsyncHttpClient client = asyncHttpClient(cfg)) {
             ExecutionException e = assertThrows(ExecutionException.class,
-                    () -> client.prepareGet(server.url()).execute().get(30, TimeUnit.SECONDS));
+                    () -> client.prepareGet(blackHole.url()).execute().get(30, TimeUnit.SECONDS));
             assertInstanceOf(TimeoutException.class, e.getCause());
 
-            Socket peer = server.awaitFirstConnection(10, TimeUnit.SECONDS);
+            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.
@@ -244,6 +244,8 @@ void requestTimeoutClosesTheConnectingSocketWithoutWaitingForHandshakeTimeout()
             } 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");
@@ -286,13 +288,18 @@ public void handle(String target, org.eclipse.jetty.server.Request baseRequest,
             Future<Response> inFlight = client.prepareGet(server.getHttpUrl() + "/foo").execute();
             assertTrue(started.await(20, TimeUnit.SECONDS), "sanity: the first request reached the server");
 
-            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());
-
-            release.countDown();
+            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");
         }
     }
 
@@ -334,13 +341,18 @@ private static final class BlackHoleServer implements Closeable {
         private final List<Socket> accepted = Collections.synchronizedList(new ArrayList<>());
         private final BlockingQueue<Socket> 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 (!Thread.currentThread().isInterrupted()) {
+                    while (!closed) {
                         Socket socket = serverSocket.accept();
+                        if (closed) {
+                            socket.close();
+                            return;
+                        }
                         accepted.add(socket);
                         firstAccepted.offer(socket);
                     }
@@ -352,8 +364,16 @@ private static final class BlackHoleServer implements Closeable {
             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() {
-            return "https://" + serverSocket.getInetAddress().getHostAddress() + ':' + serverSocket.getLocalPort() + "/foo";
+            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 {
@@ -362,12 +382,18 @@ Socket awaitFirstConnection(long timeout, TimeUnit unit) throws InterruptedExcep
 
         @Override
         public void close() {
+            closed = true;
             try {
                 serverSocket.close();
             } catch (IOException ignored) {
                 // best effort test cleanup
             }
-            thread.interrupt();
+            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 {

client/src/test/java/org/asynchttpclient/netty/channel/NettyConnectListenerPermitLeakTest.java

@@ -296,6 +296,10 @@ void requestTimeoutAbortDuringTheHandshakeDoesNotStrandThePermit() throws Except
             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();

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.
+ *
+ * <p>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<Object> newFuture() {
+        Request request = new RequestBuilder().setUrl("https://example.com:12345").build();
+        return new NettyResponseFuture<>(request, new AsyncCompletionHandler<Object>() {
+            @Override
+            public Object onCompleted(Response response) {
+                return null;
+            }
+        }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null);
+    }
+
+    @Test
+    void abortCauseSurvivesAnAbortInducedByTheCloseItTriggers() throws Exception {
+        NettyResponseFuture<Object> 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();
+        }
+    }
+}