Skip to content

fix: surface data when received non-grpc header, end stream without trailers #9217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chengxilo wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix: surface data when received non-grpc header, end stream without trailers #9217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions internal/transport/http2_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
142 changes: 142 additions & 0 deletions internal/transport/transport_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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("<html>bad gateway</html>")); 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: "<html>bad gateway</html>",
},
}

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 {
}

Expand Down
59 changes: 52 additions & 7 deletions test/end2end_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -6820,6 +6829,7 @@ func (s) TestAuthorityHeader(t *testing.T) {
}

func (s) TestHTTPServerSendsNonGRPCHeaderSurfaceFurtherData(t *testing.T) {
const htmlPayload = `<html><body>Hello World</body></html>`
const nonGRPCDataMaxLen = 1024
tests := []struct {
name string
Expand Down Expand Up @@ -6853,12 +6863,12 @@ func (s) TestHTTPServerSendsNonGRPCHeaderSurfaceFurtherData(t *testing.T) {
"content-type", "text/html",
},
},
payload: []byte(`<html><body>Hello World</body></html>`),
payload: []byte(htmlPayload),
},
},
wantCode: codes.Unknown,
wantErr: `unexpected HTTP status code received from server: 200 (OK); transport: received unexpected content-type "text/html"
data: "<html><body>Hello World</body></html>"`,
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",
Expand Down Expand Up @@ -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 {
Expand Down
Loading