server: respond correctly to client headers with END_STREAM flag set #3803
Conversation
| break | ||
| } | ||
| } | ||
| if err := stream.SendMsg(nil); err == nil { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Added an error check here. The error is an RPC status error Internal.
test/end2end_test.go
Outdated
| if err != nil { | ||
| break | ||
| } |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
test/end2end_test.go
Outdated
| st.wantRSTStream(http2.ErrCodeStreamClosed) | ||
| <-sentMessage |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
test/end2end_test.go
Outdated
| @@ -4646,13 +4646,19 @@ func (s) TestClientInitialHeaderEndStream(t *testing.T) { | |||
| } | |||
|
|
|||
| 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) | |||
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Migrated to the chan struct{} pattern.
test/end2end_test.go
Outdated
| t.Error(errUnexpectedData) | ||
| return errUnexpectedData | ||
| t.Error("unexpected data received in func server method") | ||
| return nil |
There was a problem hiding this comment.
Removed the return nil in the if block
test/end2end_test.go
Outdated
| errUnexpectedData := errors.New("unexpected data received in func server method") | ||
| t.Error(errUnexpectedData) | ||
| return errUnexpectedData | ||
| t.Error("unexpected data received in func server method") |
There was a problem hiding this comment.
Print the data? It might help with debugging if it ever happens.
There was a problem hiding this comment.
Added a print of the data.
| defer func() { | ||
| handlerDone <- true | ||
| }() | ||
| <-frameCheckingDone |
There was a problem hiding this comment.
Comment here to explain why we need to block until the server has sent the RST_STREAM. And in the test below, too
There was a problem hiding this comment.
Added a comment.
| if err == io.EOF { | ||
| break | ||
| } | ||
| if err != nil { |
There was a problem hiding this comment.
Is this call not reliably failing now that we know it has sent the RST_STREAM?
There was a problem hiding this comment.
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.
test/end2end_test.go
Outdated
| defer func() { | ||
| handlerDone <- true | ||
| close(handlerDone) | ||
| }() |
There was a problem hiding this comment.
Just defer close(handlerDone) now.
There was a problem hiding this comment.
Made this change.
test/end2end_test.go
Outdated
| }() | ||
| // Block on serverTester receiving RST_STREAM. This ensures server has closed stream before stream.Recv(). |
There was a problem hiding this comment.
Please make sure to wrap comments at 80 columns for readability.
There was a problem hiding this comment.
Wrapped comments
test/end2end_test.go
Outdated
| te := newTest(t, e) | ||
| ts := &funcServer{streamingInputCall: func(stream testpb.TestService_StreamingInputCallServer) error { | ||
| defer func() { | ||
| handlerDone <- true | ||
| close(handlerDone) |
There was a problem hiding this comment.
Made this change.
dfawley
changed the title
Added logic in transport layer for appropriately handling an end stream flag set in a stream's initial http2 header frame, as well as for handling data frames on the same stream after such a frame. Added a test for testing this logic.
Fixes #2867