Skip to content
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

Jump to bottom

server: respond correctly to client headers with END_STREAM flag set #3803

Merged
merged 13 commits into from Aug 21, 2020
Merged

server: respond correctly to client headers with END_STREAM flag set #3803

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
ddfcffe
Added appropriate handling for initial header end stream flag.
GarrettGutierrez1 Aug 10, 2020
16540ed
Merge branch 'master' into end-stream-flag-bugfix
GarrettGutierrez1 Aug 10, 2020
8dee198
Added test for unexpected client data send after send and close
GarrettGutierrez1 Aug 11, 2020
823ab91
Changed test name
GarrettGutierrez1 Aug 11, 2020
b3675e1
Fixed test to check for error when sending after stream close
GarrettGutierrez1 Aug 11, 2020
c923ada
Added error code checking to RST_STREAM frame check
GarrettGutierrez1 Aug 11, 2020
cb544ea
Merge branch 'master' into end-stream-flag-bugfix
GarrettGutierrez1 Aug 11, 2020
cd4c0fa
Added sychronization and error checking to tests
GarrettGutierrez1 Aug 13, 2020
44f802d
Improved synchronization
GarrettGutierrez1 Aug 17, 2020
7b5f2ab
Merge branch 'master' into end-stream-flag-bugfix
GarrettGutierrez1 Aug 17, 2020
8f6a20f
moved to chan struct{} pattern
GarrettGutierrez1 Aug 21, 2020
9ed72b9
Merge branch 'master' into end-stream-flag-bugfix
GarrettGutierrez1 Aug 21, 2020
7d2dbd3
wrapped comments
GarrettGutierrez1 Aug 21, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions internal/transport/http2_server.go
View file
Open in desktop
Expand Up @@ -611,6 +611,10 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
if !ok {
return
}
if s.getState() == streamReadDone {
t.closeStream(s, true, http2.ErrCodeStreamClosed, false)
return
}
if size > 0 {
if err := s.fc.onData(size); err != nil {
t.closeStream(s, true, http2.ErrCodeFlowControl, false)
Expand Down
107 changes: 102 additions & 5 deletions test/end2end_test.go
View file
Open in desktop
Expand Up @@ -4535,7 +4535,7 @@ func testClientRequestBodyErrorUnexpectedEOF(t *testing.T, e env) {
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall")
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall", false)
// Say we have 5 bytes coming, but set END_STREAM flag:
st.writeData(1, true, []byte{0, 0, 0, 0, 5})
st.wantAnyFrame() // wait for server to crash (it used to crash)
Expand All @@ -4559,7 +4559,7 @@ func testClientRequestBodyErrorCloseAfterLength(t *testing.T, e env) {
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall")
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall", false)
// say we're sending 5 bytes, but then close the connection instead.
st.writeData(1, false, []byte{0, 0, 0, 0, 5})
st.cc.Close()
Expand All @@ -4582,7 +4582,7 @@ func testClientRequestBodyErrorCancel(t *testing.T, e env) {
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall")
st.writeHeadersGRPC(1, "/grpc.testing.TestService/UnaryCall", false)
// Say we have 5 bytes coming, but cancel it instead.
st.writeRSTStream(1, http2.ErrCodeCancel)
st.writeData(1, false, []byte{0, 0, 0, 0, 5})
Expand All @@ -4595,7 +4595,7 @@ func testClientRequestBodyErrorCancel(t *testing.T, e env) {
}

// And now send an uncanceled (but still invalid), just to get a response.
st.writeHeadersGRPC(3, "/grpc.testing.TestService/UnaryCall")
st.writeHeadersGRPC(3, "/grpc.testing.TestService/UnaryCall", false)
st.writeData(3, true, []byte{0, 0, 0, 0, 0})
<-gotCall
st.wantAnyFrame()
Expand All @@ -4619,7 +4619,7 @@ func testClientRequestBodyErrorCancelStreamingInput(t *testing.T, e env) {
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
st.writeHeadersGRPC(1, "/grpc.testing.TestService/StreamingInputCall")
st.writeHeadersGRPC(1, "/grpc.testing.TestService/StreamingInputCall", false)
// Say we have 5 bytes coming, but cancel it instead.
st.writeData(1, false, []byte{0, 0, 0, 0, 5})
st.writeRSTStream(1, http2.ErrCodeCancel)
Expand All @@ -4636,6 +4636,103 @@ func testClientRequestBodyErrorCancelStreamingInput(t *testing.T, e env) {
})
}

func (s) TestClientInitialHeaderEndStream(t *testing.T) {
dfawley marked this conversation as resolved.
Show resolved Hide resolved
for _, e := range listTestEnv() {
if e.httpHandler {
continue
}
testClientInitialHeaderEndStream(t, e)
}
}

func testClientInitialHeaderEndStream(t *testing.T, e env) {
// To ensure RST_STREAM is sent for illegal data write and not normal stream close.
frameCheckingDone := make(chan bool, 1)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The typical way to implement this pattern is to make it a: chan struct{} (with no buffer) and close the channel when you want to indicate the thing happened.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Migrated to the chan struct{} pattern.

// To ensure goroutine for test does not end before RPC handler performs error checking.
handlerDone := make(chan bool, 1)
te := newTest(t, e)
ts := &funcServer{streamingInputCall: func(stream testpb.TestService_StreamingInputCallServer) error {
defer func() {
handlerDone <- true
}()
<-frameCheckingDone
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment here to explain why we need to block until the server has sent the RST_STREAM. And in the test below, too

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a comment.

_, err := stream.Recv()
if err == nil {
t.Error("unexpected data received in func server method")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Print the data? It might help with debugging if it ever happens.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a print of the data.

return nil
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is unnecessary.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed the return nil in the if block

} else if status.Code(err) != codes.Canceled {
t.Errorf("expected status canceled error, instead received '%v'", err)
}
return nil
}}
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
// Send a headers with END_STREAM flag, but then write data.
st.writeHeadersGRPC(1, "/grpc.testing.TestService/StreamingInputCall", true)
st.writeData(1, false, []byte{0, 0, 0, 0, 0})
st.wantAnyFrame()
st.wantAnyFrame()
st.wantRSTStream(http2.ErrCodeStreamClosed)
dfawley marked this conversation as resolved.
Show resolved Hide resolved
frameCheckingDone <- true
<-handlerDone
})
}

func (s) TestClientSendDataAfterCloseSend(t *testing.T) {
for _, e := range listTestEnv() {
if e.httpHandler {
continue
}
testClientSendDataAfterCloseSend(t, e)
}
}

func testClientSendDataAfterCloseSend(t *testing.T, e env) {
// To ensure RST_STREAM is sent for illegal data write prior to execution of RPC handler.
frameCheckingDone := make(chan bool, 1)
// To ensure goroutine for test does not end before RPC handler performs error checking.
handlerDone := make(chan bool, 1)
te := newTest(t, e)
ts := &funcServer{streamingInputCall: func(stream testpb.TestService_StreamingInputCallServer) error {
defer func() {
handlerDone <- true
}()
<-frameCheckingDone
for {
_, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
Comment on lines +4707 to +4710
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this call not reliably failing now that we know it has sent the RST_STREAM?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's correct. It always returns with err != nil on the first call but sometimes it is io.EOF and sometimes it is a cancelled error.

if status.Code(err) != codes.Canceled {
t.Errorf("expected status canceled error, instead received '%v'", err)
}
break
}
}
if err := stream.SendMsg(nil); err == nil {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's test the actual error we expect to get from the stream here instead of "any error is fine".

I have a feeling it will be an RPC status error now, but maybe that's not right (#3575). We can just expect the wrong thing for now and leave a comment that it will need to be updated when #3575 is fixed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added an error check here. The error is an RPC status error Internal.

t.Error("expected error sending message on stream after stream closed due to illegal data")
} else if status.Code(err) != codes.Internal {
t.Errorf("expected internal error, instead received '%v'", err)
}
return nil
}}
te.startServer(ts)
defer te.tearDown()
te.withServerTester(func(st *serverTester) {
st.writeHeadersGRPC(1, "/grpc.testing.TestService/StreamingInputCall", false)
// Send data with END_STREAM flag, but then write more data.
st.writeData(1, true, []byte{0, 0, 0, 0, 0})
st.writeData(1, false, []byte{0, 0, 0, 0, 0})
st.wantAnyFrame()
st.wantAnyFrame()
st.wantRSTStream(http2.ErrCodeStreamClosed)
frameCheckingDone <- true
<-handlerDone
})
}

func (s) TestClientResourceExhaustedCancelFullDuplex(t *testing.T) {
for _, e := range listTestEnv() {
if e.httpHandler {
Expand Down
19 changes: 17 additions & 2 deletions test/servertester.go
View file
Open in desktop
Expand Up @@ -138,6 +138,21 @@ func (st *serverTester) writeSettingsAck() {
}
}

func (st *serverTester) wantRSTStream(errCode http2.ErrCode) *http2.RSTStreamFrame {
f, err := st.readFrame()
if err != nil {
st.t.Fatalf("Error while expecting an RST frame: %v", err)
}
sf, ok := f.(*http2.RSTStreamFrame)
if !ok {
st.t.Fatalf("got a %T; want *http2.RSTStreamFrame", f)
}
if sf.ErrCode != errCode {
st.t.Fatalf("expected RST error code '%v', got '%v'", errCode.String(), sf.ErrCode.String())
}
return sf
}

func (st *serverTester) wantSettings() *http2.SettingsFrame {
f, err := st.readFrame()
if err != nil {
Expand Down Expand Up @@ -227,7 +242,7 @@ func (st *serverTester) encodeHeader(headers ...string) []byte {
return st.headerBuf.Bytes()
}

func (st *serverTester) writeHeadersGRPC(streamID uint32, path string) {
func (st *serverTester) writeHeadersGRPC(streamID uint32, path string, endStream bool) {
st.writeHeaders(http2.HeadersFrameParam{
StreamID: streamID,
BlockFragment: st.encodeHeader(
Expand All @@ -236,7 +251,7 @@ func (st *serverTester) writeHeadersGRPC(streamID uint32, path string) {
"content-type", "application/grpc",
"te", "trailers",
),
EndStream: false,
EndStream: endStream,
EndHeaders: true,
})
}
Expand Down