drpcstream: install per-stream windows from a flow-control option

Add Options.FlowControl to enable per-stream flow control. When Enabled,
NewWithOptions installs the send window (seeded with StreamWindow as
initial credit) and the receive window (HighWater + GrantThreshold), so
data writes become credit-gated and grants are emitted -- all driven by
configuration rather than injected by tests.

This is static, symmetric enablement: both peers must enable it with
compatible sizes. Advertise-on-enable (safe under mixed enablement), the
manager wiring that turns it on for a whole connection, and the
connection-level window come in later commits.

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

Author: suj-krishnan

drpcstream/flow_control_option_test.go

@@ -0,0 +1,70 @@
+// Copyright (C) 2026 Cockroach Labs.
+// See LICENSE for copying information.
+
+package drpcstream
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/zeebo/assert"
+
+	"storj.io/drpc/drpcwire"
+)
+
+// Enabling flow control via Options installs the send and receive windows,
+// seeding the send window with the configured stream window as initial credit.
+func TestStream_FlowControlEnabledInstallsWindows(t *testing.T) {
+	mw := testMuxWriter(t)
+	st := NewWithOptions(context.Background(), 1, mw, NewBufferPool(), Options{
+		SplitSize: 64 << 10,
+		FlowControl: FlowControl{
+			Enabled:        true,
+			StreamWindow:   256 << 10,
+			HighWater:      4 << 20,
+			GrantThreshold: 128 << 10,
+		},
+	})
+	assert.That(t, st.sendw != nil)
+	assert.That(t, st.recvw != nil)
+	assert.Equal(t, st.sendw.available(), int64(256<<10))
+}
+
+// Without flow control enabled, no windows are installed and behavior is
+// unchanged (ungated).
+func TestStream_FlowControlDisabledNoWindows(t *testing.T) {
+	mw := testMuxWriter(t)
+	st := NewWithOptions(context.Background(), 1, mw, NewBufferPool(), Options{SplitSize: 64 << 10})
+	assert.That(t, st.sendw == nil)
+	assert.That(t, st.recvw == nil)
+}
+
+// End to end via the option: an option-installed window gates a data write that
+// exceeds the initial credit, and an incoming grant resumes it.
+func TestStream_FlowControlOptionGatesAndResumes(t *testing.T) {
+	mw := testMuxWriter(t)
+	st := NewWithOptions(context.Background(), 1, mw, NewBufferPool(), Options{
+		SplitSize:   64 << 10,
+		FlowControl: FlowControl{Enabled: true, StreamWindow: 4, HighWater: 1 << 20, GrantThreshold: 2},
+	})
+
+	done := make(chan error, 1)
+	go func() { done <- st.RawWrite(drpcwire.KindMessage, []byte("hello")) }() // 5 bytes > 4
+
+	select {
+	case <-done:
+		t.Fatal("write returned before sufficient credit")
+	case <-time.After(blockShort):
+	}
+
+	// A grant from the peer tops up the send window and resumes the write.
+	assert.NoError(t, st.HandleFrame(drpcwire.WindowUpdateFrame(st.ID(), 1)))
+
+	select {
+	case err := <-done:
+		assert.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("write did not resume after grant")
+	}
+}

drpcstream/stream.go

@@ -30,10 +30,35 @@ type Options struct {
 	// more allocations. 0 is unlimited.
 	MaximumBufferSize int
 
+	// FlowControl configures per-stream flow control. It is disabled by default.
+	FlowControl FlowControl
+
 	// Internal contains options that are for internal use only.
 	Internal drpcopts.Stream
 }
 
+// FlowControl configures per-stream flow control. When Enabled, the stream
+// installs a send window (data writes acquire credit before going on the wire)
+// and a receive window (grants are returned as data is dispatched and
+// consumed). It is the caller's responsibility to size these so the liveness
+// rule StreamWindow >= 2*SplitSize holds.
+type FlowControl struct {
+	// Enabled turns per-stream flow control on for the stream.
+	Enabled bool
+
+	// StreamWindow is the per-stream send window: the initial send credit and
+	// the nominal window size.
+	StreamWindow int64
+
+	// HighWater is the receive-side buffered-byte high-water mark above which
+	// grants are withheld.
+	HighWater int64
+
+	// GrantThreshold is the amount of returnable credit that must accrue before
+	// a grant is emitted, coalescing many frames into one KindWindowUpdate.
+	GrantThreshold int64
+}
+
 // Stream represents an rpc actively happening on a transport.
 type Stream struct {
 	ctx  streamCtx
@@ -123,6 +148,14 @@ func NewWithOptions(
 
 	s.recvQueue.init(pool)
 
+	// When flow control is enabled, install the per-stream windows. The send
+	// window starts with StreamWindow credit (statically agreed with the peer);
+	// the receive window drives grant emission.
+	if opts.FlowControl.Enabled {
+		s.sendw = newSendWindow(opts.FlowControl.StreamWindow)
+		s.recvw = newRecvWindow(opts.FlowControl.HighWater, opts.FlowControl.GrantThreshold)
+	}
+
 	return s
 }