Bind the connection permit to the channel on connect

NettyConnectListener.onSuccess takes the per-host permit off the future
with takePartitionKeyLock(), a getAndSet(null), after which the future
can no longer release it: a later abort() finds the field null and
returns nothing. Ownership only passed to the channel when
attachSemaphoreToChannelClose ran, which on the TLS paths happened after
the handshake completed. Every exit taken before that point dropped the
token in a dead local and leaked the permit permanently.

That window is not just a race. A TLS handshake that fails - bad
certificate, peer reset, or Netty's own handshake timeout - always
leaves through it, as does an AsyncHandler callback throwing in the
same window. Since onFailure may then retry, and the retry finds a null
partitionKeyLock and acquires a fresh permit, a run of failures against
one origin drains maxConnectionsPerHost and pins it: every later
request to that host fails with TooManyConnectionsPerHostException with
no live connection to account for it. The combined limiter releases the
global slot through the same call, so one bad origin also starves
unrelated hosts. Only recreating the client recovers.

Bind the permit to channel.closeFuture() the moment it leaves the
future, and route every release through one getAndSet so it happens
exactly once - a double release would push the semaphore above the cap
and can prematurely prune a live per-host entry. Close-release is now
the default rather than a per-branch opt-in, so the failure paths are
covered by the close they already perform, and a new branch cannot
reintroduce the leak by forgetting to opt in.

Also publish the channel on the future before the handshake starts.
Until then future.channel() is null and NettyRequestSender.abort only
closes a non-null channel, so a request timeout firing mid-handshake
could not close the socket and left it, and its permit, alive until
handshakeTimeout. The field is now volatile: it is written on the event
loop but read by the timer thread in TimeoutTimerTask.expire and by
cancel() on the caller thread.

Fixes #2189

Author: hyperxpro

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

@@ -117,8 +117,11 @@ public final class NettyResponseFuture<V> implements ListenableFuture<V> {
     // volatile where we don't need CAS ops
     private volatile long touch = unpreciseMillisTime();
     private volatile ChannelState channelState = ChannelState.NEW;
+    // Written on the event loop but read off it: the HashedWheelTimer thread reads it in
+    // TimeoutTimerTask.expire to close the channel on a request timeout, and cancel() reads it on the
+    // caller thread. Without volatile those readers can miss the write and skip the close (issue #2189).
+    private volatile Channel channel;
     // state mutated only inside the event loop
-    private Channel channel;
     private boolean keepAlive = true;
     private Request targetRequest;
     private Request currentRequest;

client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java

@@ -34,6 +34,7 @@
 
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Non Blocking connect.
@@ -78,19 +79,26 @@ private void writeRequest(Channel channel) {
         Channels.setAttribute(channel, future);
 
         channelManager.registerOpenChannel(channel);
-        future.attachChannel(channel, false);
         requestSender.writeRequest(future, channel);
     }
 
     public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
-        // Take the semaphore lock from the future. For HTTP/1.1, we'll transfer it to channel.closeFuture().
-        // For HTTP/2, we release it immediately after ALPN negotiation since the connection is multiplexed.
-        final Object partitionKeyLock = (connectionSemaphore != null) ? future.takePartitionKeyLock() : null;
+        // takePartitionKeyLock() is a getAndSet(null): the future can no longer release this permit, so bind
+        // it to the channel here, before anything below can abandon that channel. Everything downstream --
+        // a failed or timed-out TLS handshake, a crashing AsyncHandler callback, a connect retry -- closes
+        // the channel but cannot reach the token, so deferring this to the individual branches leaked the
+        // permit permanently on every one of those paths (issue #2189). The HTTP/2 branches release it
+        // earlier and leave this listener a no-op.
+        final Object partitionKeyLock = connectionSemaphore != null ? future.takePartitionKeyLock() : null;
+        final AtomicReference<Object> permit = new AtomicReference<>(partitionKeyLock);
+        if (partitionKeyLock != null) {
+            channel.closeFuture().addListener(f -> releasePermitOnce(permit));
+        }
 
         Channels.setActiveToken(channel);
 
         if (futureIsAlreadyCompleted(channel)) {
-            releaseSemaphoreImmediately(partitionKeyLock);
+            releasePermitOnce(permit);
             return;
         }
 
@@ -102,10 +110,15 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
             // futureIsAlreadyCompleted check above can pass while the holder is already null.
             // Drop this connection rather than NPE-ing on setResolvedRemoteAddress below.
             Channels.silentlyCloseChannel(channel);
-            releaseSemaphoreImmediately(partitionKeyLock);
+            releasePermitOnce(permit);
             return;
         }
 
+        // Publish the channel before the (possibly long) TLS handshake. Until this runs future.channel() is
+        // null, and NettyRequestSender.abort only closes a non-null channel -- so a request timeout firing
+        // mid-handshake could not close the socket, stranding it until handshakeTimeout (issue #2189).
+        future.attachChannel(channel, false);
+
         Request request = future.getTargetRequest();
         Uri uri = request.getUri();
         // don't set a null resolved address - if the remoteAddress is null we keep
@@ -153,7 +166,6 @@ protected void onSuccess(Channel value) {
                         return;
                     }
                     // After SSL handshake to proxy, continue with normal proxy request
-                    attachSemaphoreToChannelClose(channel, partitionKeyLock);
                     writeRequest(channel);
                 }
 
@@ -215,9 +227,7 @@ protected void onSuccess(Channel value) {
                     }
                     if (http2Negotiated && !uri.isWebSocket()) {
                         channelManager.upgradePipelineToHttp2(channel.pipeline());
-                        registerHttp2AndManageSemaphore(channel, partitionKeyLock);
-                    } else {
-                        attachSemaphoreToChannelClose(channel, partitionKeyLock);
+                        registerHttp2AndManageSemaphore(channel, permit);
                     }
                     writeRequest(channel);
                 }
@@ -240,31 +250,21 @@ protected void onFailure(Throwable cause) {
             // excluded for the same RFC 8441 reason as the TLS path above — it stays on HTTP/1.1.
             if (!uri.isSecured() && channelManager.isHttp2CleartextEnabled() && !uri.isWebSocket()) {
                 channelManager.upgradePipelineToHttp2(channel.pipeline());
-                registerHttp2AndManageSemaphore(channel, partitionKeyLock);
-            } else {
-                attachSemaphoreToChannelClose(channel, partitionKeyLock);
+                registerHttp2AndManageSemaphore(channel, permit);
             }
             writeRequest(channel);
         }
     }
 
     /**
-     * Attaches the semaphore lock to the channel's close future (HTTP/1.1 behavior).
-     * The semaphore slot is released when the connection closes.
-     */
-    private void attachSemaphoreToChannelClose(Channel channel, Object partitionKeyLock) {
-        if (connectionSemaphore != null && partitionKeyLock != null) {
-            channel.closeFuture().addListener(f -> connectionSemaphore.releaseChannelLock(partitionKeyLock));
-        }
-    }
-
-    /**
-     * Releases the semaphore lock immediately (HTTP/2 behavior).
-     * HTTP/2 connections are multiplexed, so the semaphore should not be held
-     * for the lifetime of the connection.
+     * Returns the per-host permit this connection holds, at most once. The channel closeFuture, the
+     * early-exit paths and the HTTP/2 immediate release all race to call this; a double release would
+     * push the semaphore above {@code maxConnectionsPerHost}, and an unmatched one can prematurely
+     * prune a live entry from the per-host map.
      */
-    private void releaseSemaphoreImmediately(Object partitionKeyLock) {
-        if (connectionSemaphore != null && partitionKeyLock != null) {
+    private void releasePermitOnce(AtomicReference<Object> permit) {
+        Object partitionKeyLock = permit.getAndSet(null);
+        if (partitionKeyLock != null && connectionSemaphore != null) {
             connectionSemaphore.releaseChannelLock(partitionKeyLock);
         }
     }
@@ -281,27 +281,22 @@ private void releaseSemaphoreImmediately(Object partitionKeyLock) {
      * a further request fails to acquire a permit and multiplexes onto a sibling-IP connection via
      * {@link ChannelManager#pollHttp2SiblingConnection(Object)} rather than opening another (issue #2214).
      */
-    private void registerHttp2AndManageSemaphore(Channel channel, Object partitionKeyLock) {
+    private void registerHttp2AndManageSemaphore(Channel channel, AtomicReference<Object> permit) {
         // Register under the future's partition key so the H2 connection is found by the same key the
         // pool is polled with, including the IP-aware key used by LoadBalance.ROUND_ROBIN. Read the key
         // once: it is volatile (repinned on IP failover) and both uses below must agree.
         Object partitionKey = future.getPartitionKey();
         channelManager.registerHttp2Connection(partitionKey, channel);
         if (partitionKey instanceof RoundRobinPartitionKey) {
             // The permit must be freed as soon as the connection stops serving new requests: on close, or
-            // at drain start after GOAWAY (issue #2214). Both paths go through releasePermitOnce(), which
-            // guarantees a single release; a double release would push the semaphore above the cap. The
-            // state attribute is always present after upgradePipelineToHttp2, the fallback is just
-            // belt and braces.
+            // at drain start after GOAWAY (issue #2214). The closeFuture listener installed in onSuccess
+            // covers close; this adds the drain hook. Both funnel through releasePermitOnce.
             Http2ConnectionState state = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get();
-            if (state != null && connectionSemaphore != null && partitionKeyLock != null) {
-                state.setPermitRelease(() -> connectionSemaphore.releaseChannelLock(partitionKeyLock));
-                channel.closeFuture().addListener(f -> state.releasePermitOnce());
-            } else {
-                attachSemaphoreToChannelClose(channel, partitionKeyLock);
+            if (state != null) {
+                state.setPermitRelease(() -> releasePermitOnce(permit));
             }
         } else {
-            releaseSemaphoreImmediately(partitionKeyLock);
+            releasePermitOnce(permit);
         }
     }
 

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

@@ -27,6 +27,7 @@
 
 import java.net.InetAddress;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.asynchttpclient.Dsl.config;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -81,9 +82,18 @@ private EmbeddedChannel registerRrConnectionHoldingPermit(ConnectionSemaphore se
         Http2ConnectionState state = new Http2ConnectionState();
         channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state);
         channelManager.registerHttp2Connection(registryKey, channel);
-        // Mirror NettyConnectListener.registerHttp2AndManageSemaphore's round-robin branch.
-        state.setPermitRelease(() -> semaphore.releaseChannelLock(baseKey));
-        channel.closeFuture().addListener(f -> state.releasePermitOnce());
+        // Mirror NettyConnectListener's round-robin wiring: the drain hook installed by
+        // registerHttp2AndManageSemaphore, plus the closeFuture release installed in onSuccess. Both funnel
+        // through a single getAndSet so the permit is returned exactly once.
+        AtomicReference<Object> permit = new AtomicReference<>(baseKey);
+        Runnable release = () -> {
+            Object key = permit.getAndSet(null);
+            if (key != null) {
+                semaphore.releaseChannelLock(key);
+            }
+        };
+        state.setPermitRelease(release);
+        channel.closeFuture().addListener(f -> release.run());
         return channel;
     }