drpcstream: add per-stream send-window credit primitive

Add sendWindow, the per-stream flow-control credit balance on the sender.
acquire spends credit, blocking until enough is available; grant adds
credit; close terminates the window. acquire returns early on close (with
the close error) or context cancellation (with ctx.Err()), consuming no
credit in either case; both are checked before any debit, so a canceled
context never consumes credit or lets a frame proceed even when credit is
available.

Grants are no-revoke: grant takes an unsigned delta (the wire type), so a
grant can only ever raise the balance, and the addition saturates at
math.MaxInt64 so a large or malicious delta can never wrap it negative.
The balance is a signed int64; acquire never drives it below zero, but the
enablement layer may debit credit for bytes sent before flow control
begins enforcing, leaving it negative, in which case acquire blocks until
grants restore it.

This is the primitive only -- it is not yet wired into the stream write
path (that gate comes in a later commit).

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

Author: suj-krishnan

drpcstream/send_window.go

@@ -15,14 +15,13 @@ import (
 // now. acquire spends credit (blocking until enough is available), grant adds
 // credit, and close terminates the window.
 //
-// Grants are no-revoke: grant is strictly additive, and the receiver never
-// takes back credit it has already issued. The balance is a signed int64;
+// avail is a signed int64;
 // acquire never drives it below zero, but the enablement layer may debit credit
 // for bytes sent before flow control begins enforcing, leaving it negative, in
 // which case acquire blocks until grants restore it.
 type sendWindow struct {
 	mu     sync.Mutex
-	avail  int64         // available credit; signed (may be negative — see doc)
+	avail  int64         // available credit; signed (maybe negative during enablement)
 	closed bool          // set once by close; no further acquires succeed
 	err    error         // terminal error returned by acquire after close
 	notify chan struct{} // closed+replaced to wake parked acquirers
@@ -42,9 +41,7 @@ func (w *sendWindow) available() int64 {
 
 // acquire debits n bytes of credit, blocking until available, and returns nil.
 // It returns early (consuming no credit) if the window is closed or ctx is
-// canceled -- both checked before any debit and before the n <= 0 no-op, so a
-// terminated window never reports success, even for an empty (n == 0) frame. A
-// non-positive n acquires nothing.
+// canceled
 func (w *sendWindow) acquire(ctx context.Context, n int64) error {
 	for {
 		w.mu.Lock()
@@ -91,26 +88,21 @@ func (w *sendWindow) grant(n uint64) {
 	w.mu.Unlock()
 }
 
-// applyGrant returns avail + n clamped to math.MaxInt64. n is the wire delta
-// (unsigned), so the result never drops below avail. A negative avail
-// (pre-enforcement debt) is repaid before any positive credit accrues, so a
-// large n saturates only the true sum, not the raw delta.
+// applyGrant returns avail + n with an upper bound of math.MaxInt64.
+// n is the wire delta (unsigned), so the result never drops below avail.
 func applyGrant(avail int64, n uint64) int64 {
 	if avail >= 0 {
-		// math.MaxInt64-avail is the remaining headroom (>= 0, fits int64).
-		// Comparing in uint64 can't overflow, and n <= headroom makes int64(n)
-		// safe and the sum <= math.MaxInt64.
 		if n > uint64(math.MaxInt64-avail) {
 			return math.MaxInt64
 		}
 		return avail + int64(n)
 	}
+	// A negative avail (before FC enablement) is repaid before any positive
+	// credit starts accruing.
 	deficit := uint64(-avail) // |avail| as uint64; -math.MinInt64 wraps to its magnitude
 	if n <= deficit {
 		return -int64(deficit - n) // debt only partly repaid; result still <= 0
 	}
-	// rem is the credit left after clearing the debt; it can exceed int64, so
-	// clamp before the conversion rather than let it wrap negative.
 	if rem := n - deficit; rem <= uint64(math.MaxInt64) {
 		return int64(rem)
 	}