netty: Associate netty stream eagerly to avoid client hang

In #12185, RPCs were randomly hanging. In #12207 this was tracked down
to the headers promise completing successfully, but the netty stream
was null. This was because the headers write hadn't completed but
stream.close() had been called by goingAway().
This commit is contained in:
Eric Anderson 2025-07-15 15:00:24 -07:00
parent a37d3eb349
commit 2e96fbf1e8
2 changed files with 33 additions and 0 deletions

View File

@ -773,6 +773,19 @@ class NettyClientHandler extends AbstractNettyHandler {
}
}
});
// When the HEADERS are not buffered because of MAX_CONCURRENT_STREAMS in
// StreamBufferingEncoder, the stream is created immediately even if the bytes of the HEADERS
// are delayed because the OS may have too much buffered and isn't accepting the write. The
// write promise is also delayed until flush(). However, we need to associate the netty stream
// with the transport state so that goingAway() and forcefulClose() and able to notify the
// stream of failures.
//
// This leaves a hole when MAX_CONCURRENT_STREAMS is reached, as http2Stream will be null, but
// it is better than nothing.
Http2Stream http2Stream = connection().stream(streamId);
if (http2Stream != null) {
http2Stream.setProperty(streamKey, stream);
}
}
/**

View File

@ -453,6 +453,26 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase<NettyClientHand
assertTrue(future.isDone());
}
@Test
public void receivedAbruptGoAwayShouldFailRacingQueuedIoStreamid() throws Exception {
// Purposefully avoid flush(), since we want the write to not actually complete.
// EmbeddedChannel doesn't support flow control, so this is the next closest approximation.
ChannelFuture future = channel().write(
newCreateStreamCommand(grpcHeaders, streamTransportState));
// Read a GOAWAY that indicates our stream can't be sent
channelRead(goAwayFrame(0, 0 /* NO_ERROR */, Unpooled.copiedBuffer("this is a test", UTF_8)));
ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
verify(streamListener).closed(captor.capture(), same(REFUSED),
ArgumentMatchers.<Metadata>notNull());
assertEquals(Status.UNAVAILABLE.getCode(), captor.getValue().getCode());
assertEquals(
"Abrupt GOAWAY closed sent stream. HTTP/2 error code: NO_ERROR, "
+ "debug data: this is a test",
captor.getValue().getDescription());
assertTrue(future.isDone());
}
@Test
public void receivedGoAway_shouldFailBufferedStreamsExceedingMaxConcurrentStreams()
throws Exception {