diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go index 0990df425741..e555e3950eeb 100644 --- a/internal/transport/http2_client.go +++ b/internal/transport/http2_client.go @@ -1258,6 +1258,13 @@ func (t *http2Client) handleData(f *parsedDataFrame) { // The server has closed the stream without sending trailers. Record that // the read direction is closed, and set the status appropriately. if f.StreamEnded() { + // If we were collecting non-gRPC response data, finalize that status with + // whatever body we've buffered so far instead of discarding it. + if s.nonGRPCStatus != nil { + st := s.finalizeNonGRPCStatus() + t.closeStream(s, st.Err(), true, http2.ErrCodeProtocol, st, nil, true) + return + } // If client received END_STREAM from server while stream was still // active, send RST_STREAM. rstStream := s.getState() == streamActive diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go index 79f72ec24ff3..55f9acfd1d37 100644 --- a/internal/transport/transport_test.go +++ b/internal/transport/transport_test.go @@ -4045,6 +4045,148 @@ func (s) TestDeleteStreamMetricsIncrementedOnlyOnce(t *testing.T) { } } +// Tests that when a non-gRPC response is followed by an empty DATA frame with +// END_STREAM, the client surfaces the original non-gRPC error instead of discard +// the buffer we collected and returning an internal error with "server closed the +// stream without sending trailers". +func (s) TestNonGRPCStatus_EmptyDataEndStream(t *testing.T) { + tests := []struct { + name string + send func(t *testing.T, framer *http2.Framer, streamID uint32) + wantCode codes.Code + wantSubstr string + }{ + { + name: "headers then empty DATA END_STREAM", + send: func(t *testing.T, framer *http2.Framer, streamID uint32) { + var buf bytes.Buffer + henc := hpack.NewEncoder(&buf) + henc.WriteField(hpack.HeaderField{Name: ":status", Value: "401"}) + henc.WriteField(hpack.HeaderField{Name: "content-type", Value: "text/html"}) + if err := framer.WriteHeaders(http2.HeadersFrameParam{ + StreamID: streamID, + BlockFragment: buf.Bytes(), + EndHeaders: true, + EndStream: false, + }); err != nil { + t.Fatalf("Failed to write headers: %v", err) + } + if err := framer.WriteData(streamID, true, nil); err != nil { + t.Fatalf("Failed to write empty DATA: %v", err) + } + }, + wantCode: codes.Unauthenticated, + wantSubstr: "unexpected HTTP status code received from server: 401", + }, + { + name: "headers then body DATA then empty DATA END_STREAM", + send: func(t *testing.T, framer *http2.Framer, streamID uint32) { + var buf bytes.Buffer + henc := hpack.NewEncoder(&buf) + henc.WriteField(hpack.HeaderField{Name: ":status", Value: "502"}) + henc.WriteField(hpack.HeaderField{Name: "content-type", Value: "text/html"}) + if err := framer.WriteHeaders(http2.HeadersFrameParam{ + StreamID: streamID, + BlockFragment: buf.Bytes(), + EndHeaders: true, + EndStream: false, + }); err != nil { + t.Fatalf("Failed to write headers: %v", err) + } + if err := framer.WriteData(streamID, false, []byte("bad gateway")); err != nil { + t.Fatalf("Failed to write body DATA: %v", err) + } + if err := framer.WriteData(streamID, true, nil); err != nil { + t.Fatalf("Failed to write empty DATA: %v", err) + } + }, + wantCode: codes.Unavailable, + wantSubstr: "bad gateway", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + lis, err := net.Listen("tcp", "localhost:0") + if err != nil { + t.Fatalf("Failed to listen: %v", err) + } + t.Cleanup(func() { lis.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + seenHeaders := make(chan uint32, 1) + go func() { + conn, err := lis.Accept() + if err != nil { + t.Errorf("Failed to accept: %v", err) + return + } + defer conn.Close() + + if _, err := io.ReadFull(conn, make([]byte, len(clientPreface))); err != nil { + t.Errorf("Failed to read client preface: %v", err) + return + } + framer := http2.NewFramer(conn, conn) + frame, err := framer.ReadFrame() + if err != nil { + t.Errorf("Failed to read SETTINGS: %v", err) + return + } + if _, ok := frame.(*http2.SettingsFrame); !ok { + t.Errorf("Want SETTINGS, got %T", frame) + return + } + if err := framer.WriteSettings(); err != nil { + t.Errorf("Failed to write SETTINGS: %v", err) + return + } + if err := framer.WriteSettingsAck(); err != nil { + t.Errorf("Failed to write SETTINGS ACK: %v", err) + return + } + + for { + frame, err = framer.ReadFrame() + if err != nil { + return + } + if hf, ok := frame.(*http2.HeadersFrame); ok { + seenHeaders <- hf.StreamID + break + } + } + + streamID := <-seenHeaders + tc.send(t, framer, streamID) + }() + + copts := ConnectOptions{BufferPool: mem.DefaultBufferPool()} + ct, err := NewHTTP2Client(ctx, ctx, resolver.Address{Addr: lis.Addr().String()}, copts, func(GoAwayInfo) {}) + if err != nil { + t.Fatalf("NewHTTP2Client: %v", err) + } + t.Cleanup(func() { ct.Close(errors.New("test done")) }) + + stream, err := ct.NewStream(ctx, &CallHdr{}, nil) + if err != nil { + t.Fatalf("NewStream: %v", err) + } + + <-stream.Done() + st := stream.Status() + if st.Code() != tc.wantCode { + t.Errorf("Status code = %v, want %v (message: %v)", st.Code(), tc.wantCode, st.Message()) + } + if !strings.Contains(st.Message(), tc.wantSubstr) { + t.Errorf("Status message = %q, want substring %q", st.Message(), tc.wantSubstr) + } + }) + } +} + type fakeReadRequester struct { } diff --git a/test/end2end_test.go b/test/end2end_test.go index c67f3cbd38d3..d7f0ac47a2f7 100644 --- a/test/end2end_test.go +++ b/test/end2end_test.go @@ -6340,9 +6340,10 @@ func (s) TestRPCWaitsForResolver(t *testing.T) { } type httpServerResponse struct { - headers [][]string - payload []byte - trailers [][]string + headers [][]string + payload []byte + trailers [][]string + separateEndStream bool // send payload without END_STREAM, then an empty DATA with END_STREAM } type httpServer struct { @@ -6451,11 +6452,19 @@ func (s *httpServer) start(t *testing.T, lis net.Listener) { writer.Flush() } if hasPayload { - if err = s.writePayload(framer, sid, response.payload, !hasTrailers); err != nil { + endStream := !hasTrailers && !response.separateEndStream + if err = s.writePayload(framer, sid, response.payload, endStream); err != nil { t.Errorf("Error at server-side while writing payload. Err: %v", err) return } writer.Flush() + if response.separateEndStream && !hasTrailers { + if err = s.writePayload(framer, sid, nil, true); err != nil { + t.Errorf("Error at server-side while writing empty END_STREAM DATA. Err: %v", err) + return + } + writer.Flush() + } } for i, trailer := range response.trailers { if err = s.writeHeader(framer, sid, trailer, i == len(response.trailers)-1); err != nil { @@ -6820,6 +6829,7 @@ func (s) TestAuthorityHeader(t *testing.T) { } func (s) TestHTTPServerSendsNonGRPCHeaderSurfaceFurtherData(t *testing.T) { + const htmlPayload = `Hello World` const nonGRPCDataMaxLen = 1024 tests := []struct { name string @@ -6853,12 +6863,12 @@ func (s) TestHTTPServerSendsNonGRPCHeaderSurfaceFurtherData(t *testing.T) { "content-type", "text/html", }, }, - payload: []byte(`Hello World`), + payload: []byte(htmlPayload), }, }, wantCode: codes.Unknown, - wantErr: `unexpected HTTP status code received from server: 200 (OK); transport: received unexpected content-type "text/html" -data: "Hello World"`, + wantErr: fmt.Sprintf(`unexpected HTTP status code received from server: 200 (OK); transport: received unexpected content-type "text/html" +data: %q`, htmlPayload), }, { name: "non-gRPC content-type with bytes payload length more than nonGRPCDataMaxLen", @@ -6891,6 +6901,41 @@ data: ` + strconv.Quote(strings.Repeat("a", nonGRPCDataMaxLen)), wantErr: `unexpected HTTP status code received from server: 502 (Bad Gateway); malformed header: missing HTTP content-type data: "hello"`, }, + { + name: "non-gRPC content-type with empty DATA END_STREAM", + responses: []httpServerResponse{ + { + headers: [][]string{ + { + ":status", "401", + "content-type", "text/html", + }, + }, + payload: []byte{}, + }, + }, + wantCode: codes.Unauthenticated, + wantErr: `unexpected HTTP status code received from server: 401 (Unauthorized); transport: received unexpected content-type "text/html" +data: ""`, + }, + { + name: "non-gRPC content-type with payload and separate END_STREAM", + responses: []httpServerResponse{ + { + headers: [][]string{ + { + ":status", "401", + "content-type", "text/html", + }, + }, + payload: []byte(htmlPayload), + separateEndStream: true, + }, + }, + wantCode: codes.Unauthenticated, + wantErr: fmt.Sprintf(`unexpected HTTP status code received from server: 401 (Unauthorized); transport: received unexpected content-type "text/html" +data: %q`, htmlPayload), + }, } for _, test := range tests {