transport: Propagate status code on receiving RST_STREAM during message read

[arjan-bal]

Fixes: #8281

RCA: #8281 (comment)

This PR uses a status error instead of an io.UnexpectedEOF to fail RPCs that have partially read a gRPC message.

RELEASE NOTES:

• grpc: Fix bug that causes RPCs to fail with status INTERNAL instead of CANCELLED or DEADLINE_EXCEEDED when the client receives an RST_STREAM frame while reading a gRPC message.

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 82.27%. Comparing base (515f377) to head (4bf921e).
Report is 25 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #8289      +/-   ##
==========================================
+ Coverage   82.14%   82.27%   +0.13%     
==========================================
  Files         417      419       +2     
  Lines       41344    41992     +648     
==========================================
+ Hits        33961    34550     +589     
- Misses       5957     5982      +25     
- Partials     1426     1460      +34     

Files with missing lines	Coverage Δ
internal/transport/http2_client.go	92.37% <100.00%> (-0.24%)	⬇️

... and 57 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[arjan-bal]

@dfawley I've changed the code to use a status error for unblocking the reader. This bypasses the conversion to io.ErrUnexpectedEOF and doesn't require reading the status in csAttempt.recvMsg().

[dfawley]

Where is that conversion happening?

[arjan-bal]

Here:

grpc-go/rpc_util.go

Lines 674 to 680 in a43eba6

	data, err := p.r.Read(int(length))
	if err != nil {
	if err == io.EOF {
	err = io.ErrUnexpectedEOF
	}
	return 0, nil, err
	}

[dfawley]

Where is that conversion happening?

[dfawley]

Why is this in a goroutine, but the ping ack is written inline?

Also...can we just not reply to the ping and then remove the lock?

[arjan-bal]

Removed the ping handling and the mutex. Removed the separate goroutine used for writing data frames.
The removed things were unnecessary. I copied the loop that reads frames from an existing test and didn't remove the extra stuff.

grpc-go/internal/transport/transport_test.go

Lines 1414 to 1450 in 950a7cf

	for {
	frame, err := sfr.ReadFrame()
	if err != nil {
	return
	}
	switch frame := frame.(type) {
	case *http2.HeadersFrame:
	// When the client creates a stream, violate the stream flow control.
	go func() {
	buf := make([]byte, http2MaxFrameLen)
	for {
	mu.Lock()
	if err := sfr.WriteData(1, false, buf); err != nil {
	mu.Unlock()
	return
	}
	mu.Unlock()
	// This for loop is capable of hogging the CPU and cause starvation
	// in Go versions prior to 1.9,
	// in single CPU environment. Explicitly relinquish processor.
	runtime.Gosched()
	}
	}()
	case *http2.RSTStreamFrame:
	if frame.Header().StreamID != 1 || http2.ErrCode(frame.ErrCode) != http2.ErrCodeFlowControl {
	t.Errorf("RST stream received with streamID: %d and code: %v, want streamID: 1 and code: http2.ErrCodeFlowControl", frame.Header().StreamID, http2.ErrCode(frame.ErrCode))
	}
	close(success)
	return
	case *http2.PingFrame:
	mu.Lock()
	sfr.WritePing(true, frame.Data)
	mu.Unlock()
	default:
	}
	}
	}()

[dfawley]

const?