transport: fix server Recv hang on initial HEADERS with EndStream (#9391)

Fix server-side Recv hang when client sends HEADERS with EndStream set.

When a client initiates a stream with a HEADERS frame that has
EndStream=true, the server-side stream state is set to streamReadDone,
but the reader was not notified. This caused server-side Recv() calls to
hang indefinitely waiting for data.

This change writes an io.EOF error to the stream's receive buffer when
the stream is created with EndStream=true, ensuring Recv() returns
immediately with io.EOF. A new transport test has been added to verify
this behavior, and existing tests have been updated.

RELEASE NOTES:
* transport: fix an issue where server-side `Recv()` hung indefinitely
when a client initiated a stream with `HEADERS` having `EndStream` set.

Author: eshitachandwani

internal/transport/http2_server.go

@@ -531,6 +531,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	if frame.StreamEnded() {
 		// s is just created by the caller. No lock needed.
 		s.state = streamReadDone
+		s.write(recvMsg{err: io.EOF})
 	}
 	if timeoutSet {
 		s.ctx, s.cancel = context.WithTimeout(ctx, timeout)

test/end2end_test.go

@@ -4765,14 +4765,20 @@ func testClientInitialHeaderEndStream(t *testing.T, e env) {
 	te := newTest(t, e)
 	ts := &funcServer{streamingInputCall: func(stream testgrpc.TestService_StreamingInputCallServer) error {
 		defer close(handlerDone)
-		// Block on serverTester receiving RST_STREAM. This ensures server has closed
-		// stream before stream.Recv().
+		// Block on serverTester receiving RST_STREAM. This ensures server has
+		// closed stream before stream.Recv().
 		<-frameCheckingDone
-		data, err := stream.Recv()
-		if err == nil {
-			t.Errorf("unexpected data received in func server method: '%v'", data)
+		// Depending on whether the context cancellation (due to the illegal data
+		// RST_STREAM) or the buffered EOF (from the initial HEADERS END_STREAM) is
+		// selected first in recvBufferReader, stream.Recv() can return either
+		// io.EOF or Canceled.
+		if _, err := stream.Recv(); err != io.EOF && status.Code(err) != codes.Canceled {
+			t.Errorf("stream.Recv() returned error = %v, expected EOF or canceled error", err)
+		}
+		if err := stream.SendMsg(nil); err == nil {
+			t.Error("stream.SendMsg() returned nil, expected cancel error")
 		} else if status.Code(err) != codes.Canceled {
-			t.Errorf("expected canceled error, instead received '%v'", err)
+			t.Errorf("stream.SendMsg() returned error = %v, expected cancel error", err)
 		}
 		return nil
 	}}

test/transport_test.go

@@ -24,6 +24,7 @@ import (
 	"net"
 	"sync"
 	"testing"
+	"time"
 
 	"golang.org/x/net/http2"
 	"google.golang.org/grpc"
@@ -302,3 +303,53 @@ func (s) TestCancelWhileServerWaitingForFlowControl(t *testing.T) {
 		t.Fatalf("Failed to read from the stream: %v", err)
 	}
 }
+
+// Tests that when a client sends a HEADERS frame with EndStream=true, the
+// server-side stream receives an io.EOF on Recv() and does not hang waiting
+// for data frames.
+func (s) TestHeadersEndStreamNoHang(t *testing.T) {
+	receivedErr := make(chan error, 1)
+	ss := &stubserver.StubServer{
+		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+			_, err := stream.Recv()
+			receivedErr <- err
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	conn, err := net.DialTimeout("tcp", ss.Address, defaultTestTimeout)
+	if err != nil {
+		t.Fatalf("Failed to dial: %v", err)
+	}
+	defer conn.Close()
+
+	st := newServerTesterFromConn(t, conn)
+	st.greet()
+
+	// Send HEADERS with EndStream = true and no grpc-timeout header.
+	st.writeHeaders(http2.HeadersFrameParam{
+		StreamID: 1,
+		BlockFragment: st.encodeHeader(
+			":method", "POST",
+			":path", "/grpc.testing.TestService/FullDuplexCall",
+			":authority", "localhost",
+			"content-type", "application/grpc",
+			"te", "trailers",
+		),
+		EndStream:  true,
+		EndHeaders: true,
+	})
+
+	select {
+	case err := <-receivedErr:
+		if err != io.EOF {
+			t.Errorf("Streaming handler returned error = %v, expected io.EOF", err)
+		}
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timed out waiting for Recv() on the server to complete")
+	}
+}