From 553ab87564b5424431cad4a949d8fa84c6601837 Mon Sep 17 00:00:00 2001
From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com>
Date: Mon, 20 Jul 2026 10:48:27 +0200
Subject: [PATCH 1/3] Batch HTTP/2 upload body flushes

Avoid flushing every intermediate HTTP/2 DATA chunk produced by
Http2BodyWriter. The pump now flushes when it yields for source
suspension, when backpressure makes the stream unwritable, or when
it writes the terminal DATA frame.

This preserves bounded in-flight write behavior while reducing
per-chunk flush overhead for multi-chunk uploads. Add a focused unit
test for the single terminal flush path while the stream remains
writable.

Codex on behalf of Pavel Ptashyts

Co-Authored-By: Codex <codex@openai.com>
---
 .../netty/request/body/Http2BodyWriter.java   | 14 ++-
 .../request/body/Http2BodyWriterTest.java     | 97 +++++++++++++++++++
 2 files changed, 107 insertions(+), 4 deletions(-)
 create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java

diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
index 77ebdb8cf..5e6a03a36 100644
--- a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
+++ b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
@@ -38,8 +38,9 @@
  * <p>
  * This writer is a one-chunk-at-a-time pump driven entirely on the stream channel's event loop:
  * <ul>
- *   <li>It produces and writes exactly one chunk, then flushes, so frames actually leave the process
- *       instead of accumulating.</li>
+ *   <li>It writes chunks while the stream remains writable, flushing when the pump yields for flow control,
+ *       a source suspension, or the terminal DATA frame, so frames leave the process without one flush per
+ *       chunk.</li>
  *   <li>It only produces the next chunk while {@link Http2StreamChannel#isWritable()} is {@code true}.
  *       A stream child channel becomes unwritable when the HTTP/2 flow-control window is exhausted or
  *       the local high-water mark is reached (child writes go through {@code incrementPendingOutboundBytes}).
@@ -176,7 +177,8 @@ static void start(Http2StreamChannel channel, ChunkSource source) {
 
     /**
      * Produces and writes chunks until the channel goes unwritable (then parks for
-     * {@code channelWritabilityChanged}) or the body is exhausted. Always runs on the event loop.
+     * {@code channelWritabilityChanged}), the source suspends, or the body is exhausted. Always runs on the
+     * event loop. Any written frames are flushed before the pump parks or completes.
      */
     private void pump() {
         if (done) {
@@ -196,6 +198,7 @@ private void pump() {
                     // No data available yet but the body is not finished (feedable body). Park until the
                     // source signals more via the onResume callback. Any already-buffered `pending` chunk is
                     // retained (O(1)); we deliberately do not flush an early endStream.
+                    channel.flush();
                     suspended = true;
                     return;
                 }
@@ -219,9 +222,12 @@ private void pump() {
                 pending = next;
                 if (toWrite != null) {
                     writeFrame(toWrite, false);
-                    channel.flush();
 
                     if (!channel.isWritable()) {
+                        channel.flush();
+                        if (channel.isWritable()) {
+                            continue;
+                        }
                         // Flow-control window exhausted / high-water mark reached: stop producing and resume
                         // from channelWritabilityChanged. `pending` (one chunk) is retained until then.
                         ensureResumeHandler();
diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
new file mode 100644
index 000000000..4748c3fac
--- /dev/null
+++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
@@ -0,0 +1,97 @@
+/*
+ *    Copyright (c) 2014-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.netty.request.body;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class Http2BodyWriterTest {
+
+    @Test
+    public void multiChunkBodyBatchesFlushUntilTerminalFrame() {
+        Http2StreamChannel channel = mock(Http2StreamChannel.class);
+        EventLoop eventLoop = mock(EventLoop.class);
+        when(eventLoop.inEventLoop()).thenReturn(true);
+        when(channel.eventLoop()).thenReturn(eventLoop);
+        when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+        when(channel.isWritable()).thenReturn(true);
+        when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+
+        AtomicInteger terminalFrames = new AtomicInteger();
+        when(channel.write(any())).thenAnswer(invocation -> {
+            Object msg = invocation.getArgument(0);
+            if (((DefaultHttp2DataFrame) msg).isEndStream()) {
+                terminalFrames.incrementAndGet();
+            }
+            ReferenceCountUtil.release(msg);
+            return succeededFuture(channel);
+        });
+
+        FixedChunkSource source = new FixedChunkSource(4);
+
+        Http2BodyWriter.start(channel, source);
+
+        assertEquals(1, source.closed);
+        assertEquals(1, terminalFrames.get());
+        verify(channel, times(4)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(1)).flush();
+    }
+
+    private static ChannelFuture succeededFuture(Http2StreamChannel channel) {
+        return new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE).setSuccess();
+    }
+
+    private static final class FixedChunkSource implements Http2BodyWriter.ChunkSource {
+        private int chunks;
+        private int closed;
+
+        FixedChunkSource(int chunks) {
+            this.chunks = chunks;
+        }
+
+        @Override
+        public ByteBuf nextChunk(ByteBufAllocator alloc) {
+            if (chunks == 0) {
+                return null;
+            }
+            chunks--;
+            return alloc.buffer(1).writeByte(chunks);
+        }
+
+        @Override
+        public void close() {
+            closed++;
+        }
+    }
+}

From 2bb7a215f23159bdebe6fdfbb7c117624c15b695 Mon Sep 17 00:00:00 2001
From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com>
Date: Mon, 20 Jul 2026 13:22:59 +0200
Subject: [PATCH 2/3] Cover HTTP/2 upload yield paths

Flush batching is correct only if every point that parks the body
pump flushes already-written DATA frames and can resume without losing
the pending chunk.

Cover source suspension and persistent channel backpressure in addition
to the writable happy path. Correct the new test file's license year
to match the repository contribution rules.

Codex on behalf of Pavel Ptashyts

Co-Authored-By: Codex <codex@openai.com>
---
 .../request/body/Http2BodyWriterTest.java     | 107 +++++++++++++++++-
 1 file changed, 106 insertions(+), 1 deletion(-)

diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
index 4748c3fac..a8928c366 100644
--- a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
+++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
@@ -1,5 +1,5 @@
 /*
- *    Copyright (c) 2014-2026 AsyncHttpClient Project. All rights reserved.
+ *    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.
@@ -19,6 +19,9 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.DefaultChannelPromise;
 import io.netty.channel.EventLoop;
 import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
@@ -28,8 +31,10 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -68,6 +73,76 @@ public void multiChunkBodyBatchesFlushUntilTerminalFrame() {
         verify(channel, times(1)).flush();
     }
 
+    @Test
+    public void sourceSuspensionFlushesBeforeParking() {
+        Http2StreamChannel channel = mock(Http2StreamChannel.class);
+        EventLoop eventLoop = mock(EventLoop.class);
+        when(eventLoop.inEventLoop()).thenReturn(true);
+        when(channel.eventLoop()).thenReturn(eventLoop);
+        when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+        when(channel.isWritable()).thenReturn(true);
+        when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+        when(channel.write(any())).thenAnswer(invocation -> {
+            ReferenceCountUtil.release(invocation.getArgument(0));
+            return succeededFuture(channel);
+        });
+
+        SuspendingChunkSource source = new SuspendingChunkSource();
+        Http2BodyWriter.start(channel, source);
+
+        assertEquals(0, source.closed);
+        verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(1)).flush();
+
+        source.finish();
+
+        assertEquals(1, source.closed);
+        verify(channel, times(2)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(2)).flush();
+    }
+
+    @Test
+    public void unwritableChannelFlushesAndResumesWhenWritable() throws Exception {
+        Http2StreamChannel channel = mock(Http2StreamChannel.class);
+        EventLoop eventLoop = mock(EventLoop.class);
+        ChannelPipeline pipeline = mock(ChannelPipeline.class);
+        ChannelHandlerContext pipelineContext = mock(ChannelHandlerContext.class);
+        AtomicReference<ChannelHandler> resumeHandler = new AtomicReference<>();
+        when(eventLoop.inEventLoop()).thenReturn(true);
+        when(channel.eventLoop()).thenReturn(eventLoop);
+        when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
+        when(channel.isWritable()).thenReturn(false, false, true);
+        when(channel.closeFuture()).thenReturn(new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE));
+        when(channel.pipeline()).thenReturn(pipeline);
+        when(pipeline.addLast(any(ChannelHandler.class))).thenAnswer(invocation -> {
+            resumeHandler.set(invocation.getArgument(0));
+            return pipeline;
+        });
+        when(pipeline.context(any(ChannelHandler.class))).thenReturn(pipelineContext);
+        when(channel.write(any())).thenAnswer(invocation -> {
+            ReferenceCountUtil.release(invocation.getArgument(0));
+            return succeededFuture(channel);
+        });
+
+        FixedChunkSource source = new FixedChunkSource(3);
+        Http2BodyWriter.start(channel, source);
+
+        assertEquals(0, source.closed);
+        assertNotNull(resumeHandler.get());
+        verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(1)).flush();
+
+        ChannelHandlerContext eventContext = mock(ChannelHandlerContext.class);
+        when(eventContext.channel()).thenReturn(channel);
+        ((io.netty.channel.ChannelInboundHandlerAdapter) resumeHandler.get())
+                .channelWritabilityChanged(eventContext);
+
+        assertEquals(1, source.closed);
+        verify(channel, times(3)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(2)).flush();
+        verify(pipeline).remove(resumeHandler.get());
+    }
+
     private static ChannelFuture succeededFuture(Http2StreamChannel channel) {
         return new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE).setSuccess();
     }
@@ -94,4 +169,34 @@ public void close() {
             closed++;
         }
     }
+
+    private static final class SuspendingChunkSource implements Http2BodyWriter.ChunkSource {
+        private int state;
+        private int closed;
+        private Runnable resume;
+
+        @Override
+        public ByteBuf nextChunk(ByteBufAllocator alloc) {
+            if (state < 2) {
+                state++;
+                return alloc.buffer(1).writeByte(state);
+            }
+            return state == 2 ? Http2BodyWriter.SUSPEND : null;
+        }
+
+        @Override
+        public void onResume(Runnable resume) {
+            this.resume = resume;
+        }
+
+        void finish() {
+            state = 3;
+            resume.run();
+        }
+
+        @Override
+        public void close() {
+            closed++;
+        }
+    }
 }

From 531b85e8c20339d015a642cfd00b4036a981c2d1 Mon Sep 17 00:00:00 2001
From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com>
Date: Mon, 20 Jul 2026 13:59:37 +0200
Subject: [PATCH 3/3] Guard HTTP/2 body pump reentry

Prevent synchronous writability callbacks from re-entering the
HTTP/2 body pump and keep the pump parked until its terminal DATA
write completes.

Extend the writer test to trigger a nested callback during terminal
flush and verify that one endStream frame is written and source
cleanup waits for write completion.

Codex on behalf of Pavel Ptashyts

Co-Authored-By: Codex <codex@openai.com>
---
 .../netty/request/body/Http2BodyWriter.java   | 13 ++++++--
 .../request/body/Http2BodyWriterTest.java     | 31 ++++++++++++++++---
 2 files changed, 38 insertions(+), 6 deletions(-)

diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
index 5e6a03a36..7df5b0694 100644
--- a/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
+++ b/client/src/main/java/org/asynchttpclient/netty/request/body/Http2BodyWriter.java
@@ -124,6 +124,11 @@ default void onResume(Runnable resume) {
     private ByteBuf pending;
     private boolean done;
 
+    // flush() can synchronously fire channelWritabilityChanged. Prevent that callback from re-entering the
+    // pump, and prevent later callbacks from writing a second endStream frame before the first one completes.
+    private boolean pumping;
+    private boolean terminalWritePending;
+
     // Transient handler that resumes the pump when the channel becomes writable again. Added lazily the
     // first time the pump parks, removed by finish().
     private WritabilityResumeHandler resumeHandler;
@@ -181,9 +186,10 @@ static void start(Http2StreamChannel channel, ChunkSource source) {
      * event loop. Any written frames are flushed before the pump parks or completes.
      */
     private void pump() {
-        if (done) {
+        if (done || pumping || terminalWritePending) {
             return;
         }
+        pumping = true;
         try {
             while (true) {
                 if (done) {
@@ -211,6 +217,7 @@ private void pump() {
                     ByteBuf terminal = last != null ? last
                             // Empty body — preserve existing behaviour: a single empty DATA frame ends the stream.
                             : channel.alloc().buffer(0);
+                    terminalWritePending = true;
                     writeLastFrame(terminal);
                     channel.flush();
                     return;
@@ -238,6 +245,8 @@ private void pump() {
             }
         } catch (Throwable t) {
             finish(t);
+        } finally {
+            pumping = false;
         }
     }
 
@@ -314,7 +323,7 @@ private void finish(Throwable cause) {
     private final class WritabilityResumeHandler extends ChannelInboundHandlerAdapter {
         @Override
         public void channelWritabilityChanged(ChannelHandlerContext ctx) {
-            if (!done && ctx.channel().isWritable()) {
+            if (!done && !terminalWritePending && ctx.channel().isWritable()) {
                 pump();
             }
             ctx.fireChannelWritabilityChanged();
diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
index a8928c366..c2de0a1f4 100644
--- a/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
+++ b/client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java
@@ -30,6 +30,7 @@
 import io.netty.util.concurrent.ImmediateEventExecutor;
 import org.junit.jupiter.api.Test;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -102,12 +103,16 @@ public void sourceSuspensionFlushesBeforeParking() {
     }
 
     @Test
-    public void unwritableChannelFlushesAndResumesWhenWritable() throws Exception {
+    public void unwritableChannelResumesWithoutReentrantTerminalWrite() throws Exception {
         Http2StreamChannel channel = mock(Http2StreamChannel.class);
         EventLoop eventLoop = mock(EventLoop.class);
         ChannelPipeline pipeline = mock(ChannelPipeline.class);
         ChannelHandlerContext pipelineContext = mock(ChannelHandlerContext.class);
+        ChannelHandlerContext eventContext = mock(ChannelHandlerContext.class);
         AtomicReference<ChannelHandler> resumeHandler = new AtomicReference<>();
+        AtomicBoolean nestedWritabilityCallbackFired = new AtomicBoolean();
+        AtomicInteger terminalFrames = new AtomicInteger();
+        DefaultChannelPromise terminalWrite = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
         when(eventLoop.inEventLoop()).thenReturn(true);
         when(channel.eventLoop()).thenReturn(eventLoop);
         when(channel.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
@@ -119,10 +124,26 @@ public void unwritableChannelFlushesAndResumesWhenWritable() throws Exception {
             return pipeline;
         });
         when(pipeline.context(any(ChannelHandler.class))).thenReturn(pipelineContext);
+        when(eventContext.channel()).thenReturn(channel);
         when(channel.write(any())).thenAnswer(invocation -> {
-            ReferenceCountUtil.release(invocation.getArgument(0));
+            DefaultHttp2DataFrame frame = invocation.getArgument(0);
+            if (frame.isEndStream()) {
+                terminalFrames.incrementAndGet();
+                ReferenceCountUtil.release(frame);
+                return terminalWrite;
+            }
+            ReferenceCountUtil.release(frame);
             return succeededFuture(channel);
         });
+        when(channel.flush()).thenAnswer(invocation -> {
+            ChannelHandler handler = resumeHandler.get();
+            if (handler != null
+                    && terminalFrames.get() != 0
+                    && nestedWritabilityCallbackFired.compareAndSet(false, true)) {
+                ((io.netty.channel.ChannelInboundHandlerAdapter) handler).channelWritabilityChanged(eventContext);
+            }
+            return channel;
+        });
 
         FixedChunkSource source = new FixedChunkSource(3);
         Http2BodyWriter.start(channel, source);
@@ -132,11 +153,13 @@ public void unwritableChannelFlushesAndResumesWhenWritable() throws Exception {
         verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
         verify(channel, times(1)).flush();
 
-        ChannelHandlerContext eventContext = mock(ChannelHandlerContext.class);
-        when(eventContext.channel()).thenReturn(channel);
         ((io.netty.channel.ChannelInboundHandlerAdapter) resumeHandler.get())
                 .channelWritabilityChanged(eventContext);
 
+        assertEquals(0, source.closed);
+        assertEquals(1, terminalFrames.get());
+        terminalWrite.setSuccess();
+
         assertEquals(1, source.closed);
         verify(channel, times(3)).write(any(DefaultHttp2DataFrame.class));
         verify(channel, times(2)).flush();