1package daemonmux
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "runtime"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/domain"
16)
17
18// errFakeCarrierClosed is the synthetic carrier error a fakeCarrier returns
19// once Close released its blocked I/O. It mirrors the adapter prompt-close
20// contract: Close unblocks an in-flight Send or Receive.
21var errFakeCarrierClosed = errors.New("fakeCarrier: closed")
22
23// fakeFrame is one inbound carrier event: a complete envelope payload, or the
24// error the next Receive must report.
25type fakeFrame struct {
26 payload []byte
27 err error
28}
29
30// fakeCarrier is the deterministic FramedCarrier the pump tests drive. Inbound
31// frames are delivered in order through a buffered channel; outbound frames are
32// recorded in the order the writer sent them. Close closes a channel that every
33// blocked Receive and Send observes, so it unblocks both callers exactly once.
34type fakeCarrier struct {
35 mu sync.Mutex
36 calls int
37 // hold, when non-nil, makes Send wait for its release before recording.
38 hold chan struct{}
39 holdOnce sync.Once
40 closeErr error
41 closeOnce sync.Once
42 entered atomic.Int64
43
44 inbound chan fakeFrame
45 sent chan []byte
46 closed chan struct{}
47}
48
49func newFakeCarrier() *fakeCarrier {
50 return &fakeCarrier{
51 inbound: make(chan fakeFrame, 1024),
52 sent: make(chan []byte, 1024),
53 closed: make(chan struct{}),
54 }
55}
56
57func (c *fakeCarrier) Receive(ctx context.Context) ([]byte, error) {
58 select {
59 case frame := <-c.inbound:
60 if frame.err != nil {
61 return nil, frame.err
62 }
63 return frame.payload, nil
64 case <-c.closed:
65 return nil, errFakeCarrierClosed
66 case <-ctx.Done():
67 return nil, ctx.Err()
68 }
69}
70
71func (c *fakeCarrier) Send(ctx context.Context, payload []byte) error {
72 c.entered.Add(1)
73 c.mu.Lock()
74 hold := c.hold
75 c.mu.Unlock()
76 if hold != nil {
77 select {
78 case <-hold:
79 case <-c.closed:
80 return errFakeCarrierClosed
81 case <-ctx.Done():
82 return ctx.Err()
83 }
84 }
85 select {
86 case c.sent <- append([]byte(nil), payload...):
87 return nil
88 case <-c.closed:
89 return errFakeCarrierClosed
90 case <-ctx.Done():
91 return ctx.Err()
92 }
93}
94
95func (c *fakeCarrier) Close() error {
96 c.closeOnce.Do(func() {
97 c.mu.Lock()
98 c.calls++
99 c.mu.Unlock()
100 close(c.closed)
101 })
102 c.mu.Lock()
103 defer c.mu.Unlock()
104 return c.closeErr
105}
106
107func (c *fakeCarrier) closeCalls() int {
108 c.mu.Lock()
109 defer c.mu.Unlock()
110 return c.calls
111}
112
113// setHold makes the next Send wait until release is called.
114func (c *fakeCarrier) setHold(hold chan struct{}) {
115 c.mu.Lock()
116 defer c.mu.Unlock()
117 c.hold = hold
118}
119
120// release unblocks a held Send exactly once.
121func (c *fakeCarrier) release(hold chan struct{}) {
122 c.holdOnce.Do(func() { close(hold) })
123}
124
125// enteredSends reports how many Send calls the writer has started.
126func (c *fakeCarrier) enteredSends() int64 { return c.entered.Load() }
127
128// deliver queues one inbound frame for the next Receive, in order.
129func (c *fakeCarrier) deliver(t *testing.T, frame fakeFrame) {
130 t.Helper()
131 select {
132 case c.inbound <- frame:
133 case <-time.After(5 * time.Second):
134 t.Fatal("fakeCarrier: inbound delivery timed out")
135 }
136}
137
138// nextSent returns the next outbound frame the writer recorded, in order.
139func (c *fakeCarrier) nextSent(t *testing.T) []byte {
140 t.Helper()
141 select {
142 case payload := <-c.sent:
143 return payload
144 case <-time.After(5 * time.Second):
145 t.Fatal("fakeCarrier: expected an outbound frame")
146 return nil
147 }
148}
149
150// requireNoSent fails when the writer recorded an outbound frame within d.
151func (c *fakeCarrier) requireNoSent(t *testing.T, d time.Duration) {
152 t.Helper()
153 select {
154 case payload := <-c.sent:
155 t.Fatalf("fakeCarrier: unexpected outbound frame %q", payload)
156 case <-time.After(d):
157 }
158}
159
160// newTestPump builds a pump over a fresh fakeCarrier under the default
161// ceilings and closes it when the test ends.
162func newTestPump(t *testing.T, inbound EnvelopeDirection) (*Pump, *fakeCarrier) {
163 t.Helper()
164 return newTestPumpWithCeilings(t, inbound, DefaultMuxCeilings())
165}
166
167func newTestPumpWithCeilings(t *testing.T, inbound EnvelopeDirection, ceilings MuxCeilings) (*Pump, *fakeCarrier) {
168 t.Helper()
169 carrier := newFakeCarrier()
170 pump, err := NewPump(carrier, inbound, ceilings)
171 require.NoError(t, err)
172 t.Cleanup(func() { _ = pump.Close() })
173 return pump, carrier
174}
175
176// openFor returns one complete, encodable attachment Open for a physical
177// identity: the wire codec validates the whole request, not just the reference.
178func openFor(id PhysicalStreamID) Open {
179 open := testOpen()
180 open.Ref = testRef(id)
181 return open
182}
183
184func deliverClientFrame(t *testing.T, carrier *fakeCarrier, message ClientMessage) {
185 t.Helper()
186 carrier.deliver(t, fakeFrame{payload: mustEncodeClient(t, message)})
187}
188
189func deliverServerFrame(t *testing.T, carrier *fakeCarrier, message ServerMessage) {
190 t.Helper()
191 carrier.deliver(t, fakeFrame{payload: mustEncodeServer(t, message)})
192}
193
194// requireEngineEventually waits until the engine satisfies cond.
195func requireEngineEventually(t *testing.T, pump *Pump, cond func(*StreamEngine) bool, msgAndArgs ...any) {
196 t.Helper()
197 require.Eventually(t, func() bool { return cond(pump.Engine()) }, 5*time.Second, time.Millisecond, msgAndArgs...)
198}
199
200// requireStreamTerminalEventually waits until one stream has reached its
201// terminal state with Done closed, then returns that settled read.
202func requireStreamTerminalEventually(t *testing.T, pump *Pump, id PhysicalStreamID) StreamStatus {
203 t.Helper()
204 var status StreamStatus
205 require.Eventually(t, func() bool {
206 current, ok := pump.Engine().Status(id)
207 if !ok {
208 return false
209 }
210 status = current
211 return current.State == StreamTerminal && channelClosed(current.Done)
212 }, 5*time.Second, time.Millisecond, "stream %d did not reach terminal", id)
213 return status
214}
215
216// requireStreamClosingEventually waits until one undrained stream has settled
217// into closing with its preserved chunk still queued and charged and Done still
218// open, then returns that settled read. The transition is stable until the
219// chunk is drained, so the returned read cannot be a stale intermediate state.
220func requireStreamClosingEventually(t *testing.T, pump *Pump, id PhysicalStreamID) StreamStatus {
221 t.Helper()
222 var status StreamStatus
223 require.Eventually(t, func() bool {
224 current, ok := pump.Engine().Status(id)
225 if !ok {
226 return false
227 }
228 status = current
229 return current.State == StreamClosing &&
230 current.QueuedChunks == 1 &&
231 current.QueuedBytes == 1 &&
232 !channelClosed(current.Done)
233 }, 5*time.Second, time.Millisecond, "stream %d did not settle into closing with a preserved chunk", id)
234 return status
235}
236
237// takeEventually waits until one stream's inbound queue yields a chunk.
238func takeEventually(t *testing.T, pump *Pump, id PhysicalStreamID) ([]byte, bool) {
239 t.Helper()
240 var chunk []byte
241 require.Eventually(t, func() bool {
242 taken, ok := pump.Take(id)
243 if ok {
244 chunk = taken
245 }
246 return ok
247 }, 5*time.Second, time.Millisecond)
248 return chunk, chunk != nil
249}
250
251// decodeReset decodes one outbound Reset frame in the given direction.
252func decodeReset(t *testing.T, payload []byte, direction EnvelopeDirection) Reset {
253 t.Helper()
254 var message any
255 var err error
256 if direction == DirectionServer {
257 message, err = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
258 } else {
259 message, err = DecodeClient(payload, testEnvelopeCeiling, testChunkCeiling)
260 }
261 require.NoError(t, err)
262 reset, ok := message.(Reset)
263 require.True(t, ok, "outbound frame is not a Reset")
264 return reset
265}
266
267func TestNewPumpValidation(t *testing.T) {
268 t.Run("missing carrier", func(t *testing.T) {
269 pump, err := NewPump(nil, DirectionClient, DefaultMuxCeilings())
270 require.ErrorIs(t, err, ErrPumpConfig)
271 require.Nil(t, pump)
272 })
273 t.Run("unknown inbound direction", func(t *testing.T) {
274 for _, direction := range []EnvelopeDirection{0, EnvelopeDirection(200)} {
275 pump, err := NewPump(newFakeCarrier(), direction, DefaultMuxCeilings())
276 require.ErrorIs(t, err, ErrPumpConfig)
277 require.Nil(t, pump)
278 }
279 })
280 t.Run("invalid ceilings", func(t *testing.T) {
281 pump, err := NewPump(newFakeCarrier(), DirectionClient, MuxCeilings{})
282 require.ErrorIs(t, err, ErrInvalidCeilings)
283 require.Nil(t, pump)
284 })
285 t.Run("valid", func(t *testing.T) {
286 pump, carrier := newTestPumpWithCeilings(t, DirectionServer, MuxCeilings{
287 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
288 StreamChunkLimit: MinMuxChunkBytes,
289 MaxStreams: MinMuxStreams,
290 MaxAggregateBytes: MinMuxAggregateBytes,
291 })
292 require.NotNil(t, pump.Engine())
293 require.False(t, channelClosed(pump.Done()))
294 require.NoError(t, pump.Err())
295 require.Equal(t, domain.RemoteFailureNone, pump.FailureKind())
296 require.NoError(t, pump.Close())
297 require.True(t, channelClosed(pump.Done()))
298 require.Equal(t, 1, carrier.closeCalls())
299 require.ErrorIs(t, pump.SendData(1, []byte("x"), nil), ErrPhysicalClosed)
300 require.ErrorIs(t, pump.Send(Data{Physical: 1, Data: []byte("x")}), ErrDataNeedsCredit, "stream data never bypasses credit")
301 require.ErrorIs(t, pump.Send(&Data{Physical: 1, Data: []byte("x")}), ErrDataNeedsCredit, "nor in pointer form")
302 })
303}
304
305// TestPumpTwoAndHundredStreams drives inbound and outbound traffic over two
306// and a hundred concurrent streams: every inbound envelope is applied, every
307// locally admitted stream's frame is written, and closing every stream leaves
308// the physical connection open.
309func TestPumpTwoAndHundredStreams(t *testing.T) {
310 for _, count := range []int{2, 100} {
311 t.Run(fmt.Sprintf("%d streams", count), func(t *testing.T) {
312 pump, carrier := newTestPump(t, DirectionClient)
313 pump.Start(context.Background())
314
315 ids := make([]PhysicalStreamID, count)
316 for i := range ids {
317 ids[i] = PhysicalStreamID(i + 1)
318 deliverClientFrame(t, carrier, openFor(ids[i]))
319 }
320 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == count })
321
322 for _, id := range ids {
323 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
324 deliverClientFrame(t, carrier, Data{Physical: id, Data: []byte{byte(id)}})
325 }
326 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.AggregateBytes() == count })
327
328 taken, ok := pump.Take(ids[0])
329 require.True(t, ok)
330 require.Equal(t, []byte{byte(ids[0])}, taken)
331 require.Equal(t, count-1, pump.Engine().AggregateBytes())
332
333 for _, id := range ids {
334 require.NoError(t, pump.SendData(id, []byte("out"), nil))
335 }
336 seen := make(map[PhysicalStreamID]int, count)
337 for i := 0; i < count; i++ {
338 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
339 require.NoError(t, err)
340 data, ok := message.(Data)
341 require.True(t, ok)
342 require.Equal(t, []byte("out"), data.Data)
343 seen[data.Physical]++
344 }
345 require.Len(t, seen, count)
346
347 for _, id := range ids {
348 deliverClientFrame(t, carrier, Close{Physical: id})
349 }
350 // The drained stream settles at once; every other stream is closing
351 // and still holds its one unread, charged chunk until it is drained.
352 // The reader applies the closes asynchronously and in order, so each
353 // stream is observed on its own eventual transition: a global Live
354 // precondition would pass as soon as the drained stream settles,
355 // before a later stream has applied its own close.
356 terminal := requireStreamTerminalEventually(t, pump, ids[0])
357 require.Equal(t, StreamTerminal, terminal.State)
358 for _, id := range ids[1:] {
359 status := requireStreamClosingEventually(t, pump, id)
360 require.Equal(t, 1, status.QueuedChunks)
361 require.Equal(t, 1, status.QueuedBytes)
362 }
363 require.Equal(t, count-1, pump.Engine().Live())
364 require.Equal(t, count-1, pump.Engine().AggregateBytes())
365 require.False(t, pump.Engine().Closed())
366 require.False(t, channelClosed(pump.Done()))
367
368 // Draining every preserved chunk releases its bytes and terminalizes
369 // the stream; the physical connection stays open.
370 for _, id := range ids[1:] {
371 chunk, ok := takeEventually(t, pump, id)
372 require.True(t, ok)
373 require.Equal(t, []byte{byte(id)}, chunk)
374 }
375 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 0 })
376 require.Zero(t, pump.Engine().AggregateBytes())
377 require.False(t, pump.Engine().Closed())
378 require.False(t, channelClosed(pump.Done()))
379 for _, id := range ids {
380 status := mustStatus(t, pump.Engine(), id)
381 require.Equal(t, StreamTerminal, status.State)
382 require.True(t, channelClosed(status.Done))
383 require.NoError(t, status.Err)
384 }
385 })
386 }
387}
388
389// TestPumpBlockedConsumerDoesNotBlockReaderOrSibling proves a logical consumer
390// that never drains one stream's inbound queue neither blocks the reader nor a
391// sibling: the stalled stream is reset by its own bound while the sibling keeps
392// receiving.
393func TestPumpBlockedConsumerDoesNotBlockReaderOrSibling(t *testing.T) {
394 pump, carrier := newTestPump(t, DirectionClient)
395 pump.Start(context.Background())
396
397 const siblings = 2
398 for i := 1; i <= siblings; i++ {
399 deliverClientFrame(t, carrier, openFor(PhysicalStreamID(i)))
400 }
401 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == siblings })
402 for i := 1; i <= siblings; i++ {
403 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(PhysicalStreamID(i))}))
404 }
405
406 // A peer ignores flow control and fills stream 1's whole window while the
407 // consumer never Takes it.
408 chunk, full := floodWindow(pump.Ceilings())
409 for i := 0; i < full; i++ {
410 deliverClientFrame(t, carrier, Data{Physical: 1, Data: chunk})
411 }
412 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
413 status, ok := e.Status(1)
414 return ok && status.QueuedChunks == full
415 })
416
417 // One more chunk exceeds the granted window; a sibling chunk still lands,
418 // so neither the reader nor the sibling was blocked.
419 deliverClientFrame(t, carrier, Data{Physical: 1, Data: chunk})
420 deliverClientFrame(t, carrier, Data{Physical: 2, Data: []byte("sibling")})
421
422 taken, ok := takeEventually(t, pump, 2)
423 require.True(t, ok)
424 require.Equal(t, []byte("sibling"), taken)
425
426 stalled := mustStatus(t, pump.Engine(), 1)
427 require.Equal(t, StreamTerminal, stalled.State)
428 require.ErrorIs(t, stalled.Err, ErrStreamQueueFull)
429 require.Equal(t, domain.RemoteFailureInvalidResponse, stalled.FailureKind)
430 require.Zero(t, stalled.QueuedChunks)
431 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
432 require.False(t, pump.Engine().Closed())
433 require.False(t, channelClosed(pump.Done()))
434
435 reset := decodeReset(t, carrier.nextSent(t), DirectionServer)
436 require.Equal(t, PhysicalStreamID(1), reset.Physical)
437 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
438}
439
440// TestPumpBlockedWriterDoesNotBlockReader proves the reader keeps applying
441// inbound frames while the writer is blocked inside a slow carrier Send.
442func TestPumpBlockedWriterDoesNotBlockReader(t *testing.T) {
443 pump, carrier := newTestPump(t, DirectionClient)
444 pump.Start(context.Background())
445
446 deliverClientFrame(t, carrier, openFor(1))
447 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
448 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
449
450 hold := make(chan struct{})
451 carrier.setHold(hold)
452 require.NoError(t, pump.Send(Reset{Physical: 1}))
453 require.Eventually(t, func() bool { return carrier.enteredSends() >= 1 }, 5*time.Second, time.Millisecond)
454
455 for i := 0; i < 3; i++ {
456 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte{byte(i)}})
457 }
458 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.AggregateBytes() == 3 })
459
460 // Nothing reaches the wire until the carrier releases the blocked write.
461 carrier.requireNoSent(t, 50*time.Millisecond)
462 carrier.release(hold)
463
464 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
465 require.NoError(t, err)
466 _, ok := message.(Reset)
467 require.True(t, ok)
468}
469
470// TestPumpFairWriterProgress proves the writer preserves the scheduler's
471// deterministic byte-deficit round-robin order and each stream's own order.
472func TestPumpFairWriterProgress(t *testing.T) {
473 pump, carrier := newTestPump(t, DirectionClient)
474 pump.Start(context.Background())
475
476 const count = 4
477 ids := make([]PhysicalStreamID, count)
478 for i := range ids {
479 ids[i] = PhysicalStreamID(i + 1)
480 deliverClientFrame(t, carrier, openFor(ids[i]))
481 }
482 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == count })
483
484 const rounds = 2
485 for round := 0; round < rounds; round++ {
486 for _, id := range ids {
487 require.NoError(t, pump.SendData(id, []byte{byte(id), byte(round)}, nil))
488 }
489 }
490
491 seen := make(map[PhysicalStreamID]int, count)
492 for served := 0; served < rounds*count; served++ {
493 want := ids[served%count]
494 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
495 require.NoError(t, err)
496 data, ok := message.(Data)
497 require.True(t, ok)
498 require.Equal(t, want, data.Physical, "dequeue %d", served)
499 require.Equal(t, []byte{byte(want), byte(seen[want])}, data.Data, "per-stream order preserved")
500 seen[want]++
501 }
502 for _, id := range ids {
503 require.Equal(t, rounds, seen[id])
504 }
505}
506
507// TestPumpResetIsolation proves one stream-local refusal schedules exactly one
508// Reset for its own stream, settles it locally, and leaves the physical
509// connection and every sibling healthy.
510func TestPumpResetIsolation(t *testing.T) {
511 t.Run("semantic refusal", func(t *testing.T) {
512 pump, carrier := newTestPump(t, DirectionClient)
513 pump.Start(context.Background())
514
515 for _, id := range []PhysicalStreamID{1, 2, 3} {
516 deliverClientFrame(t, carrier, openFor(id))
517 }
518 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 3 })
519 for _, id := range []PhysicalStreamID{2, 3} {
520 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
521 }
522
523 // Stream 1 is still opening: inbound data is a stream-local state
524 // refusal.
525 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("early")})
526 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
527 status, ok := e.Status(1)
528 return ok && status.State == StreamTerminal
529 })
530 settled := mustStatus(t, pump.Engine(), 1)
531 require.ErrorContains(t, settled.Err, "invalid stream state")
532 require.Equal(t, domain.RemoteFailureInvalidResponse, settled.FailureKind)
533
534 deliverClientFrame(t, carrier, Data{Physical: 2, Data: []byte("sibling")})
535 taken, ok := takeEventually(t, pump, 2)
536 require.True(t, ok)
537 require.Equal(t, []byte("sibling"), taken)
538
539 reset := decodeReset(t, carrier.nextSent(t), DirectionServer)
540 require.Equal(t, PhysicalStreamID(1), reset.Physical)
541 require.True(t, reset.HasError)
542 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
543
544 // Exactly one Reset: a later frame for the settled stream is
545 // discarded, so no second frame is scheduled.
546 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("late")})
547 carrier.requireNoSent(t, 50*time.Millisecond)
548
549 require.False(t, pump.Engine().Closed())
550 require.False(t, channelClosed(pump.Done()))
551 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
552 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 3).State)
553 require.Equal(t, 2, pump.Engine().Live())
554 })
555
556 t.Run("queue overflow", func(t *testing.T) {
557 pump, carrier := newTestPump(t, DirectionClient)
558 pump.Start(context.Background())
559
560 for _, id := range []PhysicalStreamID{1, 2} {
561 deliverClientFrame(t, carrier, openFor(id))
562 }
563 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 2 })
564 for _, id := range []PhysicalStreamID{1, 2} {
565 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
566 }
567
568 chunk, full := floodWindow(pump.Ceilings())
569 for i := 0; i < full; i++ {
570 deliverClientFrame(t, carrier, Data{Physical: 1, Data: chunk})
571 }
572 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
573 status, ok := e.Status(1)
574 return ok && status.QueuedChunks == full
575 })
576 deliverClientFrame(t, carrier, Data{Physical: 1, Data: chunk})
577 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
578 status, ok := e.Status(1)
579 return ok && status.State == StreamTerminal
580 })
581
582 deliverClientFrame(t, carrier, Data{Physical: 2, Data: []byte("sibling")})
583 taken, ok := takeEventually(t, pump, 2)
584 require.True(t, ok)
585 require.Equal(t, []byte("sibling"), taken)
586
587 reset := decodeReset(t, carrier.nextSent(t), DirectionServer)
588 require.Equal(t, PhysicalStreamID(1), reset.Physical)
589 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
590 require.False(t, pump.Engine().Closed())
591 })
592}
593
594// TestPumpMalformedPhysicalFatal proves a malformed outer envelope is
595// physical-fatal: Done closes before every stream is terminalized, the carrier
596// is closed, and both goroutines stop.
597func TestPumpMalformedPhysicalFatal(t *testing.T) {
598 tight := MuxCeilings{
599 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
600 StreamChunkLimit: MinMuxChunkBytes,
601 MaxStreams: MinMuxStreams,
602 MaxAggregateBytes: MinMuxAggregateBytes,
603 }
604 cases := []struct {
605 name string
606 ceilings MuxCeilings
607 payload []byte
608 }{
609 {"garbage payload", DefaultMuxCeilings(), []byte{0xde, 0xad, 0xbe, 0xef}},
610 {"empty payload", DefaultMuxCeilings(), nil},
611 {"wrong direction", DefaultMuxCeilings(), mustEncodeServer(t, Opened{Ref: testRef(1)})},
612 {"chunk above negotiated ceiling", tight, mustEncodeClient(t, Data{Physical: 1, Data: []byte("ab")})},
613 }
614 for _, tc := range cases {
615 t.Run(tc.name, func(t *testing.T) {
616 pump, carrier := newTestPumpWithCeilings(t, DirectionClient, tc.ceilings)
617 pump.Start(context.Background())
618 deliverClientFrame(t, carrier, openFor(1))
619 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
620
621 carrier.deliver(t, fakeFrame{payload: tc.payload})
622 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Closed() })
623
624 require.Error(t, pump.Err())
625 require.Equal(t, domain.RemoteFailureInvalidResponse, pump.FailureKind())
626 require.True(t, channelClosed(pump.Engine().Done()))
627 status := mustStatus(t, pump.Engine(), 1)
628 require.Equal(t, StreamTerminal, status.State)
629 require.True(t, channelClosed(status.Done))
630 require.Equal(t, domain.RemoteFailureInvalidResponse, status.FailureKind)
631
632 require.Eventually(t, func() bool {
633 return channelClosed(pump.readerDone) && channelClosed(pump.writerDone)
634 }, 5*time.Second, time.Millisecond)
635 require.Eventually(t, func() bool { return carrier.closeCalls() == 1 }, 5*time.Second, time.Millisecond)
636 })
637 }
638}
639
640// TestPumpCarrierLossFanoutOrdering proves a carrier error is physical-fatal,
641// that the physical Done closes before every stream Done, and that every live
642// stream observes the same transport cause.
643func TestPumpCarrierLossFanoutOrdering(t *testing.T) {
644 pump, carrier := newTestPump(t, DirectionClient)
645 pump.Start(context.Background())
646
647 const count = 3
648 for i := 1; i <= count; i++ {
649 deliverClientFrame(t, carrier, openFor(PhysicalStreamID(i)))
650 }
651 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == count })
652
653 // Each watcher records whether the physical Done was already closed when
654 // its stream's Done fired.
655 ordered := make(chan bool, count)
656 for i := 1; i <= count; i++ {
657 done := mustStatus(t, pump.Engine(), PhysicalStreamID(i)).Done
658 go func(streamDone <-chan struct{}) {
659 select {
660 case <-streamDone:
661 case <-time.After(5 * time.Second):
662 ordered <- false
663 return
664 }
665 select {
666 case <-pump.Done():
667 ordered <- true
668 case <-time.After(5 * time.Second):
669 ordered <- false
670 }
671 }(done)
672 }
673
674 cause := errors.New("carrier lost")
675 carrier.deliver(t, fakeFrame{err: cause})
676 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Closed() })
677
678 for i := 0; i < count; i++ {
679 require.True(t, <-ordered, "stream Done fired before the physical Done")
680 }
681 require.ErrorIs(t, pump.Err(), cause)
682 require.Equal(t, domain.RemoteFailureTransport, pump.FailureKind())
683 require.Zero(t, pump.Engine().Live())
684 for i := 1; i <= count; i++ {
685 status := mustStatus(t, pump.Engine(), PhysicalStreamID(i))
686 require.Equal(t, StreamTerminal, status.State)
687 require.True(t, channelClosed(status.Done))
688 require.ErrorIs(t, status.Err, cause)
689 require.Equal(t, domain.RemoteFailureTransport, status.FailureKind)
690 }
691 require.Eventually(t, func() bool {
692 return channelClosed(pump.readerDone) && channelClosed(pump.writerDone)
693 }, 5*time.Second, time.Millisecond)
694 require.Equal(t, 1, carrier.closeCalls())
695}
696
697// TestPumpCancellationNoLeak proves cancelling the run context is an orderly
698// close: the physical connection settles with a nil cause, streams settle with
699// nil, the carrier is closed, and no pump goroutine survives.
700func TestPumpCancellationNoLeak(t *testing.T) {
701 before := runtime.NumGoroutine()
702 pump, carrier := newTestPump(t, DirectionClient)
703 ctx, cancel := context.WithCancel(context.Background())
704 pump.Start(ctx)
705
706 for _, id := range []PhysicalStreamID{1, 2} {
707 deliverClientFrame(t, carrier, openFor(id))
708 }
709 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 2 })
710
711 cancel()
712 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Closed() })
713 require.NoError(t, pump.Err())
714 require.Equal(t, domain.RemoteFailureNone, pump.FailureKind())
715 for _, id := range []PhysicalStreamID{1, 2} {
716 status := mustStatus(t, pump.Engine(), id)
717 require.Equal(t, StreamTerminal, status.State)
718 require.True(t, channelClosed(status.Done))
719 require.NoError(t, status.Err)
720 require.Equal(t, domain.RemoteFailureNone, status.FailureKind)
721 }
722 require.Eventually(t, func() bool { return channelClosed(pump.readerDone) }, 5*time.Second, time.Millisecond)
723 require.Eventually(t, func() bool { return channelClosed(pump.writerDone) }, 5*time.Second, time.Millisecond)
724 require.Eventually(t, func() bool { return carrier.closeCalls() == 1 }, 5*time.Second, time.Millisecond)
725 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+2 }, 5*time.Second, time.Millisecond,
726 "pump goroutines leaked after cancellation")
727}
728
729// TestPumpConcurrentClose proves concurrent Close calls are idempotent, each
730// returns the same result, the carrier is closed exactly once, and Close
731// unblocks a writer blocked inside the carrier and joins both goroutines
732// within the adapter prompt-close contract.
733func TestPumpConcurrentClose(t *testing.T) {
734 pump, carrier := newTestPump(t, DirectionClient)
735 pump.Start(context.Background())
736
737 deliverClientFrame(t, carrier, openFor(1))
738 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
739
740 hold := make(chan struct{})
741 carrier.setHold(hold)
742 require.NoError(t, pump.Send(Reset{Physical: 1}))
743 require.Eventually(t, func() bool { return carrier.enteredSends() >= 1 }, 5*time.Second, time.Millisecond)
744
745 const closers = 12
746 errs := make([]error, closers)
747 var wg sync.WaitGroup
748 for i := 0; i < closers; i++ {
749 wg.Add(1)
750 go func(index int) {
751 defer wg.Done()
752 errs[index] = pump.Close()
753 }(i)
754 }
755 joined := make(chan struct{})
756 go func() {
757 wg.Wait()
758 close(joined)
759 }()
760 select {
761 case <-joined:
762 case <-time.After(5 * time.Second):
763 t.Fatal("concurrent Close did not unblock the carrier and join both goroutines")
764 }
765
766 for i, err := range errs {
767 require.NoError(t, err, "closer %d", i)
768 }
769 require.True(t, channelClosed(pump.Done()))
770 require.NoError(t, pump.Err())
771 require.Equal(t, domain.RemoteFailureNone, pump.FailureKind())
772 require.Equal(t, 1, carrier.closeCalls())
773 require.True(t, channelClosed(pump.readerDone))
774 require.True(t, channelClosed(pump.writerDone))
775 require.ErrorIs(t, pump.SendData(1, []byte("x"), nil), ErrPhysicalClosed)
776}
777
778// TestPumpCloseBeforeStart proves Close never starts or joins goroutines that
779// were never launched, and that Start after Close is a no-op.
780func TestPumpCloseBeforeStart(t *testing.T) {
781 pump, carrier := newTestPump(t, DirectionClient)
782 require.NoError(t, pump.Close())
783 require.True(t, channelClosed(pump.Done()))
784 require.NoError(t, pump.Err())
785 require.Equal(t, 1, carrier.closeCalls())
786
787 pump.Start(context.Background())
788 require.False(t, channelClosed(pump.readerDone))
789 require.False(t, channelClosed(pump.writerDone))
790 require.NoError(t, pump.Close())
791 require.Equal(t, 1, carrier.closeCalls())
792}
793
794// TestPumpSendDirection proves Send refuses a message whose direction does not
795// match the pump's local side before it encodes or queues anything.
796func TestPumpSendDirection(t *testing.T) {
797 t.Run("daemon side writes server messages", func(t *testing.T) {
798 pump, carrier := newTestPump(t, DirectionClient)
799 pump.Start(context.Background())
800 deliverClientFrame(t, carrier, openFor(1))
801 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
802 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
803
804 require.ErrorIs(t, pump.Send(Open{Ref: testRef(2)}), ErrWrongDirection)
805 require.ErrorIs(t, pump.Send(nil), ErrInvalidMessage)
806 require.NoError(t, pump.Send(Opened{Ref: testRef(1)}))
807 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
808 require.NoError(t, err)
809 _, ok := message.(Opened)
810 require.True(t, ok)
811 carrier.requireNoSent(t, 20*time.Millisecond)
812 })
813
814 t.Run("client side writes client messages", func(t *testing.T) {
815 pump, carrier := newTestPump(t, DirectionServer)
816 pump.Start(context.Background())
817 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(1)}))
818
819 require.ErrorIs(t, pump.Send(Opened{Ref: testRef(1)}), ErrWrongDirection)
820 require.NoError(t, pump.Send(openFor(1)))
821 message, err := DecodeClient(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
822 require.NoError(t, err)
823 _, ok := message.(Open)
824 require.True(t, ok)
825
826 // The client side reads daemon frames: an inbound Opened confirms the
827 // locally admitted stream through the reader.
828 deliverServerFrame(t, carrier, Opened{Ref: testRef(1)})
829 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
830 status, ok := e.Status(1)
831 return ok && status.State == StreamOpen
832 })
833 })
834}
835
836// TestPumpFlushWaitsForInflightSend proves Flush does not return while a send
837// is in flight inside the carrier, and returns once the writer finished it.
838func TestPumpFlushWaitsForInflightSend(t *testing.T) {
839 pump, carrier := newTestPump(t, DirectionClient)
840 pump.Start(context.Background())
841 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(1)}))
842 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
843
844 hold := make(chan struct{})
845 carrier.setHold(hold)
846 require.NoError(t, pump.Send(ServerMessage(Close{Physical: 1})))
847 require.Eventually(t, func() bool { return carrier.enteredSends() >= 1 }, 5*time.Second, time.Millisecond)
848
849 flushed := make(chan error, 1)
850 go func() { flushed <- pump.Flush(context.Background()) }()
851 select {
852 case <-flushed:
853 t.Fatal("Flush returned while a send was still in flight")
854 case <-time.After(50 * time.Millisecond):
855 }
856
857 carrier.release(hold)
858 select {
859 case err := <-flushed:
860 require.NoError(t, err)
861 case <-time.After(5 * time.Second):
862 t.Fatal("Flush did not return after the write drained")
863 }
864 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
865 require.NoError(t, err)
866 closed, ok := message.(Close)
867 require.True(t, ok)
868 require.Equal(t, PhysicalStreamID(1), closed.Physical)
869}
870
871// TestPumpFlushPublishesFinalClose proves a flushes-before-close caller makes
872// the final Close observable on the wire: Send accepted means queued only, and
873// Flush is the explicit barrier that drains it.
874func TestPumpFlushPublishesFinalClose(t *testing.T) {
875 pump, carrier := newTestPump(t, DirectionClient)
876 pump.Start(context.Background())
877 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(1)}))
878 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
879
880 require.NoError(t, pump.Send(ServerMessage(Close{Physical: 1})))
881 require.NoError(t, pump.Flush(context.Background()))
882 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
883 require.NoError(t, err)
884 closed, ok := message.(Close)
885 require.True(t, ok)
886 require.Equal(t, PhysicalStreamID(1), closed.Physical)
887
888 // Flush of an already-drained pump returns immediately.
889 require.NoError(t, pump.Flush(context.Background()))
890}
891
892// TestPumpFlushHonoursContext proves Flush returns the context error when the
893// deadline wins before the scheduler drains.
894func TestPumpFlushHonoursContext(t *testing.T) {
895 pump, carrier := newTestPump(t, DirectionClient)
896 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(1)}))
897 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
898 // No writer is started, so the queued frame can never drain.
899 require.NoError(t, pump.SendData(1, []byte("queued"), nil))
900
901 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
902 defer cancel()
903 require.ErrorIs(t, pump.Flush(ctx), context.DeadlineExceeded)
904
905 // Starting the pump drains the frame and a fresh Flush succeeds.
906 pump.Start(context.Background())
907 require.NoError(t, pump.Flush(context.Background()))
908 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
909 require.NoError(t, err)
910 data, ok := message.(Data)
911 require.True(t, ok)
912 require.Equal(t, []byte("queued"), data.Data)
913}
914
915// TestPumpFlushCancellationMayDrop proves Close is not a flush: a frame queued
916// behind a blocked write is dropped when the context is cancelled rather than
917// drained, so a caller that must publish a final frame flushes first.
918func TestPumpFlushCancellationMayDrop(t *testing.T) {
919 pump, carrier := newTestPump(t, DirectionClient)
920 ctx, cancel := context.WithCancel(context.Background())
921 pump.Start(ctx)
922 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(1)}))
923 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
924
925 hold := make(chan struct{})
926 carrier.setHold(hold)
927 require.NoError(t, pump.SendData(1, []byte("first"), nil))
928 require.Eventually(t, func() bool { return carrier.enteredSends() >= 1 }, 5*time.Second, time.Millisecond)
929 // The final Close is queued behind the blocked write.
930 require.NoError(t, pump.Send(ServerMessage(Close{Physical: 1})))
931
932 cancel()
933 require.Eventually(t, func() bool { return channelClosed(pump.writerDone) }, 5*time.Second, time.Millisecond)
934 carrier.release(hold)
935 carrier.requireNoSent(t, 50*time.Millisecond)
936}
937
938// TestPumpSendDataBackpressure proves a sender that exhausted its stream
939// credit waits instead of overflowing or resetting the stream, resumes once
940// the peer returns credit, and that the stream stays open throughout.
941func TestPumpSendDataBackpressure(t *testing.T) {
942 pump, carrier := newTestPump(t, DirectionClient)
943 pump.Start(context.Background())
944 deliverClientFrame(t, carrier, openFor(1))
945 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
946 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
947
948 chunk, full := floodWindow(pump.Ceilings())
949 for i := 0; i < full; i++ {
950 require.NoError(t, pump.SendData(1, chunk, nil))
951 }
952 for i := 0; i < full; i++ {
953 carrier.nextSent(t)
954 }
955
956 sent := make(chan error, 1)
957 go func() { sent <- pump.SendData(1, chunk, nil) }()
958 select {
959 case err := <-sent:
960 t.Fatalf("SendData returned %v without credit", err)
961 case <-time.After(50 * time.Millisecond):
962 }
963 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 1).State, "a slow peer never resets the stream")
964
965 deliverClientFrame(t, carrier, WindowUpdate{Physical: 1, Credit: chunkCredit(len(chunk))})
966 require.NoError(t, <-sent)
967 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
968 require.NoError(t, err)
969 require.Equal(t, Data{Physical: 1, Data: chunk}, message)
970 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 1).State)
971}
972
973// TestPumpSendDataCancel proves a sender blocked on credit returns when its
974// cancel channel closes or its stream settles.
975func TestPumpSendDataCancel(t *testing.T) {
976 cases := []struct {
977 name string
978 settle func(*testing.T, *Pump, *fakeCarrier, chan struct{})
979 want error
980 }{
981 {"cancel", func(_ *testing.T, _ *Pump, _ *fakeCarrier, cancel chan struct{}) { close(cancel) }, ErrLogicalClosed},
982 {"peer reset", func(t *testing.T, _ *Pump, carrier *fakeCarrier, _ chan struct{}) {
983 deliverClientFrame(t, carrier, Reset{Physical: 1})
984 }, errStreamSettled},
985 {"peer close with unread data", func(t *testing.T, pump *Pump, carrier *fakeCarrier, _ chan struct{}) {
986 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("unread")})
987 deliverClientFrame(t, carrier, Close{Physical: 1})
988 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
989 status, ok := e.Status(1)
990 return ok && status.State == StreamClosing
991 })
992 }, errStreamSettled},
993 {"physical failure", func(t *testing.T, _ *Pump, carrier *fakeCarrier, _ chan struct{}) {
994 carrier.deliver(t, fakeFrame{err: errors.New("carriage lost")})
995 }, ErrPhysicalClosed},
996 {"credit beyond window resets the stream", func(t *testing.T, pump *Pump, carrier *fakeCarrier, _ chan struct{}) {
997 deliverClientFrame(t, carrier, WindowUpdate{Physical: 1, Credit: pump.Ceilings().StreamWindow()})
998 }, errStreamSettled},
999 }
1000 for _, tc := range cases {
1001 t.Run(tc.name, func(t *testing.T) {
1002 pump, carrier := newTestPump(t, DirectionClient)
1003 pump.Start(context.Background())
1004 deliverClientFrame(t, carrier, openFor(1))
1005 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
1006 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
1007 chunk, full := floodWindow(pump.Ceilings())
1008 for i := 0; i < full; i++ {
1009 require.NoError(t, pump.SendData(1, chunk, nil))
1010 }
1011
1012 cancel := make(chan struct{})
1013 sent := make(chan error, 1)
1014 go func() { sent <- pump.SendData(1, chunk, cancel) }()
1015 tc.settle(t, pump, carrier, cancel)
1016 select {
1017 case err := <-sent:
1018 require.ErrorIs(t, err, tc.want)
1019 case <-time.After(5 * time.Second):
1020 t.Fatal("blocked SendData never returned")
1021 }
1022 })
1023 }
1024}
1025
1026// TestPumpReturnsCreditInBatches proves the receiver grants consumed credit
1027// back with one WindowUpdate per batch, not one per chunk.
1028func TestPumpReturnsCreditInBatches(t *testing.T) {
1029 pump, carrier := newTestPump(t, DirectionClient)
1030 pump.Start(context.Background())
1031 deliverClientFrame(t, carrier, openFor(1))
1032 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 1 })
1033 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
1034
1035 chunk, full := floodWindow(pump.Ceilings())
1036 for i := 0; i < full; i++ {
1037 deliverClientFrame(t, carrier, Data{Physical: 1, Data: chunk})
1038 }
1039 var granted uint64
1040 updates := 0
1041 for i := 0; i < full; i++ {
1042 _, ok := takeEventually(t, pump, 1)
1043 require.True(t, ok)
1044 }
1045 for granted < pump.Ceilings().creditReturnThreshold() {
1046 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
1047 require.NoError(t, err)
1048 update, ok := message.(WindowUpdate)
1049 require.True(t, ok, "got %T", message)
1050 granted += update.Credit
1051 updates++
1052 }
1053 require.Less(t, updates, full, "credit is batched")
1054}
1055
1056// TestPumpInboundTerminalRetiresQueuedOutbound proves an inbound Close or Reset
1057// discards the stream's queued outbound frames and retires the scheduler
1058// record, and that 3000 stream cycles stay bounded with no post-terminal
1059// dequeue.
1060func TestPumpInboundTerminalRetiresQueuedOutbound(t *testing.T) {
1061 pump, _ := newTestPump(t, DirectionClient)
1062 const churn = 3000
1063 for i := 1; i <= churn; i++ {
1064 id := PhysicalStreamID(i)
1065 require.NoError(t, pump.Engine().Open(Open{Ref: testRef(id)}))
1066 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
1067 require.NoError(t, pump.SendData(id, []byte("out"), nil))
1068 require.Equal(t, 1, pump.scheduler.QueuedFrames(id))
1069
1070 if i%2 == 0 {
1071 require.NoError(t, pump.applyClient(Close{Physical: id}))
1072 } else {
1073 require.NoError(t, pump.applyClient(Reset{Physical: id}))
1074 }
1075 require.Zero(t, pump.scheduler.QueuedFrames(id), "queued outbound discarded")
1076 require.Zero(t, pump.scheduler.AggregateBytes())
1077 require.False(t, pump.Engine().Closed())
1078 }
1079 require.Zero(t, pump.Engine().Live())
1080 require.Zero(t, pump.Engine().AggregateBytes())
1081 require.LessOrEqual(t, len(pump.scheduler.streams), MaxRetiredSchedulerRecords)
1082 _, ok := pump.scheduler.Dequeue()
1083 require.False(t, ok, "no post-terminal dequeue")
1084}
1085
1086// watchEntries reports how many inbound watch entries the pump retains.
1087func watchEntries(pump *Pump) int {
1088 pump.watchMu.Lock()
1089 defer pump.watchMu.Unlock()
1090 return len(pump.watch)
1091}
1092
1093// TestPumpWatchUnknownAndTerminalClosed proves Watch hands back an
1094// already-closed channel for zero, unknown, and terminal identities without
1095// retaining a watch entry, and returns one stable open channel for a live
1096// stream that rotates on every applied frame.
1097func TestPumpWatchUnknownAndTerminalClosed(t *testing.T) {
1098 pump, _ := newTestPump(t, DirectionClient)
1099
1100 require.True(t, channelClosed(pump.Watch(0)), "zero identity is closed")
1101 require.True(t, channelClosed(pump.Watch(9)), "unknown identity is closed")
1102 require.Zero(t, watchEntries(pump), "unknown identities retain no entry")
1103
1104 require.NoError(t, pump.Engine().Open(openFor(1)))
1105 live := pump.Watch(1)
1106 require.False(t, channelClosed(live))
1107 require.Equal(t, 1, watchEntries(pump))
1108 require.Equal(t, live, pump.Watch(1), "a live watch channel is stable")
1109
1110 // An applied frame rotates the watch channel: the old one closes and a
1111 // fresh open one is installed.
1112 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
1113 require.NoError(t, pump.applyClient(Data{Physical: 1, Data: []byte("x")}))
1114 rotated := pump.Watch(1)
1115 require.NotEqual(t, live, rotated, "an applied frame rotates the channel")
1116 require.True(t, channelClosed(live))
1117
1118 // A closing stream keeps its queue readable: the applied Close rotates the
1119 // watch channel, but the stream is not terminal until its preserved chunk is
1120 // taken.
1121 require.NoError(t, pump.applyClient(Close{Physical: 1}))
1122 require.True(t, channelClosed(rotated), "the close frame rotated the current channel")
1123 require.Equal(t, StreamClosing, mustStatus(t, pump.Engine(), 1).State)
1124 closingWatch := pump.Watch(1)
1125 require.False(t, channelClosed(closingWatch), "a closing stream is still readable")
1126 require.Equal(t, closingWatch, pump.Watch(1), "the closing watch channel is stable")
1127
1128 chunk, ok := pump.Take(1)
1129 require.True(t, ok)
1130 require.Equal(t, []byte("x"), chunk)
1131 require.True(t, channelClosed(closingWatch), "draining the last chunk rotated the watch channel")
1132 require.True(t, channelClosed(pump.Watch(1)), "terminal identity is closed")
1133 require.Zero(t, watchEntries(pump), "terminal Watch retains no entry")
1134
1135 // An evicted identity is closed and retains no entry either.
1136 for i := 2; i <= MaxRetiredStreamRecords+2; i++ {
1137 id := PhysicalStreamID(i)
1138 require.NoError(t, pump.Engine().Open(openFor(id)))
1139 _, _ = pump.Engine().Close(Close{Physical: id})
1140 }
1141 _, tracked := pump.Engine().Status(1)
1142 require.False(t, tracked, "stream 1 record was evicted")
1143 require.True(t, channelClosed(pump.Watch(1)), "evicted identity is closed")
1144 require.Zero(t, watchEntries(pump), "evicted Watch retains no entry")
1145}