Repository: netty/netty
Issue: AbstractHttp2StreamChannel fails to flush window update under certain read patterns

### Expected behavior

Window updates get flushed by `AbstractHttp2StreamChannel`.

### Actual behavior

Under certain conditions that doesn't happen.

### Steps to reproduce

### Minimal yet complete reproducer code (or URL to code)
I have a test case that is really ugly and should probably be condensed but it illustrates the issue. I've annotated it with values observed with a debugger during it's execution.
```java
    static ByteBuf bb(int size) {
        ByteBuf buf = UnpooledByteBufAllocator.DEFAULT.buffer();
        for (int i = 0; i < size; i++) {
            buf.writeByte('a');
        }
        return buf;
    }


    class FlushSniffer extends ChannelOutboundHandlerAdapter {

        private boolean didFlush;

        public boolean checkFlush() {
            boolean r = didFlush;
            didFlush = false;
            return r;
        }

        @Override
        public void flush(ChannelHandlerContext ctx) throws Exception {
            didFlush = true;
            super.flush(ctx);
        }
    }

    @Test
    public void windowUpdatesAreFlushed() {
        LastInboundHandler inboundHandler = new LastInboundHandler();
        FlushSniffer flushSniffer = new FlushSniffer();
        parentChannel.pipeline().addFirst(flushSniffer);

        Http2StreamChannel childChannel = newInboundStream(3, false, inboundHandler);
        assertTrue(childChannel.config().isAutoRead());
        childChannel.config().setAutoRead(false);
        assertFalse(childChannel.config().isAutoRead());

        Http2HeadersFrame headersFrame = inboundHandler.readInbound();
        assertNotNull(headersFrame);
        // From the headers.
        assertTrue(flushSniffer.checkFlush());

        // childChannel.readState == IN_PROGRESS due to initial auto-read.
        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(1), 0, false);
        // readState == IDLE, flowControlledBytes = 1

        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(1020), 0, false);
        // readState == IDLE, flowControlledBytes = 1, 1 message(1020) buffered.
        assertTrue(flushSniffer.checkFlush());

        // This will trigger a read of the second message (1020 bytes).
        childChannel.read();
        assertTrue(flushSniffer.checkFlush()); // fails.
        // This should have flushed the window update of 1
        // flowControlledBytes = 1020

        childChannel.read();
        // triggered a flow control window updatte in DefaultHttp2LocalFlowCoontroller:465
        // flowControlledBytes = 0 <- we wrote the window update but didn't flush!
        assertTrue(flushSniffer.checkFlush()); // fails: no flush was called.

        Http2DataFrame data = inboundHandler.blockingReadInbound();
        release(data);
    }
```

There are a few problems that I see so far. 

* One is that on `beginRead()` we will write any flow bytes as an update frame and enter `doBeginRead()` but if there are no messages to be had we break out of the loop without flushing [here](https://github.com/netty/netty/blob/netty-4.1.43.Final/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java#L805-L810).
* A second problem is that when we do write the window update we check if the promise is complete and if it is, we assume it's been flushed [here](https://github.com/netty/netty/blob/netty-4.1.43.Final/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java#L848-L853). However, the Http2FrameCodec intercepts these frames and always just sets the promise to done [here](https://github.com/netty/netty/blob/4.1/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameCodec.java#L313). This may cause an explicit `flush()` call to be ineffective as the stream channel attempts to be clever about it and elid unnecessary flushes [here](https://github.com/netty/netty/blob/netty-4.1.43.Final/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java#L1034).

### Netty version
branch 4.1, SHA e0d73bca4dd36910c7cdbe3d577c51aeb8287cc6 (Fri Feb 28)

Issue comments:
And I'm happy to work on a patch once I get a 👍 that someone else agrees this is a problem.
@bryce-anderson yes this sounds like a bug.
so yep please work on a fix @bryce-anderson :)

Pull request review comments:
This feels like something that almost certainly exists somewhere else.
This change isn't strictly necessary for this patch but I spent a fair amount of time trying to read through this for the nth time and decided to clean it up so the control flow (when the method returns etc, _**not**_ H2 flow control) is more readable. If desired, we can move it to a different patch.
These are also not strictly necessary but part of the general cleanup I was attempting to do so this is more readable in the future.
Just use bytebuf.writeZero(size); ?
Would you prefer I replace the body of this method with
`UnpooledByteBufAllocator.DEFAULT.buffer().writeZero(size)` or inline it in the best?
I dont care... I was just mention it as you said there is most likely something like this already :)
nit: private final static 

Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java
index 547d724f556..92abf19a246 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java
@@ -558,10 +558,7 @@ void fireChildRead(Http2Frame frame) {
             // read (unknown, reset) and the trade off is less conditionals for the hot path (headers/data) at the
             // cost of additional readComplete notifications on the rare path.
             if (allocHandle.continueReading()) {
-                if (!readCompletePending) {
-                    readCompletePending = true;
-                    addChannelToReadCompletePendingQueue();
-                }
+                maybeAddChannelToReadCompletePendingQueue();
             } else {
                 unsafe.notifyReadComplete(allocHandle, true);
             }
@@ -807,6 +804,9 @@ void doBeginRead() {
                     if (readEOS) {
                         unsafe.closeForcibly();
                     }
+                    // We need to double check that there is nothing left to flush such as a
+                    // window update frame.
+                    flush();
                     break;
                 }
                 final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
@@ -822,10 +822,7 @@ void doBeginRead() {
                     // currently reading it is possible that more frames will be delivered to this child channel. In
                     // the case that this child channel still wants to read we delay the channelReadComplete on this
                     // child channel until the parent is done reading.
-                    if (!readCompletePending) {
-                        readCompletePending = true;
-                        addChannelToReadCompletePendingQueue();
-                    }
+                    maybeAddChannelToReadCompletePendingQueue();
                 } else {
                     notifyReadComplete(allocHandle, true);
                 }
@@ -841,6 +838,10 @@ private void updateLocalWindowIfNeeded() {
                 int bytes = flowControlledBytes;
                 flowControlledBytes = 0;
                 ChannelFuture future = write0(parentContext(), new DefaultHttp2WindowUpdateFrame(bytes).stream(stream));
+                // window update frames are commonly swallowed by the Http2FrameCodec and the promise is synchronously
+                // completed but the flow controller _may_ have generated a wire level WINDOW_UPDATE. Therefore we need,
+                // to assume there was a write done that needs to be flushed or we risk flow control starvation.
+                writeDoneAndNoFlush = true;
                 // Add a listener which will notify and teardown the stream
                 // when a window update fails if needed or check the result of the future directly if it was completed
                 // already.
@@ -849,7 +850,6 @@ private void updateLocalWindowIfNeeded() {
                     windowUpdateFrameWriteComplete(future, AbstractHttp2StreamChannel.this);
                 } else {
                     future.addListener(windowUpdateFrameWriteListener);
-                    writeDoneAndNoFlush = true;
                 }
             }
         }
@@ -920,58 +920,57 @@ public void write(Object msg, final ChannelPromise promise) {
             try {
                 if (msg instanceof Http2StreamFrame) {
                     Http2StreamFrame frame = validateStreamFrame((Http2StreamFrame) msg).stream(stream());
-                    if (!firstFrameWritten && !isStreamIdValid(stream().id())) {
-                        if (!(frame instanceof Http2HeadersFrame)) {
-                            ReferenceCountUtil.release(frame);
-                            promise.setFailure(
-                                    new IllegalArgumentException("The first frame must be a headers frame. Was: "
-                                            + frame.name()));
-                            return;
-                        }
-                        firstFrameWritten = true;
-                        ChannelFuture f = write0(parentContext(), frame);
-                        if (f.isDone()) {
-                            firstWriteComplete(f, promise);
-                        } else {
-                            final long bytes = FlowControlledFrameSizeEstimator.HANDLE_INSTANCE.size(msg);
-                            incrementPendingOutboundBytes(bytes, false);
-                            f.addListener(new ChannelFutureListener() {
-                                @Override
-                                public void operationComplete(ChannelFuture future) {
-                                    firstWriteComplete(future, promise);
-                                    decrementPendingOutboundBytes(bytes, false);
-                                }
-                            });
-                            writeDoneAndNoFlush = true;
-                        }
-                        return;
-                    }
+                    writeHttp2StreamFrame(frame, promise);
                 } else {
                     String msgStr = msg.toString();
                     ReferenceCountUtil.release(msg);
                     promise.setFailure(new IllegalArgumentException(
                             "Message must be an " + StringUtil.simpleClassName(Http2StreamFrame.class) +
                                     ": " + msgStr));
-                    return;
                 }
+            } catch (Throwable t) {
+                promise.tryFailure(t);
+            }
+        }
 
-                ChannelFuture f = write0(parentContext(), msg);
-                if (f.isDone()) {
-                    writeComplete(f, promise);
+        private void writeHttp2StreamFrame(Http2StreamFrame frame, final ChannelPromise promise) {
+            if (!firstFrameWritten && !isStreamIdValid(stream().id()) && !(frame instanceof Http2HeadersFrame)) {
+                ReferenceCountUtil.release(frame);
+                promise.setFailure(
+                    new IllegalArgumentException("The first frame must be a headers frame. Was: "
+                        + frame.name()));
+                return;
+            }
+
+            final boolean firstWrite;
+            if (firstFrameWritten) {
+                firstWrite = false;
+            } else {
+                firstWrite = firstFrameWritten = true;
+            }
+
+            ChannelFuture f = write0(parentContext(), frame);
+            if (f.isDone()) {
+                if (firstWrite) {
+                    firstWriteComplete(f, promise);
                 } else {
-                    final long bytes = FlowControlledFrameSizeEstimator.HANDLE_INSTANCE.size(msg);
-                    incrementPendingOutboundBytes(bytes, false);
-                    f.addListener(new ChannelFutureListener() {
-                        @Override
-                        public void operationComplete(ChannelFuture future) {
+                    writeComplete(f, promise);
+                }
+            } else {
+                final long bytes = FlowControlledFrameSizeEstimator.HANDLE_INSTANCE.size(frame);
+                incrementPendingOutboundBytes(bytes, false);
+                f.addListener(new ChannelFutureListener() {
+                    @Override
+                    public void operationComplete(ChannelFuture future) {
+                        if (firstWrite) {
+                            firstWriteComplete(future, promise);
+                        } else {
                             writeComplete(future, promise);
-                            decrementPendingOutboundBytes(bytes, false);
                         }
-                    });
-                    writeDoneAndNoFlush = true;
-                }
-            } catch (Throwable t) {
-                promise.tryFailure(t);
+                        decrementPendingOutboundBytes(bytes, false);
+                    }
+                });
+                writeDoneAndNoFlush = true;
             }
         }
 
@@ -1084,6 +1083,13 @@ public ChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         }
     }
 
+    private void maybeAddChannelToReadCompletePendingQueue() {
+        if (!readCompletePending) {
+            readCompletePending = true;
+            addChannelToReadCompletePendingQueue();
+        }
+    }
+
     protected void flush0(ChannelHandlerContext ctx) {
         ctx.flush();
     }


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java
index 8dcc579797c..c1207f686e0 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java
@@ -22,6 +22,7 @@
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.channel.embedded.EmbeddedChannel;
@@ -63,7 +64,6 @@
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyShort;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.never;
@@ -1152,6 +1152,66 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 3);
     }
 
+    private static final class FlushSniffer extends ChannelOutboundHandlerAdapter {
+
+        private boolean didFlush;
+
+        public boolean checkFlush() {
+            boolean r = didFlush;
+            didFlush = false;
+            return r;
+        }
+
+        @Override
+        public void flush(ChannelHandlerContext ctx) throws Exception {
+            didFlush = true;
+            super.flush(ctx);
+        }
+    }
+
+    @Test
+    public void windowUpdatesAreFlushed() {
+        LastInboundHandler inboundHandler = new LastInboundHandler();
+        FlushSniffer flushSniffer = new FlushSniffer();
+        parentChannel.pipeline().addFirst(flushSniffer);
+
+        Http2StreamChannel childChannel = newInboundStream(3, false, inboundHandler);
+        assertTrue(childChannel.config().isAutoRead());
+        childChannel.config().setAutoRead(false);
+        assertFalse(childChannel.config().isAutoRead());
+
+        Http2HeadersFrame headersFrame = inboundHandler.readInbound();
+        assertNotNull(headersFrame);
+
+        assertTrue(flushSniffer.checkFlush());
+
+        // Write some bytes to get the channel into the idle state with buffered data and also verify we
+        // do not dispatch it until we receive a read() call.
+        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(16 * 1024), 0, false);
+        frameInboundWriter.writeInboundData(childChannel.stream().id(), bb(16 * 1024), 0, false);
+        assertTrue(flushSniffer.checkFlush());
+
+        verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+        // only the first one was read because it was legacy auto-read behavior.
+        verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 1);
+        assertFalse(flushSniffer.checkFlush());
+
+        // Trigger a read of the second frame.
+        childChannel.read();
+        verifyFramesMultiplexedToCorrectChannel(childChannel, inboundHandler, 1);
+        // We expect a flush here because the StreamChannel will flush the smaller increment but the
+        // connection will collect the bytes and decide not to send a wire level frame until more are consumed.
+        assertTrue(flushSniffer.checkFlush());
+        verify(frameWriter, never()).writeWindowUpdate(eqCodecCtx(), anyInt(), anyInt(), anyChannelPromise());
+
+        // Call read one more time which should trigger the writing of the flow control update.
+        childChannel.read();
+        verify(frameWriter).writeWindowUpdate(eqCodecCtx(), eq(0), eq(32 * 1024), anyChannelPromise());
+        verify(frameWriter).writeWindowUpdate(
+            eqCodecCtx(), eq(childChannel.stream().id()), eq(32 * 1024), anyChannelPromise());
+        assertTrue(flushSniffer.checkFlush());
+    }
+
     private static void verifyFramesMultiplexedToCorrectChannel(Http2StreamChannel streamChannel,
                                                                 LastInboundHandler inboundHandler,
                                                                 int numFrames) {
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
index 6fa3449709a..ff420bf6dad 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
@@ -687,6 +687,10 @@ static ByteBuf bb(String s) {
         return ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, s);
     }
 
+    static ByteBuf bb(int size) {
+        return UnpooledByteBufAllocator.DEFAULT.buffer().writeZero(size);
+    }
+
     static void assertEqualsAndRelease(Http2Frame expected, Http2Frame actual) {
         try {
             assertEquals(expected, actual);