http-netty: reset cancelled HTTP/2 streams whose request body is still writing

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/TrailersOnlyErrorTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2021 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2021-2026 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,6 +15,7 @@
  */
 package io.servicetalk.grpc.netty;
 
+import io.servicetalk.client.api.TransportObserverConnectionFactoryFilter;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.grpc.api.GrpcClientBuilder;
 import io.servicetalk.grpc.api.GrpcServerBuilder;
@@ -38,8 +39,13 @@
 import io.servicetalk.http.api.StreamingHttpResponseFactory;
 import io.servicetalk.http.api.StreamingHttpServiceFilter;
 import io.servicetalk.http.utils.BeforeFinallyHttpOperator;
+import io.servicetalk.transport.api.ConnectionInfo;
+import io.servicetalk.transport.api.ConnectionObserver;
+import io.servicetalk.transport.api.ConnectionObserver.MultiplexedObserver;
+import io.servicetalk.transport.api.ConnectionObserver.StreamObserver;
 import io.servicetalk.transport.api.HostAndPort;
 import io.servicetalk.transport.api.ServerContext;
+import io.servicetalk.transport.api.TransportObserver;
 import io.servicetalk.transport.netty.internal.ExecutionContextExtension;
 
 import io.grpc.examples.helloworld.Greeter;
@@ -60,19 +66,22 @@
 import static io.servicetalk.concurrent.api.Publisher.from;
 import static io.servicetalk.concurrent.api.Publisher.never;
 import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION;
+import static io.servicetalk.concurrent.internal.TestTimeoutConstants.DEFAULT_TIMEOUT_SECONDS;
 import static io.servicetalk.grpc.api.GrpcHeaderNames.GRPC_STATUS;
 import static io.servicetalk.grpc.api.GrpcStatusCode.UNIMPLEMENTED;
 import static io.servicetalk.grpc.api.GrpcStatusCode.UNKNOWN;
 import static io.servicetalk.test.resources.TestUtils.assertNoAsyncErrors;
 import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress;
 import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort;
 import static java.util.Collections.singleton;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -144,6 +153,25 @@ void testServiceThrows() throws Exception {
         responseLatch.await();  // Make sure all responses complete
     }
 
+    @Test
+    void serviceThrowsWithOpenRequestStreamResetsStream() throws Exception {
+        final CountDownLatch streamClosed = new CountDownLatch(1);
+        final TesterService service = mockTesterService();
+        setupServiceThrows(service);
+
+        try (ServerContext serverContext = GrpcServers.forAddress(localAddress(0))
+                .initializeHttp(TrailersOnlyErrorTest::applyCtx)
+                .listenAndAwait(new Tester.ServiceFactory(service));
+             TesterClient client = GrpcClients.forAddress(serverHostAndPort(serverContext))
+                     .initializeHttp(builder -> applyCtx(builder).appendConnectionFactoryFilter(
+                             new TransportObserverConnectionFactoryFilter<>(onStreamClosed(streamClosed))))
+                     .build(new Tester.ClientFactory())) {
+            // The request stream never completes, so only the client can end the stream after the Trailers-Only error.
+            verifyException(client.testBiDiStream(never()).toFuture(), UNKNOWN);
+            assertTrue(streamClosed.await(DEFAULT_TIMEOUT_SECONDS, SECONDS));
+        }
+    }
+
     @Test
     void testServiceThrowsBlockingClient() throws Exception {
         final BlockingQueue<Throwable> asyncErrors = new LinkedBlockingDeque<>();
@@ -260,6 +288,33 @@ private static void verifyException(final Executable exchange, final GrpcStatusC
         assertThat(e.status().code(), is(expectedCode));
     }
 
+    private static TransportObserver onStreamClosed(final CountDownLatch latch) {
+        final StreamObserver streamObserver = new StreamObserver() {
+            @Override
+            public void streamClosed(final Throwable error) {
+                latch.countDown();
+            }
+
+            @Override
+            public void streamClosed() {
+                latch.countDown();
+            }
+        };
+        final MultiplexedObserver multiplexedObserver = new MultiplexedObserver() {
+            @Override
+            public StreamObserver onNewStream() {
+                return streamObserver;
+            }
+        };
+        final ConnectionObserver connectionObserver = new ConnectionObserver() {
+            @Override
+            public MultiplexedObserver multiplexedConnectionEstablished(final ConnectionInfo info) {
+                return multiplexedObserver;
+            }
+        };
+        return (localAddress, remoteAddress) -> connectionObserver;
+    }
+
     private static TesterService mockTesterService() {
         TesterService service = mock(TesterService.class);
         when(service.closeAsync()).thenReturn(completed());

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2019, 2021-2022 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2026 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -158,16 +158,15 @@ public void cancel() {
                         // seen by the transport before cancel. We have no way of knowing at this layer if this indeed
                         // happen. Therefore, we close the connection manually to guarantee closure.
                         //
-                        // For H2 and above, connection are multiplexed and use virtual streams for each
-                        // request-response exchange. At the time users own a Cancellable, the stream already owns
-                        // OnStreamClosedRunnable in H2ClientParentConnectionContext. It will update the concurrency
-                        // controller state if cancellation results in stream closure instead of completion.
-                        if (connectionContext().protocol().major() < 2) {
-                            LOGGER.debug("{} {} request was cancelled before receiving the full response, " +
-                                            "closing this {} connection to stop receiving more data",
-                                    connectionContext, requestMetaData, connectionContext.protocol());
-                            closeAsync().subscribe();
-                        }
+                        // For H2 and above, this connection is a single stream, so closing it resets only this
+                        // exchange. The transport resets a stream whose response is unread, but not one whose response
+                        // completed while its request body was still being written: that stream would stay open and
+                        // hold a concurrent-stream slot on both peers. OnStreamClosedRunnable in
+                        // H2ClientParentConnectionContext updates the concurrency controller when the stream closes.
+                        LOGGER.debug("{} {} request was cancelled before it completed, closing this {} {}",
+                                connectionContext, requestMetaData, connectionContext.protocol(),
+                                connectionContext.protocol().major() < 2 ? "connection" : "stream");
+                        closeAsync().subscribe();
                     }
                 })
                 .firstAndTail(this::newSplicedResponse);