1package daemonmux
2
3import (
4 "errors"
5 "fmt"
6 "sync"
7 "testing"
8
9 "github.com/stretchr/testify/assert"
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14)
15
16func testEngine() *StreamEngine { return NewStreamEngine() }
17
18// admitTestStream admits one stream in opening state.
19func admitTestStream(t *testing.T, e *StreamEngine, id PhysicalStreamID) StreamRef {
20 t.Helper()
21 ref := testRef(id)
22 require.NoError(t, e.Open(Open{Ref: ref}))
23 return ref
24}
25
26// openTestStream admits and confirms one stream.
27func openTestStream(t *testing.T, e *StreamEngine, id PhysicalStreamID) StreamRef {
28 t.Helper()
29 ref := admitTestStream(t, e, id)
30 require.NoError(t, e.Opened(Opened{Ref: ref}))
31 return ref
32}
33
34func mustStatus(t *testing.T, e *StreamEngine, id PhysicalStreamID) StreamStatus {
35 t.Helper()
36 status, ok := e.Status(id)
37 require.True(t, ok, "stream %d is not tracked", id)
38 return status
39}
40
41func channelClosed(ch <-chan struct{}) bool {
42 select {
43 case <-ch:
44 return true
45 default:
46 return false
47 }
48}
49
50func TestNewStreamEngine(t *testing.T) {
51 e := NewStreamEngine()
52 require.Zero(t, e.Live())
53 require.Zero(t, e.AggregateBytes())
54 require.False(t, e.Closed())
55 require.NoError(t, e.Err())
56 require.Equal(t, domain.RemoteFailureNone, e.FailureKind())
57 require.NotNil(t, e.Done())
58}
59
60func TestStreamStateStrings(t *testing.T) {
61 require.Equal(t, "opening", StreamOpening.String())
62 require.Equal(t, "open", StreamOpen.String())
63 require.Equal(t, "closing", StreamClosing.String())
64 require.Equal(t, "terminal", StreamTerminal.String())
65 require.Equal(t, "unknown", StreamState(0).String())
66 require.Equal(t, "unknown", StreamState(200).String())
67}
68
69// TestStreamEngineOpenAdmission is the admission table: complete references,
70// strictly increasing identities that are never reused, and one shared
71// pending-plus-live ceiling.
72func TestStreamEngineOpenAdmission(t *testing.T) {
73 t.Run("incomplete identities refused", func(t *testing.T) {
74 e := testEngine()
75 cases := []struct {
76 name string
77 ref StreamRef
78 }{
79 {"zero ref", StreamRef{}},
80 {"zero physical", StreamRef{Physical: 0, Epoch: 7, Connection: testConnectionID(0x11), Client: 3}},
81 {"zero epoch", StreamRef{Physical: 1, Epoch: 0, Connection: testConnectionID(0x11), Client: 3}},
82 {"zero connection", StreamRef{Physical: 1, Epoch: 7, Client: 3}},
83 {"zero client", StreamRef{Physical: 1, Epoch: 7, Connection: testConnectionID(0x11), Client: 0}},
84 }
85 for _, tc := range cases {
86 require.ErrorIs(t, e.Open(Open{Ref: tc.ref}), ErrInvalidStream, tc.name)
87 }
88 require.Zero(t, e.Live())
89 _, ok := e.Status(1)
90 require.False(t, ok)
91 })
92
93 t.Run("strictly increasing and never reused", func(t *testing.T) {
94 e := testEngine()
95 admitTestStream(t, e, 5)
96 require.ErrorIs(t, e.Open(Open{Ref: testRef(5)}), ErrStreamIDReused)
97 require.ErrorIs(t, e.Open(Open{Ref: testRef(1)}), ErrStreamIDReused)
98 require.NoError(t, e.Opened(Opened{Ref: testRef(5)}))
99 admitTestStream(t, e, 6)
100
101 disposition, err := e.Close(Close{Physical: 5})
102 require.NoError(t, err)
103 require.Equal(t, StreamAccepted, disposition)
104 require.ErrorIs(t, e.Open(Open{Ref: testRef(5)}), ErrStreamIDReused, "a settled ID is never reused")
105 require.ErrorIs(t, e.Open(Open{Ref: testRef(6)}), ErrStreamIDReused)
106 require.NoError(t, e.Open(Open{Ref: testRef(7)}))
107 require.Equal(t, 2, e.Live())
108 })
109
110 t.Run("pending and live share the ceiling", func(t *testing.T) {
111 e := testEngine()
112 for i := 1; i <= int(MaxMuxStreams); i++ {
113 require.NoError(t, e.Open(Open{Ref: testRef(PhysicalStreamID(i))}))
114 }
115 require.Equal(t, int(MaxMuxStreams), e.Live())
116 require.ErrorIs(t, e.Open(Open{Ref: testRef(PhysicalStreamID(MaxMuxStreams + 1))}), ErrTooManyStreams)
117
118 // Confirming a pending stream does not free a slot: pending and live
119 // share the one ceiling.
120 require.NoError(t, e.Opened(Opened{Ref: testRef(1)}))
121 require.ErrorIs(t, e.Open(Open{Ref: testRef(PhysicalStreamID(MaxMuxStreams + 2))}), ErrTooManyStreams)
122
123 // Settling one stream frees exactly one slot for a fresh higher ID.
124 disposition, err := e.Reset(Reset{Physical: 1})
125 require.NoError(t, err)
126 require.Equal(t, StreamAccepted, disposition)
127 require.Equal(t, int(MaxMuxStreams)-1, e.Live())
128 require.NoError(t, e.Open(Open{Ref: testRef(PhysicalStreamID(MaxMuxStreams + 2))}))
129 require.ErrorIs(t, e.Open(Open{Ref: testRef(PhysicalStreamID(MaxMuxStreams + 3))}), ErrTooManyStreams)
130 })
131}
132
133// TestStreamEngineFutureAndRetired proves the high-water fence: an identity
134// above it was never opened and is refused, while traffic at or below it that
135// is no longer live belongs to a retired stream and is ignored.
136func TestStreamEngineFutureAndRetired(t *testing.T) {
137 t.Run("future identity refused", func(t *testing.T) {
138 e := testEngine()
139 openTestStream(t, e, 2)
140 future := PhysicalStreamID(3)
141 require.ErrorIs(t, e.Opened(Opened{Ref: testRef(future)}), ErrFutureStream)
142 require.ErrorIs(t, e.Refused(Refused{Ref: testRef(future)}), ErrFutureStream)
143 disposition, err := e.Data(Data{Physical: future, Data: []byte{'x'}})
144 require.ErrorIs(t, err, ErrFutureStream)
145 require.Zero(t, disposition)
146 disposition, err = e.Close(Close{Physical: future})
147 require.ErrorIs(t, err, ErrFutureStream)
148 require.Zero(t, disposition)
149 disposition, err = e.Reset(Reset{Physical: future})
150 require.ErrorIs(t, err, ErrFutureStream)
151 require.Zero(t, disposition)
152 require.Equal(t, 1, e.Live())
153 })
154
155 t.Run("retired traffic ignored including gaps", func(t *testing.T) {
156 e := testEngine()
157 openTestStream(t, e, 5)
158 disposition, err := e.Reset(Reset{Physical: 5})
159 require.NoError(t, err)
160 require.Equal(t, StreamAccepted, disposition)
161 openTestStream(t, e, 9)
162
163 // 6 and 8 were never admitted but sit below the high-water mark: they
164 // are retired traffic, not future streams.
165 for _, id := range []PhysicalStreamID{5, 6, 8} {
166 disposition, err := e.Data(Data{Physical: id, Data: []byte{'x'}})
167 require.NoError(t, err)
168 require.Equal(t, StreamDiscarded, disposition)
169 disposition, err = e.Close(Close{Physical: id})
170 require.NoError(t, err)
171 require.Equal(t, StreamDiscarded, disposition)
172 disposition, err = e.Reset(Reset{Physical: id})
173 require.NoError(t, err)
174 require.Equal(t, StreamDiscarded, disposition)
175 }
176
177 // A retired identity can never be confirmed or refused again.
178 require.ErrorIs(t, e.Opened(Opened{Ref: testRef(5)}), ErrStreamState)
179 require.ErrorIs(t, e.Refused(Refused{Ref: testRef(5)}), ErrStreamState)
180 require.ErrorIs(t, e.Opened(Opened{Ref: testRef(6)}), ErrStreamState)
181 require.Equal(t, 1, e.Live())
182 })
183}
184
185type engineStep struct {
186 name string
187 run func(*StreamEngine, PhysicalStreamID) (StreamDisposition, error)
188 want StreamDisposition
189 wantErr error
190}
191
192func stepAdmit(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
193 if err := e.Open(Open{Ref: testRef(id)}); err != nil {
194 return 0, err
195 }
196 return StreamAccepted, nil
197}
198
199func stepOpened(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
200 if err := e.Opened(Opened{Ref: testRef(id)}); err != nil {
201 return 0, err
202 }
203 return StreamAccepted, nil
204}
205
206func stepRefused(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
207 if err := e.Refused(Refused{Ref: testRef(id)}); err != nil {
208 return 0, err
209 }
210 return StreamAccepted, nil
211}
212
213func stepData(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
214 return e.Data(Data{Physical: id, Data: []byte{'d'}})
215}
216
217func stepClose(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
218 return e.Close(Close{Physical: id})
219}
220
221func stepReset(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
222 return e.Reset(Reset{Physical: id})
223}
224
225// stepTake drains one queued chunk, reporting StreamAccepted when a chunk was
226// taken and StreamDiscarded when the queue was already empty. It is the drain
227// half of the orderly-close lifecycle: taking the last chunk of a closing
228// stream terminalizes it.
229func stepTake(e *StreamEngine, id PhysicalStreamID) (StreamDisposition, error) {
230 if _, ok := e.Take(id); ok {
231 return StreamAccepted, nil
232 }
233 return StreamDiscarded, nil
234}
235
236// TestStreamEngineLifecycle is the stream state table: opening -> open ->
237// terminal, with legal Opened/Refused only from opening, data only while open,
238// and late traffic discarded.
239func TestStreamEngineLifecycle(t *testing.T) {
240 cases := []struct {
241 name string
242 steps []engineStep
243 }{
244 {
245 name: "confirmed then closed",
246 steps: []engineStep{
247 {"admit", stepAdmit, StreamAccepted, nil},
248 {"data while opening", stepData, 0, ErrStreamState},
249 {"opened", stepOpened, StreamAccepted, nil},
250 {"opened twice", stepOpened, 0, ErrStreamState},
251 {"refused when open", stepRefused, 0, ErrStreamState},
252 {"close", stepClose, StreamAccepted, nil},
253 {"late data", stepData, StreamDiscarded, nil},
254 {"late close", stepClose, StreamDiscarded, nil},
255 {"late reset", stepReset, StreamDiscarded, nil},
256 {"opened after close", stepOpened, 0, ErrStreamState},
257 {"refused after close", stepRefused, 0, ErrStreamState},
258 },
259 },
260 {
261 name: "orderly close preserves queued data until drained",
262 steps: []engineStep{
263 {"admit", stepAdmit, StreamAccepted, nil},
264 {"opened", stepOpened, StreamAccepted, nil},
265 {"data when open", stepData, StreamAccepted, nil},
266 {"close with unread data", stepClose, StreamAccepted, nil},
267 {"close twice", stepClose, StreamDiscarded, nil},
268 {"data after close refused", stepData, 0, ErrStreamState},
269 {"drain the preserved chunk", stepTake, StreamAccepted, nil},
270 {"late close", stepClose, StreamDiscarded, nil},
271 {"late data", stepData, StreamDiscarded, nil},
272 {"late reset", stepReset, StreamDiscarded, nil},
273 {"opened after close", stepOpened, 0, ErrStreamState},
274 },
275 },
276 {
277 name: "refused from opening",
278 steps: []engineStep{
279 {"admit", stepAdmit, StreamAccepted, nil},
280 {"refused", stepRefused, StreamAccepted, nil},
281 {"refused twice", stepRefused, 0, ErrStreamState},
282 {"opened after refusal", stepOpened, 0, ErrStreamState},
283 {"data after refusal", stepData, StreamDiscarded, nil},
284 },
285 },
286 {
287 name: "close while opening is terminal",
288 steps: []engineStep{
289 {"admit", stepAdmit, StreamAccepted, nil},
290 {"close", stepClose, StreamAccepted, nil},
291 {"data after close", stepData, StreamDiscarded, nil},
292 {"opened after close", stepOpened, 0, ErrStreamState},
293 },
294 },
295 {
296 name: "reset while opening is terminal and idempotent",
297 steps: []engineStep{
298 {"admit", stepAdmit, StreamAccepted, nil},
299 {"reset", stepReset, StreamAccepted, nil},
300 {"reset idempotent", stepReset, StreamDiscarded, nil},
301 {"close idempotent", stepClose, StreamDiscarded, nil},
302 },
303 },
304 }
305 for _, tc := range cases {
306 t.Run(tc.name, func(t *testing.T) {
307 e := testEngine()
308 id := PhysicalStreamID(4)
309 for _, s := range tc.steps {
310 disposition, err := s.run(e, id)
311 if s.wantErr != nil {
312 require.ErrorIs(t, err, s.wantErr, s.name)
313 require.Zero(t, disposition, s.name)
314 continue
315 }
316 require.NoError(t, err, s.name)
317 require.Equal(t, s.want, disposition, s.name)
318 }
319 require.Zero(t, e.Live())
320 require.Equal(t, StreamTerminal, mustStatus(t, e, id).State)
321 })
322 }
323}
324
325// TestStreamEngineRejectedAdmissionLeavesNothing proves a refused admission
326// records nothing and cannot resurrect a settled identity.
327func TestStreamEngineRejectedAdmissionLeavesNothing(t *testing.T) {
328 e := testEngine()
329 require.ErrorIs(t, e.Open(Open{}), ErrInvalidStream)
330 require.Zero(t, e.Live())
331 _, ok := e.Status(1)
332 require.False(t, ok)
333
334 admitTestStream(t, e, 1)
335 require.ErrorIs(t, e.Open(Open{Ref: testRef(1)}), ErrStreamIDReused)
336 require.Equal(t, 1, e.Live())
337 require.Equal(t, StreamOpening, mustStatus(t, e, 1).State)
338 _, ok = e.Status(2)
339 require.False(t, ok)
340}
341
342// TestStreamEngineRefFence proves Opened and Refused are legal only for the
343// opening stream whose full reference was recorded at admission.
344func TestStreamEngineRefFence(t *testing.T) {
345 e := testEngine()
346 ref := admitTestStream(t, e, 4)
347
348 mismatched := ref
349 mismatched.Epoch++
350 require.ErrorIs(t, e.Opened(Opened{Ref: mismatched}), ErrStreamRefMismatch)
351 require.Equal(t, StreamOpening, mustStatus(t, e, 4).State)
352 require.ErrorIs(t, e.Refused(Refused{Ref: mismatched}), ErrStreamRefMismatch)
353 require.Equal(t, StreamOpening, mustStatus(t, e, 4).State)
354
355 require.NoError(t, e.Opened(Opened{Ref: ref}))
356 require.Equal(t, StreamOpen, mustStatus(t, e, 4).State)
357
358 // Legality is checked before reference identity: a settled stream refuses
359 // both a matching and a mismatched confirmation.
360 require.ErrorIs(t, e.Opened(Opened{Ref: ref}), ErrStreamState)
361 require.ErrorIs(t, e.Opened(Opened{Ref: mismatched}), ErrStreamState)
362}
363
364// TestStreamEngineRefusalRetained proves a refusal settles the stream
365// orderly and retains the typed refusal for the caller.
366func TestStreamEngineRefusalRetained(t *testing.T) {
367 e := testEngine()
368 admitTestStream(t, e, 2)
369 detail := testErrorDetail()
370 require.NoError(t, e.Refused(Refused{Ref: testRef(2), Error: detail}))
371
372 status := mustStatus(t, e, 2)
373 require.Equal(t, StreamTerminal, status.State)
374 require.True(t, status.Refused)
375 require.Equal(t, detail, status.Refusal)
376 require.True(t, channelClosed(status.Done))
377 require.NoError(t, status.Err)
378 require.Equal(t, domain.RemoteFailureNone, status.FailureKind)
379 require.Zero(t, e.Live())
380}
381
382// TestStreamEngineTwoAndHundredStreams proves admission, confirmation, data,
383// and orderly close for a small and a large concurrent set. A stream whose
384// already-queued chunk was drained before the Close terminalizes immediately;
385// every other stream preserves its queued chunk while closing until Take
386// drains it.
387func TestStreamEngineTwoAndHundredStreams(t *testing.T) {
388 for _, count := range []int{2, 100} {
389 t.Run(fmt.Sprintf("%d streams", count), func(t *testing.T) {
390 e := testEngine()
391 for i := 1; i <= count; i++ {
392 ref := openTestStream(t, e, PhysicalStreamID(i))
393 require.Equal(t, ref, mustStatus(t, e, PhysicalStreamID(i)).Ref)
394 }
395 require.Equal(t, count, e.Live())
396
397 for i := 1; i <= count; i++ {
398 disposition, err := e.Data(Data{Physical: PhysicalStreamID(i), Data: []byte{byte(i)}})
399 require.NoError(t, err)
400 require.Equal(t, StreamAccepted, disposition)
401 }
402 require.Equal(t, count, e.AggregateBytes())
403
404 taken, ok := e.Take(PhysicalStreamID(1))
405 require.True(t, ok)
406 require.Equal(t, []byte{1}, taken)
407 require.Equal(t, count-1, e.AggregateBytes())
408 require.Zero(t, mustStatus(t, e, 1).QueuedChunks)
409
410 for i := 1; i <= count; i++ {
411 disposition, err := e.Close(Close{Physical: PhysicalStreamID(i)})
412 require.NoError(t, err)
413 require.Equal(t, StreamAccepted, disposition)
414 }
415 // The drained stream is terminal; every other stream is closing and
416 // still holds exactly its one unread, charged chunk.
417 require.Equal(t, count-1, e.Live())
418 require.Equal(t, count-1, e.AggregateBytes())
419 require.False(t, e.Closed(), "closing every stream never terminalizes the physical connection")
420 terminal := mustStatus(t, e, 1)
421 require.Equal(t, StreamTerminal, terminal.State)
422 require.True(t, channelClosed(terminal.Done))
423 for i := 2; i <= count; i++ {
424 status := mustStatus(t, e, PhysicalStreamID(i))
425 require.Equal(t, StreamClosing, status.State, "stream %d", i)
426 require.False(t, channelClosed(status.Done), "stream %d", i)
427 require.Equal(t, 1, status.QueuedChunks, "stream %d", i)
428 }
429
430 // Draining every preserved chunk releases its bytes and terminalizes
431 // the stream.
432 for i := 2; i <= count; i++ {
433 chunk, ok := e.Take(PhysicalStreamID(i))
434 require.True(t, ok, "stream %d", i)
435 require.Equal(t, []byte{byte(i)}, chunk, "stream %d", i)
436 }
437 require.Zero(t, e.Live())
438 require.Zero(t, e.AggregateBytes())
439 require.False(t, e.Closed(), "closing every stream never terminalizes the physical connection")
440 for i := 1; i <= count; i++ {
441 status := mustStatus(t, e, PhysicalStreamID(i))
442 require.Equal(t, StreamTerminal, status.State)
443 require.True(t, channelClosed(status.Done))
444 require.NoError(t, status.Err)
445 }
446 })
447 }
448}
449
450// TestStreamEngineIsolation proves one stream's close or reset terminalizes
451// neither the physical connection nor a sibling.
452func TestStreamEngineIsolation(t *testing.T) {
453 e := testEngine()
454 openTestStream(t, e, 1)
455 openTestStream(t, e, 2)
456 openTestStream(t, e, 3)
457 _, err := e.Data(Data{Physical: 1, Data: []byte("a")})
458 require.NoError(t, err)
459
460 disposition, err := e.Close(Close{Physical: 2})
461 require.NoError(t, err)
462 require.Equal(t, StreamAccepted, disposition)
463 require.False(t, e.Closed())
464 require.NoError(t, e.Err())
465 require.Equal(t, StreamOpen, mustStatus(t, e, 1).State)
466 require.Equal(t, StreamOpen, mustStatus(t, e, 3).State)
467 require.Equal(t, 2, e.Live())
468 require.Equal(t, 1, e.AggregateBytes())
469
470 disposition, err = e.Reset(Reset{Physical: 3})
471 require.NoError(t, err)
472 require.Equal(t, StreamAccepted, disposition)
473 require.False(t, e.Closed())
474 require.Equal(t, StreamOpen, mustStatus(t, e, 1).State)
475 require.Equal(t, 1, e.Live())
476
477 disposition, err = e.Data(Data{Physical: 1, Data: []byte("b")})
478 require.NoError(t, err)
479 require.Equal(t, StreamAccepted, disposition)
480 require.Equal(t, 2, e.AggregateBytes())
481}
482
483// floodWindow returns a maximum chunk and how many of them exactly fill one
484// stream's receive window under ceilings: one more is a peer flow-control
485// violation.
486func floodWindow(ceilings MuxCeilings) ([]byte, int) {
487 chunk := make([]byte, ceilings.StreamChunkLimit)
488 return chunk, int(ceilings.StreamWindow() / chunkCredit(len(chunk)))
489}
490
491// TestStreamEngineQueueOverflow proves a peer that sends past its granted
492// window resets exactly its own stream, frees that stream's budget, and leaves
493// a sibling progressing.
494func TestStreamEngineQueueOverflow(t *testing.T) {
495 e := testEngine()
496 openTestStream(t, e, 1)
497 openTestStream(t, e, 2)
498 _, err := e.Data(Data{Physical: 2, Data: []byte("sibling")})
499 require.NoError(t, err)
500
501 chunk, full := floodWindow(DefaultMuxCeilings())
502 for i := 0; i < full; i++ {
503 disposition, err := e.Data(Data{Physical: 1, Data: chunk})
504 require.NoError(t, err)
505 require.Equal(t, StreamAccepted, disposition)
506 }
507 require.Equal(t, full, mustStatus(t, e, 1).QueuedChunks)
508
509 disposition, err := e.Data(Data{Physical: 1, Data: chunk})
510 require.ErrorIs(t, err, ErrStreamQueueFull)
511 require.Zero(t, disposition)
512
513 status := mustStatus(t, e, 1)
514 require.Equal(t, StreamTerminal, status.State)
515 require.True(t, channelClosed(status.Done))
516 require.ErrorIs(t, status.Err, ErrStreamQueueFull)
517 require.Equal(t, domain.RemoteFailureInvalidResponse, status.FailureKind)
518 require.Zero(t, status.QueuedChunks)
519 require.Zero(t, status.QueuedBytes)
520 require.Equal(t, len("sibling"), e.AggregateBytes(), "the overflowed stream released its budget")
521 require.Equal(t, 1, e.Live())
522 require.False(t, e.Closed())
523
524 sibling := mustStatus(t, e, 2)
525 require.Equal(t, StreamOpen, sibling.State)
526 require.False(t, channelClosed(sibling.Done))
527 disposition, err = e.Data(Data{Physical: 2, Data: []byte("more")})
528 require.NoError(t, err)
529 require.Equal(t, StreamAccepted, disposition)
530 require.Equal(t, len("sibling")+len("more"), e.AggregateBytes())
531
532 // Cancellation of an already terminal stream stays idempotent and never
533 // rewrites its cause.
534 disposition, err = e.Reset(Reset{Physical: 1})
535 require.NoError(t, err)
536 require.Equal(t, StreamDiscarded, disposition)
537 require.ErrorIs(t, mustStatus(t, e, 1).Err, ErrStreamQueueFull)
538}
539
540// TestStreamEngineQueueByteBounds proves the aggregate byte ceiling resets
541// only the offending stream.
542func TestStreamEngineQueueByteBounds(t *testing.T) {
543 t.Run("aggregate byte bound", func(t *testing.T) {
544 e := testEngine()
545 // Bypass the codec-level chunk ceiling and widen the window so this
546 // test reaches the defensive aggregate byte bound with oversized
547 // chunks.
548 e.maxChunkBytes = 16 << 20
549 e.window = 32 << 20
550 chunk := make([]byte, 16<<20)
551 for i := 1; i <= 4; i++ {
552 openTestStream(t, e, PhysicalStreamID(i))
553 disposition, err := e.Data(Data{Physical: PhysicalStreamID(i), Data: chunk})
554 require.NoError(t, err)
555 require.Equal(t, StreamAccepted, disposition)
556 }
557 require.Equal(t, int(MaxMuxAggregateBytes), e.AggregateBytes())
558
559 openTestStream(t, e, 5)
560 disposition, err := e.Data(Data{Physical: 5, Data: chunk})
561 require.ErrorIs(t, err, ErrStreamQueueFull)
562 require.Zero(t, disposition)
563 require.Equal(t, StreamTerminal, mustStatus(t, e, 5).State)
564 require.Equal(t, int(MaxMuxAggregateBytes), e.AggregateBytes(), "siblings keep their budget")
565 for i := 1; i <= 4; i++ {
566 require.Equal(t, StreamOpen, mustStatus(t, e, PhysicalStreamID(i)).State)
567 }
568 require.Equal(t, 4, e.Live())
569
570 // Taking one sibling chunk releases exactly its bytes, so a fresh
571 // stream takes the freed capacity.
572 _, ok := e.Take(PhysicalStreamID(1))
573 require.True(t, ok)
574 require.Equal(t, int(MaxMuxAggregateBytes)-len(chunk), e.AggregateBytes())
575 openTestStream(t, e, 6)
576 disposition, err = e.Data(Data{Physical: 6, Data: chunk})
577 require.NoError(t, err)
578 require.Equal(t, StreamAccepted, disposition)
579 require.Equal(t, int(MaxMuxAggregateBytes), e.AggregateBytes())
580 })
581}
582
583// TestStreamEngineTake proves the FIFO pop returns an engine-owned copy and
584// releases the chunk's bytes.
585func TestStreamEngineTake(t *testing.T) {
586 e := testEngine()
587 openTestStream(t, e, 1)
588 chunk := []byte("payload")
589 _, err := e.Data(Data{Physical: 1, Data: chunk})
590 require.NoError(t, err)
591
592 chunk[0] = 'P' // mutating the caller's slice must not change the queue
593 taken, ok := e.Take(PhysicalStreamID(1))
594 require.True(t, ok)
595 require.Equal(t, []byte("payload"), taken)
596 require.Zero(t, e.AggregateBytes())
597
598 _, ok = e.Take(PhysicalStreamID(1))
599 require.False(t, ok)
600 _, ok = e.Take(PhysicalStreamID(9))
601 require.False(t, ok)
602}
603
604// TestStreamEngineResetCause proves a reset is abnormal: it carries the typed
605// cause when present and the terminal sentinel otherwise.
606func TestStreamEngineResetCause(t *testing.T) {
607 t.Run("typed detail", func(t *testing.T) {
608 e := testEngine()
609 openTestStream(t, e, 1)
610 detail := ErrorDetail{
611 Code: ports.BrokerErrorUnavailable, Text: "reset by peer",
612 AdmissionCode: 3, FailureKind: domain.RemoteFailureTimeout,
613 }
614 disposition, err := e.Reset(Reset{Physical: 1, HasError: true, Error: detail})
615 require.NoError(t, err)
616 require.Equal(t, StreamAccepted, disposition)
617
618 status := mustStatus(t, e, 1)
619 require.Error(t, status.Err)
620 require.Equal(t, domain.RemoteFailureTimeout, status.FailureKind)
621 var brokerErr ports.BrokerError
622 require.ErrorAs(t, status.Err, &brokerErr)
623 require.Equal(t, ports.BrokerErrorUnavailable, brokerErr.Code)
624 require.Equal(t, "reset by peer", brokerErr.Text)
625 })
626
627 t.Run("no detail", func(t *testing.T) {
628 e := testEngine()
629 openTestStream(t, e, 1)
630 disposition, err := e.Reset(Reset{Physical: 1})
631 require.NoError(t, err)
632 require.Equal(t, StreamAccepted, disposition)
633
634 status := mustStatus(t, e, 1)
635 require.ErrorIs(t, status.Err, ErrTerminalFailure)
636 require.Equal(t, domain.RemoteFailureTransport, status.FailureKind)
637 })
638}
639
640// TestStreamEnginePhysicalFailureOrdering proves a physical failure closes
641// Done before terminalizing every live stream, that every stream observes the
642// physical outcome, and that a stream settled beforehand keeps its own cause.
643func TestStreamEnginePhysicalFailureOrdering(t *testing.T) {
644 e := testEngine()
645 const live = 5
646 for i := 1; i <= live; i++ {
647 openTestStream(t, e, PhysicalStreamID(i))
648 }
649 _, err := e.Reset(Reset{Physical: 5, HasError: true, Error: testErrorDetail()})
650 require.NoError(t, err)
651 siblingCause := mustStatus(t, e, 5).Err
652 require.Error(t, siblingCause)
653
654 cause := fmt.Errorf("physical loss: %w", errTerminalSentinel)
655 observed := make([]bool, live-1)
656 var wg sync.WaitGroup
657 for i := 1; i < live; i++ {
658 status := mustStatus(t, e, PhysicalStreamID(i))
659 wg.Add(1)
660 go func(i int, done <-chan struct{}) {
661 defer wg.Done()
662 <-done
663 // A reader woken by a stream must observe the physical
664 // connection already terminal.
665 observed[i-1] = e.Closed()
666 }(i, status.Done)
667 }
668 e.Fail(domain.RemoteFailureTimeout, cause)
669 wg.Wait()
670
671 require.True(t, e.Closed())
672 require.Equal(t, cause, e.Err())
673 require.ErrorIs(t, e.Err(), errTerminalSentinel)
674 require.Equal(t, domain.RemoteFailureTimeout, e.FailureKind())
675 require.Zero(t, e.Live())
676 require.Zero(t, e.AggregateBytes())
677
678 for i := 1; i < live; i++ {
679 status := mustStatus(t, e, PhysicalStreamID(i))
680 require.Equal(t, StreamTerminal, status.State)
681 require.True(t, channelClosed(status.Done))
682 require.Equal(t, cause, status.Err)
683 require.Equal(t, domain.RemoteFailureTimeout, status.FailureKind)
684 require.True(t, observed[i-1], "stream %d was terminalized before the physical Done", i)
685 }
686
687 // The stream settled before the loss keeps its own, independent cause.
688 settled := mustStatus(t, e, 5)
689 require.Equal(t, StreamTerminal, settled.State)
690 require.Equal(t, siblingCause, settled.Err)
691}
692
693// TestStreamEngineTerminate proves an orderly physical close terminalizes
694// every stream with a nil error and refuses further admission.
695func TestStreamEngineTerminate(t *testing.T) {
696 e := testEngine()
697 for i := 1; i <= 3; i++ {
698 openTestStream(t, e, PhysicalStreamID(i))
699 }
700 _, err := e.Data(Data{Physical: 1, Data: []byte("queued")})
701 require.NoError(t, err)
702
703 e.Terminate()
704 require.True(t, e.Closed())
705 require.NoError(t, e.Err())
706 require.Equal(t, domain.RemoteFailureNone, e.FailureKind())
707 require.Zero(t, e.Live())
708 require.Zero(t, e.AggregateBytes())
709 for i := 1; i <= 3; i++ {
710 status := mustStatus(t, e, PhysicalStreamID(i))
711 require.Equal(t, StreamTerminal, status.State)
712 require.True(t, channelClosed(status.Done))
713 require.NoError(t, status.Err)
714 }
715
716 require.ErrorIs(t, e.Open(Open{Ref: testRef(9)}), ErrPhysicalClosed)
717 e.Terminate()
718 require.True(t, e.Closed())
719}
720
721// TestStreamEnginePhysicalClosedFrames proves frames after a physical terminal
722// outcome stay classified: settled streams discard traffic, admission is
723// closed, and a future identity is still refused.
724func TestStreamEnginePhysicalClosedFrames(t *testing.T) {
725 e := testEngine()
726 openTestStream(t, e, 1)
727 e.Fail(domain.RemoteFailureTransport, errors.New("gone"))
728
729 for _, run := range []struct {
730 name string
731 run func() (StreamDisposition, error)
732 }{
733 {"data", func() (StreamDisposition, error) { return e.Data(Data{Physical: 1, Data: []byte{'x'}}) }},
734 {"close", func() (StreamDisposition, error) { return e.Close(Close{Physical: 1}) }},
735 {"reset", func() (StreamDisposition, error) { return e.Reset(Reset{Physical: 1}) }},
736 } {
737 disposition, err := run.run()
738 require.NoError(t, err, run.name)
739 require.Equal(t, StreamDiscarded, disposition, run.name)
740 }
741
742 require.ErrorIs(t, e.Open(Open{Ref: testRef(2)}), ErrPhysicalClosed)
743 _, err := e.Data(Data{Physical: 3, Data: []byte{'x'}})
744 require.ErrorIs(t, err, ErrFutureStream)
745}
746
747// TestStreamEngineConcurrentStreams exercises concurrent confirmation, data,
748// orderly close, abnormal reset, and admission over one engine.
749func TestStreamEngineConcurrentStreams(t *testing.T) {
750 const streams = 64
751 const admitted = 32
752 e := testEngine()
753 for i := 1; i <= streams; i++ {
754 admitTestStream(t, e, PhysicalStreamID(i))
755 }
756
757 var wg sync.WaitGroup
758 start := make(chan struct{})
759 for i := 1; i <= streams; i++ {
760 wg.Add(1)
761 go func(i int) {
762 defer wg.Done()
763 <-start
764 id := PhysicalStreamID(i)
765 assert.NoError(t, e.Opened(Opened{Ref: testRef(id)}))
766 disposition, err := e.Data(Data{Physical: id, Data: []byte{byte(i)}})
767 assert.NoError(t, err)
768 assert.Equal(t, StreamAccepted, disposition)
769 if i%2 == 0 {
770 disposition, err = e.Close(Close{Physical: id})
771 } else {
772 disposition, err = e.Reset(Reset{Physical: id})
773 }
774 assert.NoError(t, err)
775 assert.Equal(t, StreamAccepted, disposition)
776 }(i)
777 }
778 // One goroutine keeps admitting fresh, strictly increasing identities
779 // while the others settle theirs.
780 wg.Add(1)
781 go func() {
782 defer wg.Done()
783 <-start
784 for i := 1; i <= admitted; i++ {
785 id := PhysicalStreamID(streams + i)
786 assert.NoError(t, e.Open(Open{Ref: testRef(id)}))
787 assert.NoError(t, e.Opened(Opened{Ref: testRef(id)}))
788 disposition, err := e.Close(Close{Physical: id})
789 assert.NoError(t, err)
790 assert.Equal(t, StreamAccepted, disposition)
791 }
792 }()
793 close(start)
794 wg.Wait()
795
796 // The even streams were closed with one chunk still unread, so they are
797 // closing and still charged; draining each queue terminalizes it. Odd
798 // streams and the admitted-only goroutine are already terminal.
799 for i := 1; i <= streams+admitted; i++ {
800 for {
801 if _, ok := e.Take(PhysicalStreamID(i)); !ok {
802 break
803 }
804 }
805 }
806
807 require.Zero(t, e.Live())
808 require.Zero(t, e.AggregateBytes())
809 require.False(t, e.Closed())
810 for i := 1; i <= streams+admitted; i++ {
811 require.Equal(t, StreamTerminal, mustStatus(t, e, PhysicalStreamID(i)).State)
812 }
813}
814
815// TestStreamEngineConcurrentPhysicalFailure races stream work against a
816// physical failure and proves the engine settles consistently.
817func TestStreamEngineConcurrentPhysicalFailure(t *testing.T) {
818 const streams = 64
819 e := testEngine()
820 for i := 1; i <= streams; i++ {
821 openTestStream(t, e, PhysicalStreamID(i))
822 }
823 cause := errors.New("physical loss")
824
825 var wg sync.WaitGroup
826 start := make(chan struct{})
827 for i := 1; i <= streams; i++ {
828 wg.Add(1)
829 go func(i int) {
830 defer wg.Done()
831 <-start
832 id := PhysicalStreamID(i)
833 switch i % 3 {
834 case 0:
835 _, _ = e.Data(Data{Physical: id, Data: []byte{byte(i)}})
836 case 1:
837 _, _ = e.Close(Close{Physical: id})
838 default:
839 _, _ = e.Reset(Reset{Physical: id})
840 }
841 }(i)
842 }
843 wg.Add(1)
844 go func() {
845 defer wg.Done()
846 <-start
847 e.Fail(domain.RemoteFailureTimeout, cause)
848 }()
849 close(start)
850 wg.Wait()
851
852 require.True(t, e.Closed())
853 require.Equal(t, cause, e.Err())
854 require.Zero(t, e.Live())
855 require.Zero(t, e.AggregateBytes())
856 for i := 1; i <= streams; i++ {
857 status := mustStatus(t, e, PhysicalStreamID(i))
858 require.Equal(t, StreamTerminal, status.State)
859 require.True(t, channelClosed(status.Done))
860 }
861}
862
863// TestStreamEngineNilSafety proves the nil-receiver paths.
864func TestStreamEngineNilSafety(t *testing.T) {
865 var e *StreamEngine
866 require.ErrorIs(t, e.Open(Open{Ref: testRef(1)}), ErrPhysicalClosed)
867 require.ErrorIs(t, e.Opened(Opened{Ref: testRef(1)}), ErrPhysicalClosed)
868 require.ErrorIs(t, e.Refused(Refused{Ref: testRef(1)}), ErrPhysicalClosed)
869 _, err := e.Data(Data{Physical: 1, Data: []byte{'x'}})
870 require.ErrorIs(t, err, ErrPhysicalClosed)
871 _, err = e.Close(Close{Physical: 1})
872 require.ErrorIs(t, err, ErrPhysicalClosed)
873 _, err = e.Reset(Reset{Physical: 1})
874 require.ErrorIs(t, err, ErrPhysicalClosed)
875 require.Zero(t, e.Live())
876 require.Zero(t, e.AggregateBytes())
877 require.True(t, e.Closed())
878 require.Nil(t, e.Done())
879 require.NoError(t, e.Err())
880 require.Equal(t, domain.RemoteFailureNone, e.FailureKind())
881 _, ok := e.Status(1)
882 require.False(t, ok)
883 _, ok = e.Take(1)
884 require.False(t, ok)
885 e.Fail(domain.RemoteFailureTransport, errors.New("ignored"))
886 e.Terminate()
887}
888
889// TestNewStreamEngineWithCeilings proves the constructor validates the
890// negotiated advertisement, enforces the negotiated maxima, and copies the
891// value so a caller cannot mutate live enforcement.
892func TestNewStreamEngineWithCeilings(t *testing.T) {
893 t.Run("invalid advertisement refused", func(t *testing.T) {
894 bad := DefaultMuxCeilings()
895 bad.MaxStreams = 0
896 engine, err := NewStreamEngineWithCeilings(bad)
897 require.ErrorIs(t, err, ErrInvalidCeilings)
898 require.Nil(t, engine)
899
900 bad = DefaultMuxCeilings()
901 bad.StreamChunkLimit = MaxMuxChunkBytes + 1
902 _, err = NewStreamEngineWithCeilings(bad)
903 require.ErrorIs(t, err, ErrInvalidCeilings)
904
905 _, err = NewStreamEngineWithCeilings(MuxCeilings{})
906 require.ErrorIs(t, err, ErrInvalidCeilings)
907 })
908
909 t.Run("negotiated small limits enforced", func(t *testing.T) {
910 ceilings := MuxCeilings{
911 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
912 StreamChunkLimit: 32768,
913 MaxStreams: 2,
914 MaxAggregateBytes: MinMuxAggregateBytes,
915 }
916 e, err := NewStreamEngineWithCeilings(ceilings)
917 require.NoError(t, err)
918 openTestStream(t, e, 1)
919 openTestStream(t, e, 2)
920 require.ErrorIs(t, e.Open(Open{Ref: testRef(3)}), ErrTooManyStreams)
921
922 // The negotiated chunk ceiling is refused statelessly, mutating
923 // nothing.
924 _, err = e.Data(Data{Physical: 1, Data: make([]byte, 32769)})
925 require.ErrorIs(t, err, ErrTooLarge)
926 require.Equal(t, StreamOpen, mustStatus(t, e, 1).State)
927 require.Zero(t, e.AggregateBytes())
928
929 // Two full chunks fit the aggregate floor; a third would exceed it and
930 // resets exactly the offending stream.
931 _, err = e.Data(Data{Physical: 1, Data: make([]byte, 32768)})
932 require.NoError(t, err)
933 require.Equal(t, 32768, e.AggregateBytes())
934 _, err = e.Data(Data{Physical: 2, Data: make([]byte, 32768)})
935 require.NoError(t, err)
936 require.Equal(t, 65536, e.AggregateBytes())
937 _, err = e.Data(Data{Physical: 1, Data: make([]byte, 32768)})
938 require.ErrorIs(t, err, ErrStreamQueueFull)
939 require.Equal(t, StreamTerminal, mustStatus(t, e, 1).State)
940 require.Equal(t, 32768, e.AggregateBytes(), "the reset stream released its bytes")
941 require.Equal(t, StreamOpen, mustStatus(t, e, 2).State)
942 })
943
944 t.Run("ceilings are copied", func(t *testing.T) {
945 ceilings := DefaultMuxCeilings()
946 ceilings.MaxStreams = 1
947 e, err := NewStreamEngineWithCeilings(ceilings)
948 require.NoError(t, err)
949 // A caller mutating its own copy must not raise the engine's ceiling.
950 ceilings.MaxStreams = MaxMuxStreams
951 openTestStream(t, e, 1)
952 require.ErrorIs(t, e.Open(Open{Ref: testRef(2)}), ErrTooManyStreams)
953 })
954}
955
956// TestStreamEngineBoundedRetiredRecords proves a churned long-lived engine
957// keeps a bounded table: the oldest retired records are evicted, while the
958// high-water mark still classifies them as retired rather than future.
959func TestStreamEngineBoundedRetiredRecords(t *testing.T) {
960 e := testEngine()
961 const churn = 4000
962 for i := 1; i <= churn; i++ {
963 id := PhysicalStreamID(i)
964 openTestStream(t, e, id)
965 _, err := e.Close(Close{Physical: id})
966 require.NoError(t, err)
967 }
968 require.LessOrEqual(t, len(e.streams), MaxRetiredStreamRecords)
969 require.Len(t, e.retired, MaxRetiredStreamRecords)
970
971 // The newest identity is retained; the oldest record was evicted but its
972 // high-water mark still classifies late traffic as retired.
973 require.Equal(t, StreamTerminal, mustStatus(t, e, PhysicalStreamID(churn)).State)
974 _, ok := e.Status(PhysicalStreamID(1))
975 require.False(t, ok)
976 disposition, err := e.Data(Data{Physical: PhysicalStreamID(1), Data: []byte{'x'}})
977 require.NoError(t, err)
978 require.Equal(t, StreamDiscarded, disposition)
979
980 // An identity above the high-water mark is still refused, not discarded.
981 _, err = e.Data(Data{Physical: PhysicalStreamID(churn + 1), Data: []byte{'x'}})
982 require.ErrorIs(t, err, ErrFutureStream)
983}
984
985// TestStreamEnginePhysicalFirstWins proves the physical terminal publication
986// is first-wins: whichever of Fail or Terminate publishes first fixes the
987// connection and every live stream's outcome.
988func TestStreamEnginePhysicalFirstWins(t *testing.T) {
989 t.Run("failure wins an orderly close", func(t *testing.T) {
990 e := testEngine()
991 openTestStream(t, e, 1)
992 cause := errors.New("physical loss")
993 e.Fail(domain.RemoteFailureTimeout, cause)
994 e.Terminate()
995 require.True(t, e.Closed())
996 require.Equal(t, cause, e.Err())
997 require.Equal(t, domain.RemoteFailureTimeout, e.FailureKind())
998 status := mustStatus(t, e, 1)
999 require.Equal(t, cause, status.Err)
1000 require.Equal(t, domain.RemoteFailureTimeout, status.FailureKind)
1001 })
1002
1003 t.Run("orderly close wins a failure", func(t *testing.T) {
1004 e := testEngine()
1005 openTestStream(t, e, 1)
1006 e.Terminate()
1007 e.Fail(domain.RemoteFailureTimeout, errors.New("late"))
1008 require.True(t, e.Closed())
1009 require.NoError(t, e.Err())
1010 require.Equal(t, domain.RemoteFailureNone, e.FailureKind())
1011 status := mustStatus(t, e, 1)
1012 require.NoError(t, status.Err)
1013 require.Equal(t, domain.RemoteFailureNone, status.FailureKind)
1014 })
1015}
1016
1017// TestStreamEngineClosingPreservesQueuedData proves the orderly-close
1018// semantics: a Close that finds unread data marks the stream closing instead of
1019// discarding it, no new data is accepted, already-accepted chunks stay readable
1020// in order, and the first Take that empties the queue terminalizes the stream
1021// and releases its slot and bytes. A sibling is untouched throughout.
1022func TestStreamEngineClosingPreservesQueuedData(t *testing.T) {
1023 e := testEngine()
1024 closingRef := openTestStream(t, e, 1)
1025 siblingRef := openTestStream(t, e, 2)
1026
1027 for _, chunk := range []struct {
1028 physical PhysicalStreamID
1029 data string
1030 }{{1, "first"}, {1, "second"}, {2, "sibling"}} {
1031 disposition, err := e.Data(Data{Physical: chunk.physical, Data: []byte(chunk.data)})
1032 require.NoError(t, err)
1033 require.Equal(t, StreamAccepted, disposition)
1034 }
1035 require.Equal(t, len("first")+len("second")+len("sibling"), e.AggregateBytes())
1036
1037 // The orderly Close preserves both accepted chunks.
1038 disposition, err := e.Close(Close{Physical: 1})
1039 require.NoError(t, err)
1040 require.Equal(t, StreamAccepted, disposition)
1041 status := mustStatus(t, e, 1)
1042 require.Equal(t, StreamClosing, status.State)
1043 require.Equal(t, closingRef, status.Ref)
1044 require.False(t, channelClosed(status.Done), "a closing stream is not terminal yet")
1045 require.Equal(t, 2, status.QueuedChunks)
1046 require.Equal(t, len("first")+len("second"), status.QueuedBytes)
1047 require.Equal(t, 2, e.Live(), "the closing stream still holds its admission slot")
1048 require.Equal(t, len("first")+len("second")+len("sibling"), e.AggregateBytes())
1049
1050 // No new data is accepted while closing, and a second Close changes nothing.
1051 _, err = e.Data(Data{Physical: 1, Data: []byte("late")})
1052 require.ErrorIs(t, err, ErrStreamState)
1053 disposition, err = e.Close(Close{Physical: 1})
1054 require.NoError(t, err)
1055 require.Equal(t, StreamDiscarded, disposition)
1056 require.Equal(t, 2, mustStatus(t, e, 1).QueuedChunks, "a duplicate close preserved the queue")
1057
1058 // Draining preserves order and releases bytes chunk by chunk; only the last
1059 // take terminalizes the stream.
1060 chunk, ok := e.Take(1)
1061 require.True(t, ok)
1062 require.Equal(t, []byte("first"), chunk)
1063 require.Equal(t, StreamClosing, mustStatus(t, e, 1).State, "still closing with one chunk left")
1064 require.Equal(t, 2, e.Live())
1065 require.Equal(t, len("second")+len("sibling"), e.AggregateBytes())
1066
1067 chunk, ok = e.Take(1)
1068 require.True(t, ok)
1069 require.Equal(t, []byte("second"), chunk)
1070 status = mustStatus(t, e, 1)
1071 require.Equal(t, StreamTerminal, status.State)
1072 require.True(t, channelClosed(status.Done))
1073 require.Zero(t, status.QueuedChunks)
1074 require.NoError(t, status.Err)
1075 require.Equal(t, 1, e.Live(), "the drained stream released its slot")
1076 require.Equal(t, len("sibling"), e.AggregateBytes(), "only the sibling's bytes remain")
1077
1078 // A drained closing stream accepts no second drain and no new frame.
1079 _, ok = e.Take(1)
1080 require.False(t, ok)
1081 disposition, err = e.Close(Close{Physical: 1})
1082 require.NoError(t, err)
1083 require.Equal(t, StreamDiscarded, disposition)
1084
1085 // The sibling is untouched and still holds its own queued chunk.
1086 siblingStatus := mustStatus(t, e, 2)
1087 require.Equal(t, StreamOpen, siblingStatus.State)
1088 require.Equal(t, siblingRef, siblingStatus.Ref)
1089 require.Equal(t, 1, siblingStatus.QueuedChunks)
1090 chunk, ok = e.Take(2)
1091 require.True(t, ok)
1092 require.Equal(t, []byte("sibling"), chunk)
1093}
1094
1095// TestStreamEngineClosingResetDiscardsImmediately proves a Reset abandons a
1096// closing stream's preserved queue at once: the bytes, the admission slot, and
1097// the terminal outcome are released without waiting for a consumer.
1098func TestStreamEngineClosingResetDiscardsImmediately(t *testing.T) {
1099 e := testEngine()
1100 openTestStream(t, e, 1)
1101 _, err := e.Data(Data{Physical: 1, Data: []byte("queued")})
1102 require.NoError(t, err)
1103 disposition, err := e.Close(Close{Physical: 1})
1104 require.NoError(t, err)
1105 require.Equal(t, StreamAccepted, disposition)
1106 require.Equal(t, StreamClosing, mustStatus(t, e, 1).State)
1107
1108 disposition, err = e.Reset(Reset{Physical: 1})
1109 require.NoError(t, err)
1110 require.Equal(t, StreamAccepted, disposition)
1111 status := mustStatus(t, e, 1)
1112 require.Equal(t, StreamTerminal, status.State)
1113 require.True(t, channelClosed(status.Done))
1114 require.Zero(t, status.QueuedChunks)
1115 require.Zero(t, status.QueuedBytes)
1116 require.Zero(t, e.AggregateBytes())
1117 require.Zero(t, e.Live())
1118 _, ok := e.Take(1)
1119 require.False(t, ok, "a reset discarded the preserved queue")
1120}
1121
1122// TestStreamEngineClosingAccountingBounded proves the closing accounting is
1123// balanced across many cycles and stays bounded: a closing stream keeps its
1124// slot and bytes until the draining Take releases them, and the retired record
1125// set never grows past its retention bound.
1126func TestStreamEngineClosingAccountingBounded(t *testing.T) {
1127 e := testEngine()
1128 const churn = MaxRetiredStreamRecords * 3
1129 for i := 1; i <= churn; i++ {
1130 id := PhysicalStreamID(i)
1131 openTestStream(t, e, id)
1132 _, err := e.Data(Data{Physical: id, Data: []byte{byte(i)}})
1133 require.NoError(t, err)
1134 disposition, err := e.Close(Close{Physical: id})
1135 require.NoError(t, err)
1136 require.Equal(t, StreamAccepted, disposition)
1137 require.Equal(t, StreamClosing, mustStatus(t, e, id).State)
1138 require.Equal(t, 1, e.Live())
1139 require.Equal(t, 1, e.AggregateBytes())
1140
1141 chunk, ok := e.Take(id)
1142 require.True(t, ok)
1143 require.Equal(t, []byte{byte(i)}, chunk)
1144 require.Equal(t, StreamTerminal, mustStatus(t, e, id).State)
1145 require.Zero(t, e.Live())
1146 require.Zero(t, e.AggregateBytes())
1147 }
1148 require.Len(t, e.retired, MaxRetiredStreamRecords)
1149 require.LessOrEqual(t, len(e.streams), MaxRetiredStreamRecords)
1150}