Long running streams fail

[daveoxley]

What version of gRPC are you using?

1.66.2
But seen on all versions of 1.66.x so far. Issue did not occur on 1.65

What version of Go are you using (go version)?

N/A

What operating system (Linux, Windows, …) and version?

N/A

What did you do?

Long running stream fails with an error. Test case in #7642.

What did you expect to see?

Stream completing successfully:

=== RUN   TestLongRunningStream
--- PASS: TestLongRunningStream (127.39s)
PASS

What did you see instead?

Errors vary between the 2 below:

=== RUN   TestLongRunningStream
    stream_test.go:70: Failed to receive response: rpc error: code = Internal desc = received 4294967270-bytes data exceeding the limit 65535 bytes
--- FAIL: TestLongRunningStream (17.32s)

and

=== RUN   TestLongRunningStream
    stream_test.go:70: Failed to receive response: rpc error: code = Internal desc = unexpected EOF
--- FAIL: TestLongRunningStream (8.14s)

[arjan-bal]

After using git bisect on the range from tag 1.65.0-dev to HEAD of master, the commit causing the regression seems to be 9ab8b62

Command:

# Run 10 times in parallel for higher confidence
seq 1 10 | xargs -n1 -P10 -I{}  go test ./test/stream/ -run TestLongRunningStream -count 1 -v || echo "A test failed"

I then verified that the test flakes on 9ab8b62 and passed 30/30 times on the commit immediately before it.

[zhouyan]

I have been seeing the same problems as well since upgrade to 1.66 and beyond.

the error I got was control flow error, saying received data exceeding limit. The received data is around 4GB according to the logs.

[arjan-bal]

It seems like the amount of bytes being read > the amount of bytes received. This is causing an unsigned 32 bit integer to underflow resulting in 4294967270 which is -26.

I think I found the bug. The headers are being read incrementally here:

grpc-go/internal/transport/transport.go

Lines 559 to 579 in 8ea3460

	func (s *Stream) ReadHeader(header []byte) (err error) {
	// Don't request a read if there was an error earlier
	if er := s.trReader.er; er != nil {
	return er
	}
	s.requestRead(len(header))
	for len(header) != 0 {
	n, err := s.trReader.ReadHeader(header)
	header = header[n:]
	if len(header) == 0 {
	err = nil
	}
	if err != nil {
	if n > 0 && err == io.EOF {
	err = io.ErrUnexpectedEOF
	}
	return err
	}
	}
	return nil
	}

We keep filling the underlying header slice until its full, but we do so by moving the start of the slice forward. In the following function that is called by the function above to perform reads and update the flow control window, we do t.windowHandler(len(header)).

grpc-go/internal/transport/transport.go

Lines 622 to 630 in 8ea3460

	func (t *transportReader) ReadHeader(header []byte) (int, error) {
	n, err := t.reader.ReadHeader(header)
	if err != nil {
	t.er = err
	return 0, err
	}
	t.windowHandler(len(header))
	return n, nil
	}

The code has read n bytes, but it says it read len(header) bytes. len(header) is the total number of bytes that remain to be read, which is not always equal the the bytes there were read. I was able to get the test to pass after changing the line to t.windowHandler(n).