Repository: grpc/grpc-go

Issue:
Long running streams fail
### 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 https://github.com/grpc/grpc-go/pull/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)
```

Hints from issue comments:
@arjan-bal: After using [git bisect](https://git-scm.com/docs/git-bisect) on the range from tag `1.65.0-dev` to HEAD of master, the commit causing the regression seems to be https://github.com/grpc/grpc-go/commit/9ab8b62505c85a5e03be1ae97949e960e24b6d4f

Command:
```sh
# 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 https://github.com/grpc/grpc-go/commit/9ab8b62505c85a5e03be1ae97949e960e24b6d4f 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:
https://github.com/grpc/grpc-go/blob/8ea3460acace12cd1f2f0502784d42c21eeadf42/internal/transport/transport.go#L559-L579

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))`.
https://github.com/grpc/grpc-go/blob/8ea3460acace12cd1f2f0502784d42c21eeadf42/internal/transport/transport.go#L622-L630

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)`.@arjan-bal: The fix has been merged in https://github.com/grpc/grpc-go/pull/7660, keeping the issue open to track the cherrypick into release branches.@arjan-bal: The fix is part of the latest release: https://github.com/grpc/grpc-go/releases/tag/v1.67.1

Closing the issue now.

Solution patch:
diff --git a/internal/transport/transport.go b/internal/transport/transport.go
index 9a3bb3a6..e12cb0bc 100644
--- a/internal/transport/transport.go
+++ b/internal/transport/transport.go
@@ -625,7 +625,7 @@ func (t *transportReader) ReadHeader(header []byte) (int, error) {
 		t.er = err
 		return 0, err
 	}
-	t.windowHandler(len(header))
+	t.windowHandler(n)
 	return n, nil
 }
 



Test patch:
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index 05f0b0b2..65efb30c 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -2845,3 +2845,34 @@ func (s) TestClientCloseReturnsEarlyWhenGoAwayWriteHangs(t *testing.T) {
 	isGreetingDone.Store(true)
 	ct.Close(errors.New("manually closed by client"))
 }
+
+// TestReadHeaderMultipleBuffers tests the stream when the gRPC headers are
+// split across multiple buffers. It verifies that the reporting of the
+// number of bytes read for flow control is correct.
+func (s) TestReadHeaderMultipleBuffers(t *testing.T) {
+	headerLen := 5
+	recvBuffer := newRecvBuffer()
+	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, 3)})
+	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, headerLen-3)})
+	bytesRead := 0
+	s := Stream{
+		requestRead: func(int) {},
+		trReader: &transportReader{
+			reader: &recvBufferReader{
+				recv: recvBuffer,
+			},
+			windowHandler: func(i int) {
+				bytesRead += i
+			},
+		},
+	}
+
+	header := make([]byte, headerLen)
+	err := s.ReadHeader(header)
+	if err != nil {
+		t.Fatalf("ReadHeader(%v) = %v", header, err)
+	}
+	if bytesRead != headerLen {
+		t.Errorf("bytesRead = %d, want = %d", bytesRead, headerLen)
+	}
+}