9c7eb05 feat(brokeripc): credit flow control on broker logical streams · vev

M docs/protocol.md +18 -3


@@ -109,8 +109,8 @@ Timeout or reply loss after dispatch is outcome unknown and is never replayed.
109	109		## Broker negotiation
110	110
111	111		Broker connections are a separate conversation from the session protocol.
112		-	`internal/adapters/brokerwire` owns it: client tags 101-111 and server tags
113		-	201-209 are disjoint from every session tag, and each direction is its own
	112	+	`internal/adapters/brokerwire` owns it: client tags 101-114 and server tags
	113	+	201-211 are disjoint from every session tag, and each direction is its own
114	114		closed directional `oneof` union. A broker connection opens with exactly one
115	115		broker preamble (`PreambleRequest` client → server, `PreambleResponse` server
116	116		→ client) using broker roles 3/4. The preamble carries the same magic

@@ -127,7 +127,14 @@ connection dispatcher.
127	127
128	128		`internal/adapters/brokeripc` carries this conversation over `broker.sock`
129	129		in the per-user runtime directory, with same-user peer-credential admission
130		-	and bounded per-connection queues. The remote helpers `_broker-mux-stdio`,
	130	+	and bounded per-connection queues. Each logical stream is credit flow
	131	+	controlled in both directions: a sender spends `StreamChunkCredit` (payload
	132	+	plus 64 bytes) per data frame out of a 4 MiB window
	133	+	(`brokerwire.StreamWindowBytes`) and waits when it has too little, and the
	134	+	receiver returns the credit of frames its consumer has read in
	135	+	`StreamWindowUpdate` batches. A slow consumer therefore slows its own stream
	136	+	instead of losing it; data or credit beyond the window is a flow-control
	137	+	violation that settles only that stream. The remote helpers `_broker-mux-stdio`,
131	138		`_broker-mux-quic-bootstrap`, and `_broker-mux-quic-proxy` bridge a remote
132	139		daemonmux socket to stdio or to one freshly authenticated QUIC stream.
133	140

@@ -162,6 +169,12 @@ base. `ValidateOutput` enforces the epoch/base/new chain
162	169		claims its carriage's preferred window, or `MaxOutputWindow` when the
163	170		carriage states none, so output pipelines instead of waiting a round trip
164	171		per frame.
	172	+	- Unacknowledged Output data is also bounded by `MaxOutputWindowBytes`
	173	+	(2 MiB). State frames are charged to their own state, side effects to the
	174	+	next state, and a cumulative ACK releases what it covers. The next state
	175	+	frame waits when either window is full, except that a frame may always be
	176	+	sent while no state frame is in flight. Images share the remaining byte
	177	+	budget and wait for a later frame when it is too small.
165	178		- The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps
166	179		applying + acknowledging daemon frames even while a picker lease owns
167	180		the terminal (see below) — admission without physical write.

@@ -247,6 +260,8 @@ than `KillResult` and `CommandResult`.
247	260		\| Queued envelopes per connection \| 8 messages, 32 MiB bytes (`streamframe` defaults), then backpressure; sync sends uncharged, async charged once \| `adapters/streamframe` \|
248	261		\| Encoding buffer \| bounded per-connection framing buffers owned by `streamframe` \| `adapters/streamframe` \|
249	262		\| Output states in flight \| 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH \| daemon, negotiated down \|
	263	+	\| Unacknowledged Output bytes \| 2 MiB (`MaxOutputWindowBytes`); one frame always allowed when none is in flight \| daemon \|
	264	+	\| Broker stream window \| 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` \| `adapters/brokeripc` \|
250	265		\| QUIC unauthenticated peers \| 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected \| `adapters/quic` \|
251	266		\| QUIC bootstrap \| 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close \| `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` \|
252	267		\| Handshake, dial → first committed publication \| 15 s, single absolute deadline propagated unchanged, never reset \| both ends \|

 ## Broker negotiation
 Broker connections are a separate conversation from the session protocol.
-`internal/adapters/brokerwire` owns it: client tags 101-111 and server tags
+`internal/adapters/brokerwire` owns it: client tags 101-114 and server tags
-201-209 are disjoint from every session tag, and each direction is its own
+201-211 are disjoint from every session tag, and each direction is its own
 closed directional `oneof` union. A broker connection opens with exactly one
 broker preamble (`PreambleRequest` client → server, `PreambleResponse` server
 → client) using broker roles 3/4. The preamble carries the same magic
 `internal/adapters/brokeripc` carries this conversation over `broker.sock`
 in the per-user runtime directory, with same-user peer-credential admission
-and bounded per-connection queues. The remote helpers `_broker-mux-stdio`,
+and bounded per-connection queues. Each logical stream is credit flow
+controlled in both directions: a sender spends `StreamChunkCredit` (payload
+plus 64 bytes) per data frame out of a 4 MiB window
+(`brokerwire.StreamWindowBytes`) and waits when it has too little, and the
+receiver returns the credit of frames its consumer has read in
+`StreamWindowUpdate` batches. A slow consumer therefore slows its own stream
+instead of losing it; data or credit beyond the window is a flow-control
+violation that settles only that stream. The remote helpers `_broker-mux-stdio`,
 `_broker-mux-quic-bootstrap`, and `_broker-mux-quic-proxy` bridge a remote
 daemonmux socket to stdio or to one freshly authenticated QUIC stream.
   claims its carriage's preferred window, or `MaxOutputWindow` when the
   carriage states none, so output pipelines instead of waiting a round trip
   per frame.
+- Unacknowledged Output data is also bounded by `MaxOutputWindowBytes`
+  (2 MiB). State frames are charged to their own state, side effects to the
+  next state, and a cumulative ACK releases what it covers. The next state
+  frame waits when either window is full, except that a frame may always be
+  sent while no state frame is in flight. Images share the remaining byte
+  budget and wait for a later frame when it is too small.
 - The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps
   applying + acknowledging daemon frames even while a picker lease owns
   the terminal (see below) — admission without physical write.
 | Queued envelopes per connection | 8 messages, 32 MiB bytes (`streamframe` defaults), then backpressure; sync sends uncharged, async charged once | `adapters/streamframe` |
 | Encoding buffer | bounded per-connection framing buffers owned by `streamframe` | `adapters/streamframe` |
 | Output states in flight | 8 (`MaxOutputWindow`) preferred for QUIC/IPC/SSH | daemon, negotiated down |
+| Unacknowledged Output bytes | 2 MiB (`MaxOutputWindowBytes`); one frame always allowed when none is in flight | daemon |
+| Broker stream window | 4 MiB per stream and direction (`brokerwire.StreamWindowBytes`), returned by `StreamWindowUpdate` | `adapters/brokeripc` |
 | QUIC unauthenticated peers | 4 connections, 3 s auth deadline, one bidirectional stream each; extra/unidirectional streams rejected | `adapters/quic` |
 | QUIC bootstrap | 32-byte token + 16-byte nonce (base64), ≤4 KiB readiness (schema v1, host-independent port, SHA-256 pin, ≤15 s expiry), one bounded (512 B) auth record, atomic one-time consumption, token erasure on close | `adapters/quic` bootstrap, `app/_broker-mux-quic-bootstrap` + `_broker-mux-quic-proxy` |
 | Handshake, dial → first committed publication | 15 s, single absolute deadline propagated unchanged, never reset | both ends |

M internal/adapters/brokeripc/client.go +18 -3


@@ -364,9 +364,24 @@ func (c *client) dispatch(message brokerwire.ServerMessage) error {
364	364		// lookup; its data is discarded, exactly like a retired stream.
365	365		return nil
366	366		}
367		-	if err := st.deliver(m.Data); err != nil {
368		-	if errors.Is(err, ErrStreamBackpressure) {
369		-	st.fail(ErrStreamBackpressure)
	367	+	if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
	368	+	st.fail(ErrStreamCredit)
	369	+	}
	370	+	return nil
	371	+	case brokerwire.StreamWindowUpdate:
	372	+	if !c.scopeMatches(m.Epoch, m.Connection) {
	373	+	return errors.Join(ErrScopeMismatch, ErrProtocol)
	374	+	}
	375	+	disposition, err := c.conn.StreamWindowUpdate(m.Stream)
	376	+	if err != nil {
	377	+	return errors.Join(ErrProtocol, err)
	378	+	}
	379	+	if disposition == brokerwire.StreamDiscarded {
	380	+	return nil
	381	+	}
	382	+	if st := c.lookupStream(m.Stream); st != nil {
	383	+	if err := st.pipe.granted(m.Credit); err != nil {
	384	+	st.fail(err)
370	385		}
371	386		}
372	387		return nil


@@ -364,9 +364,24 @@ func (c *client) dispatch(message brokerwire.ServerMessage) error {
364		// lookup; its data is discarded, exactly like a retired stream.	364		// lookup; its data is discarded, exactly like a retired stream.
365		return nil	365		return nil
366		}	366		}
367	-	if err := st.deliver(m.Data); err != nil {	367	+	if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
368	-	if errors.Is(err, ErrStreamBackpressure) {	368	+	st.fail(ErrStreamCredit)
369	-	st.fail(ErrStreamBackpressure)	369	+	}
			370	+	return nil
			371	+	case brokerwire.StreamWindowUpdate:
			372	+	if !c.scopeMatches(m.Epoch, m.Connection) {
			373	+	return errors.Join(ErrScopeMismatch, ErrProtocol)
			374	+	}
			375	+	disposition, err := c.conn.StreamWindowUpdate(m.Stream)
			376	+	if err != nil {
			377	+	return errors.Join(ErrProtocol, err)
			378	+	}
			379	+	if disposition == brokerwire.StreamDiscarded {
			380	+	return nil
			381	+	}
			382	+	if st := c.lookupStream(m.Stream); st != nil {
			383	+	if err := st.pipe.granted(m.Credit); err != nil {
			384	+	st.fail(err)
370		}	385		}
371		}	386		}
372		return nil	387		return nil

M internal/adapters/brokeripc/config.go +1 -23


@@ -15,13 +15,6 @@ const (
15	15		// listener. A listener acquires one slot before accepting, so the bound is
16	16		// enforced before a socket is admitted rather than after.
17	17		DefaultMaxClients = 64
18		-	// DefaultStreamInboundChunks bounds one logical stream's inbound chunk
19		-	// queue: frames already accepted from the wire and not yet consumed by the
20		-	// local session carriage.
21		-	DefaultStreamInboundChunks = 64
22		-	// DefaultStreamInboundBytes bounds one logical stream's inbound chunk queue
23		-	// in bytes (4 MiB).
24		-	DefaultStreamInboundBytes = 4 << 20
25	18		)
26	19
27	20		// Config bounds one endpoint. The zero value is valid and selects every

@@ -29,10 +22,6 @@ const (
29	22		type Config struct {
30	23		// MaxClients bounds concurrent accepted client connections.
31	24		MaxClients int
32		-	// StreamInboundChunks bounds one logical stream's inbound chunk queue.
33		-	StreamInboundChunks int
34		-	// StreamInboundBytes bounds one logical stream's inbound chunk bytes.
35		-	StreamInboundBytes uint64
36	25		// HandshakeTimeout bounds one accepted connection's broker preamble, broker
37	26		// admission, and the wait for its Register. The client dialer also bounds
38	27		// its whole setup (Unix dial, preamble, Register send, and wait for

@@ -62,12 +51,6 @@ func (c Config) withDefaults() Config {
62	51		if c.MaxClients <= 0 {
63	52		c.MaxClients = DefaultMaxClients
64	53		}
65		-	if c.StreamInboundChunks <= 0 {
66		-	c.StreamInboundChunks = DefaultStreamInboundChunks
67		-	}
68		-	if c.StreamInboundBytes == 0 {
69		-	c.StreamInboundBytes = DefaultStreamInboundBytes
70		-	}
71	54		if c.HandshakeTimeout <= 0 {
72	55		c.HandshakeTimeout = protocol.HandshakeTimeout
73	56		}

@@ -80,16 +63,11 @@ func (c Config) withDefaults() Config {
80	63		// validate refuses a configuration whose explicit bounds cannot be honored.
81	64		// The zero value is always valid because withDefaults has already been applied.
82	65		func (c Config) validate() error {
83		-	if c.MaxClients <= 0 \|\| c.StreamInboundChunks <= 0 \|\| c.StreamInboundBytes == 0 \|\| c.HandshakeTimeout <= 0 {
	66	+	if c.MaxClients <= 0 \|\| c.HandshakeTimeout <= 0 {
84	67		return ErrConfig
85	68		}
86	69		if c.MaxPendingOperations <= 0 \|\| c.MaxPendingOperations > brokerwire.MaxPendingOperations {
87	70		return ErrConfig
88	71		}
89		-	// A stream inbound byte bound below one maximum chunk can never accept a
90		-	// full stream frame, which would starve a legal peer.
91		-	if c.StreamInboundBytes < brokerwire.MaxStreamChunkBytes {
92		-	return ErrConfig
93		-	}
94	72		return nil
95	73		}

 	// listener. A listener acquires one slot before accepting, so the bound is
 	// enforced before a socket is admitted rather than after.
 	DefaultMaxClients = 64
-	// DefaultStreamInboundChunks bounds one logical stream's inbound chunk
-	// queue: frames already accepted from the wire and not yet consumed by the
-	// local session carriage.
-	DefaultStreamInboundChunks = 64
-	// DefaultStreamInboundBytes bounds one logical stream's inbound chunk queue
-	// in bytes (4 MiB).
-	DefaultStreamInboundBytes = 4 << 20
 )
 // Config bounds one endpoint. The zero value is valid and selects every
 type Config struct {
 	// MaxClients bounds concurrent accepted client connections.
 	MaxClients int
-	// StreamInboundChunks bounds one logical stream's inbound chunk queue.
-	StreamInboundChunks int
-	// StreamInboundBytes bounds one logical stream's inbound chunk bytes.
-	StreamInboundBytes uint64
 	// HandshakeTimeout bounds one accepted connection's broker preamble, broker
 	// admission, and the wait for its Register. The client dialer also bounds
 	// its whole setup (Unix dial, preamble, Register send, and wait for
 	if c.MaxClients <= 0 {
 		c.MaxClients = DefaultMaxClients
 	}
-	if c.StreamInboundChunks <= 0 {
-		c.StreamInboundChunks = DefaultStreamInboundChunks
-	}
-	if c.StreamInboundBytes == 0 {
-		c.StreamInboundBytes = DefaultStreamInboundBytes
-	}
 	if c.HandshakeTimeout <= 0 {
 		c.HandshakeTimeout = protocol.HandshakeTimeout
 	}
 // validate refuses a configuration whose explicit bounds cannot be honored.
 // The zero value is always valid because withDefaults has already been applied.
 func (c Config) validate() error {
-	if c.MaxClients <= 0 || c.StreamInboundChunks <= 0 || c.StreamInboundBytes == 0 || c.HandshakeTimeout <= 0 {
+	if c.MaxClients <= 0 || c.HandshakeTimeout <= 0 {
 		return ErrConfig
 	}
 	if c.MaxPendingOperations <= 0 || c.MaxPendingOperations > brokerwire.MaxPendingOperations {
 		return ErrConfig
 	}
-	// A stream inbound byte bound below one maximum chunk can never accept a
-	// full stream frame, which would starve a legal peer.
-	if c.StreamInboundBytes < brokerwire.MaxStreamChunkBytes {
-		return ErrConfig
-	}
 	return nil
 }

M internal/adapters/brokeripc/errors.go +4 -4


@@ -33,10 +33,10 @@ var (
33	33		// ordering: a duplicate Register, a frame for an unallocated stream, or
34	34		// stream data before the stream was established.
35	35		ErrProtocol = errors.New("brokeripc: broker protocol violation")
36		-	// ErrStreamBackpressure reports a logical stream whose local consumer is
37		-	// not draining its bounded inbound queue. The affected stream is settled
38		-	// alone; the connection and every sibling stream stay up.
39		-	ErrStreamBackpressure = errors.New("brokeripc: stream consumer is not keeping up")
	36	+	// ErrStreamCredit reports a peer that sent stream data beyond the credit it
	37	+	// was granted, or granted credit beyond one stream window. The affected
	38	+	// stream is settled alone; the connection and every sibling stream stay up.
	39	+	ErrStreamCredit = errors.New("brokeripc: stream flow-control violation")
40	40		// ErrStreamGone reports a stream the peer retired while this side was still
41	41		// starting or using it.
42	42		ErrStreamGone = errors.New("brokeripc: logical stream is gone")


@@ -33,10 +33,10 @@ var (
33		// ordering: a duplicate Register, a frame for an unallocated stream, or	33		// ordering: a duplicate Register, a frame for an unallocated stream, or
34		// stream data before the stream was established.	34		// stream data before the stream was established.
35		ErrProtocol = errors.New("brokeripc: broker protocol violation")	35		ErrProtocol = errors.New("brokeripc: broker protocol violation")
36	-	// ErrStreamBackpressure reports a logical stream whose local consumer is	36	+	// ErrStreamCredit reports a peer that sent stream data beyond the credit it
37	-	// not draining its bounded inbound queue. The affected stream is settled	37	+	// was granted, or granted credit beyond one stream window. The affected
38	-	// alone; the connection and every sibling stream stay up.	38	+	// stream is settled alone; the connection and every sibling stream stay up.
39	-	ErrStreamBackpressure = errors.New("brokeripc: stream consumer is not keeping up")	39	+	ErrStreamCredit = errors.New("brokeripc: stream flow-control violation")
40		// ErrStreamGone reports a stream the peer retired while this side was still	40		// ErrStreamGone reports a stream the peer retired while this side was still
41		// starting or using it.	41		// starting or using it.
42		ErrStreamGone = errors.New("brokeripc: logical stream is gone")	42		ErrStreamGone = errors.New("brokeripc: logical stream is gone")

M internal/adapters/brokeripc/limits_test.go +12 -44


@@ -147,71 +147,39 @@ func TestConfigDefaultsAndValidation(t *testing.T) {
147	147		cfg := Config{}.withDefaults()
148	148		require.NoError(t, cfg.validate())
149	149		require.Equal(t, DefaultMaxClients, cfg.MaxClients)
150		-	require.Equal(t, DefaultStreamInboundChunks, cfg.StreamInboundChunks)
151		-	require.Equal(t, uint64(DefaultStreamInboundBytes), cfg.StreamInboundBytes)
152	150		require.Equal(t, brokerwire.MaxPendingOperations, cfg.MaxPendingOperations)
153	151
154		-	require.ErrorIs(t, (Config{StreamInboundBytes: 1}).withDefaults().validate(), ErrConfig)
155	152		require.ErrorIs(t, (Config{MaxPendingOperations: brokerwire.MaxPendingOperations + 1}).withDefaults().validate(), ErrConfig)
156	153		// A negative explicit bound is replaced by its default rather than refused.
157		-	require.NoError(t, (Config{MaxClients: -1, StreamInboundChunks: -1}).withDefaults().validate())
	154	+	require.NoError(t, (Config{MaxClients: -1}).withDefaults().validate())
158	155		}
159	156
160		-	// TestStreamPipeBoundsAndClose proves the per-stream byte channel is bounded,
161		-	// reports backpressure instead of growing, and releases a parked reader on
162		-	// close.
163		-	func TestStreamPipeBoundsAndClose(t *testing.T) {
164		-	pipe := newStreamPipe(2, 4096, func([]byte) error { return nil })
165		-	require.NoError(t, pipe.deliver([]byte("one")))
166		-	require.NoError(t, pipe.deliver([]byte("two")))
167		-	require.ErrorIs(t, pipe.deliver([]byte("three")), ErrStreamBackpressure)
168		-
169		-	buf := make([]byte, 3)
170		-	n, err := pipe.Read(buf)
171		-	require.NoError(t, err)
172		-	require.Equal(t, "one", string(buf[:n]))
173		-	require.NoError(t, pipe.deliver([]byte("three")))
174		-
175		-	// An orderly Close keeps what is already queued ("two" and "three"): Read
176		-	// drains it before observing io.EOF, so a chunk delivered just ahead of an
177		-	// orderly close is never discarded underneath its reader (see
178		-	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose for the focused
179		-	// coverage of this behavior).
180		-	require.NoError(t, pipe.Close())
181		-	require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)
182		-	drained := make([]byte, 8)
183		-	n, err = pipe.Read(drained)
184		-	require.NoError(t, err)
185		-	require.Equal(t, "two", string(drained[:n]))
186		-	n, err = pipe.Read(drained)
187		-	require.NoError(t, err)
188		-	require.Equal(t, "three", string(drained[:n]))
189		-	_, err = pipe.Read(drained)
190		-	require.ErrorIs(t, err, io.EOF)
191		-
192		-	// A reader parked on an empty pipe is released by closeWith with its cause.
193		-	parked := newStreamPipe(1, 4096, func([]byte) error { return nil })
	157	+	// TestStreamPipeCloseReleasesParkedReader proves a reader parked on an empty
	158	+	// pipe is released by closeWith with its cause.
	159	+	func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
	160	+	parked := newTestPipe()
194	161		done := make(chan error, 1)
195	162		go func() {
196		-	_, err := parked.Read(buf)
	163	+	_, err := parked.Read(make([]byte, 3))
197	164		done <- err
198	165		}()
199		-	time.Sleep(20 * time.Millisecond)
200		-	parked.closeWith(ErrStreamBackpressure)
	166	+	parked.closeWith(ErrStreamCredit)
201	167		select {
202	168		case err := <-done:
203		-	require.ErrorIs(t, err, ErrStreamBackpressure)
	169	+	require.ErrorIs(t, err, ErrStreamCredit)
204	170		case <-time.After(5 * time.Second):
205	171		t.Fatal("closeWith must release a parked reader")
206	172		}
	173	+	require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)
207	174		}
208	175
209	176		// TestStreamPipeOutboundError proves a failing outbound hook surfaces instead of
210	177		// silently dropping bytes.
211	178		func TestStreamPipeOutboundError(t *testing.T) {
212		-	pipe := newStreamPipe(1, 4096, func([]byte) error { return ErrStreamBackpressure })
	179	+	failure := errors.New("outbound failed")
	180	+	pipe := newStreamPipe(func([]byte) error { return failure }, func(uint64) error { return nil })
213	181		_, err := pipe.Write([]byte("payload"))
214		-	require.ErrorIs(t, err, ErrStreamBackpressure)
	182	+	require.ErrorIs(t, err, failure)
215	183		require.NoError(t, pipe.Close())
216	184		_, err = pipe.Write([]byte("payload"))
217	185		require.ErrorIs(t, err, ErrStreamGone)

 	cfg := Config{}.withDefaults()
 	require.NoError(t, cfg.validate())
 	require.Equal(t, DefaultMaxClients, cfg.MaxClients)
-	require.Equal(t, DefaultStreamInboundChunks, cfg.StreamInboundChunks)
-	require.Equal(t, uint64(DefaultStreamInboundBytes), cfg.StreamInboundBytes)
 	require.Equal(t, brokerwire.MaxPendingOperations, cfg.MaxPendingOperations)
-	require.ErrorIs(t, (Config{StreamInboundBytes: 1}).withDefaults().validate(), ErrConfig)
 	require.ErrorIs(t, (Config{MaxPendingOperations: brokerwire.MaxPendingOperations + 1}).withDefaults().validate(), ErrConfig)
 	// A negative explicit bound is replaced by its default rather than refused.
-	require.NoError(t, (Config{MaxClients: -1, StreamInboundChunks: -1}).withDefaults().validate())
+	require.NoError(t, (Config{MaxClients: -1}).withDefaults().validate())
 }
-// TestStreamPipeBoundsAndClose proves the per-stream byte channel is bounded,
+// TestStreamPipeCloseReleasesParkedReader proves a reader parked on an empty
-// reports backpressure instead of growing, and releases a parked reader on
+// pipe is released by closeWith with its cause.
-// close.
+func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
-func TestStreamPipeBoundsAndClose(t *testing.T) {
+	parked := newTestPipe()
-	pipe := newStreamPipe(2, 4096, func([]byte) error { return nil })
-	require.NoError(t, pipe.deliver([]byte("one")))
-	require.NoError(t, pipe.deliver([]byte("two")))
-	require.ErrorIs(t, pipe.deliver([]byte("three")), ErrStreamBackpressure)
-	buf := make([]byte, 3)
-	n, err := pipe.Read(buf)
-	require.NoError(t, err)
-	require.Equal(t, "one", string(buf[:n]))
-	require.NoError(t, pipe.deliver([]byte("three")))
-	// An orderly Close keeps what is already queued ("two" and "three"): Read
-	// drains it before observing io.EOF, so a chunk delivered just ahead of an
-	// orderly close is never discarded underneath its reader (see
-	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose for the focused
-	// coverage of this behavior).
-	require.NoError(t, pipe.Close())
-	require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)
-	drained := make([]byte, 8)
-	n, err = pipe.Read(drained)
-	require.NoError(t, err)
-	require.Equal(t, "two", string(drained[:n]))
-	n, err = pipe.Read(drained)
-	require.NoError(t, err)
-	require.Equal(t, "three", string(drained[:n]))
-	_, err = pipe.Read(drained)
-	require.ErrorIs(t, err, io.EOF)
-	// A reader parked on an empty pipe is released by closeWith with its cause.
-	parked := newStreamPipe(1, 4096, func([]byte) error { return nil })
 	done := make(chan error, 1)
 	go func() {
-		_, err := parked.Read(buf)
+		_, err := parked.Read(make([]byte, 3))
 		done <- err
 	}()
-	time.Sleep(20 * time.Millisecond)
+	parked.closeWith(ErrStreamCredit)
-	parked.closeWith(ErrStreamBackpressure)
 	select {
 	case err := <-done:
-		require.ErrorIs(t, err, ErrStreamBackpressure)
+		require.ErrorIs(t, err, ErrStreamCredit)
 	case <-time.After(5 * time.Second):
 		t.Fatal("closeWith must release a parked reader")
 	}
+	require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)
 }
 // TestStreamPipeOutboundError proves a failing outbound hook surfaces instead of
 // silently dropping bytes.
 func TestStreamPipeOutboundError(t *testing.T) {
-	pipe := newStreamPipe(1, 4096, func([]byte) error { return ErrStreamBackpressure })
+	failure := errors.New("outbound failed")
+	pipe := newStreamPipe(func([]byte) error { return failure }, func(uint64) error { return nil })
 	_, err := pipe.Write([]byte("payload"))
-	require.ErrorIs(t, err, ErrStreamBackpressure)
+	require.ErrorIs(t, err, failure)
 	require.NoError(t, pipe.Close())
 	_, err = pipe.Write([]byte("payload"))
 	require.ErrorIs(t, err, ErrStreamGone)

M internal/adapters/brokeripc/pipe.go +131 -50


@@ -3,53 +3,80 @@ package brokeripc
3	3		import (
4	4		"io"
5	5		"sync"
	6	+
	7	+	"github.com/bnema/vev/internal/adapters/brokerwire"
6	8		)
7	9
8		-	// streamPipe is the private ordered, bounded byte channel between one broker
9		-	// wire session and the sessionwire framing of one logical stream.
	10	+	// streamPipe is the private ordered, credit-controlled byte channel between one
	11	+	// broker wire session and the sessionwire framing of one logical stream.
10	12		//
11		-	// The wire reader never blocks on a slow consumer: a chunk the peer sent is
12		-	// appended to a bounded queue and refused with ErrStreamBackpressure once the
13		-	// bound is reached, so only that stream is settled and every sibling and the
14		-	// physical connection stay untouched. The consumer never blocks the writer:
15		-	// Read serves already-queued chunks or parks on the wake channel, and Close
16		-	// releases both ends promptly. Write hands each byte slice to the injected
17		-	// outbound hook, which owns chunking it under the negotiated stream chunk
18		-	// ceiling and enqueueing the frames on the connection.
	13	+	// Each direction of a stream has a brokerwire.StreamWindowBytes credit window.
	14	+	// The sender spends brokerwire.StreamChunkCredit per data frame before writing
	15	+	// it and waits while it has too little, so a slow consumer slows its own
	16	+	// sender instead of failing the stream. The receiver returns the credit of
	17	+	// every frame its consumer has fully read with one StreamWindowUpdate per
	18	+	// batch. Because a conforming peer never exceeds the window, the wire reader
	19	+	// never blocks: it appends each frame to the queue, and a frame beyond the
	20	+	// granted credit is a peer violation (ErrStreamCredit) that settles only that
	21	+	// stream. Close releases the reader, a writer waiting for credit, and the
	22	+	// writer's peer promptly.
19	23		type streamPipe struct {
20		-	mu sync.Mutex
21		-	queue [][]byte
22		-	queued int
23		-	closed bool
24		-	err error
25		-	maxChunk int
26		-	maxBytes int
	24	+	mu sync.Mutex
	25	+	queue []pipeChunk
	26	+	closed bool
	27	+	err error
	28	+
	29	+	// recvUsed is the credit held by queued frames, recvReturn the credit of
	30	+	// fully read frames not yet granted back, and returnAt the batch size.
	31	+	recvUsed uint64
	32	+	recvReturn uint64
	33	+	returnAt uint64
	34	+	// sendCredit is what the peer still allows this side to send.
	35	+	sendCredit uint64
	36	+	creditWake chan struct{}
27	37
28	38		wake chan struct{}
29	39		closedCh chan struct{}
30	40
31	41		outbound func([]byte) error
	42	+	grant func(uint64) error
32	43		}
33	44
34		-	// newStreamPipe builds one bounded stream carriage. outbound must not be nil.
35		-	func newStreamPipe(maxChunks int, maxBytes uint64, outbound func([]byte) error) *streamPipe {
36		-	if maxChunks <= 0 {
37		-	maxChunks = DefaultStreamInboundChunks
38		-	}
39		-	if maxBytes == 0 {
40		-	maxBytes = DefaultStreamInboundBytes
41		-	}
	45	+	// pipeChunk is one queued inbound frame and the credit it holds until it is
	46	+	// fully read.
	47	+	type pipeChunk struct {
	48	+	data []byte
	49	+	cost uint64
	50	+	}
	51	+
	52	+	// newStreamPipe builds one stream carriage. outbound writes one outbound byte
	53	+	// slice; grant returns consumed inbound credit to the peer. Neither may be nil.
	54	+	func newStreamPipe(outbound func([]byte) error, grant func(uint64) error) *streamPipe {
	55	+	window := brokerwire.StreamWindowBytes
42	56		return &streamPipe{
43		-	maxChunk: maxChunks,
44		-	maxBytes: int(maxBytes),
45		-	wake: make(chan struct{}, 1),
46		-	closedCh: make(chan struct{}),
47		-	outbound: outbound,
	57	+	returnAt: streamCreditReturnAt(window),
	58	+	sendCredit: window,
	59	+	creditWake: make(chan struct{}),
	60	+	wake: make(chan struct{}, 1),
	61	+	closedCh: make(chan struct{}),
	62	+	outbound: outbound,
	63	+	grant: grant,
48	64		}
49	65		}
50	66
51		-	// Read serves the next queued chunk, blocking until one arrives or the pipe
52		-	// closes. A partial copy retains the unread remainder for the next call.
	67	+	// streamCreditReturnAt is the consumed credit a receiver batches before it
	68	+	// grants it back: half the window, lowered so a sender blocked on a maximum
	69	+	// frame is always released. A sender waits only once more than
	70	+	// window-StreamChunkCredit(max) is unreturned, so granting at or below that
	71	+	// point can never deadlock.
	72	+	func streamCreditReturnAt(window uint64) uint64 {
	73	+	cost := brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
	74	+	return max(min(window/2, window-cost+1), 1)
	75	+	}
	76	+
	77	+	// Read serves the next queued frame, blocking until one arrives or the pipe
	78	+	// closes. A partial copy retains the unread remainder for the next call. The
	79	+	// credit of each fully read frame is returned to the peer in batches.
53	80		func (p *streamPipe) Read(b []byte) (int, error) {
54	81		if len(b) == 0 {
55	82		return 0, nil

@@ -57,16 +84,26 @@ func (p *streamPipe) Read(b []byte) (int, error) {
57	84		for {
58	85		p.mu.Lock()
59	86		if len(p.queue) > 0 {
60		-	chunk := p.queue[0]
61		-	n := copy(b, chunk)
62		-	if n == len(chunk) {
63		-	p.queue[0] = nil
	87	+	chunk := &p.queue[0]
	88	+	n := copy(b, chunk.data)
	89	+	var grant uint64
	90	+	if n == len(chunk.data) {
	91	+	p.recvUsed -= chunk.cost
	92	+	p.recvReturn += chunk.cost
	93	+	p.queue[0] = pipeChunk{}
64	94		p.queue = p.queue[1:]
	95	+	if p.recvReturn >= p.returnAt && !p.closed {
	96	+	grant, p.recvReturn = p.recvReturn, 0
	97	+	}
65	98		} else {
66		-	p.queue[0] = chunk[n:]
	99	+	chunk.data = chunk.data[n:]
67	100		}
68		-	p.queued -= n
69	101		p.mu.Unlock()
	102	+	if grant > 0 {
	103	+	// A failed grant means the connection itself is failing; its
	104	+	// terminal path settles this stream, so there is nothing to retry.
	105	+	_ = p.grant(grant)
	106	+	}
70	107		return n, nil
71	108		}
72	109		if p.closed {

@@ -85,8 +122,9 @@ func (p *streamPipe) Read(b []byte) (int, error) {
85	122		}
86	123		}
87	124
88		-	// Write hands one byte slice to the outbound hook. It returns ErrClosed once the
89		-	// pipe reached its terminal outcome.
	125	+	// Write hands one byte slice to the outbound hook, which spends send credit
	126	+	// per frame through reserve. It returns ErrStreamGone once the pipe reached its
	127	+	// terminal outcome.
90	128		func (p *streamPipe) Write(b []byte) (int, error) {
91	129		p.mu.Lock()
92	130		closed := p.closed

@@ -100,25 +138,68 @@ func (p *streamPipe) Write(b []byte) (int, error) {
100	138		return len(b), nil
101	139		}
102	140
103		-	// deliver accepts one inbound chunk from the connection reader. It returns
104		-	// ErrStreamBackpressure when the bound is reached and ErrStreamGone once the
105		-	// pipe is closed; the caller settles the affected stream in both cases.
	141	+	// reserve spends the send credit of one outbound frame of n bytes, waiting for
	142	+	// the peer to grant more when it is short. It returns ErrStreamGone once the
	143	+	// pipe closes, so a writer waiting for credit is always released.
	144	+	func (p *streamPipe) reserve(n int) error {
	145	+	cost := brokerwire.StreamChunkCredit(n)
	146	+	for {
	147	+	p.mu.Lock()
	148	+	if p.closed {
	149	+	p.mu.Unlock()
	150	+	return ErrStreamGone
	151	+	}
	152	+	if p.sendCredit >= cost {
	153	+	p.sendCredit -= cost
	154	+	p.mu.Unlock()
	155	+	return nil
	156	+	}
	157	+	wake := p.creditWake
	158	+	p.mu.Unlock()
	159	+	select {
	160	+	case <-wake:
	161	+	case <-p.closedCh:
	162	+	}
	163	+	}
	164	+	}
	165	+
	166	+	// granted adds credit the peer returned. Credit beyond the window is a peer
	167	+	// flow-control violation.
	168	+	func (p *streamPipe) granted(credit uint64) error {
	169	+	p.mu.Lock()
	170	+	defer p.mu.Unlock()
	171	+	if p.closed {
	172	+	return nil
	173	+	}
	174	+	if credit > brokerwire.StreamWindowBytes-p.sendCredit {
	175	+	return ErrStreamCredit
	176	+	}
	177	+	p.sendCredit += credit
	178	+	close(p.creditWake)
	179	+	p.creditWake = make(chan struct{})
	180	+	return nil
	181	+	}
	182	+
	183	+	// deliver accepts one inbound frame from the connection reader. It returns
	184	+	// ErrStreamCredit when the peer sent beyond the credit it was granted and
	185	+	// ErrStreamGone once the pipe is closed; the caller settles the affected stream
	186	+	// in both cases.
106	187		func (p *streamPipe) deliver(chunk []byte) error {
107	188		if len(chunk) == 0 {
108	189		return nil
109	190		}
	191	+	cost := brokerwire.StreamChunkCredit(len(chunk))
110	192		p.mu.Lock()
111	193		if p.closed {
112	194		p.mu.Unlock()
113	195		return ErrStreamGone
114	196		}
115		-	if len(p.queue) >= p.maxChunk \|\| p.queued+len(chunk) > p.maxBytes {
	197	+	if p.recvUsed+p.recvReturn+cost > brokerwire.StreamWindowBytes {
116	198		p.mu.Unlock()
117		-	return ErrStreamBackpressure
	199	+	return ErrStreamCredit
118	200		}
119		-	owned := append([]byte(nil), chunk...)
120		-	p.queue = append(p.queue, owned)
121		-	p.queued += len(owned)
	201	+	p.queue = append(p.queue, pipeChunk{data: append([]byte(nil), chunk...), cost: cost})
	202	+	p.recvUsed += cost
122	203		p.mu.Unlock()
123	204		select {
124	205		case p.wake <- struct{}{}:

@@ -128,7 +209,7 @@ func (p *streamPipe) deliver(chunk []byte) error {
128	209		}
129	210
130	211		// closeWith settles the pipe exactly once with cause. A nil cause is an orderly
131		-	// close; every blocked Read observes it promptly. On a nil cause, any chunks
	212	+	// close; every blocked Read observes it promptly. On a nil cause, any frames
132	213		// already queued are kept: Read serves them first and only returns io.EOF once
133	214		// the queue is drained, so a reply that arrived just before an orderly peer
134	215		// close is never discarded underneath its reader. An error cause is a failure:

@@ -144,7 +225,7 @@ func (p *streamPipe) closeWith(cause error) {
144	225		p.err = cause
145	226		if cause != nil {
146	227		p.queue = nil
147		-	p.queued = 0
	228	+	p.recvUsed = 0
148	229		}
149	230		p.mu.Unlock()
150	231		close(p.closedCh)


@@ -3,53 +3,80 @@ package brokeripc
3		import (	3		import (
4		"io"	4		"io"
5		"sync"	5		"sync"
			6	+
			7	+	"github.com/bnema/vev/internal/adapters/brokerwire"
6		)	8		)
7			9
8	-	// streamPipe is the private ordered, bounded byte channel between one broker	10	+	// streamPipe is the private ordered, credit-controlled byte channel between one
9	-	// wire session and the sessionwire framing of one logical stream.	11	+	// broker wire session and the sessionwire framing of one logical stream.
10		//	12		//
11	-	// The wire reader never blocks on a slow consumer: a chunk the peer sent is	13	+	// Each direction of a stream has a brokerwire.StreamWindowBytes credit window.
12	-	// appended to a bounded queue and refused with ErrStreamBackpressure once the	14	+	// The sender spends brokerwire.StreamChunkCredit per data frame before writing
13	-	// bound is reached, so only that stream is settled and every sibling and the	15	+	// it and waits while it has too little, so a slow consumer slows its own
14	-	// physical connection stay untouched. The consumer never blocks the writer:	16	+	// sender instead of failing the stream. The receiver returns the credit of
15	-	// Read serves already-queued chunks or parks on the wake channel, and Close	17	+	// every frame its consumer has fully read with one StreamWindowUpdate per
16	-	// releases both ends promptly. Write hands each byte slice to the injected	18	+	// batch. Because a conforming peer never exceeds the window, the wire reader
17	-	// outbound hook, which owns chunking it under the negotiated stream chunk	19	+	// never blocks: it appends each frame to the queue, and a frame beyond the
18	-	// ceiling and enqueueing the frames on the connection.	20	+	// granted credit is a peer violation (ErrStreamCredit) that settles only that
			21	+	// stream. Close releases the reader, a writer waiting for credit, and the
			22	+	// writer's peer promptly.
19		type streamPipe struct {	23		type streamPipe struct {
20	-	mu sync.Mutex	24	+	mu sync.Mutex
21	-	queue [][]byte	25	+	queue []pipeChunk
22	-	queued int	26	+	closed bool
23	-	closed bool	27	+	err error
24	-	err error	28	+
25	-	maxChunk int	29	+	// recvUsed is the credit held by queued frames, recvReturn the credit of
26	-	maxBytes int	30	+	// fully read frames not yet granted back, and returnAt the batch size.
			31	+	recvUsed uint64
			32	+	recvReturn uint64
			33	+	returnAt uint64
			34	+	// sendCredit is what the peer still allows this side to send.
			35	+	sendCredit uint64
			36	+	creditWake chan struct{}
27			37
28		wake chan struct{}	38		wake chan struct{}
29		closedCh chan struct{}	39		closedCh chan struct{}
30			40
31		outbound func([]byte) error	41		outbound func([]byte) error
			42	+	grant func(uint64) error
32		}	43		}
33			44
34	-	// newStreamPipe builds one bounded stream carriage. outbound must not be nil.	45	+	// pipeChunk is one queued inbound frame and the credit it holds until it is
35	-	func newStreamPipe(maxChunks int, maxBytes uint64, outbound func([]byte) error) *streamPipe {	46	+	// fully read.
36	-	if maxChunks <= 0 {	47	+	type pipeChunk struct {
37	-	maxChunks = DefaultStreamInboundChunks	48	+	data []byte
38	-	}	49	+	cost uint64
39	-	if maxBytes == 0 {	50	+	}
40	-	maxBytes = DefaultStreamInboundBytes	51	+
41	-	}	52	+	// newStreamPipe builds one stream carriage. outbound writes one outbound byte
			53	+	// slice; grant returns consumed inbound credit to the peer. Neither may be nil.
			54	+	func newStreamPipe(outbound func([]byte) error, grant func(uint64) error) *streamPipe {
			55	+	window := brokerwire.StreamWindowBytes
42		return &streamPipe{	56		return &streamPipe{
43	-	maxChunk: maxChunks,	57	+	returnAt: streamCreditReturnAt(window),
44	-	maxBytes: int(maxBytes),	58	+	sendCredit: window,
45	-	wake: make(chan struct{}, 1),	59	+	creditWake: make(chan struct{}),
46	-	closedCh: make(chan struct{}),	60	+	wake: make(chan struct{}, 1),
47	-	outbound: outbound,	61	+	closedCh: make(chan struct{}),
			62	+	outbound: outbound,
			63	+	grant: grant,
48		}	64		}
49		}	65		}
50			66
51	-	// Read serves the next queued chunk, blocking until one arrives or the pipe	67	+	// streamCreditReturnAt is the consumed credit a receiver batches before it
52	-	// closes. A partial copy retains the unread remainder for the next call.	68	+	// grants it back: half the window, lowered so a sender blocked on a maximum
			69	+	// frame is always released. A sender waits only once more than
			70	+	// window-StreamChunkCredit(max) is unreturned, so granting at or below that
			71	+	// point can never deadlock.
			72	+	func streamCreditReturnAt(window uint64) uint64 {
			73	+	cost := brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
			74	+	return max(min(window/2, window-cost+1), 1)
			75	+	}
			76	+
			77	+	// Read serves the next queued frame, blocking until one arrives or the pipe
			78	+	// closes. A partial copy retains the unread remainder for the next call. The
			79	+	// credit of each fully read frame is returned to the peer in batches.
53		func (p *streamPipe) Read(b []byte) (int, error) {	80		func (p *streamPipe) Read(b []byte) (int, error) {
54		if len(b) == 0 {	81		if len(b) == 0 {
55		return 0, nil	82		return 0, nil

@@ -57,16 +84,26 @@ func (p *streamPipe) Read(b []byte) (int, error) {
57		for {	84		for {
58		p.mu.Lock()	85		p.mu.Lock()
59		if len(p.queue) > 0 {	86		if len(p.queue) > 0 {
60	-	chunk := p.queue[0]	87	+	chunk := &p.queue[0]
61	-	n := copy(b, chunk)	88	+	n := copy(b, chunk.data)
62	-	if n == len(chunk) {	89	+	var grant uint64
63	-	p.queue[0] = nil	90	+	if n == len(chunk.data) {
			91	+	p.recvUsed -= chunk.cost
			92	+	p.recvReturn += chunk.cost
			93	+	p.queue[0] = pipeChunk{}
64		p.queue = p.queue[1:]	94		p.queue = p.queue[1:]
			95	+	if p.recvReturn >= p.returnAt && !p.closed {
			96	+	grant, p.recvReturn = p.recvReturn, 0
			97	+	}
65		} else {	98		} else {
66	-	p.queue[0] = chunk[n:]	99	+	chunk.data = chunk.data[n:]
67		}	100		}
68	-	p.queued -= n
69		p.mu.Unlock()	101		p.mu.Unlock()
			102	+	if grant > 0 {
			103	+	// A failed grant means the connection itself is failing; its
			104	+	// terminal path settles this stream, so there is nothing to retry.
			105	+	_ = p.grant(grant)
			106	+	}
70		return n, nil	107		return n, nil
71		}	108		}
72		if p.closed {	109		if p.closed {

@@ -85,8 +122,9 @@ func (p *streamPipe) Read(b []byte) (int, error) {
85		}	122		}
86		}	123		}
87			124
88	-	// Write hands one byte slice to the outbound hook. It returns ErrClosed once the	125	+	// Write hands one byte slice to the outbound hook, which spends send credit
89	-	// pipe reached its terminal outcome.	126	+	// per frame through reserve. It returns ErrStreamGone once the pipe reached its
			127	+	// terminal outcome.
90		func (p *streamPipe) Write(b []byte) (int, error) {	128		func (p *streamPipe) Write(b []byte) (int, error) {
91		p.mu.Lock()	129		p.mu.Lock()
92		closed := p.closed	130		closed := p.closed

@@ -100,25 +138,68 @@ func (p *streamPipe) Write(b []byte) (int, error) {
100		return len(b), nil	138		return len(b), nil
101		}	139		}
102			140
103	-	// deliver accepts one inbound chunk from the connection reader. It returns	141	+	// reserve spends the send credit of one outbound frame of n bytes, waiting for
104	-	// ErrStreamBackpressure when the bound is reached and ErrStreamGone once the	142	+	// the peer to grant more when it is short. It returns ErrStreamGone once the
105	-	// pipe is closed; the caller settles the affected stream in both cases.	143	+	// pipe closes, so a writer waiting for credit is always released.
			144	+	func (p *streamPipe) reserve(n int) error {
			145	+	cost := brokerwire.StreamChunkCredit(n)
			146	+	for {
			147	+	p.mu.Lock()
			148	+	if p.closed {
			149	+	p.mu.Unlock()
			150	+	return ErrStreamGone
			151	+	}
			152	+	if p.sendCredit >= cost {
			153	+	p.sendCredit -= cost
			154	+	p.mu.Unlock()
			155	+	return nil
			156	+	}
			157	+	wake := p.creditWake
			158	+	p.mu.Unlock()
			159	+	select {
			160	+	case <-wake:
			161	+	case <-p.closedCh:
			162	+	}
			163	+	}
			164	+	}
			165	+
			166	+	// granted adds credit the peer returned. Credit beyond the window is a peer
			167	+	// flow-control violation.
			168	+	func (p *streamPipe) granted(credit uint64) error {
			169	+	p.mu.Lock()
			170	+	defer p.mu.Unlock()
			171	+	if p.closed {
			172	+	return nil
			173	+	}
			174	+	if credit > brokerwire.StreamWindowBytes-p.sendCredit {
			175	+	return ErrStreamCredit
			176	+	}
			177	+	p.sendCredit += credit
			178	+	close(p.creditWake)
			179	+	p.creditWake = make(chan struct{})
			180	+	return nil
			181	+	}
			182	+
			183	+	// deliver accepts one inbound frame from the connection reader. It returns
			184	+	// ErrStreamCredit when the peer sent beyond the credit it was granted and
			185	+	// ErrStreamGone once the pipe is closed; the caller settles the affected stream
			186	+	// in both cases.
106		func (p *streamPipe) deliver(chunk []byte) error {	187		func (p *streamPipe) deliver(chunk []byte) error {
107		if len(chunk) == 0 {	188		if len(chunk) == 0 {
108		return nil	189		return nil
109		}	190		}
			191	+	cost := brokerwire.StreamChunkCredit(len(chunk))
110		p.mu.Lock()	192		p.mu.Lock()
111		if p.closed {	193		if p.closed {
112		p.mu.Unlock()	194		p.mu.Unlock()
113		return ErrStreamGone	195		return ErrStreamGone
114		}	196		}
115	-	if len(p.queue) >= p.maxChunk \|\| p.queued+len(chunk) > p.maxBytes {	197	+	if p.recvUsed+p.recvReturn+cost > brokerwire.StreamWindowBytes {
116		p.mu.Unlock()	198		p.mu.Unlock()
117	-	return ErrStreamBackpressure	199	+	return ErrStreamCredit
118		}	200		}
119	-	owned := append([]byte(nil), chunk...)	201	+	p.queue = append(p.queue, pipeChunk{data: append([]byte(nil), chunk...), cost: cost})
120	-	p.queue = append(p.queue, owned)	202	+	p.recvUsed += cost
121	-	p.queued += len(owned)
122		p.mu.Unlock()	203		p.mu.Unlock()
123		select {	204		select {
124		case p.wake <- struct{}{}:	205		case p.wake <- struct{}{}:

@@ -128,7 +209,7 @@ func (p *streamPipe) deliver(chunk []byte) error {
128		}	209		}
129			210
130		// closeWith settles the pipe exactly once with cause. A nil cause is an orderly	211		// closeWith settles the pipe exactly once with cause. A nil cause is an orderly
131	-	// close; every blocked Read observes it promptly. On a nil cause, any chunks	212	+	// close; every blocked Read observes it promptly. On a nil cause, any frames
132		// already queued are kept: Read serves them first and only returns io.EOF once	213		// already queued are kept: Read serves them first and only returns io.EOF once
133		// the queue is drained, so a reply that arrived just before an orderly peer	214		// the queue is drained, so a reply that arrived just before an orderly peer
134		// close is never discarded underneath its reader. An error cause is a failure:	215		// close is never discarded underneath its reader. An error cause is a failure:

@@ -144,7 +225,7 @@ func (p *streamPipe) closeWith(cause error) {
144		p.err = cause	225		p.err = cause
145		if cause != nil {	226		if cause != nil {
146		p.queue = nil	227		p.queue = nil
147	-	p.queued = 0	228	+	p.recvUsed = 0
148		}	229		}
149		p.mu.Unlock()	230		p.mu.Unlock()
150		close(p.closedCh)	231		close(p.closedCh)

M internal/adapters/brokeripc/pipe_test.go +98 -19


@@ -5,6 +5,7 @@ import (
5	5		"errors"
6	6		"io"
7	7		"testing"
	8	+	"time"
8	9
9	10		"github.com/bnema/vev/internal/adapters/brokerwire"
10	11		"github.com/bnema/vev/internal/protocol"

@@ -71,7 +72,7 @@ func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
71	72
72	73		for _, tt := range tests {
73	74		t.Run(tt.name, func(t *testing.T) {
74		-	p := newStreamPipe(0, 0, func([]byte) error { return nil })
	75	+	p := newTestPipe()
75	76		for _, chunk := range tt.chunks {
76	77		require.NoError(t, p.deliver(chunk))
77	78		}

@@ -109,26 +110,104 @@ func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
109	110		}
110	111		}
111	112
112		-	// TestStreamPipeHoldsAFullOutputWindow pins the client pipe headroom for the
113		-	// session output window. A client claims protocol.MaxOutputWindow unacked
114		-	// Output frames; with a terminal that is not draining, all of them may sit in
115		-	// this pipe at once, and overflow fails the whole stream
116		-	// (ErrStreamBackpressure). The frame size is a typical text repaint: a
117		-	// near-full page of a 500x140 terminal measured about 70 KB. Uncompressed diff
118		-	// frames with per-cell truecolor SGR can be far larger (estimated near 1 MB at
119		-	// 300x80), and eight of those can still overflow the default 4 MiB bound.
120		-	func TestStreamPipeHoldsAFullOutputWindow(t *testing.T) {
121		-	const frameBytes = 72 << 10
	113	+	// newTestPipe is a pipe whose outbound writes and credit grants go nowhere.
	114	+	func newTestPipe() *streamPipe {
	115	+	return newStreamPipe(func([]byte) error { return nil }, func(uint64) error { return nil })
	116	+	}
	117	+
	118	+	// TestStreamWindowCoversTheSessionOutputWindow pins the invariant that lets a
	119	+	// broker stream carry a full session output window without the daemon ever
	120	+	// waiting on the stream instead of on its own ACK window: the daemon's
	121	+	// unacknowledged byte budget plus one maximum Output fits in one stream window.
	122	+	// When it does not, credit still keeps every stream correct, only slower.
	123	+	func TestStreamWindowCoversTheSessionOutputWindow(t *testing.T) {
	124	+	require.GreaterOrEqual(t, brokerwire.StreamWindowBytes, uint64(protocol.MaxOutputWindowBytes),
	125	+	"one stream window must hold the daemon's whole unacknowledged byte budget")
	126	+	}
	127	+
	128	+	// TestStreamPipeCredit covers the receive and send sides of stream credit:
	129	+	// a peer may send up to one window, the consumer's reads return credit in
	130	+	// batches, a sender waits instead of failing while it has none, and grants or
	131	+	// data beyond the window are flow-control violations.
	132	+	func TestStreamPipeCredit(t *testing.T) {
122	133		chunk := int(brokerwire.MaxStreamChunkBytes)
123		-	p := newStreamPipe(0, 0, func([]byte) error { return nil })
124		-	payload := bytes.Repeat([]byte{'x'}, frameBytes)
125		-	for frame := range protocol.MaxOutputWindow {
126		-	for data := payload; len(data) > 0; {
127		-	n := min(len(data), chunk)
128		-	require.NoError(t, p.deliver(data[:n]), "frame %d overflowed the client pipe", frame)
129		-	data = data[n:]
	134	+	cost := brokerwire.StreamChunkCredit(chunk)
	135	+	perWindow := int(brokerwire.StreamWindowBytes / cost)
	136	+
	137	+	t.Run("a full window is accepted and one more frame is a violation", func(t *testing.T) {
	138	+	p := newTestPipe()
	139	+	for range perWindow {
	140	+	require.NoError(t, p.deliver(make([]byte, chunk)))
130	141		}
131		-	}
	142	+	require.ErrorIs(t, p.deliver(make([]byte, chunk)), ErrStreamCredit)
	143	+	})
	144	+
	145	+	t.Run("reads return credit in batches", func(t *testing.T) {
	146	+	var grants []uint64
	147	+	p := newStreamPipe(func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
	148	+	for range perWindow {
	149	+	require.NoError(t, p.deliver(make([]byte, chunk)))
	150	+	}
	151	+	buf := make([]byte, chunk)
	152	+	for range perWindow {
	153	+	_, err := io.ReadFull(p, buf)
	154	+	require.NoError(t, err)
	155	+	}
	156	+	require.NotEmpty(t, grants)
	157	+	total := uint64(0)
	158	+	for _, g := range grants {
	159	+	require.GreaterOrEqual(t, g, streamCreditReturnAt(brokerwire.StreamWindowBytes))
	160	+	total += g
	161	+	}
	162	+	require.LessOrEqual(t, total, uint64(perWindow)*cost)
	163	+	// Once the returned credit reaches the peer, the window is open again.
	164	+	require.NoError(t, p.deliver(make([]byte, chunk)))
	165	+	})
	166	+
	167	+	t.Run("a sender waits for credit and resumes on a grant", func(t *testing.T) {
	168	+	p := newTestPipe()
	169	+	for range perWindow {
	170	+	require.NoError(t, p.reserve(chunk))
	171	+	}
	172	+	reserved := make(chan error, 1)
	173	+	go func() { reserved <- p.reserve(chunk) }()
	174	+	select {
	175	+	case err := <-reserved:
	176	+	t.Fatalf("reserve returned %v without credit", err)
	177	+	case <-time.After(20 * time.Millisecond):
	178	+	}
	179	+	require.NoError(t, p.granted(cost))
	180	+	select {
	181	+	case err := <-reserved:
	182	+	require.NoError(t, err)
	183	+	case <-time.After(5 * time.Second):
	184	+	t.Fatal("a grant must release a sender waiting for credit")
	185	+	}
	186	+	})
	187	+
	188	+	t.Run("close releases a sender waiting for credit", func(t *testing.T) {
	189	+	p := newTestPipe()
	190	+	for range perWindow {
	191	+	require.NoError(t, p.reserve(chunk))
	192	+	}
	193	+	reserved := make(chan error, 1)
	194	+	go func() { reserved <- p.reserve(chunk) }()
	195	+	p.closeWith(ErrStreamGone)
	196	+	select {
	197	+	case err := <-reserved:
	198	+	require.ErrorIs(t, err, ErrStreamGone)
	199	+	case <-time.After(5 * time.Second):
	200	+	t.Fatal("close must release a sender waiting for credit")
	201	+	}
	202	+	})
	203	+
	204	+	t.Run("a grant beyond the window is a violation", func(t *testing.T) {
	205	+	p := newTestPipe()
	206	+	require.ErrorIs(t, p.granted(1), ErrStreamCredit)
	207	+	require.NoError(t, p.reserve(chunk))
	208	+	require.NoError(t, p.granted(cost))
	209	+	require.ErrorIs(t, p.granted(1), ErrStreamCredit)
	210	+	})
132	211		}
133	212
134	213		// TestClientStreamPrefersTheFullOutputWindow pins what a real broker-routed


@@ -5,6 +5,7 @@ import (
5		"errors"	5		"errors"
6		"io"	6		"io"
7		"testing"	7		"testing"
			8	+	"time"
8			9
9		"github.com/bnema/vev/internal/adapters/brokerwire"	10		"github.com/bnema/vev/internal/adapters/brokerwire"
10		"github.com/bnema/vev/internal/protocol"	11		"github.com/bnema/vev/internal/protocol"

@@ -71,7 +72,7 @@ func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
71			72
72		for _, tt := range tests {	73		for _, tt := range tests {
73		t.Run(tt.name, func(t *testing.T) {	74		t.Run(tt.name, func(t *testing.T) {
74	-	p := newStreamPipe(0, 0, func([]byte) error { return nil })	75	+	p := newTestPipe()
75		for _, chunk := range tt.chunks {	76		for _, chunk := range tt.chunks {
76		require.NoError(t, p.deliver(chunk))	77		require.NoError(t, p.deliver(chunk))
77		}	78		}

@@ -109,26 +110,104 @@ func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
109		}	110		}
110		}	111		}
111			112
112	-	// TestStreamPipeHoldsAFullOutputWindow pins the client pipe headroom for the	113	+	// newTestPipe is a pipe whose outbound writes and credit grants go nowhere.
113	-	// session output window. A client claims protocol.MaxOutputWindow unacked	114	+	func newTestPipe() *streamPipe {
114	-	// Output frames; with a terminal that is not draining, all of them may sit in	115	+	return newStreamPipe(func([]byte) error { return nil }, func(uint64) error { return nil })
115	-	// this pipe at once, and overflow fails the whole stream	116	+	}
116	-	// (ErrStreamBackpressure). The frame size is a typical text repaint: a	117	+
117	-	// near-full page of a 500x140 terminal measured about 70 KB. Uncompressed diff	118	+	// TestStreamWindowCoversTheSessionOutputWindow pins the invariant that lets a
118	-	// frames with per-cell truecolor SGR can be far larger (estimated near 1 MB at	119	+	// broker stream carry a full session output window without the daemon ever
119	-	// 300x80), and eight of those can still overflow the default 4 MiB bound.	120	+	// waiting on the stream instead of on its own ACK window: the daemon's
120	-	func TestStreamPipeHoldsAFullOutputWindow(t *testing.T) {	121	+	// unacknowledged byte budget plus one maximum Output fits in one stream window.
121	-	const frameBytes = 72 << 10	122	+	// When it does not, credit still keeps every stream correct, only slower.
			123	+	func TestStreamWindowCoversTheSessionOutputWindow(t *testing.T) {
			124	+	require.GreaterOrEqual(t, brokerwire.StreamWindowBytes, uint64(protocol.MaxOutputWindowBytes),
			125	+	"one stream window must hold the daemon's whole unacknowledged byte budget")
			126	+	}
			127	+
			128	+	// TestStreamPipeCredit covers the receive and send sides of stream credit:
			129	+	// a peer may send up to one window, the consumer's reads return credit in
			130	+	// batches, a sender waits instead of failing while it has none, and grants or
			131	+	// data beyond the window are flow-control violations.
			132	+	func TestStreamPipeCredit(t *testing.T) {
122		chunk := int(brokerwire.MaxStreamChunkBytes)	133		chunk := int(brokerwire.MaxStreamChunkBytes)
123	-	p := newStreamPipe(0, 0, func([]byte) error { return nil })	134	+	cost := brokerwire.StreamChunkCredit(chunk)
124	-	payload := bytes.Repeat([]byte{'x'}, frameBytes)	135	+	perWindow := int(brokerwire.StreamWindowBytes / cost)
125	-	for frame := range protocol.MaxOutputWindow {	136	+
126	-	for data := payload; len(data) > 0; {	137	+	t.Run("a full window is accepted and one more frame is a violation", func(t *testing.T) {
127	-	n := min(len(data), chunk)	138	+	p := newTestPipe()
128	-	require.NoError(t, p.deliver(data[:n]), "frame %d overflowed the client pipe", frame)	139	+	for range perWindow {
129	-	data = data[n:]	140	+	require.NoError(t, p.deliver(make([]byte, chunk)))
130		}	141		}
131	-	}	142	+	require.ErrorIs(t, p.deliver(make([]byte, chunk)), ErrStreamCredit)
			143	+	})
			144	+
			145	+	t.Run("reads return credit in batches", func(t *testing.T) {
			146	+	var grants []uint64
			147	+	p := newStreamPipe(func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
			148	+	for range perWindow {
			149	+	require.NoError(t, p.deliver(make([]byte, chunk)))
			150	+	}
			151	+	buf := make([]byte, chunk)
			152	+	for range perWindow {
			153	+	_, err := io.ReadFull(p, buf)
			154	+	require.NoError(t, err)
			155	+	}
			156	+	require.NotEmpty(t, grants)
			157	+	total := uint64(0)
			158	+	for _, g := range grants {
			159	+	require.GreaterOrEqual(t, g, streamCreditReturnAt(brokerwire.StreamWindowBytes))
			160	+	total += g
			161	+	}
			162	+	require.LessOrEqual(t, total, uint64(perWindow)*cost)
			163	+	// Once the returned credit reaches the peer, the window is open again.
			164	+	require.NoError(t, p.deliver(make([]byte, chunk)))
			165	+	})
			166	+
			167	+	t.Run("a sender waits for credit and resumes on a grant", func(t *testing.T) {
			168	+	p := newTestPipe()
			169	+	for range perWindow {
			170	+	require.NoError(t, p.reserve(chunk))
			171	+	}
			172	+	reserved := make(chan error, 1)
			173	+	go func() { reserved <- p.reserve(chunk) }()
			174	+	select {
			175	+	case err := <-reserved:
			176	+	t.Fatalf("reserve returned %v without credit", err)
			177	+	case <-time.After(20 * time.Millisecond):
			178	+	}
			179	+	require.NoError(t, p.granted(cost))
			180	+	select {
			181	+	case err := <-reserved:
			182	+	require.NoError(t, err)
			183	+	case <-time.After(5 * time.Second):
			184	+	t.Fatal("a grant must release a sender waiting for credit")
			185	+	}
			186	+	})
			187	+
			188	+	t.Run("close releases a sender waiting for credit", func(t *testing.T) {
			189	+	p := newTestPipe()
			190	+	for range perWindow {
			191	+	require.NoError(t, p.reserve(chunk))
			192	+	}
			193	+	reserved := make(chan error, 1)
			194	+	go func() { reserved <- p.reserve(chunk) }()
			195	+	p.closeWith(ErrStreamGone)
			196	+	select {
			197	+	case err := <-reserved:
			198	+	require.ErrorIs(t, err, ErrStreamGone)
			199	+	case <-time.After(5 * time.Second):
			200	+	t.Fatal("close must release a sender waiting for credit")
			201	+	}
			202	+	})
			203	+
			204	+	t.Run("a grant beyond the window is a violation", func(t *testing.T) {
			205	+	p := newTestPipe()
			206	+	require.ErrorIs(t, p.granted(1), ErrStreamCredit)
			207	+	require.NoError(t, p.reserve(chunk))
			208	+	require.NoError(t, p.granted(cost))
			209	+	require.ErrorIs(t, p.granted(1), ErrStreamCredit)
			210	+	})
132		}	211		}
133			212
134		// TestClientStreamPrefersTheFullOutputWindow pins what a real broker-routed	213		// TestClientStreamPrefersTheFullOutputWindow pins what a real broker-routed

M internal/adapters/brokeripc/server.go +5 -0


@@ -763,6 +763,11 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
763	763		}
764	764		s.closeStreamByPeer(m.Stream)
765	765		return nil
	766	+	case brokerwire.StreamWindowUpdate:
	767	+	if !s.scopeMatches(m.Epoch, m.Connection) {
	768	+	return s.refuseScope()
	769	+	}
	770	+	return s.grantStreamCredit(m)
766	771		case brokerwire.StartPreview:
767	772		if !s.scopeMatches(m.Epoch, m.Connection) {
768	773		return s.refuseScope()


@@ -763,6 +763,11 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
763	}	763		}
764	s.closeStreamByPeer(m.Stream)	764		s.closeStreamByPeer(m.Stream)
765	return nil	765		return nil
		766	+	case brokerwire.StreamWindowUpdate:
		767	+	if !s.scopeMatches(m.Epoch, m.Connection) {
		768	+	return s.refuseScope()
		769	+	}
		770	+	return s.grantStreamCredit(m)
766	case brokerwire.StartPreview:	771		case brokerwire.StartPreview:
767	if !s.scopeMatches(m.Epoch, m.Connection) {	772		if !s.scopeMatches(m.Epoch, m.Connection) {
768	return s.refuseScope()	773		return s.refuseScope()

M internal/adapters/brokeripc/session_test.go +47 -22


@@ -856,43 +856,68 @@ func TestStalledSubscriberDoesNotBlockOtherClients(t *testing.T) {
856	856		requireConnectionDelegates(t, active)
857	857		}
858	858
859		-	// TestClientStreamBackpressureSettlesOnlyThatStream proves a local consumer that
860		-	// stops draining a stream's bounded inbound queue settles exactly that stream:
861		-	// the peer is told, the stream reports backpressure, and the connection and its
862		-	// operations stay up.
863		-	func TestClientStreamBackpressureSettlesOnlyThatStream(t *testing.T) {
	859	+	// TestStalledClientStreamWaitsForCredit proves a local consumer that stops
	860	+	// reading slows the broker relay instead of losing its stream: the daemon's
	861	+	// messages beyond one stream window wait at the broker, every one is delivered
	862	+	// in order once the consumer reads again, and the connection stays usable
	863	+	// meanwhile.
	864	+	func TestStalledClientStreamWaitsForCredit(t *testing.T) {
864	865		e := startEndpoint(t, Config{MaxClients: 2})
865		-	client := e.dialWith(Config{StreamInboundChunks: 1})
	866	+	client := e.dial()
866	867		_ = e.accept()
867	868		core := e.authority.last()
868	869
869		-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	870	+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
870	871		defer cancel()
871	872		stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
872	873		require.NoError(t, err)
	874	+	t.Cleanup(func() { _ = stream.Close() })
873	875		conn := core.logicalConn(1)
874	876		require.NotNil(t, conn)
	877	+	// Complete the session preamble before the daemon speaks.
	878	+	require.NoError(t, stream.SendClient(protocol.Ping{}))
	879	+	select {
	880	+	case <-conn.fromClient:
	881	+	case <-ctx.Done():
	882	+	t.Fatal("client message never reached the broker core")
	883	+	}
875	884
876		-	// Two daemon messages arrive while the local consumer never reads: the
877		-	// bounded capacity-one queue refuses the second one.
878		-	conn.toClient <- protocol.Pong{}
879		-	conn.toClient <- protocol.Pong{}
	885	+	// Well over one stream window of daemon output while nothing reads. Side
	886	+	// effect Outputs need no view context; random-looking data defeats
	887	+	// compression so every byte crosses the stream.
	888	+	const messages = 24
	889	+	data := make([]byte, 512<<10)
	890	+	for i := range data {
	891	+	data[i] = byte(i*131 + i/7)
	892	+	}
	893	+	go func() {
	894	+	for i := range messages {
	895	+	payload := append([]byte{byte(i)}, data...)
	896	+	select {
	897	+	case conn.toClient <- protocol.Output{Epoch: 1, Size: domain.Size{Cols: 80, Rows: 24}, Data: payload}:
	898	+	case <-ctx.Done():
	899	+	return
	900	+	}
	901	+	}
	902	+	}()
880	903
	904	+	// The relay is blocked on credit, not failed: the stream stays open and the
	905	+	// connection keeps serving other work.
881	906		select {
882	907		case <-stream.Done():
883		-	case <-time.After(5 * time.Second):
884		-	t.Fatal("a stalled stream consumer must settle the stream")
	908	+	t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())
	909	+	case <-time.After(200 * time.Millisecond):
885	910		}
886		-	require.ErrorIs(t, stream.(*clientStream).Err(), ErrStreamBackpressure)
887		-
888		-	require.Eventually(t, func() bool {
889		-	core.mu.Lock()
890		-	defer core.mu.Unlock()
891		-	return len(core.closedIDs) > 0
892		-	}, 5time.Second, 10time.Millisecond, "the broker must be told to retire the stalled stream")
893		-
894		-	// The connection survives a stream-local stall.
895	911		requireConnectionDelegates(t, client)
	912	+
	913	+	for i := range messages {
	914	+	message, err := stream.ReceiveServer()
	915	+	require.NoError(t, err)
	916	+	got, ok := message.(protocol.Output)
	917	+	require.True(t, ok, "message %d is %T", i, message)
	918	+	require.Len(t, got.Data, len(data)+1)
	919	+	require.Equal(t, byte(i), got.Data[0], "messages arrive in order")
	920	+	}
896	921		}
897	922
898	923		// TestOrderlyStreamCloseReportsNoError proves a local close is an orderly end:

 	requireConnectionDelegates(t, active)
 }
-// TestClientStreamBackpressureSettlesOnlyThatStream proves a local consumer that
+// TestStalledClientStreamWaitsForCredit proves a local consumer that stops
-// stops draining a stream's bounded inbound queue settles exactly that stream:
+// reading slows the broker relay instead of losing its stream: the daemon's
-// the peer is told, the stream reports backpressure, and the connection and its
+// messages beyond one stream window wait at the broker, every one is delivered
-// operations stay up.
+// in order once the consumer reads again, and the connection stays usable
-func TestClientStreamBackpressureSettlesOnlyThatStream(t *testing.T) {
+// meanwhile.
+func TestStalledClientStreamWaitsForCredit(t *testing.T) {
 	e := startEndpoint(t, Config{MaxClients: 2})
-	client := e.dialWith(Config{StreamInboundChunks: 1})
+	client := e.dial()
 	_ = e.accept()
 	core := e.authority.last()
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
 	require.NoError(t, err)
+	t.Cleanup(func() { _ = stream.Close() })
 	conn := core.logicalConn(1)
 	require.NotNil(t, conn)
+	// Complete the session preamble before the daemon speaks.
+	require.NoError(t, stream.SendClient(protocol.Ping{}))
+	select {
+	case <-conn.fromClient:
+	case <-ctx.Done():
+		t.Fatal("client message never reached the broker core")
+	}
-	// Two daemon messages arrive while the local consumer never reads: the
+	// Well over one stream window of daemon output while nothing reads. Side
-	// bounded capacity-one queue refuses the second one.
+	// effect Outputs need no view context; random-looking data defeats
-	conn.toClient <- protocol.Pong{}
+	// compression so every byte crosses the stream.
-	conn.toClient <- protocol.Pong{}
+	const messages = 24
+	data := make([]byte, 512<<10)
+	for i := range data {
+		data[i] = byte(i*131 + i/7)
+	}
+	go func() {
+		for i := range messages {
+			payload := append([]byte{byte(i)}, data...)
+			select {
+			case conn.toClient <- protocol.Output{Epoch: 1, Size: domain.Size{Cols: 80, Rows: 24}, Data: payload}:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	// The relay is blocked on credit, not failed: the stream stays open and the
+	// connection keeps serving other work.
 	select {
 	case <-stream.Done():
-	case <-time.After(5 * time.Second):
+		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())
-		t.Fatal("a stalled stream consumer must settle the stream")
+	case <-time.After(200 * time.Millisecond):
 	}
-	require.ErrorIs(t, stream.(*clientStream).Err(), ErrStreamBackpressure)
-	require.Eventually(t, func() bool {
-		core.mu.Lock()
-		defer core.mu.Unlock()
-		return len(core.closedIDs) > 0
-	}, 5*time.Second, 10*time.Millisecond, "the broker must be told to retire the stalled stream")
-	// The connection survives a stream-local stall.
 	requireConnectionDelegates(t, client)
+	for i := range messages {
+		message, err := stream.ReceiveServer()
+		require.NoError(t, err)
+		got, ok := message.(protocol.Output)
+		require.True(t, ok, "message %d is %T", i, message)
+		require.Len(t, got.Data, len(data)+1)
+		require.Equal(t, byte(i), got.Data[0], "messages arrive in order")
+	}
 }
 // TestOrderlyStreamCloseReportsNoError proves a local close is an orderly end:

M internal/adapters/brokeripc/stream.go +64 -29


@@ -42,13 +42,15 @@ func newServerStream(s *serverSession, id ports.BrokerStreamID, core ports.Broke
42	42		return nil, ErrConfig
43	43		}
44	44		st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
45		-	st.pipe = newStreamPipe(s.cfg.StreamInboundChunks, s.cfg.StreamInboundBytes, st.sendChunk)
	45	+	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
46	46		st.carriage = newCarriage(st.pipe)
47	47		return st, nil
48	48		}
49	49
50	50		// sendChunk splits one outbound byte slice under the negotiated stream chunk
51		-	// ceiling and writes each frame in order.
	51	+	// ceiling and writes each frame in order, waiting for the client's credit
	52	+	// before each one: a client that is not reading slows this relay, and through
	53	+	// the daemon link the daemon, instead of losing its stream.
52	54		func (st *serverStream) sendChunk(data []byte) error {
53	55		ceiling := int(st.session.ceilings.StreamChunkLimit)
54	56		if ceiling <= 0 {

@@ -56,6 +58,9 @@ func (st *serverStream) sendChunk(data []byte) error {
56	58		}
57	59		for len(data) > 0 {
58	60		n := min(len(data), ceiling)
	61	+	if err := st.pipe.reserve(n); err != nil {
	62	+	return err
	63	+	}
59	64		if err := st.session.send(brokerwire.ServerStreamData{
60	65		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],
61	66		}); err != nil {

@@ -66,6 +71,13 @@ func (st *serverStream) sendChunk(data []byte) error {
66	71		return nil
67	72		}
68	73
	74	+	// grantCredit returns consumed inbound credit to the client.
	75	+	func (st *serverStream) grantCredit(credit uint64) error {
	76	+	return st.session.send(brokerwire.StreamWindowUpdate{
	77	+	Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Credit: credit,
	78	+	})
	79	+	}
	80	+
69	81		// deliver accepts one inbound chunk from the connection reader.
70	82		func (st *serverStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
71	83

@@ -347,36 +359,47 @@ func (s *serverSession) closeStreamByPeer(id ports.BrokerStreamID) {
347	359		_ = s.core.CloseStream(s.scope.Connection, id)
348	360		}
349	361
350		-	// deliverStreamData routes one inbound client chunk to its bridge. Backpressure
351		-	// settles only that stream; a chunk for an identity the peer never opened is a
352		-	// protocol violation.
	362	+	// deliverStreamData routes one inbound client chunk to its bridge. Data beyond
	363	+	// the granted credit settles only that stream; a chunk for an identity the peer
	364	+	// never opened is a protocol violation.
353	365		func (s *serverSession) deliverStreamData(m brokerwire.ClientStreamData) error {
354		-	disposition, err := s.conn.StreamData(m.Stream)
355		-	if err != nil {
356		-	// Stream data for a stream that was never allocated, or before it was
357		-	// confirmed open, is a peer protocol violation.
358		-	return errors.Join(ErrProtocol, err)
	366	+	st, err := s.streamFor(m.Stream)
	367	+	if st == nil \|\| err != nil {
	368	+	return err
359	369		}
360		-	if disposition == brokerwire.StreamDiscarded {
361		-	return nil
	370	+	if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
	371	+	s.settleStream(st, ErrStreamCredit)
362	372		}
363		-	st := s.lookupStream(m.Stream)
364		-	if st == nil {
365		-	// The bridge was settled between the tracker check and the lookup (a
366		-	// stream-local failure won the race); the chunk is discarded with it.
367		-	return nil
	373	+	// Any other error means the stream already reached its terminal outcome;
	374	+	// its data is discarded.
	375	+	return nil
	376	+	}
	377	+
	378	+	// grantStreamCredit applies one client credit grant to its bridge.
	379	+	func (s *serverSession) grantStreamCredit(m brokerwire.StreamWindowUpdate) error {
	380	+	st, err := s.streamFor(m.Stream)
	381	+	if st == nil \|\| err != nil {
	382	+	return err
368	383		}
369		-	switch err := st.deliver(m.Data); {
370		-	case err == nil:
371		-	return nil
372		-	case errors.Is(err, ErrStreamBackpressure):
373		-	s.settleStream(st, ErrStreamBackpressure)
374		-	return nil
375		-	default:
376		-	// The stream already reached its terminal outcome; its data is
377		-	// discarded.
378		-	return nil
	384	+	if err := st.pipe.granted(m.Credit); err != nil {
	385	+	s.settleStream(st, err)
	386	+	}
	387	+	return nil
	388	+	}
	389	+
	390	+	// streamFor classifies one stream-scoped frame and returns its live bridge. A
	391	+	// frame for a retired stream, or one settled between the tracker check and the
	392	+	// lookup, yields no bridge and no error; a frame for a stream that was never
	393	+	// allocated, or before it was confirmed open, is a peer protocol violation.
	394	+	func (s serverSession) streamFor(id ports.BrokerStreamID) (serverStream, error) {
	395	+	disposition, err := s.conn.StreamData(id)
	396	+	if err != nil {
	397	+	return nil, errors.Join(ErrProtocol, err)
	398	+	}
	399	+	if disposition == brokerwire.StreamDiscarded {
	400	+	return nil, nil
379	401		}
	402	+	return s.lookupStream(id), nil
380	403		}
381	404
382	405		// clientStream is the client-side half of one logical stream: the sessionwire

@@ -407,7 +430,7 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
407	430		opened: make(chan error, 1),
408	431		done: make(chan struct{}),
409	432		}
410		-	st.pipe = newStreamPipe(c.cfg.StreamInboundChunks, c.cfg.StreamInboundBytes, st.sendChunk)
	433	+	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
411	434		st.carriage = newCarriage(st.pipe)
412	435		st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
413	436		if st.ClientConnection == nil {

@@ -417,7 +440,7 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
417	440		}
418	441
419	442		// sendChunk splits one outbound client byte slice under the negotiated stream
420		-	// chunk ceiling.
	443	+	// chunk ceiling, waiting for the broker's credit before each frame.
421	444		func (st *clientStream) sendChunk(data []byte) error {
422	445		ceiling := int(st.client.ceilings.StreamChunkLimit)
423	446		if ceiling <= 0 {

@@ -425,6 +448,9 @@ func (st *clientStream) sendChunk(data []byte) error {
425	448		}
426	449		for len(data) > 0 {
427	450		n := min(len(data), ceiling)
	451	+	if err := st.pipe.reserve(n); err != nil {
	452	+	return err
	453	+	}
428	454		if err := st.client.send(brokerwire.ClientStreamData{
429	455		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],
430	456		}); err != nil {

@@ -435,6 +461,15 @@ func (st *clientStream) sendChunk(data []byte) error {
435	461		return nil
436	462		}
437	463
	464	+	// grantCredit returns consumed inbound credit to the broker. It is written
	465	+	// synchronously because a dropped grant would stall the stream forever; the
	466	+	// write never waits long, since the broker's reader never blocks on a stream.
	467	+	func (st *clientStream) grantCredit(credit uint64) error {
	468	+	return st.client.send(brokerwire.StreamWindowUpdate{
	469	+	Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Credit: credit,
	470	+	})
	471	+	}
	472	+
438	473		// deliver accepts one inbound server chunk.
439	474		func (st *clientStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
440	475


@@ -42,13 +42,15 @@ func newServerStream(s *serverSession, id ports.BrokerStreamID, core ports.Broke
42		return nil, ErrConfig	42		return nil, ErrConfig
43		}	43		}
44		st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}	44		st := &serverStream{session: s, id: id, core: core, done: make(chan struct{})}
45	-	st.pipe = newStreamPipe(s.cfg.StreamInboundChunks, s.cfg.StreamInboundBytes, st.sendChunk)	45	+	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
46		st.carriage = newCarriage(st.pipe)	46		st.carriage = newCarriage(st.pipe)
47		return st, nil	47		return st, nil
48		}	48		}
49			49
50		// sendChunk splits one outbound byte slice under the negotiated stream chunk	50		// sendChunk splits one outbound byte slice under the negotiated stream chunk
51	-	// ceiling and writes each frame in order.	51	+	// ceiling and writes each frame in order, waiting for the client's credit
			52	+	// before each one: a client that is not reading slows this relay, and through
			53	+	// the daemon link the daemon, instead of losing its stream.
52		func (st *serverStream) sendChunk(data []byte) error {	54		func (st *serverStream) sendChunk(data []byte) error {
53		ceiling := int(st.session.ceilings.StreamChunkLimit)	55		ceiling := int(st.session.ceilings.StreamChunkLimit)
54		if ceiling <= 0 {	56		if ceiling <= 0 {

@@ -56,6 +58,9 @@ func (st *serverStream) sendChunk(data []byte) error {
56		}	58		}
57		for len(data) > 0 {	59		for len(data) > 0 {
58		n := min(len(data), ceiling)	60		n := min(len(data), ceiling)
			61	+	if err := st.pipe.reserve(n); err != nil {
			62	+	return err
			63	+	}
59		if err := st.session.send(brokerwire.ServerStreamData{	64		if err := st.session.send(brokerwire.ServerStreamData{
60		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],	65		Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Data: data[:n],
61		}); err != nil {	66		}); err != nil {

@@ -66,6 +71,13 @@ func (st *serverStream) sendChunk(data []byte) error {
66		return nil	71		return nil
67		}	72		}
68			73
			74	+	// grantCredit returns consumed inbound credit to the client.
			75	+	func (st *serverStream) grantCredit(credit uint64) error {
			76	+	return st.session.send(brokerwire.StreamWindowUpdate{
			77	+	Epoch: st.session.epoch, Connection: st.session.scope.Connection, Stream: st.id, Credit: credit,
			78	+	})
			79	+	}
			80	+
69		// deliver accepts one inbound chunk from the connection reader.	81		// deliver accepts one inbound chunk from the connection reader.
70		func (st *serverStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }	82		func (st *serverStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
71			83

@@ -347,36 +359,47 @@ func (s *serverSession) closeStreamByPeer(id ports.BrokerStreamID) {
347		_ = s.core.CloseStream(s.scope.Connection, id)	359		_ = s.core.CloseStream(s.scope.Connection, id)
348		}	360		}
349			361
350	-	// deliverStreamData routes one inbound client chunk to its bridge. Backpressure	362	+	// deliverStreamData routes one inbound client chunk to its bridge. Data beyond
351	-	// settles only that stream; a chunk for an identity the peer never opened is a	363	+	// the granted credit settles only that stream; a chunk for an identity the peer
352	-	// protocol violation.	364	+	// never opened is a protocol violation.
353		func (s *serverSession) deliverStreamData(m brokerwire.ClientStreamData) error {	365		func (s *serverSession) deliverStreamData(m brokerwire.ClientStreamData) error {
354	-	disposition, err := s.conn.StreamData(m.Stream)	366	+	st, err := s.streamFor(m.Stream)
355	-	if err != nil {	367	+	if st == nil \|\| err != nil {
356	-	// Stream data for a stream that was never allocated, or before it was	368	+	return err
357	-	// confirmed open, is a peer protocol violation.
358	-	return errors.Join(ErrProtocol, err)
359		}	369		}
360	-	if disposition == brokerwire.StreamDiscarded {	370	+	if err := st.deliver(m.Data); errors.Is(err, ErrStreamCredit) {
361	-	return nil	371	+	s.settleStream(st, ErrStreamCredit)
362		}	372		}
363	-	st := s.lookupStream(m.Stream)	373	+	// Any other error means the stream already reached its terminal outcome;
364	-	if st == nil {	374	+	// its data is discarded.
365	-	// The bridge was settled between the tracker check and the lookup (a	375	+	return nil
366	-	// stream-local failure won the race); the chunk is discarded with it.	376	+	}
367	-	return nil	377	+
			378	+	// grantStreamCredit applies one client credit grant to its bridge.
			379	+	func (s *serverSession) grantStreamCredit(m brokerwire.StreamWindowUpdate) error {
			380	+	st, err := s.streamFor(m.Stream)
			381	+	if st == nil \|\| err != nil {
			382	+	return err
368		}	383		}
369	-	switch err := st.deliver(m.Data); {	384	+	if err := st.pipe.granted(m.Credit); err != nil {
370	-	case err == nil:	385	+	s.settleStream(st, err)
371	-	return nil	386	+	}
372	-	case errors.Is(err, ErrStreamBackpressure):	387	+	return nil
373	-	s.settleStream(st, ErrStreamBackpressure)	388	+	}
374	-	return nil	389	+
375	-	default:	390	+	// streamFor classifies one stream-scoped frame and returns its live bridge. A
376	-	// The stream already reached its terminal outcome; its data is	391	+	// frame for a retired stream, or one settled between the tracker check and the
377	-	// discarded.	392	+	// lookup, yields no bridge and no error; a frame for a stream that was never
378	-	return nil	393	+	// allocated, or before it was confirmed open, is a peer protocol violation.
			394	+	func (s serverSession) streamFor(id ports.BrokerStreamID) (serverStream, error) {
			395	+	disposition, err := s.conn.StreamData(id)
			396	+	if err != nil {
			397	+	return nil, errors.Join(ErrProtocol, err)
			398	+	}
			399	+	if disposition == brokerwire.StreamDiscarded {
			400	+	return nil, nil
379		}	401		}
			402	+	return s.lookupStream(id), nil
380		}	403		}
381			404
382		// clientStream is the client-side half of one logical stream: the sessionwire	405		// clientStream is the client-side half of one logical stream: the sessionwire

@@ -407,7 +430,7 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
407		opened: make(chan error, 1),	430		opened: make(chan error, 1),
408		done: make(chan struct{}),	431		done: make(chan struct{}),
409		}	432		}
410	-	st.pipe = newStreamPipe(c.cfg.StreamInboundChunks, c.cfg.StreamInboundBytes, st.sendChunk)	433	+	st.pipe = newStreamPipe(st.sendChunk, st.grantCredit)
411		st.carriage = newCarriage(st.pipe)	434		st.carriage = newCarriage(st.pipe)
412		st.ClientConnection = sessionwire.NewClientConnection(st.carriage)	435		st.ClientConnection = sessionwire.NewClientConnection(st.carriage)
413		if st.ClientConnection == nil {	436		if st.ClientConnection == nil {

@@ -417,7 +440,7 @@ func newClientStream(c client, id ports.BrokerStreamID) (clientStream, error)
417		}	440		}
418			441
419		// sendChunk splits one outbound client byte slice under the negotiated stream	442		// sendChunk splits one outbound client byte slice under the negotiated stream
420	-	// chunk ceiling.	443	+	// chunk ceiling, waiting for the broker's credit before each frame.
421		func (st *clientStream) sendChunk(data []byte) error {	444		func (st *clientStream) sendChunk(data []byte) error {
422		ceiling := int(st.client.ceilings.StreamChunkLimit)	445		ceiling := int(st.client.ceilings.StreamChunkLimit)
423		if ceiling <= 0 {	446		if ceiling <= 0 {

@@ -425,6 +448,9 @@ func (st *clientStream) sendChunk(data []byte) error {
425		}	448		}
426		for len(data) > 0 {	449		for len(data) > 0 {
427		n := min(len(data), ceiling)	450		n := min(len(data), ceiling)
			451	+	if err := st.pipe.reserve(n); err != nil {
			452	+	return err
			453	+	}
428		if err := st.client.send(brokerwire.ClientStreamData{	454		if err := st.client.send(brokerwire.ClientStreamData{
429		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],	455		Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Data: data[:n],
430		}); err != nil {	456		}); err != nil {

@@ -435,6 +461,15 @@ func (st *clientStream) sendChunk(data []byte) error {
435		return nil	461		return nil
436		}	462		}
437			463
			464	+	// grantCredit returns consumed inbound credit to the broker. It is written
			465	+	// synchronously because a dropped grant would stall the stream forever; the
			466	+	// write never waits long, since the broker's reader never blocks on a stream.
			467	+	func (st *clientStream) grantCredit(credit uint64) error {
			468	+	return st.client.send(brokerwire.StreamWindowUpdate{
			469	+	Epoch: st.client.scope.Epoch, Connection: st.client.scope.Connection, Stream: st.id, Credit: credit,
			470	+	})
			471	+	}
			472	+
438		// deliver accepts one inbound server chunk.	473		// deliver accepts one inbound server chunk.
439		func (st *clientStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }	474		func (st *clientStream) deliver(chunk []byte) error { return st.pipe.deliver(chunk) }
440			475

M internal/adapters/brokerwire/codec.go +53 -0


@@ -241,6 +241,17 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
241	241		return nil, ErrInvalidMessage
242	242		}
243	243		return encodeClientEnvelope(*m, maxChunkBytes)
	244	+	case StreamWindowUpdate:
	245	+	converted, err := streamWindowUpdateToWire(m)
	246	+	if err != nil {
	247	+	return nil, err
	248	+	}
	249	+	return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
	250	+	case *StreamWindowUpdate:
	251	+	if m == nil {
	252	+	return nil, ErrInvalidMessage
	253	+	}
	254	+	return encodeClientEnvelope(*m, maxChunkBytes)
244	255		case StartPreview:
245	256		converted, err := startPreviewToWire(m)
246	257		if err != nil {

@@ -308,6 +319,8 @@ func decodeClientEnvelope(envelope *wire.BrokerClientEnvelope, maxChunkBytes uin
308	319		return nil, ErrInvalidMessage
309	320		}
310	321		return CloseStream{Epoch: epoch, Connection: connection, Stream: stream}, nil
	322	+	case *wire.BrokerClientEnvelope_StreamWindowUpdate:
	323	+	return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
311	324		case *wire.BrokerClientEnvelope_StartPreview:
312	325		return startPreviewFromWire(payload.StartPreview)
313	326		case *wire.BrokerClientEnvelope_CancelPreview:

@@ -707,6 +720,33 @@ func openStreamFromWire(message *wire.OpenStream) (OpenStream, error) {
707	720		return candidate, nil
708	721		}
709	722
	723	+	// streamWindowUpdateToWire refuses a zero or above-window grant: credit is
	724	+	// returned only for consumed data, so a valid grant never exceeds one window.
	725	+	func streamWindowUpdateToWire(m StreamWindowUpdate) (*wire.StreamWindowUpdate, error) {
	726	+	if m.Epoch == 0 \|\| m.Connection.Validate() != nil \|\| m.Stream.Validate() != nil {
	727	+	return nil, ErrInvalidMessage
	728	+	}
	729	+	if m.Credit == 0 \|\| m.Credit > StreamWindowBytes {
	730	+	return nil, ErrInvalidMessage
	731	+	}
	732	+	return &wire.StreamWindowUpdate{Ref: refToWire(m.Epoch, m.Connection, m.Stream), Credit: m.Credit}, nil
	733	+	}
	734	+
	735	+	func streamWindowUpdateFromWire(message *wire.StreamWindowUpdate) (StreamWindowUpdate, error) {
	736	+	if message == nil {
	737	+	return StreamWindowUpdate{}, ErrInvalidMessage
	738	+	}
	739	+	epoch, connection, stream, err := refFromWire(message.GetRef())
	740	+	if err != nil {
	741	+	return StreamWindowUpdate{}, ErrInvalidMessage
	742	+	}
	743	+	credit := message.GetCredit()
	744	+	if credit == 0 \|\| credit > StreamWindowBytes {
	745	+	return StreamWindowUpdate{}, ErrInvalidMessage
	746	+	}
	747	+	return StreamWindowUpdate{Epoch: epoch, Connection: connection, Stream: stream, Credit: credit}, nil
	748	+	}
	749	+
710	750		func clientStreamDataToWire(m ClientStreamData, maxChunkBytes uint64) (*wire.ClientStreamData, error) {
711	751		if m.Epoch == 0 {
712	752		return nil, ErrInvalidMessage

@@ -828,6 +868,17 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
828	868		return nil, ErrInvalidMessage
829	869		}
830	870		return encodeServerEnvelope(*m, maxChunkBytes)
	871	+	case StreamWindowUpdate:
	872	+	converted, err := streamWindowUpdateToWire(m)
	873	+	if err != nil {
	874	+	return nil, err
	875	+	}
	876	+	return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
	877	+	case *StreamWindowUpdate:
	878	+	if m == nil {
	879	+	return nil, ErrInvalidMessage
	880	+	}
	881	+	return encodeServerEnvelope(*m, maxChunkBytes)
831	882		case StreamClosed:
832	883		converted, err := streamClosedToWire(m)
833	884		if err != nil {

@@ -929,6 +980,8 @@ func decodeServerEnvelope(envelope *wire.BrokerServerEnvelope, maxChunkBytes uin
929	980		return serverStreamDataFromWire(payload.ServerStreamData, maxChunkBytes)
930	981		case *wire.BrokerServerEnvelope_StreamClosed:
931	982		return streamClosedFromWire(payload.StreamClosed)
	983	+	case *wire.BrokerServerEnvelope_StreamWindowUpdate:
	984	+	return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
932	985		case *wire.BrokerServerEnvelope_Progress:
933	986		return progressFromWire(payload.Progress)
934	987		case *wire.BrokerServerEnvelope_BrokerErrorMessage:


@@ -241,6 +241,17 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
241	return nil, ErrInvalidMessage	241		return nil, ErrInvalidMessage
242	}	242		}
243	return encodeClientEnvelope(*m, maxChunkBytes)	243		return encodeClientEnvelope(*m, maxChunkBytes)
		244	+	case StreamWindowUpdate:
		245	+	converted, err := streamWindowUpdateToWire(m)
		246	+	if err != nil {
		247	+	return nil, err
		248	+	}
		249	+	return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
		250	+	case *StreamWindowUpdate:
		251	+	if m == nil {
		252	+	return nil, ErrInvalidMessage
		253	+	}
		254	+	return encodeClientEnvelope(*m, maxChunkBytes)
244	case StartPreview:	255		case StartPreview:
245	converted, err := startPreviewToWire(m)	256		converted, err := startPreviewToWire(m)
246	if err != nil {	257		if err != nil {

@@ -308,6 +319,8 @@ func decodeClientEnvelope(envelope *wire.BrokerClientEnvelope, maxChunkBytes uin
308	return nil, ErrInvalidMessage	319		return nil, ErrInvalidMessage
309	}	320		}
310	return CloseStream{Epoch: epoch, Connection: connection, Stream: stream}, nil	321		return CloseStream{Epoch: epoch, Connection: connection, Stream: stream}, nil
		322	+	case *wire.BrokerClientEnvelope_StreamWindowUpdate:
		323	+	return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
311	case *wire.BrokerClientEnvelope_StartPreview:	324		case *wire.BrokerClientEnvelope_StartPreview:
312	return startPreviewFromWire(payload.StartPreview)	325		return startPreviewFromWire(payload.StartPreview)
313	case *wire.BrokerClientEnvelope_CancelPreview:	326		case *wire.BrokerClientEnvelope_CancelPreview:

@@ -707,6 +720,33 @@ func openStreamFromWire(message *wire.OpenStream) (OpenStream, error) {
707	return candidate, nil	720		return candidate, nil
708	}	721		}
709		722
		723	+	// streamWindowUpdateToWire refuses a zero or above-window grant: credit is
		724	+	// returned only for consumed data, so a valid grant never exceeds one window.
		725	+	func streamWindowUpdateToWire(m StreamWindowUpdate) (*wire.StreamWindowUpdate, error) {
		726	+	if m.Epoch == 0 \|\| m.Connection.Validate() != nil \|\| m.Stream.Validate() != nil {
		727	+	return nil, ErrInvalidMessage
		728	+	}
		729	+	if m.Credit == 0 \|\| m.Credit > StreamWindowBytes {
		730	+	return nil, ErrInvalidMessage
		731	+	}
		732	+	return &wire.StreamWindowUpdate{Ref: refToWire(m.Epoch, m.Connection, m.Stream), Credit: m.Credit}, nil
		733	+	}
		734	+
		735	+	func streamWindowUpdateFromWire(message *wire.StreamWindowUpdate) (StreamWindowUpdate, error) {
		736	+	if message == nil {
		737	+	return StreamWindowUpdate{}, ErrInvalidMessage
		738	+	}
		739	+	epoch, connection, stream, err := refFromWire(message.GetRef())
		740	+	if err != nil {
		741	+	return StreamWindowUpdate{}, ErrInvalidMessage
		742	+	}
		743	+	credit := message.GetCredit()
		744	+	if credit == 0 \|\| credit > StreamWindowBytes {
		745	+	return StreamWindowUpdate{}, ErrInvalidMessage
		746	+	}
		747	+	return StreamWindowUpdate{Epoch: epoch, Connection: connection, Stream: stream, Credit: credit}, nil
		748	+	}
		749	+
710	func clientStreamDataToWire(m ClientStreamData, maxChunkBytes uint64) (*wire.ClientStreamData, error) {	750		func clientStreamDataToWire(m ClientStreamData, maxChunkBytes uint64) (*wire.ClientStreamData, error) {
711	if m.Epoch == 0 {	751		if m.Epoch == 0 {
712	return nil, ErrInvalidMessage	752		return nil, ErrInvalidMessage

@@ -828,6 +868,17 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
828	return nil, ErrInvalidMessage	868		return nil, ErrInvalidMessage
829	}	869		}
830	return encodeServerEnvelope(*m, maxChunkBytes)	870		return encodeServerEnvelope(*m, maxChunkBytes)
		871	+	case StreamWindowUpdate:
		872	+	converted, err := streamWindowUpdateToWire(m)
		873	+	if err != nil {
		874	+	return nil, err
		875	+	}
		876	+	return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
		877	+	case *StreamWindowUpdate:
		878	+	if m == nil {
		879	+	return nil, ErrInvalidMessage
		880	+	}
		881	+	return encodeServerEnvelope(*m, maxChunkBytes)
831	case StreamClosed:	882		case StreamClosed:
832	converted, err := streamClosedToWire(m)	883		converted, err := streamClosedToWire(m)
833	if err != nil {	884		if err != nil {

@@ -929,6 +980,8 @@ func decodeServerEnvelope(envelope *wire.BrokerServerEnvelope, maxChunkBytes uin
929	return serverStreamDataFromWire(payload.ServerStreamData, maxChunkBytes)	980		return serverStreamDataFromWire(payload.ServerStreamData, maxChunkBytes)
930	case *wire.BrokerServerEnvelope_StreamClosed:	981		case *wire.BrokerServerEnvelope_StreamClosed:
931	return streamClosedFromWire(payload.StreamClosed)	982		return streamClosedFromWire(payload.StreamClosed)
		983	+	case *wire.BrokerServerEnvelope_StreamWindowUpdate:
		984	+	return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
932	case *wire.BrokerServerEnvelope_Progress:	985		case *wire.BrokerServerEnvelope_Progress:
933	return progressFromWire(payload.Progress)	986		return progressFromWire(payload.Progress)
934	case *wire.BrokerServerEnvelope_BrokerErrorMessage:	987		case *wire.BrokerServerEnvelope_BrokerErrorMessage:

M internal/adapters/brokerwire/limits.go +14 -0


@@ -39,8 +39,22 @@ const (
39	39		// MaxStreamChunkBytes is the largest legal stream chunk ceiling:
40	40		// 64 KiB per opaque stream frame.
41	41		MaxStreamChunkBytes uint64 = 64 << 10
	42	+
	43	+	// StreamWindowBytes is the stream-data credit each side of one logical
	44	+	// stream grants its peer per direction. A sender spends StreamChunkCredit
	45	+	// per data frame and waits when it has too little, so the receiver never
	46	+	// queues more than this per stream; StreamWindowUpdate returns the credit
	47	+	// of consumed frames.
	48	+	StreamWindowBytes uint64 = 4 << 20
	49	+	// StreamChunkCreditOverhead is the fixed credit one data frame costs on top
	50	+	// of its payload, so a burst of tiny frames is bounded in count as well.
	51	+	StreamChunkCreditOverhead uint64 = 64
42	52		)
43	53
	54	+	// StreamChunkCredit is the flow-control cost of one stream data frame of n
	55	+	// payload bytes.
	56	+	func StreamChunkCredit(n int) uint64 { return uint64(n) + StreamChunkCreditOverhead }
	57	+
44	58		// Ceilings are the immutable per-connection ceilings negotiated once by the
45	59		// broker preamble. MaxReceiveEnvelopeBytes bounds every envelope this side
46	60		// accepts; StreamChunkLimit bounds one opaque stream data frame in either


@@ -39,8 +39,22 @@ const (
39	// MaxStreamChunkBytes is the largest legal stream chunk ceiling:	39		// MaxStreamChunkBytes is the largest legal stream chunk ceiling:
40	// 64 KiB per opaque stream frame.	40		// 64 KiB per opaque stream frame.
41	MaxStreamChunkBytes uint64 = 64 << 10	41		MaxStreamChunkBytes uint64 = 64 << 10
		42	+
		43	+	// StreamWindowBytes is the stream-data credit each side of one logical
		44	+	// stream grants its peer per direction. A sender spends StreamChunkCredit
		45	+	// per data frame and waits when it has too little, so the receiver never
		46	+	// queues more than this per stream; StreamWindowUpdate returns the credit
		47	+	// of consumed frames.
		48	+	StreamWindowBytes uint64 = 4 << 20
		49	+	// StreamChunkCreditOverhead is the fixed credit one data frame costs on top
		50	+	// of its payload, so a burst of tiny frames is bounded in count as well.
		51	+	StreamChunkCreditOverhead uint64 = 64
42	)	52		)
43		53
		54	+	// StreamChunkCredit is the flow-control cost of one stream data frame of n
		55	+	// payload bytes.
		56	+	func StreamChunkCredit(n int) uint64 { return uint64(n) + StreamChunkCreditOverhead }
		57	+
44	// Ceilings are the immutable per-connection ceilings negotiated once by the	58		// Ceilings are the immutable per-connection ceilings negotiated once by the
45	// broker preamble. MaxReceiveEnvelopeBytes bounds every envelope this side	59		// broker preamble. MaxReceiveEnvelopeBytes bounds every envelope this side
46	// accepts; StreamChunkLimit bounds one opaque stream data frame in either	60		// accepts; StreamChunkLimit bounds one opaque stream data frame in either

M internal/adapters/brokerwire/messages.go +13 -0


@@ -165,6 +165,19 @@ type CloseStream struct {
165	165
166	166		func (CloseStream) brokerClientMessage() {}
167	167
	168	+	// StreamWindowUpdate grants the peer Credit more stream-data bytes on one
	169	+	// logical stream after the receiver consumed them. It travels in both
	170	+	// directions: each side grants credit for the data it receives.
	171	+	type StreamWindowUpdate struct {
	172	+	Epoch ports.BrokerEpoch
	173	+	Connection ports.BrokerConnectionID
	174	+	Stream ports.BrokerStreamID
	175	+	Credit uint64
	176	+	}
	177	+
	178	+	func (StreamWindowUpdate) brokerClientMessage() {}
	179	+	func (StreamWindowUpdate) brokerServerMessage() {}
	180	+
168	181		// StartPreview starts or replaces one connection-scoped live preview.
169	182		// Generation is the connection-scoped preview authority; Route names the
170	183		// observed daemon (the broker allocates the stream) and Preview is the bounded


@@ -165,6 +165,19 @@ type CloseStream struct {
165		165
166	func (CloseStream) brokerClientMessage() {}	166		func (CloseStream) brokerClientMessage() {}
167		167
		168	+	// StreamWindowUpdate grants the peer Credit more stream-data bytes on one
		169	+	// logical stream after the receiver consumed them. It travels in both
		170	+	// directions: each side grants credit for the data it receives.
		171	+	type StreamWindowUpdate struct {
		172	+	Epoch ports.BrokerEpoch
		173	+	Connection ports.BrokerConnectionID
		174	+	Stream ports.BrokerStreamID
		175	+	Credit uint64
		176	+	}
		177	+
		178	+	func (StreamWindowUpdate) brokerClientMessage() {}
		179	+	func (StreamWindowUpdate) brokerServerMessage() {}
		180	+
168	// StartPreview starts or replaces one connection-scoped live preview.	181		// StartPreview starts or replaces one connection-scoped live preview.
169	// Generation is the connection-scoped preview authority; Route names the	182		// Generation is the connection-scoped preview authority; Route names the
170	// observed daemon (the broker allocates the stream) and Preview is the bounded	183		// observed daemon (the broker allocates the stream) and Preview is the bounded

M internal/adapters/brokerwire/state.go +7 -0


@@ -486,6 +486,13 @@ func (c *Connection) StreamData(stream ports.BrokerStreamID) (StreamDisposition,
486	486		return c.streams.Data(c.scope, stream)
487	487		}
488	488
	489	+	// StreamWindowUpdate classifies one credit grant for a stream exactly like
	490	+	// stream data: legal while open, discarded once retired. The ready check and
	491	+	// the tracker lookup share one connection lock.
	492	+	func (c *Connection) StreamWindowUpdate(stream ports.BrokerStreamID) (StreamDisposition, error) {
	493	+	return c.StreamData(stream)
	494	+	}
	495	+
489	496		// StreamProgress reports one phase of an in-flight stream operation. The
490	497		// ready check and the tracker mutation share one connection lock.
491	498		func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {


@@ -486,6 +486,13 @@ func (c *Connection) StreamData(stream ports.BrokerStreamID) (StreamDisposition,
486	return c.streams.Data(c.scope, stream)	486		return c.streams.Data(c.scope, stream)
487	}	487		}
488		488
		489	+	// StreamWindowUpdate classifies one credit grant for a stream exactly like
		490	+	// stream data: legal while open, discarded once retired. The ready check and
		491	+	// the tracker lookup share one connection lock.
		492	+	func (c *Connection) StreamWindowUpdate(stream ports.BrokerStreamID) (StreamDisposition, error) {
		493	+	return c.StreamData(stream)
		494	+	}
		495	+
489	// StreamProgress reports one phase of an in-flight stream operation. The	496		// StreamProgress reports one phase of an in-flight stream operation. The
490	// ready check and the tracker mutation share one connection lock.	497		// ready check and the tracker mutation share one connection lock.
491	func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {	498		func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {

M internal/protocol/session.go +1 -1


@@ -17,7 +17,7 @@ import (
17	17		// daemonmux physical preamble conversation (multiplex.proto), whose
18	18		// negotiated ceilings and accepted daemon binding ride the same wire
19	19		// version as the session and broker conversations.
20		-	const Version uint16 = 65
	20	+	const Version uint16 = 66
21	21
22	22		// HandshakeTimeout bounds every transport handshake from connect through the
23	23		// first committed publication. It excludes the preceding client-local


@@ -17,7 +17,7 @@ import (
17		// daemonmux physical preamble conversation (multiplex.proto), whose	17		// daemonmux physical preamble conversation (multiplex.proto), whose
18		// negotiated ceilings and accepted daemon binding ride the same wire	18		// negotiated ceilings and accepted daemon binding ride the same wire
19		// version as the session and broker conversations.	19		// version as the session and broker conversations.
20	-	const Version uint16 = 65	20	+	const Version uint16 = 66
21			21
22		// HandshakeTimeout bounds every transport handshake from connect through the	22		// HandshakeTimeout bounds every transport handshake from connect through the
23		// first committed publication. It excludes the preceding client-local	23		// first committed publication. It excludes the preceding client-local

M internal/protocol/session_test.go +1 -1


@@ -10,7 +10,7 @@ import (
10	10		)
11	11
12	12		func TestVersionRemainsWireCompatible(t *testing.T) {
13		-	require.Equal(t, uint16(65), Version)
	13	+	require.Equal(t, uint16(66), Version)
14	14		require.Equal(t, (16<<20)-407, MaxOutputDataLen)
15	15		}
16	16


@@ -10,7 +10,7 @@ import (
10		)	10		)
11			11
12		func TestVersionRemainsWireCompatible(t *testing.T) {	12		func TestVersionRemainsWireCompatible(t *testing.T) {
13	-	require.Equal(t, uint16(65), Version)	13	+	require.Equal(t, uint16(66), Version)
14		require.Equal(t, (16<<20)-407, MaxOutputDataLen)	14		require.Equal(t, (16<<20)-407, MaxOutputDataLen)
15		}	15		}
16			16

M internal/protocol/wire/broker.pb.go +262 -163 generated file: collapsed

load diff

M internal/protocol/wire/broker_test.go +8 -2


@@ -20,7 +20,7 @@ var brokerClientTags = map[string]protoreflect.FieldNumber{
20	20		"register": 101, "subscribe": 102, "resync": 103, "unsubscribe": 104,
21	21		"add_host": 105, "remove_host": 106, "reconcile": 107, "open_stream": 108,
22	22		"client_stream_data": 109, "close_stream": 110, "update_host_policy": 111,
23		-	"start_preview": 112, "cancel_preview": 113,
	23	+	"start_preview": 112, "cancel_preview": 113, "stream_window_update": 114,
24	24		}
25	25
26	26		// brokerServerTags is the frozen server tag inventory.

@@ -28,7 +28,7 @@ var brokerServerTags = map[string]protoreflect.FieldNumber{
28	28		"registered": 201, "snapshot_part": 202, "operation_result": 203,
29	29		"stream_opened": 204, "server_stream_data": 205, "stream_closed": 206,
30	30		"progress": 207, "broker_error_message": 208, "shutdown": 209,
31		-	"preview_publication": 210,
	31	+	"preview_publication": 210, "stream_window_update": 211,
32	32		}
33	33
34	34		func oneofTags(t *testing.T, message proto.Message) map[string]protoreflect.FieldNumber {

@@ -194,6 +194,9 @@ func brokerClientSamples() map[string]*BrokerClientEnvelope {
194	194		"cancel_preview": {Payload: &BrokerClientEnvelope_CancelPreview{CancelPreview: &CancelPreview{
195	195		Scope: brokerScope(), Generation: 3,
196	196		}}},
	197	+	"stream_window_update": {Payload: &BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
	198	+	Ref: brokerRef(), Credit: 4096,
	199	+	}}},
197	200		}
198	201		}
199	202

@@ -241,6 +244,9 @@ func brokerServerSamples() map[string]*BrokerServerEnvelope {
241	244		Scope: brokerScope(), Generation: 3,
242	245		Result: &PreviewPublication_Error{Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"}},
243	246		}}},
	247	+	"stream_window_update": {Payload: &BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
	248	+	Ref: brokerRef(), Credit: 4096,
	249	+	}}},
244	250		}
245	251		}
246	252

 	"register": 101, "subscribe": 102, "resync": 103, "unsubscribe": 104,
 	"add_host": 105, "remove_host": 106, "reconcile": 107, "open_stream": 108,
 	"client_stream_data": 109, "close_stream": 110, "update_host_policy": 111,
-	"start_preview": 112, "cancel_preview": 113,
+	"start_preview": 112, "cancel_preview": 113, "stream_window_update": 114,
 }
 // brokerServerTags is the frozen server tag inventory.
 	"registered": 201, "snapshot_part": 202, "operation_result": 203,
 	"stream_opened": 204, "server_stream_data": 205, "stream_closed": 206,
 	"progress": 207, "broker_error_message": 208, "shutdown": 209,
-	"preview_publication": 210,
+	"preview_publication": 210, "stream_window_update": 211,
 }
 func oneofTags(t *testing.T, message proto.Message) map[string]protoreflect.FieldNumber {
 		"cancel_preview": {Payload: &BrokerClientEnvelope_CancelPreview{CancelPreview: &CancelPreview{
 			Scope: brokerScope(), Generation: 3,
 		}}},
+		"stream_window_update": {Payload: &BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
+			Ref: brokerRef(), Credit: 4096,
+		}}},
 	}
 }
 			Scope: brokerScope(), Generation: 3,
 			Result: &PreviewPublication_Error{Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"}},
 		}}},
+		"stream_window_update": {Payload: &BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
+			Ref: brokerRef(), Credit: 4096,
+		}}},
 	}
 }

M internal/protocol/wire/envelope_inventory_test.go +2 -2


@@ -15,8 +15,8 @@ import (
15	15		// shares a payload type except route navigation failure, and the preamble
16	16		// constants agree with the schema authority.
17	17		func TestEnvelopeInventoryProvesDirectionAndSemanticPath(t *testing.T) {
18		-	if protocol.Version != 65 {
19		-	t.Fatalf("protocol.Version = %d, want 65", protocol.Version)
	18	+	if protocol.Version != 66 {
	19	+	t.Fatalf("protocol.Version = %d, want 66", protocol.Version)
20	20		}
21	21		if ProtocolEpoch != 1 {
22	22		t.Fatalf("ProtocolEpoch = %d, want 1", ProtocolEpoch)


@@ -15,8 +15,8 @@ import (
15		// shares a payload type except route navigation failure, and the preamble	15		// shares a payload type except route navigation failure, and the preamble
16		// constants agree with the schema authority.	16		// constants agree with the schema authority.
17		func TestEnvelopeInventoryProvesDirectionAndSemanticPath(t *testing.T) {	17		func TestEnvelopeInventoryProvesDirectionAndSemanticPath(t *testing.T) {
18	-	if protocol.Version != 65 {	18	+	if protocol.Version != 66 {
19	-	t.Fatalf("protocol.Version = %d, want 65", protocol.Version)	19	+	t.Fatalf("protocol.Version = %d, want 66", protocol.Version)
20		}	20		}
21		if ProtocolEpoch != 1 {	21		if ProtocolEpoch != 1 {
22		t.Fatalf("ProtocolEpoch = %d, want 1", ProtocolEpoch)	22		t.Fatalf("ProtocolEpoch = %d, want 1", ProtocolEpoch)

M internal/protocol/wire/schema/broker.proto +11 -0


@@ -89,6 +89,7 @@ message BrokerClientEnvelope {
89	89		UpdateHostPolicy update_host_policy = 111;
90	90		StartPreview start_preview = 112;
91	91		CancelPreview cancel_preview = 113;
	92	+	StreamWindowUpdate stream_window_update = 114;
92	93		}
93	94		}
94	95

@@ -106,6 +107,7 @@ message BrokerServerEnvelope {
106	107		BrokerErrorMessage broker_error_message = 208;
107	108		Shutdown shutdown = 209;
108	109		PreviewPublication preview_publication = 210;
	110	+	StreamWindowUpdate stream_window_update = 211;
109	111		}
110	112		}
111	113

@@ -221,6 +223,15 @@ message CloseStream {
221	223		BrokerStreamRef ref = 1;
222	224		}
223	225
	226	+	// StreamWindowUpdate grants the peer credit to send more stream data on one
	227	+	// logical stream: the receiver consumed that much of its stream window. Each
	228	+	// direction of a stream has its own window, so a client grants credit for
	229	+	// ServerStreamData and the server grants credit for ClientStreamData.
	230	+	message StreamWindowUpdate {
	231	+	BrokerStreamRef ref = 1;
	232	+	uint64 credit = 2;
	233	+	}
	234	+
224	235		// Registered confirms one accepted client connection and its assigned scope.
225	236		message Registered {
226	237		BrokerScope scope = 1;


@@ -89,6 +89,7 @@ message BrokerClientEnvelope {
89	UpdateHostPolicy update_host_policy = 111;	89		UpdateHostPolicy update_host_policy = 111;
90	StartPreview start_preview = 112;	90		StartPreview start_preview = 112;
91	CancelPreview cancel_preview = 113;	91		CancelPreview cancel_preview = 113;
		92	+	StreamWindowUpdate stream_window_update = 114;
92	}	93		}
93	}	94		}
94		95

@@ -106,6 +107,7 @@ message BrokerServerEnvelope {
106	BrokerErrorMessage broker_error_message = 208;	107		BrokerErrorMessage broker_error_message = 208;
107	Shutdown shutdown = 209;	108		Shutdown shutdown = 209;
108	PreviewPublication preview_publication = 210;	109		PreviewPublication preview_publication = 210;
		110	+	StreamWindowUpdate stream_window_update = 211;
109	}	111		}
110	}	112		}
111		113

@@ -221,6 +223,15 @@ message CloseStream {
221	BrokerStreamRef ref = 1;	223		BrokerStreamRef ref = 1;
222	}	224		}
223		225
		226	+	// StreamWindowUpdate grants the peer credit to send more stream data on one
		227	+	// logical stream: the receiver consumed that much of its stream window. Each
		228	+	// direction of a stream has its own window, so a client grants credit for
		229	+	// ServerStreamData and the server grants credit for ClientStreamData.
		230	+	message StreamWindowUpdate {
		231	+	BrokerStreamRef ref = 1;
		232	+	uint64 credit = 2;
		233	+	}
		234	+
224	// Registered confirms one accepted client connection and its assigned scope.	235		// Registered confirms one accepted client connection and its assigned scope.
225	message Registered {	236		message Registered {
226	BrokerScope scope = 1;	237		BrokerScope scope = 1;