Harden the HTTP/2 connection-waiter registry against waiter failures and shutdown races

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

@@ -155,6 +155,11 @@ public class ChannelManager {
     // rather than hang (its request-timeout is not scheduled yet at this point). See NettyRequestSender's
     // HTTP/2 deferral.
     private final ConcurrentHashMap<Object, Set<Consumer<Channel>>> http2ConnectionWaiters = new ConcurrentHashMap<>();
+    // Set once, permanently, when the client closes and sweeps its waiters (failHttp2ConnectionWaiters). Read
+    // by addHttp2ConnectionWaiter to fail-closed against a request that arms a waiter in the window between the
+    // sweep and nettyTimer.stop() — such a waiter would otherwise be neither woken nor timed out, hanging its
+    // request forever. See addHttp2ConnectionWaiter and NettyRequestSender's Http2ConnectionWaiter.arm().
+    private volatile boolean waitersClosed;
 
     private AsyncHttpClientHandler wsHandler;
     private Http2Handler http2Handler;
@@ -504,38 +509,73 @@ public void registerHttp2Connection(Object partitionKey, Channel channel) {
      * (issue #2214) — the per-IP registration key on its own would never wake it. The waiter must be
      * idempotent — it may be invoked by a registration, by the client-close sweep, or removed and invoked by
      * its own timeout concurrently.
+     *
+     * @return {@code true} if the waiter was registered; {@code false} if the client is already closing, in
+     *         which case it is NOT registered and the caller must fail its request immediately rather than
+     *         arm a timeout (no connection will register and the timer that would fire the deadline is being
+     *         stopped). This closes the window where a request arms a waiter between the
+     *         {@link #failHttp2ConnectionWaiters()} sweep and {@code nettyTimer.stop()} and then hangs forever.
      */
-    public void addHttp2ConnectionWaiter(Object partitionKey, Consumer<Channel> onConnection) {
+    public boolean addHttp2ConnectionWaiter(Object partitionKey, Consumer<Channel> onConnection) {
         http2ConnectionWaiters.computeIfAbsent(baseKeyOf(partitionKey), k -> ConcurrentHashMap.newKeySet()).add(onConnection);
+        // Recheck after adding: failHttp2ConnectionWaiters sets waitersClosed BEFORE sweeping, so a waiter that
+        // is not caught by the sweep (added into a bucket the sweep already passed, or a bucket it recreated)
+        // observes the flag here and unregisters itself. Every waiter is therefore either swept or fails-closed.
+        if (waitersClosed) {
+            removeHttp2ConnectionWaiter(partitionKey, onConnection);
+            return false;
+        }
+        return true;
     }
 
     public void removeHttp2ConnectionWaiter(Object partitionKey, Consumer<Channel> onConnection) {
-        Set<Consumer<Channel>> waiters = http2ConnectionWaiters.get(baseKeyOf(partitionKey));
-        if (waiters != null) {
+        // Prune an emptied set under the bucket lock so an origin whose deferred requests only ever time out
+        // (a saturated or unreachable host that never registers an H2 connection) does not retain one empty
+        // Set per host for the client's lifetime. computeIfPresent serializes with a concurrent
+        // addHttp2ConnectionWaiter on the same key; a racing add that recreates the set is re-covered by the
+        // waiter's own re-poll in arm(), so no wakeup is lost.
+        http2ConnectionWaiters.computeIfPresent(baseKeyOf(partitionKey), (k, waiters) -> {
             waiters.remove(onConnection);
-        }
+            return waiters.isEmpty() ? null : waiters;
+        });
     }
 
     private void wakeHttp2ConnectionWaiters(Object partitionKey, Channel channel) {
         Set<Consumer<Channel>> waiters = http2ConnectionWaiters.remove(baseKeyOf(partitionKey));
         if (waiters != null) {
             for (Consumer<Channel> waiter : waiters) {
-                waiter.accept(channel);
+                notifyHttp2ConnectionWaiter(waiter, channel);
             }
         }
     }
 
     private void failHttp2ConnectionWaiters() {
+        // Mark closed before the sweep so a waiter arming concurrently is either swept below or fails-closed in
+        // addHttp2ConnectionWaiter — it can never remain armed-but-never-notified.
+        waitersClosed = true;
         for (Object key : http2ConnectionWaiters.keySet()) {
             Set<Consumer<Channel>> waiters = http2ConnectionWaiters.remove(key);
             if (waiters != null) {
                 for (Consumer<Channel> waiter : waiters) {
-                    waiter.accept(null);
+                    notifyHttp2ConnectionWaiter(waiter, null);
                 }
             }
         }
     }
 
+    // Invoking a waiter drives a full request send (writeRequest). Isolate failures: one waiter throwing must
+    // not starve the remaining waiters, and — since wakeHttp2ConnectionWaiters runs inside
+    // registerHttp2Connection, on the establishing connection's onSuccess path — must not propagate out and
+    // abort that request's own semaphore release and writeRequest (which would silently leak a connection
+    // permit and hang the establishing request).
+    private static void notifyHttp2ConnectionWaiter(Consumer<Channel> waiter, Channel channel) {
+        try {
+            waiter.accept(channel);
+        } catch (Throwable t) {
+            LOGGER.warn("HTTP/2 connection waiter failed", t);
+        }
+    }
+
     /**
      * Fails a request that was queued in {@link Http2ConnectionState} waiting for a stream slot but can
      * never get one (the connection dropped or started draining). Releases its request body first —

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -1204,7 +1204,14 @@ private final class Http2ConnectionWaiter<T> implements Consumer<Channel> {
         }
 
         void arm() {
-            channelManager.addHttp2ConnectionWaiter(waitKey, this);
+            if (!channelManager.addHttp2ConnectionWaiter(waitKey, this)) {
+                // The client is closing (the waiter sweep already ran): no connection will register and the
+                // nettyTimer that would fire our deadline is being stopped, so arming would hang the request.
+                // Fail now with the original permit exception instead — and before touching nettyTimer, whose
+                // newTimeout throws once stopped.
+                accept(null);
+                return;
+            }
             // Assign the deadline before the recheck so it exists (and is cancellable) if a wake races in.
             deadline = nettyTimer.newTimeout(t -> fireTimeout(),
                     config.getConnectTimeout().toMillis(), TimeUnit.MILLISECONDS);

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

@@ -23,6 +23,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
+import java.lang.reflect.Field;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -133,4 +135,94 @@ public void waiterFailedWithNullOnClose() {
             timer.stop();
         }
     }
+
+    @Test
+    @Timeout(unit = TimeUnit.SECONDS, value = 30)
+    public void throwingWaiterDoesNotStarveOthersOnRegistration() {
+        Timer timer = new HashedWheelTimer();
+        ChannelManager cm = newChannelManager(timer);
+        Channel channel = new EmbeddedChannel();
+        try {
+            AtomicBoolean survivorWoken = new AtomicBoolean();
+            cm.addHttp2ConnectionWaiter(KEY, c -> {
+                throw new RuntimeException("waiter blew up");
+            });
+            cm.addHttp2ConnectionWaiter(KEY, c -> survivorWoken.set(true));
+
+            // Must not propagate the throwing waiter's exception: registerHttp2Connection runs on the
+            // establishing connection's onSuccess path, which still has to release its permit and write.
+            cm.registerHttp2Connection(KEY, channel);
+
+            assertTrue(survivorWoken.get(), "a throwing waiter must not starve the other waiters for the key");
+        } finally {
+            channel.close();
+            cm.close();
+            timer.stop();
+        }
+    }
+
+    @Test
+    @Timeout(unit = TimeUnit.SECONDS, value = 30)
+    public void throwingWaiterDoesNotStarveOthersOnClose() {
+        Timer timer = new HashedWheelTimer();
+        ChannelManager cm = newChannelManager(timer);
+        try {
+            AtomicBoolean survivorFailed = new AtomicBoolean();
+            cm.addHttp2ConnectionWaiter(KEY, c -> {
+                throw new RuntimeException("waiter blew up");
+            });
+            cm.addHttp2ConnectionWaiter(KEY, c -> survivorFailed.set(true));
+
+            cm.close(); // must isolate the throwing waiter and still fail the rest
+
+            assertTrue(survivorFailed.get(), "a throwing waiter must not starve the other waiters on close");
+        } finally {
+            timer.stop();
+        }
+    }
+
+    @Test
+    @Timeout(unit = TimeUnit.SECONDS, value = 30)
+    public void waiterAddedAfterCloseFailsClosed() {
+        Timer timer = new HashedWheelTimer();
+        ChannelManager cm = newChannelManager(timer);
+        try {
+            cm.close(); // sweeps pending waiters and latches the closed state
+
+            AtomicBoolean woken = new AtomicBoolean();
+            boolean registered = cm.addHttp2ConnectionWaiter(KEY, c -> woken.set(true));
+
+            assertFalse(registered, "adding a waiter after close must fail-closed so the caller fails its request "
+                    + "instead of arming a timeout that never fires");
+            assertFalse(woken.get(), "a fail-closed add must not itself invoke the waiter");
+        } finally {
+            timer.stop();
+        }
+    }
+
+    @Test
+    @Timeout(unit = TimeUnit.SECONDS, value = 30)
+    public void removingLastWaiterPrunesEmptyEntry() throws Exception {
+        Timer timer = new HashedWheelTimer();
+        ChannelManager cm = newChannelManager(timer);
+        try {
+            Consumer<Channel> waiter = c -> {
+            };
+            cm.addHttp2ConnectionWaiter(KEY, waiter);
+            cm.removeHttp2ConnectionWaiter(KEY, waiter);
+
+            assertTrue(waiterMap(cm).isEmpty(),
+                    "removing the last waiter for a key must prune the empty set, not retain it for the client's lifetime");
+        } finally {
+            cm.close();
+            timer.stop();
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Map<Object, ?> waiterMap(ChannelManager cm) throws Exception {
+        Field field = ChannelManager.class.getDeclaredField("http2ConnectionWaiters");
+        field.setAccessible(true);
+        return (Map<Object, ?>) field.get(cm);
+    }
 }