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 7 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
84 changes: 79 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,80 @@ 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) {
te := newTest(t, e)
ts := &funcServer{streamingInputCall: func(stream testpb.TestService_StreamingInputCallServer) error {
_, err := stream.Recv()
if err == nil {
errUnexpectedData := errors.New("unexpected data received in func server method")
t.Error(errUnexpectedData)
return errUnexpectedData
}
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
})
}

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

func testClientSendDataAfterCloseSend(t *testing.T, e env) {
sentMessage := make(chan bool, 1)
te := newTest(t, e)
ts := &funcServer{streamingInputCall: func(stream testpb.TestService_StreamingInputCallServer) error {
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.

break
}
Copy link
Member

Choose a reason for hiding this comment

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

What is err when this happens? Maybe we should change that? I'm guessing this is a race between Recv returning before/after the illegal data frame is written.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The err for this when logged reads as rpc error: code = Canceled desc = context canceled. Sometimes Recv returns an err and sometimes the loop breaks because of err == io.EOF. That does sound like a race. Most of the time Recv returns the canceled desc err.

}
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 on illegal data")
}
sentMessage <- true
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)
<-sentMessage
Copy link
Member

Choose a reason for hiding this comment

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

Without this synchronization you wouldn't see the RST_STREAM at all sometimes? That sounds wrong. The server should still send a RST_STREAM even after the stream is terminated on the server's side.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Without the synchronization the handler would sometimes not encounter an error. I think this was because the goroutine for the test would finish before the goroutine for the handler would encounter an error condition.

})
}

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 *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