Fix request-body double-free and harden HTTP/2 lifecycle

Fix the cancel/timeout-vs-write request-body double-free with an atomic NettyRequest state, bound the pendingOpeners queue, keep a stream-open failure from closing the multiplexed parent, skip interceptors on interim 1xx, and build the WebSocket http/1.1 SSL context lazily. Document the setBody(ByteBuf) ownership contract and strengthen the request-body-leak regression tests.

Author: hyperxpro

client/src/main/java/org/asynchttpclient/RequestBuilderBase.java

@@ -478,6 +478,18 @@ public T setBody(ByteBuffer data) {
         return asDerivedType();
     }
 
+    /**
+     * Sets the request body from a Netty {@link ByteBuf}.
+     * <p>
+     * <strong>Ownership:</strong> the caller retains ownership of {@code data}. AsyncHttpClient sends a
+     * retained duplicate per attempt (so redirects, auth replays and retries each get their own reference and
+     * the body survives across them) and never releases {@code data} itself. The caller is responsible for
+     * releasing {@code data} once the request has completed. (This differs from older releases, which consumed
+     * and released the buffer on the first send — and double-freed it on any retry.)
+     *
+     * @param data the request body; the caller keeps ownership and must release it after the request completes
+     * @return this builder
+     */
     public T setBody(ByteBuf data) {
         resetBody();
         byteBufData = data;

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

@@ -276,7 +276,9 @@ private boolean terminateAndExit() {
      */
     private void releaseRequestIfNotHandedToChannel() {
         NettyRequest request = nettyRequest;
-        if (request != null && !request.isHandedToChannel()) {
+        if (request != null) {
+            // release() atomically no-ops if the request was already handed to the channel encoder (which then
+            // owns the release), so there is no check-then-act race with the concurrent event-loop write.
             request.release();
         }
     }

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

@@ -58,7 +58,15 @@ private static final class PendingOpener {
     private volatile int maxConcurrentStreams = Integer.MAX_VALUE;
     private final AtomicBoolean draining = new AtomicBoolean(false);
     private volatile int lastGoAwayStreamId = Integer.MAX_VALUE;
+    // Hard cap on requests queued waiting for a free stream slot. A peer that accepts the connection but never
+    // grants slots — SETTINGS_MAX_CONCURRENT_STREAMS=0, or a small limit with streams it never completes — would
+    // otherwise make every subsequent request queue forever, each pinning a NettyResponseFuture and its request
+    // body buffer, until the client OOMs. Past the cap, offerPendingOpener rejects and the caller fails the
+    // request fast. All queue mutations happen under pendingLock, so pendingCount tracks size in O(1)
+    // (ConcurrentLinkedQueue.size() is O(n)).
+    private static final int MAX_PENDING_OPENERS = 10_000;
     private final ConcurrentLinkedQueue<PendingOpener> pendingOpeners = new ConcurrentLinkedQueue<>();
+    private int pendingCount;
     private final Object pendingLock = new Object();
     private final AtomicBoolean closed = new AtomicBoolean(false);
     // Set when this connection lost the per-partition registration race (thundering herd): it is not in
@@ -106,17 +114,19 @@ public boolean offerPendingOpener(Runnable opener) {
     /**
      * Runs {@code opener} immediately if a stream slot is free, otherwise queues it for a later
      * {@link #releaseStream()}. Returns {@code false} — <em>without</em> queuing — when the connection is
-     * already draining or closed: such a connection never runs a queued opener ({@link #drainPendingOpeners}
-     * only re-offers it, and {@link #failPendingOpeners} has already drained the queue), so the caller MUST
-     * fail the request itself rather than let it sit until the request timeout fires (Issue #2160).
+     * already draining or closed, or when the pending queue is already at {@link #MAX_PENDING_OPENERS}: in
+     * each case the caller MUST fail the request itself rather than let it sit until the request timeout fires
+     * (Issue #2160). A draining/closed connection never runs a queued opener ({@link #drainPendingOpeners} only
+     * re-offers it, and {@link #failPendingOpeners} has already drained the queue); a full queue means the peer
+     * is starving slots and the request would otherwise grow heap without bound.
      * <p>
-     * Race-free against {@link #failPendingOpeners}: that method sets {@code closed} before draining under
-     * {@code pendingLock}. An opener enqueued before the drain acquires the lock is caught by the drain; an
-     * enqueue attempt sequenced after the drain observes {@code closed} here (the lock provides the
-     * happens-before) and is rejected. Either way no opener is left stranded.
+     * 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.
      *
      * @return {@code true} if the opener was run inline or queued; {@code false} if rejected because the
-     *         connection is draining/closed (caller must fail the request)
+     *         connection is draining/closed or the pending queue is full (caller must fail the request)
      */
     public boolean offerPendingOpener(NettyResponseFuture<?> future, Runnable opener) {
         synchronized (pendingLock) {
@@ -126,7 +136,11 @@ public boolean offerPendingOpener(NettyResponseFuture<?> future, Runnable opener
             if (tryAcquireStream()) {
                 opener.run();
             } else {
+                if (pendingCount >= MAX_PENDING_OPENERS) {
+                    return false;
+                }
                 pendingOpeners.add(new PendingOpener(future, opener));
+                pendingCount++;
             }
             return true;
         }
@@ -142,6 +156,7 @@ private void drainPendingOpeners() {
             // this never over-opens; every poll is under pendingLock, so a non-empty queue always yields a
             // non-null opener.
             while (!pendingOpeners.isEmpty() && tryAcquireStream()) {
+                pendingCount--;
                 pendingOpeners.poll().opener.run();
             }
         }
@@ -161,13 +176,18 @@ private void drainPendingOpeners() {
      * @param failer invoked once per orphaned request future (e.g. to abort it)
      */
     public void failPendingOpeners(Consumer<NettyResponseFuture<?>> failer) {
-        closed.set(true);
         List<PendingOpener> drained = new ArrayList<>();
         synchronized (pendingLock) {
+            // Set closed UNDER pendingLock, before draining: an offerPendingOpener that already holds the lock
+            // finishes its enqueue and is caught by the drain below; one that has not yet acquired the lock
+            // observes closed==true (lock happens-before) and is rejected. Setting it outside the lock would
+            // leave the invariant the offerPendingOpener javadoc relies on resting on luck, not the lock.
+            closed.set(true);
             PendingOpener p;
             while ((p = pendingOpeners.poll()) != null) {
                 drained.add(p);
             }
+            pendingCount = 0;
         }
         // Fail outside the lock — failer may re-enter client code.
         for (PendingOpener p : drained) {

client/src/main/java/org/asynchttpclient/netty/handler/Http2ContentDecompressor.java

@@ -134,7 +134,15 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 totalDecompressedBytes += decompressed.readableBytes();
                 if (totalDecompressedBytes > maxDecompressedBytes) {
                     decompressed.release();
-                    releaseDecompressor();
+                    // Swallow any teardown throw (finishAndReleaseAll() can re-run the decoder over a leftover
+                    // and throw a second DecompressionException) so the caller sees the bomb-limit message below,
+                    // not the raw codec error. The decoder's own cumulation is freed in channelInputClosed's
+                    // finally regardless, so swallowing here leaks nothing — same as the corrupt-body catch.
+                    try {
+                        releaseDecompressor();
+                    } catch (Throwable cleanupError) {
+                        // best-effort decoder teardown
+                    }
                     throw new DecompressionException(
                             "HTTP/2 response body exceeds the maximum decompressed size of "
                                     + maxDecompressedBytes + " bytes");

client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java

@@ -163,13 +163,20 @@ private void handleHttp2HeadersFrame(Http2HeadersFrame headersFrame, Channel cha
 
         NettyResponseStatus status = new NettyResponseStatus(future.getUri(), syntheticResponse, channel);
 
+        // RFC 9110 §15.2: a 1xx is an INTERIM response, not the final one. 100-continue must still run the
+        // interceptor chain (Continue100Interceptor resumes the deferred body), but any OTHER interim — 102
+        // Processing, 103 Early Hints — must NOT touch the chain at all: running it would persist the interim's
+        // Set-Cookie into the CookieStore and execute response filters against a non-final response, then do it
+        // again on the real response. The interim HEADERS has endStream=false, so just wait for the final frame.
+        if (statusCode > 100 && statusCode < 200) {
+            return;
+        }
+
         if (!interceptors.exitAfterIntercept(channel, future, handler, syntheticResponse, status, responseHeaders)) {
-            // RFC 9113 §8.1 / RFC 9110 §15.2: a 1xx is an INTERIM response, not the final one. Any 1xx not
-            // consumed by an interceptor (100-continue is handled in exitAfterIntercept above; 102 Processing
-            // and 103 Early Hints are not) must NOT be delivered to the AsyncHandler as the final status —
-            // that fires onStatusReceived/onHeadersReceived a second time when the real response arrives.
-            // The interim HEADERS has endStream=false, so just keep waiting for the final HEADERS frame.
-            if (statusCode >= 100 && statusCode < 200) {
+            // A 100 that the interceptor chain did not consume (no Expect/100-continue in flight) is still
+            // interim and must not be delivered to the AsyncHandler as the final status — that would fire
+            // onStatusReceived/onHeadersReceived a second time when the real response arrives.
+            if (statusCode == 100) {
                 return;
             }
             boolean abort = handler.onStatusReceived(status) == State.ABORT;

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

@@ -24,17 +24,21 @@
 public final class NettyRequest {
 
     @SuppressWarnings("rawtypes")
-    private static final AtomicIntegerFieldUpdater<NettyRequest> RELEASED_UPDATER =
-            AtomicIntegerFieldUpdater.newUpdater(NettyRequest.class, "released");
+    private static final AtomicIntegerFieldUpdater<NettyRequest> STATE_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(NettyRequest.class, "state");
+
+    // Single atomic state machine guarding who releases httpRequest (and the request-body ByteBuf it holds),
+    // so it is released EXACTLY once — no double-free, no leak — even when an abort/cancel/timeout on one
+    // thread races the channel write on the event loop. The two transitions out of OWNED_BY_AHC are mutually
+    // exclusive (CAS), so exactly one of {Netty's encoder, AHC} ends up owning the release.
+    private static final int OWNED_BY_AHC = 0;       // initial: AHC owns the release
+    private static final int HANDED_TO_CHANNEL = 1;  // handed to Netty's HTTP/1.1 encoder; IT releases
+    private static final int RELEASED = 2;           // AHC released it
 
     private final HttpRequest httpRequest;
     private final NettyBody body;
     @SuppressWarnings("unused")
-    private volatile int released;
-    // Set once httpRequest has been handed to a channel write (HTTP/1.1): from then on Netty's encoder owns
-    // and releases it, so AsyncHttpClient must NOT also release it. Until then — and always in the HTTP/2
-    // path, where httpRequest is never written to a channel but re-encoded as frames — AHC owns the release.
-    private volatile boolean handedToChannel;
+    private volatile int state;
 
     NettyRequest(HttpRequest httpRequest, NettyBody body) {
         this.httpRequest = httpRequest;
@@ -51,30 +55,35 @@ public NettyBody getBody() {
 
     /**
      * Releases the underlying HTTP/1.1 {@link HttpRequest} (and the request-body {@link io.netty.buffer.ByteBuf}
-     * it holds) exactly once. In the HTTP/2 path the {@code httpRequest} object itself is never written to a
-     * channel — its content is re-encoded as HTTP/2 frames — so AsyncHttpClient owns its release.
-     * {@code NettyRequestSender.sendHttp2Frames} releases it on the normal send path; this lets the
-     * early-abort paths (a draining/closed/queued-then-dropped connection, or a crashing {@code onRequestSend})
-     * release it too, so the request body buffer never leaks. Idempotent and thread-safe — extra calls are no-ops.
+     * it holds) — but only while AsyncHttpClient still owns it (it was never handed to a channel write). Once
+     * {@link #markHandedToChannel()} has claimed it, Netty's encoder owns the release and this becomes a no-op,
+     * so the two can never double-free. The atomic CAS also closes the check-then-release TOCTOU an external
+     * {@code isHandedToChannel()} test would leave between an abort thread and the event-loop write.
+     * <p>
+     * In the HTTP/2 path the {@code httpRequest} object is never written to a channel — its content is
+     * re-encoded as HTTP/2 frames — so AHC always owns its release and this performs it. The early-abort paths
+     * (a draining/closed/queued-then-dropped connection, or a crashing {@code onRequestSend}) call it too.
+     * Idempotent and thread-safe — extra calls are no-ops.
      */
     public void release() {
-        if (RELEASED_UPDATER.compareAndSet(this, 0, 1)) {
+        if (STATE_UPDATER.compareAndSet(this, OWNED_BY_AHC, RELEASED)) {
             ReferenceCountUtil.release(httpRequest);
         }
     }
 
     /**
-     * Marks that {@link #getHttpRequest()} has been handed to a channel write (the HTTP/1.1 path), after which
-     * Netty's encoder owns its release. This lets the abort/terminate paths release the request body only when
-     * it was never written — preventing a double-free with the encoder while still freeing it on an
-     * abort-before-write (e.g. a {@code setBody(ByteBuf)} request whose connection fails before the request is
-     * sent). Idempotent.
+     * Atomically claims {@link #getHttpRequest()} for a channel write (the HTTP/1.1 path), after which Netty's
+     * encoder owns its release. Returns {@code true} if the caller may proceed with the write; returns
+     * {@code false} when a concurrent abort/cancel/timeout has ALREADY released the request body — in which case
+     * the caller MUST NOT write the now-freed buffer. This single CAS resolves the double-free/use-after-free
+     * race a separate {@code handedToChannel} flag plus a non-atomic {@code release()} would otherwise allow
+     * (e.g. a {@code setBody(ByteBuf)} request cancelled on one thread while the event loop writes it).
      */
-    public void markHandedToChannel() {
-        handedToChannel = true;
+    public boolean markHandedToChannel() {
+        return STATE_UPDATER.compareAndSet(this, OWNED_BY_AHC, HANDED_TO_CHANNEL);
     }
 
     public boolean isHandedToChannel() {
-        return handedToChannel;
+        return state == HANDED_TO_CHANNEL;
     }
 }

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

@@ -508,10 +508,14 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
                     return;
                 }
 
-                // About to hand httpRequest to the channel — from here Netty's HTTP/1.1 encoder owns and
-                // releases it, so a later abort must not release it again (NettyRequest.release() is gated on
-                // isHandedToChannel()). Before this point an abort releases the request body itself.
-                nettyRequest.markHandedToChannel();
+                // Atomically claim httpRequest for the channel write — from here Netty's HTTP/1.1 encoder owns
+                // and releases it, so a later abort must not release it again (NettyRequest.release() then
+                // no-ops). If the claim FAILS, a concurrent abort/cancel/timeout has already released the
+                // request body on another thread; writing the freed buffer would be a use-after-free, so bail
+                // (the future is already terminal). Before this point an abort releases the request body itself.
+                if (!nettyRequest.markHandedToChannel()) {
+                    return;
+                }
 
                 // if the request has a body, we want to track progress
                 if (writeBody) {
@@ -676,7 +680,13 @@ protected void initChannel(Http2StreamChannel streamCh) {
                             state.releaseStream();
                         }
                         releaseHttp2Request(future);
-                        abort(parentChannel, future, f.cause());
+                        // 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
+                        // bookkeeping races AHC's own cap) while the parent connection is healthy with sibling
+                        // streams. abort(channel, ...) would closeChannel(parentChannel) and take them all down —
+                        // the multiplexed-connection blast radius. Keep it stream-scoped, like the draining/queued
+                        // paths above and Http2Handler.streamFailed(close=false).
+                        future.abort(f.cause());
                     }
                 });
     }

client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java

@@ -47,8 +47,10 @@
  *       the stream pipeline resumes it from {@code channelWritabilityChanged}.</li>
  *   <li>It reads one chunk ahead so the <em>final</em> DATA frame can carry {@code endStream=true};
  *       an empty body still sends a single empty DATA frame with {@code endStream=true}.</li>
- *   <li>In-flight heap is bounded to O(chunk size): at most one buffered "pending" chunk is held across
- *       a writability wait.</li>
+ *   <li>This writer holds at most one buffered "pending" chunk across a writability wait. Production stops
+ *       once the channel goes unwritable, so total in-flight heap is bounded by the channel's write
+ *       high-water mark (the already-written chunks the channel's outbound buffer still owns) plus that one
+ *       pending chunk — not by the size of the whole body.</li>
  * </ul>
  * <p>
  * <strong>Lifecycle / cleanup.</strong> Because the pump completes asynchronously (after {@code writeHttp2}