add batchDiscard

.github/workflows/ci.yml

@@ -27,7 +27,7 @@ jobs:
         os: [ubuntu-latest, macos-latest, windows-latest]
 
     steps:
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
         with:
           path: src/github.com/containerd/ttrpc
       - uses: actions/setup-go@3041bf56c941b39c61721a86cd11f3bb1338122a # v5.2.0
@@ -47,19 +47,19 @@ jobs:
   #
   project:
     name: Project Checks
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     timeout-minutes: 5
 
     steps:
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
         with:
           path: src/github.com/containerd/ttrpc
           fetch-depth: 25
       - uses: actions/setup-go@3041bf56c941b39c61721a86cd11f3bb1338122a # v5.2.0
         with:
           go-version: ${{ env.GO_VERSION }}
 
-      - uses: containerd/project-checks@434a07157608eeaa1d5c8d4dd506154204cd9401 # v1.1.0
+      - uses: containerd/project-checks@d7751f3c375b8fe4a84c02a068184ee4c1f59bc4 # v1.2.2
         with:
           working-directory: src/github.com/containerd/ttrpc
 
@@ -78,7 +78,7 @@ jobs:
     timeout-minutes: 10
     steps:
 
-    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+    - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
       with:
         path: src/github.com/containerd/ttrpc
         fetch-depth: 25
@@ -110,7 +110,7 @@ jobs:
     timeout-minutes: 5
     steps:
 
-    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+    - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
       with:
         path: src/github.com/containerd/ttrpc
         fetch-depth: 25

channel.go

@@ -111,6 +111,25 @@ func newChannel(conn net.Conn) *channel {
 	}
 }
 
+func batchDiscard(br *bufio.Reader, total int, batchSize int) (int, error) {
+	var discarded int
+	for discarded < total {
+		needDiscard := total - discarded
+		currentBatch := batchSize
+		if needDiscard < currentBatch {
+			currentBatch = needDiscard
+		}
+
+		n, err := br.Discard(currentBatch)
+		discarded += n
+
+		if err != nil {
+			return discarded, err
+		}
+	}
+	return discarded, nil
+}
+
 // recv a message from the channel. The returned buffer contains the message.
 //
 // If a valid grpc status is returned, the message header
@@ -124,7 +143,7 @@ func (ch *channel) recv() (messageHeader, []byte, error) {
 	}
 
 	if mh.Length > uint32(messageLengthMax) {
-		if _, err := ch.br.Discard(int(mh.Length)); err != nil {
+		if _, err := batchDiscard(ch.br, int(mh.Length), messageLengthMax); err != nil {
 			return mh, nil, fmt.Errorf("failed to discard after receiving oversized message: %w", err)
 		}