netty: reset stream when trailers arrive with END_STREAM unsent

netty/src/main/java/io/grpc/netty/NettyClientStream.java

@@ -342,7 +342,7 @@ public void deframeFailed(Throwable cause) {
 
     void transportHeadersReceived(Http2Headers headers, boolean endOfStream) {
       if (endOfStream) {
-        if (!isOutboundClosed()) {
+        if (!isOutboundClosed() || (http2Stream != null && http2Stream.state().localSideOpen())) {
           handler.getWriteQueue().enqueue(new CancelClientStreamCommand(this, null), true);
         }
         transportTrailersReceived(Utils.convertTrailers(headers));

netty/src/test/java/io/grpc/netty/NettyClientStreamTest.java

@@ -68,6 +68,7 @@
 import io.netty.handler.codec.http2.DefaultHttp2Headers;
 import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.util.AsciiString;
 import java.io.BufferedInputStream;
 import java.io.ByteArrayInputStream;
@@ -331,6 +332,7 @@ public void inboundTrailersAfterHalfCloseDoesNotSendRstStream() {
     stream().transportState().setId(STREAM_ID);
     stream().transportState().transportHeadersReceived(grpcResponseHeaders(), false);
     stream.halfClose();
+    when(http2Stream.state()).thenReturn(Http2Stream.State.HALF_CLOSED_LOCAL);
     stream().transportState().transportHeadersReceived(grpcResponseTrailers(Status.OK), true);
     verify(writeQueue, never()).enqueue(isA(CancelClientStreamCommand.class), eq(true));
   }

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -96,6 +96,7 @@
 import io.netty.channel.socket.SocketChannelConfig;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.http2.Http2Connection;
 import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.handler.codec.http2.StreamBufferingEncoder;
 import io.netty.handler.ssl.ClientAuth;
@@ -222,6 +223,111 @@ public void addDefaultUserAgent() throws Exception {
     assertEquals(GrpcUtil.getGrpcUserAgent("netty", null), headers.get(USER_AGENT_KEY));
   }
 
+  /**
+   * The server responds and closes the call before it has read the whole request, and then stops
+   * returning flow-control window. The client has already half-closed, so its END_STREAM frame is
+   * stuck in Netty's remote flow controller and can never be written. The client must reset the
+   * stream; otherwise the HTTP/2 stream stays open in the connection forever and keeps the whole
+   * call reachable.
+   */
+  @Test
+  public void earlyServerResponseBlockedEndOfStreamShouldNotLeakHttp2Stream() throws Exception {
+    // A server that answers every call immediately. It never requests any of the request stream,
+    // so it never returns window.
+    startServer(
+        new ServerListener() {
+          @Override
+          public ServerTransportListener transportCreated(ServerTransport transport) {
+            return new ServerTransportListener() {
+              @Override
+              public void streamCreated(ServerStream stream, String method, Metadata headers) {
+                stream.setListener(
+                    new ServerStreamListener() {
+                      @Override
+                      public void messagesAvailable(MessageProducer producer) {}
+
+                      @Override
+                      public void onReady() {}
+
+                      @Override
+                      public void halfClosed() {}
+
+                      @Override
+                      public void closed(Status status) {}
+
+                      @Override
+                      public void triggerEvent(Object event) {}
+                    });
+                stream.writeHeaders(new Metadata(), false);
+                stream.writeMessage(new ByteArrayInputStream(Rpc.MESSAGE.getBytes(UTF_8)));
+                stream.flush();
+                stream.close(Status.OK, new Metadata());
+              }
+
+              @Override
+              public Attributes transportReady(Attributes transportAttrs) {
+                return transportAttrs;
+              }
+
+              @Override
+              public void transportTerminated() {}
+            };
+          }
+
+          @Override
+          public void serverShutdown() {}
+        });
+    NettyClientTransport transport = newTransport(newNegotiator());
+    callMeMaybe(transport.start(clientTransportListener));
+
+    ClientStream stream =
+        transport.newStream(
+            Rpc.METHOD,
+            new Metadata(),
+            CallOptions.DEFAULT,
+            new ClientStreamTracer[] {new ClientStreamTracer() {}});
+    TestClientStreamListener listener = new TestClientStreamListener();
+    stream.start(listener);
+    stream.request(1);
+    // Larger than the initial stream window, so the tail of the request stays queued in Netty's
+    // remote flow controller.
+    stream.writeMessage(new ByteArrayInputStream(new byte[2 * DEFAULT_WINDOW_SIZE]));
+    stream.flush();
+    // Half-close while that tail is still queued, so END_STREAM is queued behind it. The server
+    // needs a round trip to answer, so this always happens before its trailers arrive.
+    stream.halfClose();
+
+    listener.responseFuture.get(10, TimeUnit.SECONDS);
+    listener.closedFuture.get(10, TimeUnit.SECONDS);
+
+    assertThat(activeStreams(transport)).isEmpty();
+  }
+
+  /** Describes the HTTP/2 streams still open in the client's connection. */
+  private static List<String> activeStreams(NettyClientTransport transport) throws Exception {
+    NettyClientHandler handler = transport.channel().pipeline().get(NettyClientHandler.class);
+    return transport
+        .channel()
+        .eventLoop()
+        .submit(
+            () -> {
+              List<String> descriptions = new ArrayList<>();
+              Http2Connection connection = handler.connection();
+              connection.forEachActiveStream(
+                  stream -> {
+                    descriptions.add(
+                        String.format(
+                            "stream %d is %s with remote flow control window %d",
+                            stream.id(),
+                            stream.state(),
+                            connection.remote().flowController().windowSize(stream)));
+                    return true;
+                  });
+              return descriptions;
+            })
+        .get(10, TimeUnit.SECONDS);
+  }
+
   @Test
   public void setSoLingerChannelOption() throws IOException, GeneralSecurityException {
     startServer();