Run HTTP/2 stream openers outside lock (#2274)

## Summary
- reserve HTTP/2 stream slots and dequeue pending openers while holding
`pendingLock`
- invoke immediate and queued stream-opening callbacks after releasing
the lock
- keep the empty pending-queue release path allocation-free
- add deterministic concurrency tests for both opener paths

## Motivation
`Http2ConnectionState` invoked stream openers while holding
`pendingLock`. Opening a stream can reach `AsyncHandler.onRequestSend`,
so one slow callback serialized concurrent submissions to the same
HTTP/2 connection.

The change preserves the Issue #2160 close/enqueue happens-before
relationship and atomic stream-slot accounting. Only callback execution
moves outside the monitor.

## Validation
- `./mvnw -pl client
-Dtest=org.asynchttpclient.netty.channel.Http2ConnectionStateTest test`
  - 49 tests passed
- `./mvnw -pl client
-Dtest='org.asynchttpclient.netty.channel.Http2ConnectionStateTest,org.asynchttpclient.Http2StreamOrphanRegressionTest,org.asynchttpclient.Http2MultiplexBugRegressionTest,org.asynchttpclient.Http2ResidualFixesRegressionTest'
test`
  - 68 tests passed

The local environment only provides JDK 21. The JDK 11 `./mvnw clean
verify` gate was not run locally; the repository CI matrix provides JDK
11 coverage.

## Attribution
Codex on behalf of Pavel Ptashyts

---------

Co-authored-by: Codex <codex@openai.com>

Author: pavel-ptashyts

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

@@ -130,42 +130,51 @@ public boolean offerPendingOpener(Runnable opener) {
      * Race-free against {@link #failPendingOpeners}: that method sets {@code closed} and drains the queue under
      * {@code pendingLock}. An opener enqueued before the drain runs is caught by the drain; an enqueue attempt
      * sequenced after it observes {@code closed} here (the lock provides the happens-before) and is rejected.
-     * Either way no opener is left stranded.
+     * 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.
      *
      * @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)
      */
     public boolean offerPendingOpener(NettyResponseFuture<?> future, Runnable opener) {
+        boolean runOpener = false;
         synchronized (pendingLock) {
             if (draining.get() || closed.get()) {
                 return false;
             }
             if (tryAcquireStream()) {
-                opener.run();
+                runOpener = true;
             } else {
                 if (pendingCount >= MAX_PENDING_OPENERS) {
                     return false;
                 }
                 pendingOpeners.add(new PendingOpener(future, opener));
                 pendingCount++;
             }
-            return true;
         }
+        if (runOpener) {
+            opener.run();
+        }
+        return true;
     }
 
     private void drainPendingOpeners() {
-        synchronized (pendingLock) {
-            // Open as many queued requests as there are now-free stream slots. A single stream completion
-            // frees exactly one slot (so this usually runs one opener), but a SETTINGS frame that RAISES
-            // SETTINGS_MAX_CONCURRENT_STREAMS frees several at once — drain them all here rather than waking
-            // only one and stalling the rest until the next completion (a missed-wakeup; the Issue #2160
-            // silent-timeout class). tryAcquireStream() enforces the cap and the draining/closed gate, so
-            // this never over-opens; every poll is under pendingLock, so a non-empty queue always yields a
-            // non-null opener.
-            while (!pendingOpeners.isEmpty() && tryAcquireStream()) {
+        // Drain every slot exposed by a SETTINGS increase; stopping after one can strand requests until another
+        // stream completes (Issue #2160). Reserve and dequeue one opener per iteration under pendingLock:
+        // tryAcquireStream() enforces capacity and draining/closed gates, the lock makes poll() non-null after
+        // the emptiness check, and a throwing opener cannot strand a pre-reserved batch.
+        while (true) {
+            PendingOpener pending;
+            synchronized (pendingLock) {
+                if (pendingOpeners.isEmpty() || !tryAcquireStream()) {
+                    return;
+                }
                 pendingCount--;
-                pendingOpeners.poll().opener.run();
+                pending = pendingOpeners.poll();
             }
+            // Opening a stream can invoke user callbacks. Run without pendingLock so a slow callback does not
+            // serialize unrelated submissions to this connection.
+            pending.opener.run();
         }
     }
 

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

@@ -24,6 +24,7 @@
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -270,6 +271,103 @@ public void pendingOpenerRunsOnRelease() {
         assertEquals(1, executionCount.get(), "Pending opener should have been executed on release");
     }
 
+    @Test
+    public void immediateOpenerRunsOutsidePendingLock() throws Exception {
+        Http2ConnectionState state = new Http2ConnectionState();
+        state.updateMaxConcurrentStreams(2);
+        CountDownLatch openerStarted = new CountDownLatch(1);
+        CountDownLatch releaseOpener = new CountDownLatch(1);
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+
+        try {
+            Future<Boolean> blockingOffer = executor.submit(() ->
+                    state.offerPendingOpener(blockingOpener(openerStarted, releaseOpener)));
+            assertTrue(openerStarted.await(5, TimeUnit.SECONDS), "first opener should start");
+
+            Future<Boolean> competingOffer = executor.submit(() -> state.offerPendingOpener(() -> { }));
+            assertTrue(competingOffer.get(5, TimeUnit.SECONDS),
+                    "a running opener must not hold pendingLock");
+
+            releaseOpener.countDown();
+            assertTrue(blockingOffer.get(5, TimeUnit.SECONDS));
+            state.releaseStream();
+            state.releaseStream();
+        } finally {
+            releaseOpener.countDown();
+            executor.shutdownNow();
+            assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test
+    public void queuedOpenerRunsOutsidePendingLock() throws Exception {
+        Http2ConnectionState state = new Http2ConnectionState();
+        state.updateMaxConcurrentStreams(1);
+        assertTrue(state.tryAcquireStream());
+        CountDownLatch openerStarted = new CountDownLatch(1);
+        CountDownLatch releaseOpener = new CountDownLatch(1);
+        state.addPendingOpener(blockingOpener(openerStarted, releaseOpener));
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+
+        try {
+            Future<?> drain = executor.submit(state::releaseStream);
+            assertTrue(openerStarted.await(5, TimeUnit.SECONDS), "queued opener should start");
+
+            Future<Boolean> competingOffer = executor.submit(() -> state.offerPendingOpener(() -> { }));
+            assertTrue(competingOffer.get(5, TimeUnit.SECONDS),
+                    "a drained opener must not hold pendingLock");
+
+            releaseOpener.countDown();
+            drain.get(5, TimeUnit.SECONDS);
+            state.releaseStream();
+            state.releaseStream();
+        } finally {
+            releaseOpener.countDown();
+            executor.shutdownNow();
+            assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test
+    public void raisedLimitDrainsMultiplePendingOpenersInOrder() {
+        Http2ConnectionState state = new Http2ConnectionState();
+        state.updateMaxConcurrentStreams(1);
+        assertTrue(state.tryAcquireStream());
+        List<Integer> executionOrder = new ArrayList<>();
+        state.addPendingOpener(() -> executionOrder.add(1));
+        state.addPendingOpener(() -> executionOrder.add(2));
+        state.addPendingOpener(() -> executionOrder.add(3));
+
+        state.updateMaxConcurrentStreams(4);
+
+        assertEquals(List.of(1, 2, 3), executionOrder);
+        assertEquals(4, state.getActiveStreams());
+    }
+
+    @Test
+    public void throwingBatchOpenerLeavesRemainingQueueDrainable() {
+        Http2ConnectionState state = new Http2ConnectionState();
+        state.updateMaxConcurrentStreams(1);
+        assertTrue(state.tryAcquireStream());
+        List<Integer> executionOrder = new ArrayList<>();
+        state.addPendingOpener(() -> {
+            executionOrder.add(1);
+            throw new IllegalStateException("boom");
+        });
+        state.addPendingOpener(() -> executionOrder.add(2));
+        state.addPendingOpener(() -> executionOrder.add(3));
+
+        assertThrows(IllegalStateException.class, () -> state.updateMaxConcurrentStreams(4));
+        assertEquals(List.of(1), executionOrder);
+        // A throwing public Runnable leaks its reserved slot, matching the behavior before this change.
+        assertEquals(2, state.getActiveStreams());
+
+        state.releaseStream();
+
+        assertEquals(List.of(1, 2, 3), executionOrder);
+        assertEquals(3, state.getActiveStreams());
+    }
+
     @Test
     public void multiplePendingOpenersExecuteInOrder() {
         Http2ConnectionState state = new Http2ConnectionState();
@@ -897,4 +995,18 @@ public void releasePermitOnceIsAtomicUnderConcurrency() throws InterruptedExcept
         }
         assertEquals(rounds, totalReleases.get(), "exactly one release per round");
     }
+
+    private static Runnable blockingOpener(CountDownLatch started, CountDownLatch release) {
+        return () -> {
+            started.countDown();
+            try {
+                if (!release.await(10, TimeUnit.SECONDS)) {
+                    throw new AssertionError("timed out waiting to release opener");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("interrupted while waiting to release opener", e);
+            }
+        };
+    }
 }