diff --git a/netty/src/main/java/io/grpc/netty/NettyClientStream.java b/netty/src/main/java/io/grpc/netty/NettyClientStream.java index 2939eed2e37..c55b5da2f0b 100644 --- a/netty/src/main/java/io/grpc/netty/NettyClientStream.java +++ b/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)); diff --git a/netty/src/test/java/io/grpc/netty/NettyClientStreamTest.java b/netty/src/test/java/io/grpc/netty/NettyClientStreamTest.java index 4dd24c3fd4d..145d94b55b4 100644 --- a/netty/src/test/java/io/grpc/netty/NettyClientStreamTest.java +++ b/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)); } diff --git a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java index 8cb5cca9861..f6957348746 100644 --- a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java +++ b/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 activeStreams(NettyClientTransport transport) throws Exception { + NettyClientHandler handler = transport.channel().pipeline().get(NettyClientHandler.class); + return transport + .channel() + .eventLoop() + .submit( + () -> { + List 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();