drpcwire: add KindWindowUpdate flow-control frame

Define the KindWindowUpdate control frame that carries a flow-control
credit grant: a varint byte delta in the body, for the stream named by
the frame's stream id. Add WindowUpdateFrame/ParseWindowUpdate helpers
to build and read it. Stream id 0 is reserved for a possible future
connection-level window and is unused in v1.

The control bit marks the frame as out-of-band signaling, so it is
emitted without blocking on data backpressure. It is not relied on for
backward compatibility: an unknown control frame is only dropped after
packet assembly, so it is not safely ignorable on an active stream.
Instead, grants are only ever sent once flow control is active
cluster-wide, so a peer that predates flow control never receives one.

ParseWindowUpdate enforces the whole v1 wire contract in one place, since
these frames are intercepted before packet assembly and its message-id
checks: it accepts only a self-contained control frame (Control and Done
set) naming a real stream (id != 0) with a positive delta and no trailing
bytes. A non-conforming frame yields ok=false and is dropped by the
caller, so a reserved stream-0 update from a future peer is ignored
rather than mishandled.

This is dormant scaffolding for the flow-control work: nothing emits or
acts on the frame yet.

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

Author: suj-krishnan

drpcwire/packet.go

@@ -36,6 +36,11 @@ const (
 
 	// KindInvokeMetadata includes metadata about the next Invoke packet.
 	KindInvokeMetadata Kind = 7
+
+	// KindWindowUpdate carries a flow-control credit grant: a varint byte delta
+	// for the frame's stream id. Stream id 0 is reserved for a future
+	// connection-level window and is unused in v1.
+	KindWindowUpdate Kind = 8
 )
 
 //
@@ -143,6 +148,35 @@ func AppendFrame(buf []byte, fr Frame) []byte {
 	return out
 }
 
+// WindowUpdateFrame builds a KindWindowUpdate grant frame. Callers must pass a
+// real stream id (>0; 0 is reserved) and a positive delta, the conditions
+// ParseWindowUpdate enforces.
+func WindowUpdateFrame(streamID, delta uint64) Frame {
+	return Frame{
+		Data:    AppendVarint(nil, delta),
+		ID:      ID{Stream: streamID},
+		Kind:    KindWindowUpdate,
+		Done:    true,
+		Control: true,
+	}
+}
+
+// ParseWindowUpdate returns a grant frame's stream id and delta. ok is false
+// unless the frame conforms to the wire contract, so the caller drops
+// non-conforming frames rather than acting on them.
+func ParseWindowUpdate(fr Frame) (streamID, delta uint64, ok bool) {
+	// A self-contained control frame for a real stream; the message id is
+	// unchecked since grants are intercepted before packet assembly.
+	if fr.Kind != KindWindowUpdate || !fr.Control || !fr.Done || fr.ID.Stream == 0 {
+		return 0, 0, false
+	}
+	rem, d, parsed, err := ReadVarint(fr.Data)
+	if !parsed || err != nil || len(rem) != 0 || d == 0 { // positive delta, no trailing bytes
+		return 0, 0, false
+	}
+	return fr.ID.Stream, d, true
+}
+
 //
 // packet
 //

drpcwire/packet_string.go

@@ -19,3 +19,61 @@ func TestAppendParse(t *testing.T) {
 		assert.DeepEqual(t, got, exp)
 	}
 }
+
+func TestKindWindowUpdateString(t *testing.T) {
+	assert.Equal(t, KindWindowUpdate.String(), "WindowUpdate")
+}
+
+func TestWindowUpdateFrameRoundTrip(t *testing.T) {
+	for _, tc := range []struct {
+		stream uint64
+		delta  uint64
+	}{
+		{stream: 7, delta: 128 << 10},    // per-stream
+		{stream: 123456, delta: 1 << 40}, // large delta
+	} {
+		fr := WindowUpdateFrame(tc.stream, tc.delta)
+		assert.That(t, fr.Control)
+		assert.That(t, fr.Done)
+		assert.Equal(t, fr.Kind, KindWindowUpdate)
+
+		rem, got, ok, err := ParseFrame(AppendFrame(nil, fr))
+		assert.NoError(t, err)
+		assert.That(t, ok)
+		assert.Equal(t, len(rem), 0)
+
+		sid, delta, ok := ParseWindowUpdate(got)
+		assert.That(t, ok)
+		assert.Equal(t, sid, tc.stream)
+		assert.Equal(t, delta, tc.delta)
+	}
+}
+
+func TestParseWindowUpdateRejectsNonconforming(t *testing.T) {
+	if _, _, ok := ParseWindowUpdate(WindowUpdateFrame(7, 128)); !ok {
+		t.Fatal("conforming window update did not parse")
+	}
+
+	notControl := WindowUpdateFrame(7, 128)
+	notControl.Control = false
+	notDone := WindowUpdateFrame(7, 128)
+	notDone.Done = false
+	trailing := WindowUpdateFrame(7, 128)
+	trailing.Data = append(append([]byte(nil), trailing.Data...), 0xff)
+	empty := WindowUpdateFrame(7, 128)
+	empty.Data = nil
+
+	for name, fr := range map[string]Frame{
+		"wrong kind":     {Kind: KindMessage, Control: true, Done: true, ID: ID{Stream: 7}, Data: AppendVarint(nil, 1)},
+		"not control":    notControl,
+		"not done":       notDone,
+		"stream zero":    WindowUpdateFrame(0, 1),
+		"zero delta":     WindowUpdateFrame(7, 0),
+		"trailing bytes": trailing,
+		"empty payload":  empty,
+	} {
+		if _, _, ok := ParseWindowUpdate(fr); ok {
+			t.Fatalf("expected %q to be rejected", name)
+		}
+	}
+}