Batch HTTP/2 upload body flushes

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}).
@@ -123,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;
@@ -176,12 +182,14 @@ 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) {
+        if (done || pumping || terminalWritePending) {
             return;
         }
+        pumping = true;
         try {
             while (true) {
                 if (done) {
@@ -196,6 +204,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;
                 }
@@ -208,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;
@@ -219,9 +229,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();
@@ -232,6 +245,8 @@ private void pump() {
             }
         } catch (Throwable t) {
             finish(t);
+        } finally {
+            pumping = false;
         }
     }
 
@@ -308,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();

client/src/test/java/org/asynchttpclient/netty/request/body/Http2BodyWriterTest.java

@@ -0,0 +1,225 @@
+/*
+ *    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.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.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;
+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.AtomicBoolean;
+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;
+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();
+    }
+
+    @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 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);
+        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(eventContext.channel()).thenReturn(channel);
+        when(channel.write(any())).thenAnswer(invocation -> {
+            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);
+
+        assertEquals(0, source.closed);
+        assertNotNull(resumeHandler.get());
+        verify(channel, times(1)).write(any(DefaultHttp2DataFrame.class));
+        verify(channel, times(1)).flush();
+
+        ((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();
+        verify(pipeline).remove(resumeHandler.get());
+    }
+
+    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++;
+        }
+    }
+
+    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++;
+        }
+    }
+}