Retire HTTP/2 connections at connectionTTL

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.
+     * <p>
+     * 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();
 

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.</li>
+     *       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.</li>
      *   <li>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.</li>
+     *       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.</li>
      *   <li>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

client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java

@@ -86,6 +86,9 @@ public final class NettyResponseFuture<V> implements ListenableFuture<V> {
     private static final AtomicIntegerFieldUpdater<NettyResponseFuture> ON_THROWABLE_CALLED_FIELD = AtomicIntegerFieldUpdater
             .newUpdater(NettyResponseFuture.class, "onThrowableCalled");
     @SuppressWarnings("rawtypes")
+    private static final AtomicIntegerFieldUpdater<NettyResponseFuture> HTTP2_REDISPATCHED_FIELD = AtomicIntegerFieldUpdater
+            .newUpdater(NettyResponseFuture.class, "http2Redispatched");
+    @SuppressWarnings("rawtypes")
     private static final AtomicReferenceFieldUpdater<NettyResponseFuture, TimeoutsHolder> TIMEOUTS_HOLDER_FIELD = AtomicReferenceFieldUpdater
             .newUpdater(NettyResponseFuture.class, TimeoutsHolder.class, "timeoutsHolder");
     @SuppressWarnings("rawtypes")
@@ -118,6 +121,8 @@ public final class NettyResponseFuture<V> implements ListenableFuture<V> {
     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

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,13 +102,15 @@
 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;
 import java.util.Map.Entry;
 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<ChannelOption<?>, 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<InetSocketAddress> 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;

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.
+     * <p>
+     * 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.
+     * <p>
+     * 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);