Don't close channel when receiving a STOP_SENDING frame

Motivation:

We need to ensure we correctly handle the situation when receiving a STOP_SENDING frame. Due a bug we fired an exception trough the pipeline and closed the channel.

Modifications:

- When receiving a STOP_SENDING frame just faill all writes without closing the channel

Result:

Fixes bug reported in netty/netty-incubator-codec-http3#262

Author: normanmaurer

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicChannel.java

@@ -992,7 +992,10 @@ private boolean handleWritableStreams() {
                             QuicheQuicStreamChannel streamChannel = streams.get(streamId);
                             if (streamChannel != null) {
                                 int capacity = Quiche.quiche_conn_stream_capacity(connAddr, streamId);
-                                if (capacity < 0) {
+                                if (capacity < 0 &&
+                                        // If we received a STOP_SENDING frame we still want to call writable(...)
+                                        // to force all data to be written and so failed.
+                                        capacity != Quiche.QUICHE_ERR_STREAM_STOPPED) {
                                     if (!Quiche.quiche_conn_stream_finished(connAddr, streamId)) {
                                         // Only fire an exception if the error was not caused because the stream is
                                         // considered finished.

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicStreamChannel.java

@@ -642,6 +642,12 @@ boolean writeQueued() {
                             return written;
                         }
                     } catch (Exception e) {
+                        if (e instanceof QuicException && (
+                                (QuicException) e).error() == QuicError.STREAM_STOPPED) {
+                            // Once its signaled that the stream is stopped we can just fail everything.
+                            queue.removeAndFailAll(e);
+                            break;
+                        }
                         queue.remove().setFailure(e);
                         continue;
                     }

codec-native-quic/src/test/java/io/netty/incubator/codec/quic/QuicStreamShutdownTest.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2023 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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:
+ *
+ *   https://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 io.netty.incubator.codec.quic;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QuicStreamShutdownTest extends AbstractQuicTest {
+
+    @ParameterizedTest
+    @MethodSource("newSslTaskExecutors")
+    public void testShutdownInputClosureCausesStreamStopped(Executor executor) throws Throwable {
+        Channel server = null;
+        Channel channel = null;
+        CountDownLatch latch = new CountDownLatch(2);
+        try {
+            server = QuicTestUtils.newServer(executor, new ChannelInboundHandlerAdapter(), new ChannelInboundHandlerAdapter() {
+                @Override
+                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+                    ChannelFutureListener futureListener = new ChannelFutureListener() {
+                        @Override
+                        public void operationComplete(ChannelFuture channelFuture) {
+                            Throwable cause = channelFuture.cause();
+                            if (cause instanceof QuicException &&
+                                    ((QuicException) cause).error() == QuicError.STREAM_STOPPED) {
+                                latch.countDown();
+                            }
+                        }
+                    };
+                    ByteBuf buffer = (ByteBuf) msg;
+                    ctx.write(buffer.retainedDuplicate()).addListener(futureListener);
+                    ctx.writeAndFlush(buffer).addListener(futureListener);
+                }
+            });
+            channel = QuicTestUtils.newClient(executor);
+            QuicChannel quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
+                    .handler(new ChannelInboundHandlerAdapter())
+                    .streamHandler(new ChannelInboundHandlerAdapter())
+                    .remoteAddress(server.localAddress())
+                    .connect()
+                    .get();
+
+            QuicStreamChannel streamChannel = quicChannel.createStream(QuicStreamType.BIDIRECTIONAL,
+                    new ChannelInboundHandlerAdapter()).sync().getNow();
+            streamChannel.shutdownInput().sync();
+            assertTrue(streamChannel.isInputShutdown());
+            streamChannel.writeAndFlush(Unpooled.buffer().writeLong(8)).sync();
+
+            latch.await();
+        } finally {
+            QuicTestUtils.closeIfNotNull(channel);
+            QuicTestUtils.closeIfNotNull(server);
+
+            shutdown(executor);
+        }
+    }
+}