diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 5cf086810..7e79e59e8 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -390,7 +390,18 @@ default Duration getFailedIpCooldownPeriod() { boolean isStrict302Handling(); /** - * @return the maximum time an {@link AsyncHttpClient} will keep connection in the pool, or negative value to keep connection while possible. + * The maximum age of a connection, or a value that is not positive (the default is negative) to keep connections + * for as long as they last. Resolution is milliseconds. + *

+ * An HTTP/1.1 connection older than this is closed instead of being returned to the pool, and the pool cleaner + * closes one that passes it while idle in the pool; its age counts from the first time it was pooled. An HTTP/2 + * connection is retired once it has been open for between 90% and 100% of this value, the spread keeping + * connections opened together from being replaced together: new requests then go to a new connection, while the + * streams already open on it, and the requests already waiting for one of its stream slots, finish on it before + * it closes. Retirement never interrupts an exchange in flight, so a long-lived response keeps its connection + * open past this age. + * + * @return the maximum age of a pooled HTTP/1.1 or multiplexed HTTP/2 connection */ Duration getConnectionTtl(); diff --git a/client/src/main/java/org/asynchttpclient/LoadBalance.java b/client/src/main/java/org/asynchttpclient/LoadBalance.java index e4caa9f80..95443f14e 100644 --- a/client/src/main/java/org/asynchttpclient/LoadBalance.java +++ b/client/src/main/java/org/asynchttpclient/LoadBalance.java @@ -63,17 +63,19 @@ public enum LoadBalance { * IPs, and a request pinned to any further IP cannot acquire a per-host permit — it multiplexes * onto one of those K sibling connections (multiplexing onto an existing connection takes no * permit) instead of opening another, failing with {@code TooManyConnectionsPerHostException} - * only if none is available. Those K connections are held open (kept alive by HTTP/2 PING), so - * they stay pinned to their K IPs rather than rotating across all of them; set - * {@code maxConnectionsPerHost} at least as large as the resolved-IP count for full per-IP - * spreading. The default is unlimited, so this bound only applies when a finite per-host limit is - * configured. + * only if none is available. Those K connections stay pinned to their K IPs for as long as they + * are open, rather than rotating across all of them; {@code connectionTtl}, when set, retires them + * so that the set moves across the IPs over time. Set {@code maxConnectionsPerHost} at least as + * large as the resolved-IP count for full per-IP spreading. The default is unlimited, so this + * bound only applies when a finite per-host limit is configured. *

  • Since the per-host permit is held for the connection's lifetime, each live round-robin HTTP/2 - * connection also occupies a global {@code maxConnections} slot until it closes, and pool - * {@code connectionTtl}/idle reaping do not apply (these connections live in the HTTP/2 registry, - * kept alive by PING, until GOAWAY or the connection drops). When a GOAWAY starts draining a - * connection its permit is released immediately, so a replacement can be opened without waiting - * for in-flight streams to finish.
    • + * connection also occupies a global {@code maxConnections} slot until it is retired or + * drained. These connections live in the HTTP/2 registry rather than the pool, so idle reaping + * does not apply: one stays open until the server closes it, a GOAWAY drains it, or it reaches + * {@link AsyncHttpClientConfig#getConnectionTtl() connectionTtl}, and HTTP/2 PING, when enabled, + * keeps an idle one alive. When a GOAWAY or {@code connectionTtl} retires a connection its permit + * is released immediately, so a replacement can be opened without waiting for in-flight streams to + * finish; until they do, the old and the new connection are both open. *
  • The address order comes straight from the configured * {@link io.netty.resolver.InetNameResolver}; this mode does not re-sort it. For the * rotation to map consistently across requests, use a resolver that returns the addresses diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index e0b7159a4..525fdc963 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -86,6 +86,9 @@ public final class NettyResponseFuture implements ListenableFuture { private static final AtomicIntegerFieldUpdater ON_THROWABLE_CALLED_FIELD = AtomicIntegerFieldUpdater .newUpdater(NettyResponseFuture.class, "onThrowableCalled"); @SuppressWarnings("rawtypes") + private static final AtomicIntegerFieldUpdater HTTP2_REDISPATCHED_FIELD = AtomicIntegerFieldUpdater + .newUpdater(NettyResponseFuture.class, "http2Redispatched"); + @SuppressWarnings("rawtypes") private static final AtomicReferenceFieldUpdater TIMEOUTS_HOLDER_FIELD = AtomicReferenceFieldUpdater .newUpdater(NettyResponseFuture.class, TimeoutsHolder.class, "timeoutsHolder"); @SuppressWarnings("rawtypes") @@ -118,6 +121,8 @@ public final class NettyResponseFuture implements ListenableFuture { private volatile int contentProcessed; @SuppressWarnings("unused") private volatile int onThrowableCalled; + @SuppressWarnings("unused") + private volatile int http2Redispatched; private volatile TimeoutsHolder timeoutsHolder; @SuppressWarnings("unused") private volatile @Nullable NettyResponseBodyControl responseBodyControl; @@ -622,6 +627,17 @@ public boolean incrementRetryAndCheck() { return maxRetry > 0 && CURRENT_RETRY_UPDATER.incrementAndGet(this) <= maxRetry; } + /** + * Claims the one re-dispatch this exchange is allowed when the HTTP/2 connection it was bound for was retired at + * {@code connectionTtl} before a stream could be opened. Separate from the retry count: the request was never + * sent, so re-sending it is not a retry. + * + * @return {@code true} the first time only + */ + public boolean claimHttp2Redispatch() { + return HTTP2_REDISPATCHED_FIELD.compareAndSet(this, 0, 1); + } + /** * Return true if the {@link Future} can be recovered. There is some scenario * where a connection can be closed by an unexpected IOException, and in some diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index 2095bfaab..2394fe5a4 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -67,6 +67,7 @@ import io.netty.util.concurrent.GlobalEventExecutor; import io.netty.util.concurrent.ImmediateEventExecutor; import io.netty.util.concurrent.Promise; +import io.netty.util.concurrent.ScheduledFuture; import io.netty.util.internal.PlatformDependent; import org.asynchttpclient.AsyncHandler; import org.asynchttpclient.AsyncHttpClientConfig; @@ -101,6 +102,7 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.time.Duration; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -108,6 +110,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; @@ -146,6 +149,9 @@ public class ChannelManager { // initializer instead of via Bootstrap#option to avoid Netty's synchronized per-connect options map (issue #2218). private final Map.Entry, Object>[] channelOptions; private final long handshakeTimeout; + // connectionTtl in milliseconds, bounding the age of every HTTP/2 connection; not positive means unbounded, + // the same conversion and test DefaultChannelPool applies to HTTP/1.1 connections. + private final long http2ConnectionTtl; private final @Nullable AddressResolverGroup addressResolverGroup; private final ChannelPool channelPool; @@ -217,6 +223,17 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) { this.channelPool = channelPool; openChannels = new DefaultChannelGroup("asyncHttpClient", GlobalEventExecutor.INSTANCE); handshakeTimeout = config.getHandshakeTimeout(); + // Null was never rejected, and only DefaultChannelPool dereferences it, so a client with its own pool or + // without keep-alive could be built with it. + Duration connectionTtl = config.getConnectionTtl(); + long connectionTtlMillis; + try { + connectionTtlMillis = connectionTtl != null ? connectionTtl.toMillis() : -1; + } catch (ArithmeticException tooLongForMillis) { + // A Duration beyond Long.MAX_VALUE milliseconds (ChronoUnit.FOREVER's, say) is as good as unbounded. + connectionTtlMillis = -1; + } + http2ConnectionTtl = connectionTtlMillis; // check if external EventLoopGroup is defined ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName()); @@ -1280,6 +1297,53 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception pipeline.addLast("http2-idle-state", new IdleStateHandler(0, 0, pingIntervalMs, TimeUnit.MILLISECONDS)); pipeline.addLast("http2-ping", new Http2PingHandler()); } + + // Bound the connection's age by connectionTtl. Nothing else would: the registry has no idle reaper, and PING, + // when enabled, keeps an idle connection alive. Up to a tenth earlier than configured, never later, so that + // connections opened together (one per IP in LoadBalance.ROUND_ROBIN, or a fleet started at once) are not all + // replaced at the same instant every time. + if (http2ConnectionTtl > 0) { + long delayMillis = http2ConnectionTtl - ThreadLocalRandom.current().nextLong(http2ConnectionTtl / 10 + 1); + pipeline.addLast("http2-connection-ttl", new Http2ConnectionTtlHandler(delayMillis)); + } + } + + /** + * Retires an HTTP/2 connection that reached {@code connectionTtl}. It is no longer handed out, the streams open + * on it and the openers queued for it still run on it, and it closes once it has neither. Runs on the + * connection's event loop. + */ + private void retireHttp2Connection(Channel channel) { + Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); + if (state == null) { + return; + } + // Must leave the registry, not just be flagged. While it is registered and active it stays the canonical + // connection for its key, and registerHttp2Connection marks any newcomer for that key redundant: the + // replacement would serve one request and close, and so would every connection after it until this one closed. + // Removed before it is marked retired, so that a request which sees it retired and is sent again cannot find + // it in the registry. + Object partitionKey = state.getPartitionKey(); + if (partitionKey != null) { + removeHttp2Connection(partitionKey, channel); + } + if (!state.retire()) { + return; + } + // As at GOAWAY: free the round-robin permit now, so a replacement can open while streams finish here. + state.releasePermitOnce(); + LOGGER.debug("HTTP/2 connection {} reached connectionTtl, retiring it", channel); + // No GOAWAY now. RFC 9113 section 6.8 asks for one before closing, and Netty's close() sends GOAWAY(NO_ERROR) + // when this connection is closed idle. An early one would cost requests and buy nothing: a request that looked + // this connection up just before it was retired still opens a stream on it, and a server may treat a client + // GOAWAY as the start of its own shutdown and drop streams opened after it, as Go's HTTP/2 server does. + // Until then a replacement may be open alongside. Section 9.1 asks clients not to open more than one + // connection per origin and lists replacements only for stream-identifier exhaustion, TLS rekeying and + // errors, so this overlap is a deliberate deviation of the kind LoadBalance.ROUND_ROBIN already documents, + // bounded by the streams still in flight here. + if (state.closeIfRetiredAndIdle()) { + closeChannel(channel); + } } /** @@ -1412,6 +1476,36 @@ private static final class ConnectionCounts { private long idleConnectionCount; } + /** + * Retires its HTTP/2 connection once it has been open for the given delay. The task runs on the connection's own + * event loop, so it is serialized with the GOAWAY and stream-close handling there, and it is cancelled when the + * pipeline is torn down on close, so it never outlives the connection. + */ + private final class Http2ConnectionTtlHandler extends ChannelInboundHandlerAdapter { + + private final long delayMillis; + private @Nullable ScheduledFuture expiry; + + private Http2ConnectionTtlHandler(long delayMillis) { + this.delayMillis = delayMillis; + } + + @Override + public void handlerAdded(ChannelHandlerContext ctx) { + expiry = ctx.executor().schedule(() -> retireHttp2Connection(ctx.channel()), + delayMillis, TimeUnit.MILLISECONDS); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) { + ScheduledFuture scheduled = expiry; + if (scheduled != null) { + scheduled.cancel(false); + expiry = null; + } + } + } + private static final class ClientHttp2FrameCodecBuilder extends Http2FrameCodecBuilder { private final SuspensionAwareHttp2LocalFlowController flowController; 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 4cfd3a8ef..72c2f9b27 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java @@ -28,7 +28,8 @@ /** * Tracks per-connection HTTP/2 state: active stream count, max concurrent streams, - * draining status (from GOAWAY), and pending stream openers. + * draining status (from GOAWAY), retirement (the connection reached {@code connectionTtl}), + * and pending stream openers. */ public class Http2ConnectionState { @@ -75,6 +76,15 @@ private static final class PendingOpener { // — rather than lingering open and unregistered. Distinct from draining: it does NOT block its own // opening request from acquiring a slot. private final AtomicBoolean redundant = new AtomicBoolean(false); + // Set when the connection reached connectionTtl. Distinct from draining: the server has refused nothing, so + // a retired connection still opens streams for requests that already hold it (queued openers, and a request + // that looked it up just before it was retired); it only stops being handed out, and closes once idle. + private final AtomicBoolean retired = new AtomicBoolean(false); + // Parked in activeStreams by closeIfRetiredAndIdle so that admitting a stream and deciding to close are one + // CAS on one word: either a late tryAcquireStream gets its slot and keeps the connection open, or the + // connection closes and that tryAcquireStream fails before anything is written. Checking the count and then + // closing would let a stream open on a connection that is being closed under it. + private static final int IDLE_CLOSED = Integer.MIN_VALUE; private volatile Object partitionKey; // Releases the per-host permit this connection holds in ROUND_ROBIN mode (installed by // NettyConnectListener, absent in DEFAULT mode). Fired by the GOAWAY handler so the permit is freed at @@ -89,7 +99,7 @@ public boolean tryAcquireStream() { } while (true) { int current = activeStreams.get(); - if (current >= maxConcurrentStreams) { + if (current == IDLE_CLOSED || current >= maxConcurrentStreams) { return false; } if (activeStreams.compareAndSet(current, current + 1)) { @@ -133,14 +143,18 @@ public boolean offerPendingOpener(Runnable opener) { * sequenced after it observes {@code closed} here (the lock provides the happens-before) and is rejected. * Either way no opener is left stranded. Opener callbacks always run after releasing {@code pendingLock}, * because opening a stream can invoke user code and must not serialize other request submissions. + *

    + * A retired connection also rejects: it is on its way out, so a request that would have to wait for one of its + * slots is better sent to a fresh connection. Openers queued before retirement stay queued and still run here. * * @return {@code true} if the opener was run inline or queued; {@code false} if rejected because the - * connection is draining/closed or the pending queue is full (caller must fail the request) + * connection is draining/closed/retired or the pending queue is full (caller must fail or re-dispatch + * the request) */ public boolean offerPendingOpener(NettyResponseFuture future, Runnable opener) { boolean runOpener = false; synchronized (pendingLock) { - if (draining.get() || closed.get()) { + if (draining.get() || closed.get() || retired.get()) { return false; } if (tryAcquireStream()) { @@ -249,13 +263,47 @@ public int getMaxConcurrentStreams() { } public int getActiveStreams() { - return activeStreams.get(); + int current = activeStreams.get(); + return current == IDLE_CLOSED ? 0 : current; } public boolean isDraining() { return draining.get(); } + /** + * Marks the connection as having reached {@code connectionTtl}. + * + * @return {@code true} only for the call that retired it, so the retirement work runs once + */ + public boolean retire() { + return retired.compareAndSet(false, true); + } + + public boolean isRetired() { + return retired.get(); + } + + /** + * For a retired connection with no open stream and nothing queued, permanently refuses further streams. + *

    + * The refusal and the idle check are a single CAS on the stream count, so it is exclusive with a concurrent + * {@link #tryAcquireStream()}: exactly one of them wins. The queue is checked under {@code pendingLock}, which + * every enqueue holds. + * + * @return {@code true} if the connection is now closed to streams and the caller must close the channel; + * {@code false} if it is not retired or still has work, and for every call after the one that returned + * {@code true} + */ + public boolean closeIfRetiredAndIdle() { + if (!retired.get()) { + return false; + } + synchronized (pendingLock) { + return pendingCount == 0 && activeStreams.compareAndSet(0, IDLE_CLOSED); + } + } + /** Marks this connection as a redundant duplicate that should close once its last stream ends. */ public void markRedundant() { redundant.set(true); 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 7d175d369..bfc82a0ec 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -436,6 +436,11 @@ private ListenableFuture sendRequestWithOpenChannel(NettyResponseFuture void writeRequest(NettyResponseFuture future, Channel channel) { // if the channel is dead because it was pooled and the remote server decided to close it, // we just let it go and the channelInactive do its work if (!Channels.isChannelActive(channel)) { + // Except for a retired HTTP/2 connection, usually closed idle at connectionTtl: nothing stores this request + // on it, so no channelInactive would ever complete it. Nothing was written, so send it on a fresh one. + if (isRetiredHttp2Connection(channel)) { + redispatchUnsentHttp2Request(future); + } return; } @@ -775,6 +785,14 @@ public void writeRequest(NettyResponseFuture future, Channel channel) { */ private void writeHttp2Request(NettyResponseFuture future, Channel parentChannel, Http2ConnectionState state) { if (!state.tryAcquireStream()) { + if (state.isRetired()) { + // Retired at connectionTtl after this request looked the connection up: it is closing, or full and + // on its way out. Nothing has been written, so send the request on a fresh connection instead. + // Checked before draining on purpose: a retired connection that a server GOAWAY has since drained + // still never received this request, so it is re-sent rather than failed like a draining refusal. + redispatchUnsentHttp2Request(future); + return; + } if (state.isDraining()) { // Connection is draining from GOAWAY — fail the future so it retries on a new connection. // Don't close the parent channel since it may still have active streams. sendHttp2Frames @@ -788,6 +806,10 @@ private void writeHttp2Request(NettyResponseFuture future, Channel parent // tryAcquireStream() above and this enqueue — such an opener would never run, so fail the // request now instead of leaving it to be completed only by the request timeout (Issue #2160). if (!state.offerPendingOpener(future, () -> openHttp2Stream(future, parentChannel, state))) { + if (state.isRetired()) { + redispatchUnsentHttp2Request(future); + return; + } releaseHttp2Request(future); // Fail ONLY this future (future.abort, not abort(parentChannel, ...)): the parent may be // draining-but-still-active with healthy sibling streams, and abort(channel, ...) would close @@ -809,6 +831,50 @@ private void writeHttp2Request(NettyResponseFuture future, Channel parent openHttp2Stream(future, parentChannel, state); } + /** + * Sends again, on another connection, a request that was bound for an HTTP/2 connection retired at + * {@code connectionTtl} but could not get a stream on it: the connection was closing, or every slot was taken. + *

    + * Nothing of the request reached the wire, so this is not a replay: unlike {@link #retry}, it spends none of the + * {@code maxRequestRetry} budget and does not call {@link AsyncHandler#onRetry()}. The attempt goes through + * {@link #sendNextRequest} like a retry does, so it picks the proxy again, re-polls the pool + * ({@code onConnectionPoolAttempt}, then {@code onConnectionPooled} or the resolution, connect and TLS callbacks + * of a new connection), rebuilds the request from the current {@link Request}, and re-arms its timeouts: the + * attempt gets a fresh {@code requestTimeout}, or what is left of the exchange's budget when the request + * timeout is an absolute deadline ({@code useAbsoluteRequestDeadline}). Request filters do not run again; they + * ran once, before the first attempt. The retired connection left the registry before it was marked retired, so + * the attempt cannot land on it again. It is allowed once per exchange; a second refusal fails it as before. + */ + private void redispatchUnsentHttp2Request(NettyResponseFuture future) { + releaseHttp2Request(future); + if (future.isDone()) { + // A timeout or cancel got there first. Sending it anyway could reach attachChannel on a done future, which + // closes the channel it is handed: that can be a healthy connection shared with other requests. + return; + } + if (isClosed() || !future.claimHttp2Redispatch()) { + future.abort(new IOException("HTTP/2 connection reached connectionTtl before the request could be sent")); + return; + } + // The future still points at the retired connection until the new attempt attaches its own. A timeout or + // cancel in between, during the replacement's DNS lookup and connect, would close that connection under the + // streams still finishing on it. + future.attachChannel(null, false); + try { + sendNextRequest(future.getCurrentRequest(), future); + } catch (Exception e) { + abort(null, future, e); + } + } + + private static boolean isRetiredHttp2Connection(@Nullable Channel channel) { + if (channel == null) { + return false; + } + Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); + return state != null && state.isRetired(); + } + private void openHttp2Stream(NettyResponseFuture future, Channel parentChannel, Http2ConnectionState state) { new Http2StreamChannelBootstrap(parentChannel) .handler(new ChannelInitializer() { @@ -840,7 +906,8 @@ protected void initChannel(Http2StreamChannel streamCh) { // NettyConnectListener does for the HTTP/1.1 connection semaphore) is the one place // guaranteed to run exactly once per opened stream, so activeStreams cannot leak and // wedge the connection. (The open-FAILURE branch below has no channel and releases - // inline.) The listener also closes a draining connection once its last stream ends. + // inline.) The listener also closes a draining or retired connection once its last stream + // ends. streamChannel.closeFuture().addListener(closed -> { // Safety net: free the request body when the stream ends, however it ends — // covers an Expect/100-continue request whose server answered the final response @@ -852,8 +919,11 @@ protected void initChannel(Http2StreamChannel streamCh) { state.releaseStream(); // Close the parent once it has no active streams AND it is either draining // (GOAWAY) or a redundant duplicate (#10 thundering-herd loser) — neither - // will serve further requests, so it must not linger open. - if ((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) { + // will serve further requests, so it must not linger open. A retired one (it + // reached connectionTtl) still admits late streams, so it closes only through + // the atomic idle check, which a concurrent tryAcquireStream cannot slip past. + if (((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) + || state.closeIfRetiredAndIdle()) { channelManager.closeChannel(parentChannel); } }); @@ -892,8 +962,12 @@ protected void initChannel(Http2StreamChannel streamCh) { } } else { // Stream channel was never opened — no closeFuture will fire, so release the - // acquired slot and the unsent request body inline. + // acquired slot and the unsent request body inline. If that was the last slot of a + // retired connection, no stream close is left to close it, so do it here. state.releaseStream(); + if (state.closeIfRetiredAndIdle()) { + channelManager.closeChannel(parentChannel); + } releaseHttp2Request(future); // Fail ONLY this future (future.abort, not abort(parentChannel, ...)): opening one stream // can fail for a stream-local reason (e.g. Netty rejecting it as the outbound max-streams diff --git a/client/src/test/java/org/asynchttpclient/BasicHttp2Test.java b/client/src/test/java/org/asynchttpclient/BasicHttp2Test.java index 2190941f1..9faa27634 100644 --- a/client/src/test/java/org/asynchttpclient/BasicHttp2Test.java +++ b/client/src/test/java/org/asynchttpclient/BasicHttp2Test.java @@ -60,6 +60,7 @@ import io.netty.util.concurrent.Promise; import org.asynchttpclient.handler.MaxRedirectException; import org.asynchttpclient.handler.RedirectRefusedException; +import org.asynchttpclient.netty.request.NettyRequest; import org.asynchttpclient.proxy.ProxyServer; import org.asynchttpclient.proxy.ProxyType; import org.asynchttpclient.test.EventCollectingHandler; @@ -89,6 +90,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -114,7 +117,9 @@ 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.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -773,6 +778,222 @@ public void http2RoundRobinGoawayReleasesPermitForReplacementConnection() throws } } + /** + * Records which connection an exchange ran on: {@link #opened} when it opened one, {@link #pooled} when it + * reused one, and when it was written. + */ + private static final class ConnectionRecorder extends AsyncCompletionHandlerBase { + + private final AtomicReference opened = new AtomicReference<>(); + private final AtomicReference pooled = new AtomicReference<>(); + private final CountDownLatch sent = new CountDownLatch(1); + + @Override + public void onTcpConnectSuccess(InetSocketAddress remoteAddress, Channel connection) { + opened.set(connection); + } + + @Override + public void onConnectionPooled(Channel connection) { + pooled.set(connection); + } + + @Override + public void onRequestSend(NettyRequest request) { + sent.countDown(); + } + + private Channel connection() { + Channel channel = opened.get() != null ? opened.get() : pooled.get(); + assertNotNull(channel, "the exchange ran on no connection"); + return channel; + } + } + + /** + * Issue #2370: an HTTP/2 connection is retired once it reaches {@code connectionTtl}. Idle, it closes at that + * point, and the next request opens a new connection. + */ + @Test + public void http2ConnectionIsReplacedAfterConnectionTtl() throws Exception { + try (AsyncHttpClient client = http2ClientWithConfig(b -> b.setConnectionTtl(Duration.ofMillis(300)))) { + ConnectionRecorder first = new ConnectionRecorder(); + Response firstResponse = client.prepareGet(httpsUrl("/ok")).execute(first).get(10, SECONDS); + assertEquals(HttpProtocol.HTTP_2, firstResponse.getProtocol()); + Channel firstConnection = first.connection(); + + assertTrue(firstConnection.closeFuture().await(10, SECONDS), + "an idle HTTP/2 connection must close by itself once it reaches connectionTtl"); + + ConnectionRecorder second = new ConnectionRecorder(); + Response secondResponse = client.prepareGet(httpsUrl("/ok")).execute(second).get(10, SECONDS); + assertEquals(200, secondResponse.getStatusCode()); + assertEquals(HttpProtocol.HTTP_2, secondResponse.getProtocol()); + assertNotNull(second.opened.get(), "the next request must open a new connection"); + assertNotSame(firstConnection, second.connection()); + } + } + + /** + * Issue #2370: a stream in flight when its connection reaches {@code connectionTtl} finishes normally on it, + * while requests sent after the expiry already go to a new connection; the retired one closes once the stream + * ends. + */ + @Test + public void http2StreamInFlightAtConnectionTtlCompletesWhileNewRequestsMove() throws Exception { + try (AsyncHttpClient client = http2ClientWithConfig(b -> b.setConnectionTtl(Duration.ofSeconds(1)))) { + ConnectionRecorder longRunning = new ConnectionRecorder(); + ListenableFuture longResponse = client.prepareGet(httpsUrl("/delay/3000")).execute(longRunning); + assertTrue(longRunning.sent.await(10, SECONDS)); + Channel retiring = longRunning.connection(); + + // Sequential requests ride the connection until it is retired; the first one after that opens another. + // Each round trip paces the loop, which the long-running stream's 3 s bounds well past the 1 s TTL. + ConnectionRecorder moved = null; + long deadline = System.nanoTime() + SECONDS.toNanos(10); + while (moved == null && System.nanoTime() < deadline) { + ConnectionRecorder next = new ConnectionRecorder(); + client.prepareGet(httpsUrl("/ok")).execute(next).get(10, SECONDS); + if (next.connection() != retiring) { + moved = next; + } + } + assertNotNull(moved, "requests sent after connectionTtl must move to a new connection"); + assertNotNull(moved.opened.get(), "by opening one"); + assertFalse(longResponse.isDone(), "the long-running stream is still in flight on the retired connection"); + assertTrue(retiring.isActive(), "and the retired connection stays open for it"); + + assertEquals(200, longResponse.get(10, SECONDS).getStatusCode(), + "a stream in flight at connectionTtl must complete normally"); + assertTrue(retiring.closeFuture().await(10, SECONDS), + "the retired connection must close once its last stream has ended"); + } + } + + /** + * Issue #2370: a request queued for a stream slot when its connection reaches {@code connectionTtl} is served on + * that connection once a slot frees up, rather than being failed. + */ + @Test + public void http2RequestQueuedAtConnectionTtlIsServedOnTheRetiredConnection() throws Exception { + try (AsyncHttpClient client = http2ClientWithConfig(b -> b.setConnectionTtl(Duration.ofSeconds(1)) + .setHttp2MaxConcurrentStreams(1))) { + ConnectionRecorder holder = new ConnectionRecorder(); + ListenableFuture holderResponse = client.prepareGet(httpsUrl("/delay/3000")).execute(holder); + assertTrue(holder.sent.await(10, SECONDS)); + Channel connection = holder.connection(); + + // The only slot is held for 3 s, so this waits in the queue across the 1 s expiry. + ConnectionRecorder queued = new ConnectionRecorder(); + ListenableFuture queuedResponse = client.prepareGet(httpsUrl("/ok")).execute(queued); + assertSame(connection, queued.pooled.get(), "queued on the connection the long stream holds"); + assertEquals(1, queued.sent.getCount(), "and not yet sent, since there is no free slot"); + + assertEquals(200, queuedResponse.get(10, SECONDS).getStatusCode(), + "a request queued at connectionTtl must be served, not failed"); + assertEquals(200, holderResponse.get(10, SECONDS).getStatusCode()); + assertNull(queued.opened.get(), "it was served on the retired connection, not on a new one"); + assertTrue(connection.closeFuture().await(10, SECONDS), + "the retired connection must close once both exchanges are done"); + } + } + + /** + * Holds an exchange between its pool lookup and its write, for as long as the connection it was handed for the + * first time is open, and counts retries. + */ + private static final class IdleCloseRacer extends AsyncCompletionHandlerBase { + + private final Channel heldUntilClosed; + private final AtomicInteger retries; + private final AtomicBoolean pooledOnIt = new AtomicBoolean(); + + private IdleCloseRacer(Channel heldUntilClosed, AtomicInteger retries) { + this.heldUntilClosed = heldUntilClosed; + this.retries = retries; + } + + @Override + public void onConnectionPooled(Channel connection) { + if (connection == heldUntilClosed) { + pooledOnIt.set(true); + connection.closeFuture().awaitUninterruptibly(10, SECONDS); + } + } + + @Override + public void onRetry() { + retries.incrementAndGet(); + } + } + + /** + * Issue #2370: requests that looked the connection up just before it reached {@code connectionTtl}, and reach + * their write only after it was closed idle, are sent on a new connection. Nothing was written, so they are + * neither failed nor retried, and none is left waiting for its request timeout. + */ + @Test + public void http2RequestsRacingTheIdleCloseAtConnectionTtlAreSentOnANewConnection() throws Exception { + ExecutorService callers = Executors.newFixedThreadPool(2); + try (AsyncHttpClient client = http2ClientWithConfig(b -> b.setConnectionTtl(Duration.ofSeconds(1)) + .setMaxRequestRetry(0))) { + ConnectionRecorder first = new ConnectionRecorder(); + client.prepareGet(httpsUrl("/ok")).execute(first).get(10, SECONDS); + Channel idle = first.connection(); + AtomicInteger retries = new AtomicInteger(); + + // Two, because only the first request to find the closed connection can take its one active token, and the + // closed-channel path completes nothing for any later one. + List racers = List.of(new IdleCloseRacer(idle, retries), new IdleCloseRacer(idle, retries)); + List>> submitted = new ArrayList<>(); + for (IdleCloseRacer racer : racers) { + submitted.add(callers.submit(() -> client.prepareGet(httpsUrl("/ok")).execute(racer))); + } + for (Future> request : submitted) { + assertEquals(200, request.get(10, SECONDS).get(10, SECONDS).getStatusCode()); + } + + for (IdleCloseRacer racer : racers) { + assertTrue(racer.pooledOnIt.get(), + "each request must have been handed the connection before it closed"); + } + assertFalse(idle.isOpen()); + assertEquals(0, retries.get(), "nothing was sent, so nothing may count as a retry"); + } finally { + callers.shutdownNow(); + } + } + + /** + * Issue #2370 with {@link LoadBalance#ROUND_ROBIN}: each per-IP connection is retired on its own, and replaced. + */ + @Test + public void http2RoundRobinConnectionsAreEachReplacedAfterConnectionTtl() throws Exception { + NameResolver resolver = multiIpResolver("127.0.0.1", "127.0.0.2"); + try (AsyncHttpClient client = http2ClientWithConfig(b -> b.setLoadBalance(LoadBalance.ROUND_ROBIN) + .setConnectionTtl(Duration.ofMillis(300)))) { + List firstConnections = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + ConnectionRecorder recorder = new ConnectionRecorder(); + client.executeRequest(Dsl.get(httpsUrl("/ok")).setNameResolver(resolver), recorder).get(10, SECONDS); + firstConnections.add(recorder.connection()); + } + assertNotSame(firstConnections.get(0), firstConnections.get(1), "one connection per IP"); + + for (Channel connection : firstConnections) { + assertTrue(connection.closeFuture().await(10, SECONDS), + "every per-IP connection must be retired at connectionTtl"); + } + for (int i = 0; i < 2; i++) { + ConnectionRecorder recorder = new ConnectionRecorder(); + Response response = client.executeRequest(Dsl.get(httpsUrl("/ok")).setNameResolver(resolver), recorder) + .get(10, SECONDS); + assertEquals(200, response.getStatusCode()); + assertNotNull(recorder.opened.get(), "each IP gets a new connection"); + } + } + } + /** * Creates an AHC client with a specific request timeout. */ diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2LifetimeTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2LifetimeTest.java new file mode 100644 index 000000000..988fff414 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2LifetimeTest.java @@ -0,0 +1,266 @@ +/* + * 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 io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http2.Http2Error; +import io.netty.handler.codec.http2.Http2FrameCodecBuilder; +import io.netty.handler.codec.http2.Http2GoAwayFrame; +import io.netty.util.HashedWheelTimer; +import io.netty.util.ReferenceCountUtil; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncHttpClientConfig; +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.InetAddress; +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * An HTTP/2 connection is retired once it has been open for {@code connectionTtl}, give or take the tenth taken off + * so that connections opened together do not expire together. Driven on a frozen {@link EmbeddedChannel} clock, so + * the expiry is an exact point in time rather than something to wait for. The end-to-end path over a real socket + * is covered by {@code BasicHttp2Test}. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class ChannelManagerHttp2LifetimeTest { + + private static final String TTL_HANDLER = "http2-connection-ttl"; + private static final long TTL_MILLIS = 60_000; + // The earliest the connection may be retired: a tenth before the configured TTL. + private static final long EARLIEST_EXPIRY_MILLIS = TTL_MILLIS - TTL_MILLIS / 10; + private static final String BASE = "https://host:443"; + + private final List channels = new ArrayList<>(); + private ChannelManager channelManager; + private Timer timer; + + @BeforeEach + void setUp() { + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config().setConnectionTtl(Duration.ofMillis(TTL_MILLIS)).build(), timer); + } + + @AfterEach + void tearDown() { + for (EmbeddedChannel channel : channels) { + channel.finishAndReleaseAll(); + } + channelManager.close(); + timer.stop(); + } + + /** + * An upgraded, registered HTTP/2 connection whose clock starts frozen at the upgrade, as NettyConnectListener + * leaves it. + */ + private EmbeddedChannel http2Connection(Object partitionKey) { + EmbeddedChannel channel = new EmbeddedChannel(); + channels.add(channel); + channel.freezeTime(); + channelManager.upgradePipelineToHttp2(channel.pipeline()); + channelManager.registerHttp2Connection(partitionKey, channel); + return channel; + } + + private static Http2ConnectionState stateOf(EmbeddedChannel channel) { + return channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); + } + + private static void advance(EmbeddedChannel channel, long millis) { + channel.advanceTimeBy(millis, TimeUnit.MILLISECONDS); + channel.runScheduledPendingTasks(); + } + + @Test + void noLifetimeUnlessConnectionTtlIsPositive() { + assertNoTtlHandler(config().build()); + assertNoTtlHandler(config().setConnectionTtl(Duration.ofMillis(-1)).build()); + // Below a millisecond counts as unset, as it does for the HTTP/1.1 pool. + assertNoTtlHandler(config().setConnectionTtl(Duration.ofNanos(500_000)).build()); + // Too long for milliseconds counts as unbounded rather than failing the client's construction. Keep-alive is + // off because DefaultChannelPool, which the HTTP/1.1 pool uses otherwise, cannot take such a value either. + assertNoTtlHandler(config().setKeepAlive(false).setConnectionTtl(ChronoUnit.FOREVER.getDuration()).build()); + } + + private static void assertNoTtlHandler(AsyncHttpClientConfig config) { + Timer timer = new HashedWheelTimer(); + ChannelManager manager = new ChannelManager(config, timer); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + manager.upgradePipelineToHttp2(channel.pipeline()); + assertNull(channel.pipeline().get(TTL_HANDLER), "connectionTtl " + config.getConnectionTtl()); + } finally { + channel.finishAndReleaseAll(); + manager.close(); + timer.stop(); + } + } + + @Test + void connectionIsNotRetiredBeforeTheEarliestExpiry() { + EmbeddedChannel channel = http2Connection(BASE); + assertNotNull(channel.pipeline().get(TTL_HANDLER)); + + advance(channel, EARLIEST_EXPIRY_MILLIS - 1); + + assertFalse(stateOf(channel).isRetired()); + assertSame(channel, channelManager.pollHttp2Connection(BASE), "still handed out"); + assertTrue(channel.isOpen()); + } + + @Test + void idleConnectionIsRetiredAndClosedWithGoAwayAtConnectionTtl() { + EmbeddedChannel channel = http2Connection(BASE); + + advance(channel, TTL_MILLIS); + + assertTrue(stateOf(channel).isRetired()); + assertNull(channelManager.pollHttp2Connection(BASE)); + assertFalse(channel.isOpen(), "nothing runs on it, so it closes at once"); + assertEquals(List.of(Http2Error.NO_ERROR.code()), goAwayErrorCodesSentBy(channel), + "exactly one GOAWAY(NO_ERROR), sent by the close and not earlier"); + } + + @Test + void busyConnectionIsRetiredButStaysOpenUntilItsLastStreamEnds() { + EmbeddedChannel channel = http2Connection(BASE); + Http2ConnectionState state = stateOf(channel); + assertTrue(state.tryAcquireStream()); + + advance(channel, TTL_MILLIS); + + assertTrue(state.isRetired()); + assertNull(channelManager.pollHttp2Connection(BASE), "no longer handed out"); + assertTrue(channel.isOpen(), "a stream in flight must be allowed to finish"); + assertTrue(goAwayErrorCodesSentBy(channel).isEmpty(), "no GOAWAY while the connection is still in use"); + + // What the stream's close listener in NettyRequestSender does. + state.releaseStream(); + assertTrue(state.closeIfRetiredAndIdle(), "the last stream ending makes it closable"); + } + + @Test + void replacementRegisteredAfterRetirementIsTheConnectionHandedOut() { + EmbeddedChannel retired = http2Connection(BASE); + assertTrue(stateOf(retired).tryAcquireStream()); + advance(retired, TTL_MILLIS); + + EmbeddedChannel replacement = http2Connection(BASE); + + assertFalse(stateOf(replacement).isRedundant(), + "a retired connection left in the registry would make every replacement redundant"); + assertSame(replacement, channelManager.pollHttp2Connection(BASE)); + + retired.close(); + assertSame(replacement, channelManager.pollHttp2Connection(BASE), + "the retired connection closing must not take its replacement out of the registry"); + } + + @Test + void roundRobinPermitIsReleasedAtRetirementAndNotAgainOnClose() throws Exception { + PerHostConnectionSemaphore semaphore = new PerHostConnectionSemaphore(1, 0); + semaphore.acquireChannelLock(BASE); + RoundRobinPartitionKey key = new RoundRobinPartitionKey(BASE, InetAddress.getByName("127.0.0.1")); + EmbeddedChannel channel = http2Connection(key); + // NettyConnectListener's round-robin wiring: a release hook on the state and on the closeFuture, sharing + // one getAndSet. + AtomicReference permit = new AtomicReference<>(BASE); + Runnable release = () -> { + Object held = permit.getAndSet(null); + if (held != null) { + semaphore.releaseChannelLock(held); + } + }; + stateOf(channel).setPermitRelease(release); + channel.closeFuture().addListener(f -> release.run()); + assertTrue(stateOf(channel).tryAcquireStream()); + assertEquals(0, availablePerHost(semaphore)); + + advance(channel, TTL_MILLIS); + + assertEquals(1, availablePerHost(semaphore), "a replacement must be admissible while the old streams finish"); + assertTrue(channel.isOpen()); + stateOf(channel).releaseStream(); + channel.close(); + assertEquals(1, availablePerHost(semaphore), "closing must not release the permit a second time"); + } + + private static int availablePerHost(PerHostConnectionSemaphore semaphore) { + Semaphore free = semaphore.freeChannelsPerHost.get(BASE); + return free != null ? free.availablePermits() : semaphore.maxConnectionsPerHost; + } + + @Test + void removingTheHandlerCancelsTheExpiry() { + // handlerRemoved is what the pipeline teardown of a real close runs. Closing an EmbeddedChannel would prove + // nothing here, since it cancels every task scheduled on its own loop by itself. + EmbeddedChannel channel = http2Connection(BASE); + channel.pipeline().remove(TTL_HANDLER); + + advance(channel, 2 * TTL_MILLIS); + + assertFalse(stateOf(channel).isRetired(), "the expiry of a removed handler must not run"); + assertSame(channel, channelManager.pollHttp2Connection(BASE)); + assertTrue(channel.isOpen()); + } + + /** + * Error codes of the GOAWAY frames the connection wrote, decoded by a server-side codec fed with everything the + * connection wrote. + */ + private static List goAwayErrorCodesSentBy(EmbeddedChannel client) { + List errorCodes = new ArrayList<>(); + EmbeddedChannel server = new EmbeddedChannel(Http2FrameCodecBuilder.forServer().build(), + new ChannelInboundHandlerAdapter() { + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof Http2GoAwayFrame) { + errorCodes.add(((Http2GoAwayFrame) msg).errorCode()); + } + ReferenceCountUtil.release(msg); + } + }); + try { + Object written; + while ((written = client.readOutbound()) != null) { + server.writeInbound(written); + } + } finally { + server.finishAndReleaseAll(); + } + return errorCodes; + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java index a77af3fc3..52e6af38c 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java @@ -1005,6 +1005,134 @@ public void releasePermitOnceIsAtomicUnderConcurrency() throws InterruptedExcept assertEquals(rounds, totalReleases.get(), "exactly one release per round"); } + // ------------------------------------------------------------------------- + // Retirement at connectionTtl + // ------------------------------------------------------------------------- + + @Test + public void retireTakesEffectOnce() { + Http2ConnectionState state = new Http2ConnectionState(); + assertFalse(state.isRetired()); + + assertTrue(state.retire(), "the first retire() is the one that retires the connection"); + assertFalse(state.retire(), "later calls must not run the retirement again"); + assertTrue(state.isRetired()); + } + + @Test + public void retiredConnectionStillOpensStreamsWhileSlotsAreFree() { + Http2ConnectionState retired = new Http2ConnectionState(); + retired.retire(); + assertTrue(retired.tryAcquireStream(), + "the server has refused nothing, so a request already holding a retired connection still gets a slot"); + assertEquals(1, retired.getActiveStreams()); + + // The contrast that keeps retirement from being implemented as draining. + Http2ConnectionState draining = new Http2ConnectionState(); + draining.setDraining(0); + assertFalse(draining.tryAcquireStream()); + } + + @Test + public void retiredConnectionAcceptsNoNewQueuedWork() { + Http2ConnectionState state = new Http2ConnectionState(); + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + state.retire(); + + AtomicInteger ran = new AtomicInteger(); + assertFalse(state.offerPendingOpener(ran::incrementAndGet), + "a full retired connection must turn work away, so the caller can send it elsewhere"); + state.releaseStream(); + assertEquals(0, ran.get(), "a rejected opener must never run"); + assertEquals(0, state.getActiveStreams()); + } + + @Test + public void openerQueuedBeforeRetirementStillRunsOnTheRetiredConnection() { + Http2ConnectionState state = new Http2ConnectionState(); + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + AtomicInteger ran = new AtomicInteger(); + assertTrue(state.offerPendingOpener(ran::incrementAndGet)); + assertEquals(0, ran.get(), "queued behind the stream holding the only slot"); + + state.retire(); + state.releaseStream(); + + assertEquals(1, ran.get(), "an opener queued before retirement must get the freed slot, not be dropped"); + assertEquals(1, state.getActiveStreams()); + assertFalse(state.closeIfRetiredAndIdle(), "the opener's stream keeps the connection open"); + } + + @Test + public void idleCloseNeedsRetirementNoStreamsAndAnEmptyQueue() { + Http2ConnectionState state = new Http2ConnectionState(); + assertFalse(state.closeIfRetiredAndIdle(), "a connection that is not retired is never closed by this"); + + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + assertTrue(state.offerPendingOpener(() -> { })); + state.retire(); + assertFalse(state.closeIfRetiredAndIdle(), "not idle while a stream is open"); + + // Releasing the stream hands its slot to the queued opener, whose stream then ends too. + state.releaseStream(); + assertEquals(1, state.getActiveStreams()); + assertFalse(state.closeIfRetiredAndIdle(), "not idle while the queued opener's stream is open"); + state.releaseStream(); + + assertTrue(state.closeIfRetiredAndIdle(), "retired, no stream and nothing queued: the caller must close it"); + assertEquals(0, state.getActiveStreams(), "the closed marker must not leak through the stream count"); + assertFalse(state.tryAcquireStream(), "once closed it must refuse every further stream"); + assertFalse(state.closeIfRetiredAndIdle(), "and only one caller may close it"); + } + + @Test + public void idleCloseIsRefusedWhileAnOpenerIsQueued() { + Http2ConnectionState state = new Http2ConnectionState(); + // A server that grants no slot at all: the opener stays queued with no stream open. + state.updateMaxConcurrentStreams(0); + assertTrue(state.offerPendingOpener(() -> { })); + state.retire(); + + assertEquals(0, state.getActiveStreams()); + assertFalse(state.closeIfRetiredAndIdle(), "a queued request must not lose its connection to the idle close"); + } + + @Test + public void idleCloseAndStreamAdmissionNeverBothWin() throws Exception { + // A request that looked up the connection before it was retired races the close of the now idle connection. + // Exactly one may win: a stream opened on a connection that is being closed would fail after it was sent. + int rounds = 20_000; + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + for (int i = 0; i < rounds; i++) { + Http2ConnectionState state = new Http2ConnectionState(); + state.retire(); + CyclicBarrier barrier = new CyclicBarrier(2); + Future admitted = executor.submit(() -> { + barrier.await(); + return state.tryAcquireStream(); + }); + Future closed = executor.submit(() -> { + barrier.await(); + return state.closeIfRetiredAndIdle(); + }); + boolean streamWon = admitted.get(10, TimeUnit.SECONDS); + boolean closeWon = closed.get(10, TimeUnit.SECONDS); + if (streamWon == closeWon) { + throw new AssertionError("round " + i + ": stream admitted=" + streamWon + ", idle close=" + closeWon + + "; exactly one of them must win"); + } + assertEquals(streamWon ? 1 : 0, state.getActiveStreams()); + } + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + private static Runnable blockingOpener(CountDownLatch started, CountDownLatch release) { return () -> { started.countDown(); diff --git a/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderHttp2RetiredTest.java b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderHttp2RetiredTest.java new file mode 100644 index 000000000..e80117c3c --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderHttp2RetiredTest.java @@ -0,0 +1,311 @@ +/* + * 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.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncCompletionHandlerBase; +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.Http2ConnectionState; +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.io.IOException; +import java.net.ConnectException; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.asynchttpclient.Dsl.config; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +/** + * A request that looked up an HTTP/2 connection just before the connection was retired at {@code connectionTtl}, + * and then found it closing or full, has written nothing, so it is sent again on another connection instead of + * failing. It is sent to a port nobody listens on: a {@link ConnectException} is the proof that the request was + * re-dispatched to a new connection, where failing in place would surface the retirement {@link IOException}. + * {@code maxRequestRetry} is 0 throughout, because the re-dispatch must not depend on the retry budget. + */ +@ExtendWith(NettyLeakDetectorExtension.class) +class NettyRequestSenderHttp2RetiredTest { + + private AsyncHttpClientConfig config; + private Timer timer; + private ChannelManager channelManager; + private NettyRequestSender sender; + private EmbeddedChannel parent; + private int closedPort; + + @BeforeEach + void setUp() throws IOException { + config = config().setMaxRequestRetry(0).build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, timer); + sender = new NettyRequestSender(config, channelManager, timer, mock(AsyncHttpClientState.class)); + channelManager.configureBootstraps(sender); + parent = new EmbeddedChannel(); + try (ServerSocket socket = new ServerSocket(0)) { + closedPort = socket.getLocalPort(); + } + } + + @AfterEach + void tearDown() { + parent.finishAndReleaseAll(); + channelManager.close(); + timer.stop(); + } + + private Http2ConnectionState http2State() { + Http2ConnectionState state = new Http2ConnectionState(); + parent.attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state); + return state; + } + + private NettyResponseFuture futureFor(Request request) { + return futureFor(request, new AsyncCompletionHandlerBase()); + } + + private NettyResponseFuture futureFor(Request request, AsyncCompletionHandlerBase handler) { + return new NettyResponseFuture<>(request, handler, + new NettyRequestFactory(config).newNettyRequest(request, false, null, null, null), 0, + ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + } + + private Request get() { + return new RequestBuilder("GET").setUrl("http://127.0.0.1:" + closedPort + "/").build(); + } + + private static Throwable failureOf(NettyResponseFuture future) { + ExecutionException failure = assertThrows(ExecutionException.class, () -> future.get(10, TimeUnit.SECONDS)); + return failure.getCause(); + } + + @Test + void requestLosingTheIdleCloseIsSentOnAnotherConnection() { + Http2ConnectionState state = http2State(); + state.retire(); + assertTrue(state.closeIfRetiredAndIdle(), "the connection was closed before the request got a stream"); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future), + "the request must have been sent again on a new connection"); + } + + @Test + void requestFindingTheRetiredConnectionFullIsSentOnAnotherConnection() { + Http2ConnectionState state = http2State(); + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + state.retire(); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future), + "a full retired connection must not queue the request; it goes to a new connection"); + assertEquals(1, state.getActiveStreams(), "nothing was queued on the retired connection"); + } + + @Test + void requestIsSentAgainOnlyOnce() { + Http2ConnectionState state = http2State(); + state.retire(); + assertTrue(state.closeIfRetiredAndIdle()); + NettyResponseFuture future = futureFor(get()); + assertTrue(future.claimHttp2Redispatch(), "stands in for an earlier re-dispatch of the same exchange"); + assertFalse(future.claimHttp2Redispatch()); + + sender.writeRequest(future, parent); + + Throwable cause = failureOf(future); + assertInstanceOf(IOException.class, cause); + assertFalse(cause instanceof ConnectException, "a second refusal must fail, not be sent yet again"); + } + + @Test + void requestFindingTheRetiredConnectionAlreadyClosedIsSentOnAnotherConnection() { + // The idle close ran between this request's lookup and its write: it never reaches the stream admission, and + // since nothing stores a request on an HTTP/2 connection, no channelInactive would ever complete it. + Http2ConnectionState state = http2State(); + state.retire(); + assertTrue(state.closeIfRetiredAndIdle()); + parent.close(); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future), + "the request must have been sent again on a new connection"); + } + + @Test + void requestRefusedByARetiredConnectionThatSinceDrainedIsSentOnAnotherConnection() { + // A server GOAWAY after retirement: the request was still never written, so it is re-sent rather than failed + // the way a refusal by a connection that is only draining is. + Http2ConnectionState state = http2State(); + state.retire(); + state.setDraining(0); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future)); + } + + @Test + void requestRefusedASlotThenFindingTheConnectionRetiredIsSentOnAnotherConnection() { + // Retired after this request was refused a slot but before it queued: only the check after the refused + // enqueue can see it. + RetiresWhenAskedIfDraining state = new RetiresWhenAskedIfDraining(); + parent.attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state); + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + state.armed.set(true); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future)); + } + + /** Retires itself when asked whether it is draining, which the sender asks only after refusing a slot. */ + private static final class RetiresWhenAskedIfDraining extends Http2ConnectionState { + + private final AtomicBoolean armed = new AtomicBoolean(); + + @Override + public boolean isDraining() { + boolean draining = super.isDraining(); + if (armed.get()) { + retire(); + } + return draining; + } + } + + @Test + void retiredConnectionWhoseLastStreamFailsToOpenIsClosed() { + // Retired but not yet closed, so the request gets a slot; the stream then fails to open, since this bare + // parent has no Http2MultiplexHandler. No stream close is left to close the connection, and its one-shot + // expiry has already run. + Http2ConnectionState state = http2State(); + state.retire(); + NettyResponseFuture future = futureFor(get()); + + sender.writeRequest(future, parent); + + assertTrue(future.isDone()); + assertEquals(0, state.getActiveStreams()); + assertFalse(parent.isOpen(), "the retired connection lost its last stream and must close"); + } + + @Test + void completedRequestIsNotSentAgain() { + // Completed by a timeout or cancel before the re-dispatch. Sent anyway, it would be attached to the healthy + // connection now registered for its origin, and attaching a done future closes the channel it is given. + Request request = get(); + EmbeddedChannel replacement = new EmbeddedChannel(); + try { + replacement.attr(Http2ConnectionState.HTTP2_STATE_KEY).set(new Http2ConnectionState()); + channelManager.registerHttp2Connection( + request.getChannelPoolPartitioning().getPartitionKey(request.getUri(), null, null), replacement); + Http2ConnectionState state = http2State(); + state.retire(); + assertTrue(state.closeIfRetiredAndIdle()); + NettyResponseFuture future = futureFor(request); + assertTrue(future.cancel(true)); + + sender.writeRequest(future, parent); + + assertTrue(replacement.isOpen(), "a done request must not reach the replacement connection"); + } finally { + replacement.finishAndReleaseAll(); + } + } + + @Test + void cancellingTheReDispatchedRequestLeavesTheRetiredConnectionOpen() { + // The retired connection is full and still carries a stream. The request it refused is cancelled while its + // new attempt is still looking for a connection; that must not close the retired one under its stream. + AtomicReference> cancelled = new AtomicReference<>(); + AsyncCompletionHandlerBase cancelOnNextAttempt = new AsyncCompletionHandlerBase() { + @Override + public void onConnectionPoolAttempt() { + cancelled.get().cancel(true); + } + }; + Http2ConnectionState state = http2State(); + state.setClientMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + state.retire(); + NettyResponseFuture future = futureFor(get(), cancelOnNextAttempt); + cancelled.set(future); + // As the pooled path leaves it: the request is attached to the connection it looked up. + future.attachChannel(parent, false); + + sender.writeRequest(future, parent); + + assertTrue(future.isCancelled()); + assertTrue(parent.isOpen(), "the stream still running on the retired connection must be left to finish"); + } + + @Test + void reDispatchedRequestBodyIsReleased() { + // Leak-checked by the extension: the body built for the refused attempt and the one rebuilt for the + // re-dispatch must both be released, and the caller's own buffer must be left to the caller. + ByteBuf body = ByteBufAllocator.DEFAULT.heapBuffer(); + body.writeCharSequence("payload", StandardCharsets.US_ASCII); + try { + Http2ConnectionState state = http2State(); + state.retire(); + assertTrue(state.closeIfRetiredAndIdle()); + Request request = new RequestBuilder("POST").setUrl("http://127.0.0.1:" + closedPort + "/") + .setBody(body).build(); + NettyResponseFuture future = futureFor(request); + + sender.writeRequest(future, parent); + + assertInstanceOf(ConnectException.class, failureOf(future)); + assertEquals(1, body.refCnt(), "the caller's buffer still belongs to the caller"); + } finally { + body.release(); + } + } +}