From 106d38120297b1f01528b323857b13a9cd6f76fa Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 19:48:21 +0000 Subject: [PATCH 01/13] Release the connection permit before notifying the handler --- .../asynchttpclient/netty/channel/Channels.java | 16 ++++++++++++++++ .../netty/channel/NettyConnectListener.java | 2 ++ .../netty/request/NettyRequestSender.java | 7 +++++++ .../netty/NettyRequestThrottleTimeoutTest.java | 9 +++++++-- 4 files changed, 32 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/Channels.java b/client/src/main/java/org/asynchttpclient/netty/channel/Channels.java index c56a05ba54..09f46dedae 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Channels.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Channels.java @@ -28,6 +28,7 @@ public final class Channels { private static final AttributeKey DEFAULT_ATTRIBUTE = AttributeKey.valueOf("default"); private static final AttributeKey ACTIVE_TOKEN_ATTRIBUTE = AttributeKey.valueOf("activeToken"); + private static final AttributeKey PERMIT_RELEASE_ATTRIBUTE = AttributeKey.valueOf("permitRelease"); private Channels() { // Prevent outside initialization @@ -58,6 +59,21 @@ public static boolean isActiveTokenSet(Channel channel) { return channel != null && channel.attr(ACTIVE_TOKEN_ATTRIBUTE).getAndSet(null) != null; } + static void setPermitRelease(Channel channel, Runnable release) { + channel.attr(PERMIT_RELEASE_ATTRIBUTE).set(release); + } + + /** + * Releases the connection permit this channel holds, if any. Safe to call more than once; the close + * listener remains the backstop. + */ + public static void releasePermit(Channel channel) { + Runnable release = channel.attr(PERMIT_RELEASE_ATTRIBUTE).get(); + if (release != null) { + release.run(); + } + } + public static void silentlyCloseChannel(Channel channel) { try { if (channel != null && channel.isActive()) { 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 fd12897a0f..b8ad403eb9 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java @@ -97,6 +97,8 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) { final Object partitionKeyLock = semaphore != null ? future.takePartitionKeyLock() : null; final AtomicReference permit = new AtomicReference<>(partitionKeyLock); if (partitionKeyLock != null) { + // Also reachable from the channel, so an abort can return the permit before it tells the handler. + Channels.setPermitRelease(channel, () -> releasePermitOnce(semaphore, permit)); channel.closeFuture().addListener(f -> releasePermitOnce(semaphore, permit)); } 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 bdf05133c3..4efbce36a5 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -1155,6 +1155,13 @@ public void abort(Channel channel, NettyResponseFuture future, Throwable t) { // 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. + // + // The permit goes back first: the channel is closed below either way, and a handler that sends its + // next request from onThrowable must not be refused by the connection it is being told has failed. + // An HTTP/2 connection keeps its permit until it stops serving streams. + if (channel != null && !ChannelManager.isHttp2(channel)) { + Channels.releasePermit(channel); + } if (!future.isDone()) { future.setChannelState(ChannelState.CLOSED); LOGGER.debug("Aborting Future {}\n", future); diff --git a/client/src/test/java/org/asynchttpclient/netty/NettyRequestThrottleTimeoutTest.java b/client/src/test/java/org/asynchttpclient/netty/NettyRequestThrottleTimeoutTest.java index e7d9c842c5..dbf500e305 100644 --- a/client/src/test/java/org/asynchttpclient/netty/NettyRequestThrottleTimeoutTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/NettyRequestThrottleTimeoutTest.java @@ -20,6 +20,7 @@ import org.asynchttpclient.AsyncCompletionHandler; import org.asynchttpclient.AsyncHttpClient; import org.asynchttpclient.Response; +import org.asynchttpclient.exception.TooManyConnectionsException; import org.eclipse.jetty.server.Request; import org.eclipse.jetty.server.handler.AbstractHandler; import org.junit.jupiter.api.Test; @@ -55,7 +56,7 @@ public void testRequestTimeout() throws IOException { try (AsyncHttpClient client = asyncHttpClient(config().setMaxConnections(1))) { final CountDownLatch latch = new CountDownLatch(samples); - final List tooManyConnections = Collections.synchronizedList(new ArrayList<>(2)); + final List tooManyConnections = Collections.synchronizedList(new ArrayList<>(2)); for (int i = 0; i < samples; i++) { new Thread(() -> { @@ -74,6 +75,10 @@ public Response onCompleted(Response response) { @Override public void onThrowable(Throwable t) { logger.error("onThrowable got an error", t); + // execute() reports a refused permit here, it does not throw + if (t instanceof TooManyConnectionsException) { + tooManyConnections.add(t); + } try { Thread.sleep(100); } catch (InterruptedException e) { @@ -101,7 +106,7 @@ public void onThrowable(Throwable t) { assertTrue(latch.await(30, TimeUnit.SECONDS)); }); - for (Exception e : tooManyConnections) { + for (Throwable e : tooManyConnections) { logger.error("Exception while calling execute", e); } From fb8e7e94f9bcf60d3086cf42f78af9fb68a724dc Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 19:48:56 +0000 Subject: [PATCH 02/13] Retry an unreachable peer on native transports too --- .../netty/future/StackTraceInspector.java | 7 +++++-- .../netty/future/StackTraceInspectorTest.java | 10 ++++++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java index 3532de3889..7df5922bc2 100755 --- a/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java +++ b/client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java @@ -19,6 +19,7 @@ import java.io.IOException; import java.net.ConnectException; +import java.net.NoRouteToHostException; import java.nio.channels.ClosedChannelException; public final class StackTraceInspector { @@ -45,9 +46,11 @@ private static boolean recoverOnConnectCloseException(Throwable t) { if (t instanceof ConnectTimeoutException) { return false; } - // The type covers every transport. The frames (checkConnect up to JDK 12, pollConnect after) - // still matter: NIO reports an unreachable peer as NoRouteToHostException, not a ConnectException. + // The types cover every transport: native ones report an unreachable peer as a bare + // NoRouteToHostException. The frames (checkConnect up to JDK 12, pollConnect after) keep whatever + // else NIO throws from connect completion. if (t instanceof ConnectException + || t instanceof NoRouteToHostException || exceptionInMethod(t, "sun.nio.ch.SocketChannelImpl", "checkConnect") || exceptionInMethod(t, "sun.nio.ch.Net", "pollConnect")) { return true; diff --git a/client/src/test/java/org/asynchttpclient/netty/future/StackTraceInspectorTest.java b/client/src/test/java/org/asynchttpclient/netty/future/StackTraceInspectorTest.java index c9aff90ed9..a450762450 100644 --- a/client/src/test/java/org/asynchttpclient/netty/future/StackTraceInspectorTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/future/StackTraceInspectorTest.java @@ -57,8 +57,14 @@ public void refusedNativeTransportConnectIsRecoverable() { assertTrue(StackTraceInspector.recoverOnNettyDisconnectException(annotated(refused))); } - // NoRouteToHostException is not a ConnectException, so only the frame probes match it. The running JDK - // produces just one of the two frames (checkConnect up to JDK 12, pollConnect after), hence the fake stacks. + // Native transports map EHOSTUNREACH and ENETUNREACH to a NoRouteToHostException with no sun.nio.ch frame. + @Test + public void unreachablePeerOnNativeTransportIsRecoverable() { + assertTrue(StackTraceInspector.recoverOnNettyDisconnectException(annotated(new NoRouteToHostException()))); + } + + // The running JDK produces just one of the two frames (checkConnect up to JDK 12, pollConnect after), + // hence the fake stacks. @Test public void unreachablePeerReportedFromConnectCompletionIsRecoverable() { assertTrue(StackTraceInspector.recoverOnNettyDisconnectException( From 2eb923ee610f1217a66f7de994a632bef84336df Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 19:55:25 +0000 Subject: [PATCH 03/13] Fail WebSocket sends uniformly once closing begins --- .../netty/ws/NettyWebSocket.java | 60 ++++++++++++---- .../ws/SecureWebSocketWriteFutureTest.java | 50 +++++++++++++ .../ws/WebSocketWriteFutureTest.java | 71 +++++++++++++++---- 3 files changed, 157 insertions(+), 24 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/ws/SecureWebSocketWriteFutureTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java b/client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java index 2329edacf9..581b2b0f04 100755 --- a/client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java +++ b/client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java @@ -18,6 +18,8 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufUtil; import io.netty.channel.Channel; +import io.netty.channel.ChannelPromise; +import io.netty.channel.EventLoop; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; @@ -35,11 +37,13 @@ import org.slf4j.LoggerFactory; import java.net.SocketAddress; +import java.nio.channels.ClosedChannelException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.RejectedExecutionException; import static io.netty.buffer.Unpooled.wrappedBuffer; @@ -53,6 +57,8 @@ public final class NettyWebSocket implements WebSocket { private FragmentedFrameType expectedFragmentedFrameType; // no need for volatile because only mutated in IO thread private boolean ready; + // written in the IO thread, but isOpen() reads it from any + private volatile boolean closing; private List bufferedFrames; public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders) { @@ -87,12 +93,12 @@ public Future sendTextFrame(String message) { @Override public Future sendTextFrame(String payload, boolean finalFragment, int rsv) { - return channel.writeAndFlush(new TextWebSocketFrame(finalFragment, rsv, payload)); + return send(new TextWebSocketFrame(finalFragment, rsv, payload)); } @Override public Future sendTextFrame(ByteBuf payload, boolean finalFragment, int rsv) { - return channel.writeAndFlush(new TextWebSocketFrame(finalFragment, rsv, payload)); + return send(new TextWebSocketFrame(finalFragment, rsv, payload)); } @Override @@ -107,12 +113,12 @@ public Future sendBinaryFrame(byte[] payload, boolean finalFragment, int r @Override public Future sendBinaryFrame(ByteBuf payload, boolean finalFragment, int rsv) { - return channel.writeAndFlush(new BinaryWebSocketFrame(finalFragment, rsv, payload)); + return send(new BinaryWebSocketFrame(finalFragment, rsv, payload)); } @Override public Future sendContinuationFrame(String payload, boolean finalFragment, int rsv) { - return channel.writeAndFlush(new ContinuationWebSocketFrame(finalFragment, rsv, payload)); + return send(new ContinuationWebSocketFrame(finalFragment, rsv, payload)); } @Override @@ -122,12 +128,12 @@ public Future sendContinuationFrame(byte[] payload, boolean finalFragment, @Override public Future sendContinuationFrame(ByteBuf payload, boolean finalFragment, int rsv) { - return channel.writeAndFlush(new ContinuationWebSocketFrame(finalFragment, rsv, payload)); + return send(new ContinuationWebSocketFrame(finalFragment, rsv, payload)); } @Override public Future sendPingFrame() { - return channel.writeAndFlush(new PingWebSocketFrame()); + return send(new PingWebSocketFrame()); } @Override @@ -137,12 +143,12 @@ public Future sendPingFrame(byte[] payload) { @Override public Future sendPingFrame(ByteBuf payload) { - return channel.writeAndFlush(new PingWebSocketFrame(payload)); + return send(new PingWebSocketFrame(payload)); } @Override public Future sendPongFrame() { - return channel.writeAndFlush(new PongWebSocketFrame()); + return send(new PongWebSocketFrame()); } @Override @@ -152,7 +158,7 @@ public Future sendPongFrame(byte[] payload) { @Override public Future sendPongFrame(ByteBuf payload) { - return channel.writeAndFlush(new PongWebSocketFrame(wrappedBuffer(payload))); + return send(new PongWebSocketFrame(wrappedBuffer(payload))); } @Override @@ -162,15 +168,44 @@ public Future sendCloseFrame() { @Override public Future sendCloseFrame(int statusCode, String reasonText) { - if (channel.isOpen()) { - return channel.writeAndFlush(new CloseWebSocketFrame(statusCode, reasonText)); + if (isOpen()) { + return send(new CloseWebSocketFrame(statusCode, reasonText)); } return ImmediateEventExecutor.INSTANCE.newSucceededFuture(null); } + // Writes are decided on the event loop, so one issued once the close has begun always fails the same way. + // Left to the pipeline it fails with whatever the SslHandler or the socket happens to report. A listener + // may still send from inside onClose, which is how it answers the peer's close frame. + private Future send(WebSocketFrame frame) { + EventLoop eventLoop = channel.eventLoop(); + if (eventLoop.inEventLoop()) { + return closing ? refuse(frame) : channel.writeAndFlush(frame); + } + ChannelPromise promise = channel.newPromise(); + try { + eventLoop.execute(() -> { + if (closing) { + frame.release(); + promise.setFailure(new ClosedChannelException()); + } else { + channel.writeAndFlush(frame, promise); + } + }); + } catch (RejectedExecutionException e) { + return refuse(frame); + } + return promise; + } + + private static Future refuse(WebSocketFrame frame) { + frame.release(); + return ImmediateEventExecutor.INSTANCE.newFailedFuture(new ClosedChannelException()); + } + @Override public boolean isOpen() { - return channel.isOpen(); + return !closing && channel.isOpen(); } @Override @@ -233,6 +268,7 @@ public void handleFrame(WebSocketFrame frame) { Channels.setDiscard(channel); CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame; onClose(closeFrame.statusCode(), closeFrame.reasonText()); + closing = true; Channels.silentlyCloseChannel(channel); } else if (frame instanceof PingWebSocketFrame) { diff --git a/client/src/test/java/org/asynchttpclient/ws/SecureWebSocketWriteFutureTest.java b/client/src/test/java/org/asynchttpclient/ws/SecureWebSocketWriteFutureTest.java new file mode 100644 index 0000000000..dcc0612ca1 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/ws/SecureWebSocketWriteFutureTest.java @@ -0,0 +1,50 @@ +/* + * 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.ws; + +import org.asynchttpclient.AsyncHttpClient; +import org.eclipse.jetty.server.Server; +import org.eclipse.jetty.server.ServerConnector; +import org.junit.jupiter.api.BeforeEach; + +import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; +import static org.asynchttpclient.test.TestUtils.addHttpsConnector; + +// Under TLS a closing channel stays open until close_notify is flushed, so the pipeline alone fails a late +// write with whatever the SslHandler reports. +public class SecureWebSocketWriteFutureTest extends WebSocketWriteFutureTest { + + @Override + @BeforeEach + public void setUpGlobal() throws Exception { + server = new Server(); + ServerConnector connector = addHttpsConnector(server); + server.setHandler(configureHandler()); + server.start(); + port1 = connector.getLocalPort(); + } + + @Override + protected String getTargetUrl() { + return String.format("wss://localhost:%d/", port1); + } + + @Override + protected AsyncHttpClient newClient() { + return asyncHttpClient(config().setUseInsecureTrustManager(true)); + } +} diff --git a/client/src/test/java/org/asynchttpclient/ws/WebSocketWriteFutureTest.java b/client/src/test/java/org/asynchttpclient/ws/WebSocketWriteFutureTest.java index 2c56f76ebb..4e395817f1 100644 --- a/client/src/test/java/org/asynchttpclient/ws/WebSocketWriteFutureTest.java +++ b/client/src/test/java/org/asynchttpclient/ws/WebSocketWriteFutureTest.java @@ -20,11 +20,14 @@ import org.junit.jupiter.api.Timeout; import java.nio.channels.ClosedChannelException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -38,7 +41,7 @@ public class WebSocketWriteFutureTest extends AbstractBasicWebSocketTest { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendTextMessage() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendTextFrame("TEXT").get(10, TimeUnit.SECONDS); } } @@ -46,7 +49,7 @@ public void sendTextMessage() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendTextMessageExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -59,7 +62,7 @@ public void sendTextMessageExpectFailure() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendByteMessage() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendBinaryFrame("BYTES".getBytes()).get(10, TimeUnit.SECONDS); } } @@ -67,7 +70,7 @@ public void sendByteMessage() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendByteMessageExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -80,7 +83,7 @@ public void sendByteMessageExpectFailure() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendPingMessage() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendPingFrame("PING".getBytes()).get(10, TimeUnit.SECONDS); } } @@ -88,7 +91,7 @@ public void sendPingMessage() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendPingMessageExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -101,7 +104,7 @@ public void sendPingMessageExpectFailure() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendPongMessage() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendPongFrame("PONG".getBytes()).get(10, TimeUnit.SECONDS); } } @@ -109,7 +112,7 @@ public void sendPongMessage() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void sendPongMessageExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -122,7 +125,7 @@ public void sendPongMessageExpectFailure() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void streamBytes() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendBinaryFrame("STREAM".getBytes(), true, 0).get(1, TimeUnit.SECONDS); } } @@ -130,7 +133,7 @@ public void streamBytes() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void streamBytesExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -143,7 +146,7 @@ public void streamBytesExpectFailure() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void streamText() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { getWebSocket(c).sendTextFrame("STREAM", true, 0).get(1, TimeUnit.SECONDS); } } @@ -152,7 +155,7 @@ public void streamText() throws Exception { @Test @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void streamTextExpectFailure() throws Exception { - try (AsyncHttpClient c = asyncHttpClient()) { + try (AsyncHttpClient c = newClient()) { CountDownLatch closeLatch = new CountDownLatch(1); WebSocket websocket = getWebSocket(c, closeLatch); websocket.sendCloseFrame(); @@ -162,6 +165,50 @@ public void streamTextExpectFailure() throws Exception { } } + // The server closes first, so the client's answer is sent from inside onClose. + @Test + @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) + public void closeFrameIsAnsweredFromOnClose() throws Exception { + try (AsyncHttpClient c = newClient()) { + CompletableFuture> answer = new CompletableFuture<>(); + WebSocket websocket = c.prepareGet(getTargetUrl()).execute(new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WebSocketListener() { + + @Override + public void onOpen(WebSocket websocket) { + } + + @Override + public void onError(Throwable t) { + answer.completeExceptionally(t); + } + + @Override + public void onClose(WebSocket websocket, int code, String reason) { + answer.complete(websocket.sendCloseFrame(code, reason)); + } + }).build()).get(); + websocket.sendTextFrame("CLOSE").get(TIMEOUT, TimeUnit.SECONDS); + answer.get(TIMEOUT, TimeUnit.SECONDS).get(TIMEOUT, TimeUnit.SECONDS); + } + } + + @Test + @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) + public void closedWebSocketIsNotOpen() throws Exception { + try (AsyncHttpClient c = newClient()) { + CountDownLatch closeLatch = new CountDownLatch(1); + WebSocket websocket = getWebSocket(c, closeLatch); + websocket.sendCloseFrame(); + assertTrue(closeLatch.await(TIMEOUT, TimeUnit.SECONDS), "the close handshake never completed"); + assertThrows(ExecutionException.class, () -> websocket.sendTextFrame("TEXT").get(TIMEOUT, TimeUnit.SECONDS)); + assertFalse(websocket.isOpen()); + } + } + + protected AsyncHttpClient newClient() { + return asyncHttpClient(); + } + private static void assertClosedChannel(ExecutionException e) { assertInstanceOf(ClosedChannelException.class, e.getCause()); } From 492cfb090dd0c6b2b5b15de3b833adc1bdd4f503 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:01:43 +0000 Subject: [PATCH 04/13] Never replay a request the connect listener wrote --- .../netty/channel/NettyConnectListener.java | 7 +- .../NettyConnectListenerReplayTest.java | 116 ++++++++++++++++++ 2 files changed, 121 insertions(+), 2 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/NettyConnectListenerReplayTest.java 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 b8ad403eb9..518d411cdf 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java @@ -47,6 +47,8 @@ public final class NettyConnectListener { private final NettyResponseFuture future; private final ChannelManager channelManager; private final ConnectionSemaphore connectionSemaphore; + // Shared by the addresses one connect fails over, but only the one that connects reaches writeRequest. + private volatile boolean requestWritten; public NettyConnectListener(NettyResponseFuture future, NettyRequestSender requestSender, ChannelManager channelManager, ConnectionSemaphore connectionSemaphore) { this.future = future; @@ -83,6 +85,7 @@ private void writeRequest(Channel channel) { Channels.setAttribute(channel, future); channelManager.registerOpenChannel(channel); + requestWritten = true; requestSender.writeRequest(future, channel); } @@ -353,8 +356,7 @@ private void registerHttp2AndManageSemaphore(Channel channel, ConnectionSemaphor } /** - * Must only be called before {@link #writeRequest}: it may replay the request, and replaying one that was - * already written would send it twice. + * Replays the request only while it is unwritten: replaying one that was already written would send it twice. */ public void onFailure(Channel channel, Throwable cause) { @@ -364,6 +366,7 @@ public void onFailure(Channel channel, Throwable cause) { boolean canRetry = future.incrementRetryAndCheck(); LOGGER.debug("Trying to recover from failing to connect channel {} with a retry value of {} ", channel, canRetry); if (canRetry// + && !requestWritten && cause != null // FIXME when can we have a null cause? && (future.getChannelState() != ChannelState.NEW || StackTraceInspector.recoverOnNettyDisconnectException(cause))) { diff --git a/client/src/test/java/org/asynchttpclient/netty/request/NettyConnectListenerReplayTest.java b/client/src/test/java/org/asynchttpclient/netty/request/NettyConnectListenerReplayTest.java new file mode 100644 index 0000000000..9c24a12c01 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/NettyConnectListenerReplayTest.java @@ -0,0 +1,116 @@ +/* + * 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.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +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.AsyncHttpClientState; +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.asynchttpclient.netty.channel.NettyConnectListener; +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 org.junit.jupiter.api.extension.ExtendWith; + +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +@ExtendWith(NettyLeakDetectorExtension.class) +class NettyConnectListenerReplayTest { + + private AsyncHttpClientConfig config; + private ChannelManager channelManager; + private NettyRequestSender requestSender; + private Timer timer; + + @BeforeEach + void setUp() { + config = config().setMaxRequestRetry(5).build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, timer); + requestSender = new NettyRequestSender(config, channelManager, timer, mock(AsyncHttpClientState.class)); + } + + @AfterEach + void tearDown() { + channelManager.close(); + timer.stop(); + } + + // No call site reports a failure after the write today. One that did would otherwise resend the request, + // and with retries left the caller would see a 200 for a body the server received twice. + @Test + void aFailureReportedAfterTheWriteIsNotReplayed() { + RetryCountingHandler handler = new RetryCountingHandler(); + Request request = new RequestBuilder("POST").setUrl("http://example.com:12345").setBody("body").build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, handler, + new NettyRequestFactory(config).newNettyRequest(request, false, null, null, null), 5, + ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + future.setTimeoutsHolder(new TimeoutsHolder(null, future, null, config, null)); + + NettyConnectListener listener = new NettyConnectListener<>(future, requestSender, channelManager, null); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + listener.onSuccess(channel, new InetSocketAddress("127.0.0.1", 12345)); + assertTrue(channel.outboundMessages().size() > 0, "fixture: the request must have been written"); + + // A refused connect, which is a failure onFailure does replay while the request is unwritten. + ConnectException refused = new ConnectException("Connection refused: /127.0.0.1:12345"); + refused.initCause(new ConnectException("Connection refused")); + listener.onFailure(channel, refused); + + assertEquals(0, handler.retries.get(), "a written request was replayed"); + ExecutionException e = assertThrows(ExecutionException.class, future::get); + assertInstanceOf(ConnectException.class, e.getCause()); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static final class RetryCountingHandler extends AsyncCompletionHandler { + + private final AtomicInteger retries = new AtomicInteger(); + + @Override + public void onRetry() { + retries.incrementAndGet(); + } + + @Override + public Response onCompleted(Response response) { + return response; + } + } +} From 0be377106f2410872d167a75190a8d8dc45b1561 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:01:43 +0000 Subject: [PATCH 05/13] Keep the real failure when onRetry refuses --- .../netty/request/NettyRequestSender.java | 2 +- .../handler/BodyDeferringAsyncHandlerTest.java | 10 +++++++--- 2 files changed, 8 insertions(+), 4 deletions(-) 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 4efbce36a5..3c69313124 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -1199,8 +1199,8 @@ public boolean retry(NettyResponseFuture future) { try { future.getAsyncHandler().onRetry(); } catch (Exception e) { + // Throwing is how a handler refuses a retry. The caller aborts with the failure it has. LOGGER.error("onRetry crashed", e); - abort(future.channel(), future, e); return false; } diff --git a/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java b/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java index b7313b5bc0..b643d3e146 100644 --- a/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java +++ b/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java @@ -32,6 +32,7 @@ import java.io.OutputStream; import java.io.PipedInputStream; import java.io.PipedOutputStream; +import java.net.ConnectException; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.concurrent.ExecutionException; @@ -219,7 +220,8 @@ public void deferredInputStreamTrickWithCloseConnectionAndRetry() throws Throwab try (is; cos) { copy(is, cos); } catch (Exception ex) { - assertInstanceOf(UnsupportedOperationException.class, ex.getCause()); + // The refused retry is only logged: the caller is told why the exchange failed. + assertInstanceOf(RemotelyClosedException.class, ex.getCause()); } } } @@ -227,13 +229,15 @@ public void deferredInputStreamTrickWithCloseConnectionAndRetry() throws Throwab @Test public void testConnectionRefused() throws Exception { int newPortWithoutAnyoneListening = findFreePort(); - try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) { + try (AsyncHttpClient client = asyncHttpClient(config().setMaxRequestRetry(1).setRequestTimeout(Duration.ofSeconds(10)).build())) { BoundRequestBuilder r = client.prepareGet("http://localhost:" + newPortWithoutAnyoneListening + "/testConnectionRefused"); CountingOutputStream cos = new CountingOutputStream(); BodyDeferringAsyncHandler bdah = new BodyDeferringAsyncHandler(cos); r.execute(bdah); - assertThrows(IOException.class, () -> bdah.getResponse()); + // The handler refuses retries by throwing from onRetry; that must not replace the real failure. + IOException e = assertThrows(IOException.class, () -> bdah.getResponse()); + assertInstanceOf(ConnectException.class, e.getCause()); } } From d314f7de79f7e1f151f85da2ea7f5d9ba4e2536a Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:04:29 +0000 Subject: [PATCH 06/13] Test the combined semaphore's shared timeout budget --- .../channel/CombinedConnectionSemaphore.java | 14 +++++++++-- .../netty/channel/SemaphoreTest.java | 25 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java index 6cb62c967a..1d4f23badc 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java @@ -18,6 +18,7 @@ import java.io.IOException; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; /** * A combined {@link ConnectionSemaphore} with two limits - a global limit and a per-host limit @@ -25,9 +26,18 @@ public class CombinedConnectionSemaphore extends PerHostConnectionSemaphore { protected final MaxConnectionSemaphore globalMaxConnectionSemaphore; + private final LongSupplier millisClock; + CombinedConnectionSemaphore(int maxConnections, int maxConnectionsPerHost, int acquireTimeout) { + // Monotonic, and finer than the 15 ms steps currentTimeMillis takes on Windows. + this(maxConnections, maxConnectionsPerHost, acquireTimeout, () -> TimeUnit.NANOSECONDS.toMillis(System.nanoTime())); + } + + // For tests: the split of the budget can be checked without waiting for it. + CombinedConnectionSemaphore(int maxConnections, int maxConnectionsPerHost, int acquireTimeout, LongSupplier millisClock) { super(maxConnectionsPerHost, acquireTimeout); globalMaxConnectionSemaphore = new MaxConnectionSemaphore(maxConnections, acquireTimeout); + this.millisClock = millisClock; } @Override @@ -89,9 +99,9 @@ protected long acquireGlobal(Object partitionKey) throws IOException { * Acquires the global lock and returns the remaining time, in millis, to acquire the per-host lock */ protected long acquireGlobalTimed(Object partitionKey) throws IOException { - long beforeGlobalAcquire = System.currentTimeMillis(); + long beforeGlobalAcquire = millisClock.getAsLong(); acquireGlobal(partitionKey); - long lockTime = System.currentTimeMillis() - beforeGlobalAcquire; + long lockTime = millisClock.getAsLong() - beforeGlobalAcquire; return acquireTimeout - lockTime; } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java index 0dbd4e0fa3..15f3d149e1 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java @@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerArray; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -155,6 +156,30 @@ private void checkAcquireTime(ConnectionSemaphore semaphore) { } } + // The per-host gate gets what the global gate left of the timeout, not a second full timeout. Both gates + // have a free permit here, so only the budget can refuse the acquire. + @Test + public void combinedSpendsOneTimeoutAcrossBothGates() throws IOException { + ConnectionSemaphore refused = combinedWhoseGlobalGateTakes(CHECK_ACQUIRE_TIME__TIMEOUT + 1); + assertThrows(TooManyConnectionsPerHostException.class, () -> refused.acquireChannelLock(PK)); + // the refusal handed the global permit back + refused.acquireChannelLock(PK, true); + + combinedWhoseGlobalGateTakes(CHECK_ACQUIRE_TIME__TIMEOUT / 2).acquireChannelLock(PK); + } + + private static ConnectionSemaphore combinedWhoseGlobalGateTakes(long millis) { + AtomicLong now = new AtomicLong(); + return new CombinedConnectionSemaphore(1, 1, CHECK_ACQUIRE_TIME__TIMEOUT, now::get) { + @Override + protected long acquireGlobal(Object partitionKey) throws IOException { + long remaining = super.acquireGlobal(partitionKey); + now.addAndGet(millis); + return remaining; + } + }; + } + // ---- a waiting acquire completes on release, not on its own timeout ---- // Far longer than the @Timeout below, so waiting it out cannot pass. From 2ac27938aeb6fc73571858f92c469b6161a9b4cd Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:07:20 +0000 Subject: [PATCH 07/13] Test the missing-file guard on a first request --- .../asynchttpclient/request/body/PutFileTest.java | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/client/src/test/java/org/asynchttpclient/request/body/PutFileTest.java b/client/src/test/java/org/asynchttpclient/request/body/PutFileTest.java index 3e8a1cf58a..100616ea41 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/PutFileTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/PutFileTest.java @@ -25,11 +25,15 @@ import java.io.IOException; import java.io.InputStream; import java.time.Duration; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import static org.asynchttpclient.Dsl.asyncHttpClient; import static org.asynchttpclient.Dsl.config; import static org.asynchttpclient.test.TestUtils.createTempFile; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; public class PutFileTest extends AbstractBasicTest { @@ -51,6 +55,17 @@ public void testPutSmallFile() throws Exception { put(1024); } + // No interceptor stands in front of a first request, so this is NettyFileBody's own guard. + @Test + public void testPutMissingFile() throws Exception { + File missing = new File(createTempFile(1).getParentFile(), "missing-" + System.nanoTime()); + try (AsyncHttpClient client = asyncHttpClient()) { + ExecutionException e = assertThrows(ExecutionException.class, + () -> client.preparePut(getTargetUrl()).setBody(missing).execute().get(TIMEOUT, TimeUnit.SECONDS)); + assertInstanceOf(IllegalArgumentException.class, e.getCause()); + } + } + @Override public AbstractHandler configureHandler() throws Exception { return new AbstractHandler() { From 02489ce5e4f14c630a3a09ad71e4f2d036398ac5 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:07:20 +0000 Subject: [PATCH 08/13] Give HttpToHttpsRedirectTest per-test redirect state --- .../HttpToHttpsRedirectTest.java | 20 +++---------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/client/src/test/java/org/asynchttpclient/HttpToHttpsRedirectTest.java b/client/src/test/java/org/asynchttpclient/HttpToHttpsRedirectTest.java index 862e953f3f..aef7f9db36 100644 --- a/client/src/test/java/org/asynchttpclient/HttpToHttpsRedirectTest.java +++ b/client/src/test/java/org/asynchttpclient/HttpToHttpsRedirectTest.java @@ -40,9 +40,6 @@ public class HttpToHttpsRedirectTest extends AbstractBasicTest { - // FIXME super NOT threadsafe!!! - private static final AtomicBoolean redirectDone = new AtomicBoolean(false); - @Override @BeforeEach public void setUpGlobal() throws Exception { @@ -62,19 +59,9 @@ public void tearDownGlobal() throws Exception { super.tearDownGlobal(); } - @Test - // FIXME find a way to make this threadsafe - public void runAllSequentiallyBecauseNotThreadSafe() throws Exception { - httpToHttpsRedirect(); - httpToHttpsProperConfig(); - relativeLocationUrl(); - } - // @Disabled @Test public void httpToHttpsRedirect() throws Exception { - redirectDone.getAndSet(false); - AsyncHttpClientConfig cg = config() .setMaxRedirects(5) .setFollowRedirect(true) @@ -90,8 +77,6 @@ public void httpToHttpsRedirect() throws Exception { @Test public void httpToHttpsProperConfig() throws Exception { - redirectDone.getAndSet(false); - AsyncHttpClientConfig cg = config() .setMaxRedirects(5) .setFollowRedirect(true) @@ -113,8 +98,6 @@ public void httpToHttpsProperConfig() throws Exception { @Test public void relativeLocationUrl() throws Exception { - redirectDone.getAndSet(false); - AsyncHttpClientConfig cg = config() .setMaxRedirects(5) .setFollowRedirect(true) @@ -130,6 +113,9 @@ public void relativeLocationUrl() throws Exception { private static class Relative302Handler extends AbstractHandler { + // One handler per test, so no test sees another's redirect. + private final AtomicBoolean redirectDone = new AtomicBoolean(false); + @Override public void handle(String s, Request r, HttpServletRequest httpRequest, HttpServletResponse httpResponse) throws IOException, ServletException { From ac40cebfc7f95dae4b0e643ce02521ad0be0959a Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:07:44 +0000 Subject: [PATCH 09/13] Bound CI jobs with timeout-minutes --- .github/workflows/builds.yml | 4 ++++ .github/workflows/maven.yml | 3 +++ 2 files changed, 7 insertions(+) diff --git a/.github/workflows/builds.yml b/.github/workflows/builds.yml index 0a93370605..896c561190 100644 --- a/.github/workflows/builds.yml +++ b/.github/workflows/builds.yml @@ -7,6 +7,7 @@ on: jobs: Verify: runs-on: ubuntu-latest + timeout-minutes: 15 steps: - uses: actions/checkout@v7 - uses: actions/setup-java@v6.0.1 @@ -20,6 +21,7 @@ jobs: RunOnLinux: runs-on: ubuntu-latest + timeout-minutes: 45 needs: Verify steps: - uses: actions/checkout@v7 @@ -34,6 +36,7 @@ jobs: RunOnMacOs: runs-on: macos-latest + timeout-minutes: 45 needs: Verify steps: - uses: actions/checkout@v7 @@ -48,6 +51,7 @@ jobs: RunOnWindows: runs-on: windows-latest + timeout-minutes: 45 needs: Verify steps: - uses: actions/checkout@v7 diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml index df47f04455..a8f514018e 100644 --- a/.github/workflows/maven.yml +++ b/.github/workflows/maven.yml @@ -21,6 +21,7 @@ on: jobs: compile-and-check: runs-on: ubuntu-latest + timeout-minutes: 15 steps: - uses: actions/checkout@v7 # The last entry wins the default JAVA_HOME, so the baseline check runs on JDK 11. @@ -62,6 +63,8 @@ jobs: distribution: corretto runs-on: ${{ matrix.os }} + # A test job takes under 10 minutes. Nothing retries a hung test any more, so bound it here. + timeout-minutes: 30 steps: - uses: actions/checkout@v7 - uses: actions/setup-java@v6.0.1 From 94e53e9517453d5ba9d0951fe24c00802b33cc3e Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:11:57 +0000 Subject: [PATCH 10/13] Test request deadlines on injected clocks --- .../netty/timeout/TimeoutsHolder.java | 37 ++++++++---- .../netty/timeout/TimeoutsHolderTest.java | 58 ++++++++++--------- 2 files changed, 58 insertions(+), 37 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index d7fdf28391..8c5c269d62 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -22,6 +22,7 @@ import org.asynchttpclient.Request; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.NettyRequestSender; +import org.asynchttpclient.util.DateUtils; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,8 +31,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; - -import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; +import java.util.function.LongSupplier; /** * The request and read timeouts of one exchange, armed either on the client's {@link Timer} or on the event @@ -56,6 +56,8 @@ public class TimeoutsHolder { private volatile @Nullable ReadTimeoutTimerTask readTimeoutTask; private final NettyResponseFuture nettyResponseFuture; private volatile InetSocketAddress remoteAddress; + private final LongSupplier millisClock; + private final LongSupplier nanoClock; public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) { @@ -71,6 +73,17 @@ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFutu */ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) { + this(nettyTimer, eventExecutor, nettyResponseFuture, requestSender, config, originalRemoteAddress, + DateUtils::unpreciseMillisTime, System::nanoTime); + } + + // For tests: a deadline can be checked without letting time pass. nanoClock must read the clock the + // future took its start from. + TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture nettyResponseFuture, + NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress, + LongSupplier millisClock, LongSupplier nanoClock) { + this.millisClock = millisClock; + this.nanoClock = nanoClock; this.nettyTimer = nettyTimer; this.eventExecutor = eventExecutor; this.nettyResponseFuture = nettyResponseFuture; @@ -93,8 +106,8 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N // exchange has already spent bounds it as a whole instead. Which one applies is the caller's // choice, per request or per client. Left negative when the deadline is already behind us, which is // what stops startReadTimeout arming a sibling for an exchange that is over. - requestTimeoutMillisTime = unpreciseMillisTime() - + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture) : requestTimeoutInMs); + requestTimeoutMillisTime = millisClock.getAsLong() + + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs); requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; @@ -125,7 +138,7 @@ public void start() { // absolute deadline was anchored before this holder existed, so there the remainder is the budget, // floored at zero: a task armed at zero still runs, and running is how the exchange gets failed. arm(requestTimeoutTask, absoluteDeadline - ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture), 0L) : requestTimeoutValue); + ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue); } } @@ -142,17 +155,21 @@ public void start() { * @see org.asynchttpclient.AsyncHttpClientConfig#isUseAbsoluteRequestDeadline() */ public static long remainingBudget(AsyncHttpClientConfig config, NettyResponseFuture nettyResponseFuture) { + return remainingBudget(config, nettyResponseFuture, System::nanoTime); + } + + static long remainingBudget(AsyncHttpClientConfig config, NettyResponseFuture nettyResponseFuture, LongSupplier nanoClock) { if (!nettyResponseFuture.isUseAbsoluteRequestDeadline()) { return Long.MAX_VALUE; } - return remainingBudget(requestTimeout(config, nettyResponseFuture.getTargetRequest()), nettyResponseFuture); + return remainingBudget(requestTimeout(config, nettyResponseFuture.getTargetRequest()), nettyResponseFuture, nanoClock); } - private static long remainingBudget(long requestTimeoutInMs, NettyResponseFuture nettyResponseFuture) { + private static long remainingBudget(long requestTimeoutInMs, NettyResponseFuture nettyResponseFuture, LongSupplier nanoClock) { if (requestTimeoutInMs <= -1) { return Long.MAX_VALUE; } - long spent = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - nettyResponseFuture.getStartNanos()); + long spent = TimeUnit.NANOSECONDS.toMillis(nanoClock.getAsLong() - nettyResponseFuture.getStartNanos()); return requestTimeoutInMs - spent; } @@ -207,7 +224,7 @@ public void startReadTimeout() { void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { if (requestTimeoutTask == null - || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { + || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - millisClock.getAsLong()) { // only schedule a new readTimeout if the requestTimeout doesn't happen first if (task == null) { // first call triggered from outside (else is read timeout is re-scheduling itself) @@ -239,7 +256,7 @@ private static void release(@Nullable TimeoutTimerTask task) { private long remainingRequestTimeout() { // Floored at zero rather than passed on negative: a scheduler has no use for a negative delay, and the // task has to run either way, since running is what fails the exchange. - return Math.max(requestTimeoutMillisTime - unpreciseMillisTime(), 0L); + return Math.max(requestTimeoutMillisTime - millisClock.getAsLong(), 0L); } /** diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java index 13180203c2..be4e9d8785 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java @@ -26,7 +26,10 @@ import org.junit.jupiter.api.Test; import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -42,72 +45,72 @@ public class TimeoutsHolderTest { private static final Duration BUDGET = Duration.ofMillis(600); private static final long ELAPSED_MS = 100; - // The deadline is a wall-clock reading and the budget is netted off in whole milliseconds, so an anchored - // deadline lands within a few milliseconds of itself rather than exactly on it. - private static final long TOLERANCE_MS = 30; + + // Both fake clocks move only when a test moves them. The nano one runs from the exchange's own start, + // which the future reads off System.nanoTime(). + private long nanoOrigin; + private final AtomicLong elapsedMillis = new AtomicLong(); @Test - public void anAbsoluteDeadlineStaysWhereTheExchangeStarted() throws Exception { + public void anAbsoluteDeadlineStaysWhereTheExchangeStarted() { NettyResponseFuture future = exchange(true); long firstHop = deadlineOf(future, BUDGET); - Thread.sleep(ELAPSED_MS); + elapsedMillis.addAndGet(ELAPSED_MS); long secondHop = deadlineOf(future, BUDGET); - assertTrue(Math.abs(secondHop - firstHop) <= TOLERANCE_MS, - "the second hop moved the deadline by " + (secondHop - firstHop) + " ms"); + assertEquals(firstHop, secondHop, "the second hop moved the deadline"); } @Test - public void aPerAttemptTimeoutGivesTheSecondHopItsOwnBudget() throws Exception { + public void aPerAttemptTimeoutGivesTheSecondHopItsOwnBudget() { NettyResponseFuture future = exchange(false); long firstHop = deadlineOf(future, BUDGET); - Thread.sleep(ELAPSED_MS); + elapsedMillis.addAndGet(ELAPSED_MS); long secondHop = deadlineOf(future, BUDGET); - assertTrue(secondHop - firstHop >= ELAPSED_MS / 2, - "the second hop should have started a budget of its own, moved by only " - + (secondHop - firstHop) + " ms"); + assertEquals(firstHop + ELAPSED_MS, secondHop, "the second hop should have started a budget of its own"); } @Test - public void anExchangeThatOutranItsDeadlineHasNothingLeft() throws Exception { - // A budget this small is spent by the time the sleep is over, so the next hop has nothing to run in. + public void anExchangeThatOutranItsDeadlineHasNothingLeft() { NettyResponseFuture future = exchange(true); - Thread.sleep(ELAPSED_MS); + elapsedMillis.addAndGet(ELAPSED_MS); - assertTrue(TimeoutsHolder.remainingBudget(config(Duration.ofMillis(1)), future) <= 0, + assertTrue(TimeoutsHolder.remainingBudget(config(Duration.ofMillis(1)), future, this::nanos) <= 0, "a spent deadline should leave nothing to send a further hop with"); } @Test - public void aPerAttemptExchangeIsNotBoundedAsAWhole() throws Exception { + public void aPerAttemptExchangeIsNotBoundedAsAWhole() { // Asserted on the deadline the holder computes rather than on the budget: per attempt there is no // exchange-wide budget to run out of, so the arithmetic is not what the answer rests on. NettyResponseFuture future = exchange(false); - Thread.sleep(ELAPSED_MS); + elapsedMillis.addAndGet(ELAPSED_MS); - long deadline = deadlineOf(future, BUDGET); + assertEquals(millis() + BUDGET.toMillis(), deadlineOf(future, BUDGET), + "a hop should be given the configured timeout of its own however long the exchange has run"); + } - assertTrue(deadline - System.currentTimeMillis() >= BUDGET.toMillis() - TOLERANCE_MS, - "a hop should be given the configured timeout of its own however long the exchange has run, got " - + (deadline - System.currentTimeMillis()) + " ms"); + private long millis() { + return elapsedMillis.get(); } - private static long deadlineOf(NettyResponseFuture future, Duration requestTimeout) { - return holder(future, requestTimeout).requestTimeoutMillisTime(); + private long nanos() { + return nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis.get()); } - private static TimeoutsHolder holder(NettyResponseFuture future, Duration requestTimeout) { - return new TimeoutsHolder(null, future, null, config(requestTimeout), null); + private long deadlineOf(NettyResponseFuture future, Duration requestTimeout) { + return new TimeoutsHolder(null, null, future, null, config(requestTimeout), null, this::millis, this::nanos) + .requestTimeoutMillisTime(); } private static AsyncHttpClientConfig config(Duration requestTimeout) { return new DefaultAsyncHttpClientConfig.Builder().setRequestTimeout(requestTimeout).build(); } - private static NettyResponseFuture exchange(boolean useAbsoluteRequestDeadline) { + private NettyResponseFuture exchange(boolean useAbsoluteRequestDeadline) { Request request = new RequestBuilder().setUrl("http://example.com:12345").build(); NettyResponseFuture future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { @Override @@ -116,6 +119,7 @@ public Object onCompleted(Response response) { } }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); future.setUseAbsoluteRequestDeadline(useAbsoluteRequestDeadline); + nanoOrigin = future.getStartNanos(); return future; } } From a81a54e000e56da858bc002125e7164faa0c22de Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:13:10 +0000 Subject: [PATCH 11/13] Size the piped stream test's buffer to its body --- .../asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java b/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java index b643d3e146..47b74db930 100644 --- a/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java +++ b/client/src/test/java/org/asynchttpclient/handler/BodyDeferringAsyncHandlerTest.java @@ -245,7 +245,8 @@ public void testConnectionRefused() throws Exception { public void testPipedStreams() throws Exception { try (AsyncHttpClient client = asyncHttpClient(getAsyncHttpClientConfig())) { PipedOutputStream pout = new PipedOutputStream(); - try (PipedInputStream pin = new PipedInputStream(pout)) { + // Room for the whole body: a full pipe blocks the event loop until this thread reads. + try (PipedInputStream pin = new PipedInputStream(pout, CONTENT_LENGTH_VALUE)) { BodyDeferringAsyncHandler handler = new BodyDeferringAsyncHandler(pout); ListenableFuture respFut = client.prepareGet(getTargetUrl()).execute(handler); From 72572e0a5879ebb577b7c75ef604843b85a9f1c6 Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:13:10 +0000 Subject: [PATCH 12/13] Fail the stream test when its handshake times out --- .../org/asynchttpclient/AsyncStreamLifecycleTest.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/client/src/test/java/org/asynchttpclient/AsyncStreamLifecycleTest.java b/client/src/test/java/org/asynchttpclient/AsyncStreamLifecycleTest.java index 61e3a4d118..27957bcd33 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncStreamLifecycleTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncStreamLifecycleTest.java @@ -38,6 +38,7 @@ import static org.asynchttpclient.Dsl.asyncHttpClient; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -52,6 +53,7 @@ public class AsyncStreamLifecycleTest extends AbstractBasicTest { // Counted down by the client on its first body part. The server writes the second part only after that. private volatile CountDownLatch firstPartReceived = new CountDownLatch(1); + private volatile boolean handshakeTimedOut; @Override @AfterAll @@ -75,7 +77,9 @@ public void handle(String s, Request request, HttpServletRequest req, final Http writer.write("part1"); writer.flush(); if (!firstPartReceived.await(TIMEOUT, TimeUnit.SECONDS)) { - logger.error("Client never received part1."); + // Writing part2 anyway would let the test pass without the ordering it checks. + handshakeTimedOut = true; + return; } logger.info("Delivering part2."); writer.write("part2"); @@ -96,6 +100,7 @@ public void handle(String s, Request request, HttpServletRequest req, final Http @Timeout(unit = TimeUnit.MILLISECONDS, value = 60000) public void testStream() throws Exception { firstPartReceived = new CountDownLatch(1); + handshakeTimedOut = false; try (AsyncHttpClient ahc = asyncHttpClient()) { final AtomicReference thrown = new AtomicReference<>(); final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); @@ -145,6 +150,7 @@ public Object onCompleted() { // The latch also fires on failure, so check for one before looking at the parts. assertTrue(latch.await(TIMEOUT, TimeUnit.SECONDS), () -> "Latch failed. Received so far: " + queue); assertNull(thrown.get(), () -> "Got throwable: " + thrown.get()); + assertFalse(handshakeTimedOut, "the server gave up waiting for the client to receive part1"); assertEquals(2, queue.size()); assertEquals("part1", queue.poll()); assertEquals("part2", queue.poll()); From 62d827d72da003e1c7e46b7ae174edcf2f6c971e Mon Sep 17 00:00:00 2001 From: Aayush Atharva Date: Mon, 21 Sep 2026 20:14:58 +0000 Subject: [PATCH 13/13] Assert no Jetty server outlives its test class --- .../org/asynchttpclient/AbstractBasicTest.java | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/client/src/test/java/org/asynchttpclient/AbstractBasicTest.java b/client/src/test/java/org/asynchttpclient/AbstractBasicTest.java index 5cbd1c2416..1982067ad9 100644 --- a/client/src/test/java/org/asynchttpclient/AbstractBasicTest.java +++ b/client/src/test/java/org/asynchttpclient/AbstractBasicTest.java @@ -24,7 +24,9 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.extension.AfterAllCallback; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.ExtensionContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,7 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; @TestInstance(TestInstance.Lifecycle.PER_CLASS) -@ExtendWith(NettyLeakDetectorExtension.class) +@ExtendWith({NettyLeakDetectorExtension.class, AbstractBasicTest.ServerStoppedGuard.class}) public abstract class AbstractBasicTest { protected static final Logger logger = LoggerFactory.getLogger(AbstractBasicTest.class); protected static final int TIMEOUT = 30; @@ -85,6 +87,20 @@ public void assertReplacedServerWasStopped() { } } + // An extension rather than an @AfterAll method: JUnit leaves the order of two of those in one class open, + // but runs this after all of them. + static final class ServerStoppedGuard implements AfterAllCallback { + + @Override + public void afterAll(ExtensionContext context) { + Object instance = context.getTestInstance().orElse(null); + if (instance instanceof AbstractBasicTest) { + Server server = ((AbstractBasicTest) instance).server; + assertTrue(server == null || server.isStopped(), "a Jetty server was still running at class end"); + } + } + } + protected String getTargetUrl() { return String.format("http://localhost:%d/foo/test", port1); }