From ed2446a2d3f7aeb46132a311db1054dc529830ee Mon Sep 17 00:00:00 2001 From: NianJiuZst <3235467914@qq.com> Date: Wed, 23 Sep 2026 23:10:00 +0800 Subject: [PATCH] netty: Ignore inbound frames without a transport state Guard HEADERS and DATA callbacks when the HTTP/2 stream has no associated gRPC TransportState. Return discarded DATA payload and padding to the inbound flow controller, and keep notifying keepalive for received frames. Add decoder-level regression coverage for headers, trailers, and DATA with and without END_STREAM, including buffer release and flow control. Fixes #13065. --- .../io/grpc/netty/NettyClientHandler.java | 26 +++--- .../io/grpc/netty/NettyClientHandlerTest.java | 87 ++++++++++++++++++- 2 files changed, 102 insertions(+), 11 deletions(-) diff --git a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java index 4ccc8747bb5..3a92aa2e94d 100644 --- a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java +++ b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java @@ -435,9 +435,15 @@ void returnProcessedBytes(Http2Stream stream, int bytes) { } private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream) { + if (keepAliveManager != null) { + keepAliveManager.onDataReceived(); + } // Stream 1 is reserved for the Upgrade response, so we should ignore its headers here: if (streamId != Http2CodecUtil.HTTP_UPGRADE_STREAM_ID) { NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId)); + if (stream == null) { + return; + } PerfMark.event("NettyClientHandler.onHeadersRead", stream.tag()); // check metadata size vs soft limit int h2HeadersSize = Utils.getH2HeadersSize(headers); @@ -463,23 +469,24 @@ private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream } stream.transportHeadersReceived(headers, endStream); } - - if (keepAliveManager != null) { - keepAliveManager.onDataReceived(); - } } /** * Handler for an inbound HTTP/2 DATA frame. */ - private void onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream) { + private int onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream) { flowControlPing().onDataRead(data.readableBytes(), padding); - NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId)); - PerfMark.event("NettyClientHandler.onDataRead", stream.tag()); - stream.transportDataReceived(data, endOfStream); if (keepAliveManager != null) { keepAliveManager.onDataReceived(); } + NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId)); + if (stream == null) { + // No deframer will consume this payload, so return its flow-control credit immediately. + return data.readableBytes() + padding; + } + PerfMark.event("NettyClientHandler.onDataRead", stream.tag()); + stream.transportDataReceived(data, endOfStream); + return padding; } /** @@ -1089,8 +1096,7 @@ public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) { @Override public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) throws Http2Exception { - NettyClientHandler.this.onDataRead(streamId, data, padding, endOfStream); - return padding; + return NettyClientHandler.this.onDataRead(streamId, data, padding, endOfStream); } @Override diff --git a/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java b/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java index 73050bc93d7..ad9c797c578 100644 --- a/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java @@ -84,6 +84,7 @@ import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoop; import io.netty.handler.codec.http2.DefaultHttp2Connection; +import io.netty.handler.codec.http2.DefaultHttp2FrameWriter; import io.netty.handler.codec.http2.DefaultHttp2Headers; import io.netty.handler.codec.http2.Http2Connection; import io.netty.handler.codec.http2.Http2Error; @@ -135,7 +136,10 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase setKeepaliveManagerFor = ImmutableList.of("cancelShouldSucceed", "sendFrameShouldSucceed", "channelShutdownShouldCancelBufferedStreams", - "createIncrementsIdsForActualAndBufferdStreams", "dataPingAckIsRecognized"); + "createIncrementsIdsForActualAndBufferdStreams", "dataPingAckIsRecognized", + "headersWithoutTransportStateShouldBeIgnored", "trailersWithoutTransportStateShouldBeIgnored", + "dataWithoutTransportStateShouldBeConsumed", + "dataEndStreamWithoutTransportStateShouldBeConsumed"); private Runnable tooManyPingsRunnable = new Runnable() { @Override public void run() {} }; @@ -372,6 +376,87 @@ public void sendForUnknownStreamShouldFail() throws Exception { assertFalse(future.isSuccess()); } + @Test + public void headersWithoutTransportStateShouldBeIgnored() throws Exception { + headersWithoutTransportState(false); + } + + @Test + public void trailersWithoutTransportStateShouldBeIgnored() throws Exception { + headersWithoutTransportState(true); + } + + private void headersWithoutTransportState(boolean endStream) throws Exception { + createStream(); + int streamWithoutTransportState = STREAM_ID + 2; + Http2Stream http2Stream = connection().local().createStream(streamWithoutTransportState, true); + Http2Headers headers = new DefaultHttp2Headers(); + if (endStream) { + http2Stream.headersReceived(false); + headers.set("grpc-status", "0"); + } else { + headers.status(STATUS_OK).set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC); + } + ChannelHandlerContext frameContext = newMockContext(); + new DefaultHttp2FrameWriter().writeHeaders( + frameContext, streamWithoutTransportState, headers, 0, endStream, newPromise()); + + ByteBuf frame = captureWrite(frameContext); + channelRead(frame); + + assertEquals(0, frame.refCnt()); + assertNull(lifecycleManager.getShutdownStatus()); + assertTrue(channel().isOpen()); + verify(mockKeepAliveManager).onDataReceived(); + verify(streamListener, never()).closed(any(Status.class), any(RpcProgress.class), + any(Metadata.class)); + verify(streamListener, never()).headersRead(any(Metadata.class)); + + channelRead(headersFrame(STREAM_ID, new DefaultHttp2Headers().status(STATUS_OK) + .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC))); + verify(streamListener).headersRead(any(Metadata.class)); + } + + @Test + public void dataWithoutTransportStateShouldBeConsumed() throws Exception { + dataWithoutTransportState(false); + } + + @Test + public void dataEndStreamWithoutTransportStateShouldBeConsumed() throws Exception { + dataWithoutTransportState(true); + } + + private void dataWithoutTransportState(boolean endStream) throws Exception { + createStream(); + int streamWithoutTransportState = STREAM_ID + 2; + Http2Stream http2Stream = connection().local().createStream(streamWithoutTransportState, true); + http2Stream.headersReceived(false); + Http2LocalFlowController flowController = connection().local().flowController(); + ChannelHandlerContext frameContext = newMockContext(); + ByteBuf data = content(); + new DefaultHttp2FrameWriter().writeData( + frameContext, streamWithoutTransportState, data, 5, endStream, newPromise()); + + ByteBuf frame = captureWrite(frameContext); + channelRead(frame); + + assertEquals(0, frame.refCnt()); + assertEquals(0, data.refCnt()); + assertNull(lifecycleManager.getShutdownStatus()); + assertTrue(channel().isOpen()); + assertEquals(0, flowController.unconsumedBytes(http2Stream)); + assertEquals(0, flowController.unconsumedBytes(connection().connectionStream())); + verify(mockKeepAliveManager).onDataReceived(); + verify(streamListener, never()).closed(any(Status.class), any(RpcProgress.class), + any(Metadata.class)); + verify(streamListener, never()).messagesAvailable(any(StreamListener.MessageProducer.class)); + + channelRead(headersFrame(STREAM_ID, new DefaultHttp2Headers().status(STATUS_OK) + .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC))); + verify(streamListener).headersRead(any(Metadata.class)); + } + @Test public void inboundShouldForwardToStream() throws Exception { createStream();