drpcstream: add per-stream receive-window grant policy

Add recvWindow, the per-stream receive side of flow control. It tracks
buffered bytes (in-progress reassembly plus completed, not-yet-consumed
data) and decides when to return credit to the sender: grants are
withheld while buffered is at or above the high-water mark, and the
accrued returnable credit is otherwise released once it reaches the
threshold, coalescing many frames into a single grant. Consuming reopens
a gate that the high-water mark had closed.

dispatched/consumed return the credit delta the caller should emit as a
KindWindowUpdate. Emitting the frame, wiring into the read path
(HandleFrame/MsgRecv), and the connection-level receive budget come in
later commits.

Co-Authored-By: roachdev-claude <roachdev-claude-bot@cockroachlabs.com>

Author: suj-krishnan

drpcstream/recv_window.go

@@ -0,0 +1,75 @@
+// Copyright (C) 2026 Cockroach Labs.
+// See LICENSE for copying information.
+
+package drpcstream
+
+import "sync"
+
+// recvWindow is the per-stream receive side of flow control. It tracks how many
+// bytes are currently buffered for the stream (in-progress reassembly plus
+// completed, not-yet-consumed data) and decides when to return credit to the
+// sender.
+//
+// The receiver holds no credit balance of its own; it accrues "returnable"
+// credit as bytes are dispatched off the wire and releases it as grants:
+//   - while buffered is at or above the high-water mark, grants are withheld
+//     (backpressure), though the returnable credit keeps accruing;
+//   - otherwise, accrued credit is released once it reaches the threshold,
+//     coalescing many frames into a single grant.
+//
+// dispatched and consumed each return the credit delta the caller should send
+// as a KindWindowUpdate (0 when nothing should be sent). Emitting the frame is
+// the caller's job. dispatched (reader goroutine) and consumed (application
+// goroutine) can run concurrently, so the state is mutex-guarded.
+type recvWindow struct {
+	mu        sync.Mutex
+	buffered  int64 // in-progress reassembly + completed, not-yet-consumed bytes
+	pending   int64 // returnable credit accrued but not yet granted
+	highWater int64 // withhold grants while buffered is at or above this
+	threshold int64 // release accrued credit once it reaches this
+}
+
+// newRecvWindow returns a recvWindow with the given high-water mark and grant
+// threshold.
+func newRecvWindow(highWater, threshold int64) *recvWindow {
+	return &recvWindow{highWater: highWater, threshold: threshold}
+}
+
+// dispatched records that n bytes were dispatched off the wire into the
+// stream's buffers and returns the credit delta to grant (0 if none).
+func (w *recvWindow) dispatched(n int64) int64 {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.buffered += n
+	w.pending += n
+	return w.maybeGrantLocked()
+}
+
+// consumed records that n bytes were consumed by the application and returns
+// the credit delta to grant (0 if none). Consuming can reopen a gate that was
+// closed by the high-water mark, flushing credit accrued during the pause.
+func (w *recvWindow) consumed(n int64) int64 {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.buffered -= n
+	return w.maybeGrantLocked()
+}
+
+// bufferedBytes returns the current buffered byte count.
+func (w *recvWindow) bufferedBytes() int64 {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.buffered
+}
+
+// maybeGrantLocked releases the accrued returnable credit when the high-water
+// gate is open and the threshold is met, returning the delta (0 otherwise). It
+// must be called with w.mu held.
+func (w *recvWindow) maybeGrantLocked() int64 {
+	if w.buffered < w.highWater && w.pending >= w.threshold {
+		g := w.pending
+		w.pending = 0
+		return g
+	}
+	return 0
+}

drpcstream/recv_window_test.go

@@ -0,0 +1,62 @@
+// Copyright (C) 2026 Cockroach Labs.
+// See LICENSE for copying information.
+
+package drpcstream
+
+import (
+	"testing"
+
+	"github.com/zeebo/assert"
+)
+
+// Below the high-water mark, accrued returnable credit is released as a grant
+// once it reaches the threshold, and withheld (0) below it.
+func TestRecvWindowGrantsAtThreshold(t *testing.T) {
+	w := newRecvWindow(1000, 100)
+	assert.Equal(t, w.dispatched(60), int64(0))   // pending 60 < 100
+	assert.Equal(t, w.dispatched(60), int64(120)) // pending 120 >= 100, buffered 120 < 1000
+	assert.Equal(t, w.bufferedBytes(), int64(120))
+	assert.Equal(t, w.dispatched(50), int64(0)) // pending reset by the grant; 50 < 100
+}
+
+// At or above the high-water mark, grants are withheld even when the accrued
+// credit is past the threshold.
+func TestRecvWindowWithholdsAtOrAboveHighWater(t *testing.T) {
+	w := newRecvWindow(100, 50)
+	assert.Equal(t, w.dispatched(120), int64(0)) // buffered 120 >= 100 -> withhold
+	assert.Equal(t, w.bufferedBytes(), int64(120))
+}
+
+// Consuming buffered data drops below the high-water mark and flushes the
+// credit accrued while the gate was closed (resume-on-consume).
+func TestRecvWindowResumesOnConsume(t *testing.T) {
+	w := newRecvWindow(100, 50)
+	assert.Equal(t, w.dispatched(120), int64(0)) // withheld; pending 120
+	assert.Equal(t, w.consumed(30), int64(120))  // buffered 90 < 100 -> flush pending
+	assert.Equal(t, w.bufferedBytes(), int64(90))
+}
+
+// Consuming while the accrued credit is below the threshold emits no grant.
+func TestRecvWindowConsumeBelowThresholdNoGrant(t *testing.T) {
+	w := newRecvWindow(1000, 100)
+	assert.Equal(t, w.dispatched(50), int64(0)) // pending 50 < 100
+	assert.Equal(t, w.consumed(20), int64(0))   // buffered 30, pending 50 < 100
+	assert.Equal(t, w.bufferedBytes(), int64(30))
+}
+
+// Many small dispatches coalesce into few grants, each carrying the accrued
+// amount.
+func TestRecvWindowCoalescesGrants(t *testing.T) {
+	w := newRecvWindow(1_000_000, 100)
+	grants, granted := 0, int64(0)
+	for i := 0; i < 10; i++ {
+		if g := w.dispatched(30); g > 0 {
+			grants++
+			granted += g
+		}
+	}
+	// 10 dispatches of 30 (300 bytes) with threshold 100 coalesce into 2 grants
+	// of 120 each; the remaining 60 stays accrued.
+	assert.Equal(t, grants, 2)
+	assert.Equal(t, granted, int64(240))
+}