Harden queued body read cancellation

Skip worker reads when cancellation closes a body stream before its
queued task starts. This prevents closed streams from being touched
and keeps cancelled uploads from consuming executor capacity.

Add deterministic HTTP/1.1 and HTTP/2 queue tests, plus an HTTP/2
sibling stream test that proves a blocked upload does not stall
another stream. Document executor-taking constructors as internal
client wiring.

Codex on behalf of Pavel Ptashyts

Co-Authored-By: Codex <codex@openai.com>

Author: pavel-ptashyts

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

@@ -134,6 +134,11 @@ ChannelManager channelManager() {
         return channelManager;
     }
 
+    // Visible for testing
+    @Nullable ExecutorService blockingBodyReadExecutor() {
+        return blockingBodyReadExecutor;
+    }
+
     private static Timer newNettyTimer(AsyncHttpClientConfig config) {
         ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName() + "-timer");
         HashedWheelTimer timer = new HashedWheelTimer(threadFactory, config.getHashedWheelTimerTickDuration(), TimeUnit.MILLISECONDS, config.getHashedWheelTimerSize());

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

@@ -130,6 +130,18 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa
         this(config, channelManager, nettyTimer, clientState, Runnable::run);
     }
 
+    /**
+     * Creates the request sender with the client-owned executor used for blocking request body reads.
+     * This overload is public only for cross-package client wiring; applications should configure the
+     * executor through {@link AsyncHttpClientConfig}.
+     *
+     * @param config client configuration
+     * @param channelManager channel manager
+     * @param nettyTimer request timer
+     * @param clientState client lifecycle state
+     * @param blockingBodyReadExecutor executor for blocking request body reads
+     * @since 3.0.12
+     */
     public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelManager, Timer nettyTimer,
                               AsyncHttpClientState clientState, Executor blockingBodyReadExecutor) {
         this.config = config;

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

@@ -55,10 +55,29 @@ public NettyInputStreamBody(InputStream inputStream, long contentLength) {
         this(inputStream, contentLength, Runnable::run, false);
     }
 
+    /**
+     * Creates a request body whose stream reads are dispatched through the supplied executor.
+     * This overload supports internal client wiring; applications should enable offloading through
+     * {@link org.asynchttpclient.AsyncHttpClientConfig}.
+     *
+     * @param inputStream request body stream
+     * @param blockingBodyReadExecutor executor for blocking reads
+     * @since 3.0.12
+     */
     public NettyInputStreamBody(InputStream inputStream, Executor blockingBodyReadExecutor) {
         this(inputStream, -1L, blockingBodyReadExecutor);
     }
 
+    /**
+     * Creates a request body whose stream reads are dispatched through the supplied executor.
+     * This overload supports internal client wiring; applications should enable offloading through
+     * {@link org.asynchttpclient.AsyncHttpClientConfig}.
+     *
+     * @param inputStream request body stream
+     * @param contentLength request body length, or {@code -1} when unknown
+     * @param blockingBodyReadExecutor executor for blocking reads
+     * @since 3.0.12
+     */
     public NettyInputStreamBody(InputStream inputStream, long contentLength, Executor blockingBodyReadExecutor) {
         this(inputStream, contentLength, blockingBodyReadExecutor, true);
     }
@@ -263,6 +282,9 @@ private void submitReadAfterSuspend() throws IOException {
         }
 
         private void readOffEventLoop() {
+            if (closed) {
+                return;
+            }
             int read = -1;
             IOException thrown = null;
             try {
@@ -400,6 +422,9 @@ private void submitReadAfterSuspend() throws IOException {
         }
 
         private void readOffEventLoop() {
+            if (closed) {
+                return;
+            }
             int read = -1;
             IOException thrown = null;
             try {

client/src/test/java/org/asynchttpclient/BasicHttpTest.java

@@ -55,6 +55,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -66,6 +67,9 @@
 import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.BlockingInputStream;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.CloseProbeInputStream;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.awaitExecutorState;
 import static org.asynchttpclient.Dsl.config;
 import static org.asynchttpclient.Dsl.get;
 import static org.asynchttpclient.Dsl.head;
@@ -1130,6 +1134,48 @@ public int read(byte[] buffer, int offset, int length) {
         });
     }
 
+    @RepeatedIfExceptionsTest(repeats = 5)
+    public void cancelledQueuedInputStreamIsNotReadOverHttp1() throws Throwable {
+        withServer(server).run(server -> {
+            server.enqueueEcho();
+            server.enqueueEcho();
+            byte[] bodyBytes = "{}".getBytes(StandardCharsets.ISO_8859_1);
+            BlockingInputStream blockingStream = new BlockingInputStream(bodyBytes);
+            CloseProbeInputStream queuedStream = new CloseProbeInputStream();
+
+            try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(config()
+                    .setRequestBodyStreamReadOffloadEnabled(true)
+                    .setRequestBodyStreamReadThreadsCount(1)
+                    .setRequestBodyStreamReadQueueSize(1)
+                    .build())) {
+                ThreadPoolExecutor executor = assertInstanceOf(ThreadPoolExecutor.class,
+                        client.blockingBodyReadExecutor());
+                ListenableFuture<Response> blockingUpload = client.preparePost(getTargetUrl())
+                        .setBody(new InputStreamBodyGenerator(blockingStream, bodyBytes.length))
+                        .execute();
+                assertTrue(blockingStream.readStarted.await(5, SECONDS),
+                        "first body read should occupy the worker");
+
+                ListenableFuture<Response> queuedUpload = client.preparePost(getTargetUrl())
+                        .setBody(new InputStreamBodyGenerator(queuedStream, 1))
+                        .execute();
+                awaitExecutorState(executor, 1, 1);
+
+                assertTrue(queuedUpload.cancel(true), "queued request should be cancelled");
+                assertTrue(queuedStream.closed.await(5, SECONDS),
+                        "cancellation should close the queued stream");
+
+                blockingStream.releaseRead.countDown();
+                assertEquals(200, blockingUpload.get(TIMEOUT, SECONDS).getStatusCode());
+                awaitExecutorState(executor, 0, 0);
+                assertFalse(queuedStream.readAttempted.get(),
+                        "a queued read must not start after cancellation closed its stream");
+            } finally {
+                blockingStream.releaseRead.countDown();
+            }
+        });
+    }
+
     private static final class EventLoopProbeInputStream extends InputStream {
         private final byte[] data;
         private final AtomicBoolean readAttempted = new AtomicBoolean();
@@ -1186,4 +1232,5 @@ private void recordReadThread() {
             }
         }
     }
+
 }

client/src/test/java/org/asynchttpclient/Http2StreamingBodyFlowControlTest.java

@@ -61,6 +61,7 @@
 import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -71,6 +72,9 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.asynchttpclient.Dsl.asyncHttpClient;
 import static org.asynchttpclient.Dsl.config;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.BlockingInputStream;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.CloseProbeInputStream;
+import static org.asynchttpclient.OffloadedBodyReadTestSupport.awaitExecutorState;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -362,6 +366,74 @@ public void inputStreamBodyReadsOffEventLoopOverHttp2() throws Exception {
         assertFalse(is.readOnEventLoop.get(), "InputStream.read must not run on an event-loop thread");
     }
 
+    @Test
+    public void blockedInputStreamDoesNotStallSiblingStreamOverHttp2() throws Exception {
+        startServer(-1);
+        byte[] payload = deterministicPayload(SMALL_SIZE);
+        BlockingInputStream bodyStream = new BlockingInputStream(payload);
+
+        try (AsyncHttpClient client = http2Client()) {
+            ListenableFuture<Response> blockedUpload = client.preparePost(httpsUrl("/blocked"))
+                    .setBody(new InputStreamBodyGenerator(bodyStream, payload.length))
+                    .execute();
+            assertTrue(bodyStream.readStarted.await(5, SECONDS), "blocking body read should start");
+
+            Response sibling = client.prepareGet(httpsUrl("/sibling"))
+                    .execute()
+                    .get(5, SECONDS);
+            assertEquals(200, sibling.getStatusCode(),
+                    "a sibling stream should complete while the upload read is blocked");
+
+            bodyStream.releaseRead.countDown();
+            assertEchoed(blockedUpload.get(30, SECONDS), payload);
+        } finally {
+            bodyStream.releaseRead.countDown();
+        }
+    }
+
+    @Test
+    public void cancelledQueuedInputStreamIsNotReadOverHttp2() throws Exception {
+        startServer(-1);
+        byte[] payload = deterministicPayload(SMALL_SIZE);
+        BlockingInputStream blockingStream = new BlockingInputStream(payload);
+        CloseProbeInputStream queuedStream = new CloseProbeInputStream();
+
+        DefaultAsyncHttpClientConfig clientConfig = config()
+                .setUseInsecureTrustManager(true)
+                .setHttp2Enabled(true)
+                .setMaxConnectionsPerHost(1)
+                .setRequestBodyStreamReadOffloadEnabled(true)
+                .setRequestBodyStreamReadThreadsCount(1)
+                .setRequestBodyStreamReadQueueSize(1)
+                .setRequestTimeout(Duration.ofSeconds(30))
+                .build();
+
+        try (DefaultAsyncHttpClient client = new DefaultAsyncHttpClient(clientConfig)) {
+            ThreadPoolExecutor executor = assertInstanceOf(ThreadPoolExecutor.class,
+                    client.blockingBodyReadExecutor());
+            ListenableFuture<Response> blockingUpload = client.preparePost(httpsUrl("/blocking"))
+                    .setBody(new InputStreamBodyGenerator(blockingStream, payload.length))
+                    .execute();
+            assertTrue(blockingStream.readStarted.await(5, SECONDS), "first body read should occupy the worker");
+
+            ListenableFuture<Response> queuedUpload = client.preparePost(httpsUrl("/queued"))
+                    .setBody(new InputStreamBodyGenerator(queuedStream, 1))
+                    .execute();
+            awaitExecutorState(executor, 1, 1);
+
+            assertTrue(queuedUpload.cancel(true), "queued request should be cancelled");
+            assertTrue(queuedStream.closed.await(5, SECONDS), "cancellation should close the queued stream");
+
+            blockingStream.releaseRead.countDown();
+            assertEchoed(blockingUpload.get(30, SECONDS), payload);
+            awaitExecutorState(executor, 0, 0);
+            assertFalse(queuedStream.readAttempted.get(),
+                    "a queued read must not start after cancellation closed its stream");
+        } finally {
+            blockingStream.releaseRead.countDown();
+        }
+    }
+
     @Test
     public void inputStreamRuntimeFailureFailsRequestOverHttp2() throws Exception {
         startServer(-1);
@@ -587,6 +659,7 @@ private void recordReadThread() {
         }
     }
 
+
     // Lifecycle (finding #2 from the cold audit): when the stream is closed while the body pump is parked on
     // SUSPEND (a streaming/feedable body with no data yet), no in-flight write's failure would run cleanup —
     // so the writer must close the body source from the stream's closeFuture, or the source (and any file

client/src/test/java/org/asynchttpclient/OffloadedBodyReadTestSupport.java

@@ -0,0 +1,117 @@
+/*
+ *    Copyright (c) 2026 AsyncHttpClient Project. All rights reserved.
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+package org.asynchttpclient;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+final class OffloadedBodyReadTestSupport {
+
+    private OffloadedBodyReadTestSupport() {
+    }
+
+    static void awaitExecutorState(ThreadPoolExecutor executor, int queued, int active)
+            throws InterruptedException {
+        long deadline = System.nanoTime() + SECONDS.toNanos(5);
+        while ((executor.getQueue().size() != queued || executor.getActiveCount() != active)
+                && System.nanoTime() < deadline) {
+            new CountDownLatch(1).await(10, MILLISECONDS);
+        }
+        assertEquals(queued, executor.getQueue().size(), "unexpected body-read executor queue size");
+        assertEquals(active, executor.getActiveCount(), "unexpected body-read executor active count");
+    }
+
+    static final class BlockingInputStream extends InputStream {
+        private final byte[] data;
+        private final AtomicBoolean firstRead = new AtomicBoolean(true);
+        private final AtomicInteger position = new AtomicInteger();
+        final CountDownLatch readStarted = new CountDownLatch(1);
+        final CountDownLatch releaseRead = new CountDownLatch(1);
+
+        BlockingInputStream(byte[] data) {
+            this.data = data;
+        }
+
+        @Override
+        public int read() throws IOException {
+            byte[] single = new byte[1];
+            int read = read(single, 0, 1);
+            return read < 0 ? -1 : single[0] & 0xFF;
+        }
+
+        @Override
+        public int read(byte[] buffer, int offset, int length) throws IOException {
+            awaitFirstRead();
+            int current = position.get();
+            if (current == data.length) {
+                return -1;
+            }
+            int read = Math.min(length, data.length - current);
+            System.arraycopy(data, current, buffer, offset, read);
+            position.addAndGet(read);
+            return read;
+        }
+
+        @Override
+        public void close() {
+            releaseRead.countDown();
+        }
+
+        private void awaitFirstRead() throws IOException {
+            if (firstRead.compareAndSet(true, false)) {
+                readStarted.countDown();
+                try {
+                    if (!releaseRead.await(10, SECONDS)) {
+                        throw new IOException("timed out waiting to release blocking request body read");
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new IOException("blocking request body read interrupted", e);
+                }
+            }
+        }
+    }
+
+    static final class CloseProbeInputStream extends InputStream {
+        final AtomicBoolean readAttempted = new AtomicBoolean();
+        final CountDownLatch closed = new CountDownLatch(1);
+
+        @Override
+        public int read() {
+            readAttempted.set(true);
+            return -1;
+        }
+
+        @Override
+        public int read(byte[] buffer, int offset, int length) {
+            readAttempted.set(true);
+            return -1;
+        }
+
+        @Override
+        public void close() {
+            closed.countDown();
+        }
+    }
+}