1package daemonmux
2
3import (
4 "fmt"
5 "strings"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/ports"
13)
14
15// testScheduler returns a scheduler with an open admission gate and one open
16// stream per requested physical identity, in ascending order.
17func testScheduler(t *testing.T, streams int) (*StreamEngine, *Scheduler, []PhysicalStreamID) {
18 t.Helper()
19 engine := testEngine()
20 scheduler := NewScheduler(engine)
21 ids := make([]PhysicalStreamID, streams)
22 for i := range ids {
23 ids[i] = PhysicalStreamID(i + 1)
24 openTestStream(t, engine, ids[i])
25 }
26 return engine, scheduler, ids
27}
28
29func countActive(scheduler *Scheduler, ids []PhysicalStreamID) int {
30 active := 0
31 for _, id := range ids {
32 if scheduler.QueuedFrames(id) > 0 {
33 active++
34 }
35 }
36 return active
37}
38
39// overflowOne queues one small frame for id and then overflows the aggregate
40// bound with a frame the budget cannot hold, so id's record is overflowed and
41// empty. The aggregate is restored for the next caller.
42func overflowOne(t *testing.T, scheduler *Scheduler, id PhysicalStreamID) {
43 t.Helper()
44 saved := scheduler.maxAggregateBytes
45 defer func() { scheduler.maxAggregateBytes = saved }()
46 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("a")}))
47 scheduler.maxAggregateBytes = scheduler.AggregateBytes()
48 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("b")}), ErrSchedulerFull)
49}
50
51// TestSchedulerConstants pins the negotiated bounds the scheduler is specified
52// to enforce.
53func TestSchedulerConstants(t *testing.T) {
54 require.Equal(t, 64<<20, int(MaxMuxAggregateBytes))
55 require.Equal(t, uint64(512<<10), DefaultMuxCeilings().StreamWindow())
56 require.Equal(t, int64(64<<10), SchedulerQuantum)
57 require.Equal(t, 4, MaxMuxSchedulerControlBurst)
58}
59
60func TestSchedulerStrings(t *testing.T) {
61 require.Equal(t, "data", FrameData.String())
62 require.Equal(t, "control", FrameControl.String())
63 require.Equal(t, "unknown", FrameClass(0).String())
64 require.Equal(t, "client", DirectionClient.String())
65 require.Equal(t, "server", DirectionServer.String())
66 require.Equal(t, "unknown", EnvelopeDirection(0).String())
67}
68
69func TestSchedulerNilSafety(t *testing.T) {
70 var scheduler *Scheduler
71 _, ok := scheduler.Dequeue()
72 require.False(t, ok)
73 require.False(t, scheduler.Pending())
74 require.Zero(t, scheduler.Active())
75 require.Zero(t, scheduler.AggregateBytes())
76 require.Zero(t, scheduler.QueuedFrames(1))
77 require.Zero(t, scheduler.QueuedBytes(1))
78 require.False(t, scheduler.Reset(1))
79 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: []byte("x")}), ErrInvalidMessage)
80 require.ErrorIs(t, scheduler.EnqueueServer(Opened{Ref: testRef(1)}), ErrInvalidMessage)
81}
82
83// TestSchedulerAdmissionGate proves a frame is schedulable only for a stream the
84// gate admitted and has not terminalized, and that a nil gate trusts the
85// caller.
86func TestSchedulerAdmissionGate(t *testing.T) {
87 t.Run("engine gate", func(t *testing.T) {
88 engine := testEngine()
89 scheduler := NewScheduler(engine)
90 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 9, Data: []byte("x")}), ErrSchedulerUnadmitted)
91
92 require.NoError(t, engine.Open(Open{Ref: testRef(9)}))
93 require.NoError(t, scheduler.EnqueueClient(Data{Physical: 9, Data: []byte("x")}))
94
95 require.NoError(t, engine.Opened(Opened{Ref: testRef(9)}))
96 require.NoError(t, scheduler.EnqueueClient(Data{Physical: 9, Data: []byte("y")}))
97
98 _, err := engine.Close(Close{Physical: 9})
99 require.NoError(t, err)
100 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 9, Data: []byte("z")}), ErrSchedulerUnadmitted)
101 })
102
103 t.Run("nil gate", func(t *testing.T) {
104 scheduler := NewScheduler(nil)
105 require.NoError(t, scheduler.EnqueueClient(Data{Physical: 7, Data: []byte("x")}))
106 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 0, Data: []byte("x")}), ErrInvalidMessage)
107 })
108
109 t.Run("unknown variant fails closed", func(t *testing.T) {
110 scheduler := NewScheduler(nil)
111 require.ErrorIs(t, scheduler.EnqueueClient(unknownClientMessage{}), ErrWrongDirection)
112 require.ErrorIs(t, scheduler.EnqueueServer(unknownServerMessage{}), ErrWrongDirection)
113 })
114}
115
116// TestSchedulerEnvelopeRoundTrip proves a dequeued envelope carries the
117// complete serialized directional envelope, decoded by the matching codec, and
118// that neither the bytes nor the committed message aliases caller memory.
119func TestSchedulerEnvelopeRoundTrip(t *testing.T) {
120 ceiling := DefaultMuxCeilings()
121 _, scheduler, ids := testScheduler(t, 1)
122
123 payload := []byte("payload")
124 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: payload}))
125 payload[0] = 'X' // a caller mutation must never reach the queued frame
126
127 envelope, ok := scheduler.Dequeue()
128 require.True(t, ok)
129 require.Equal(t, ids[0], envelope.Physical)
130 require.Equal(t, DirectionClient, envelope.Direction)
131 require.Equal(t, FrameData, envelope.Class)
132 require.Nil(t, envelope.Server)
133 data, ok := envelope.Client.(Data)
134 require.True(t, ok)
135 require.Equal(t, []byte("payload"), data.Data)
136
137 decoded, err := DecodeClient(envelope.Bytes, ceiling.MaxReceiveEnvelopeBytes, ceiling.StreamChunkLimit)
138 require.NoError(t, err)
139 require.Equal(t, Data{Physical: ids[0], Data: []byte("payload")}, decoded)
140
141 require.NoError(t, scheduler.EnqueueServer(Opened{Ref: testRef(ids[0])}))
142 serverEnvelope, ok := scheduler.Dequeue()
143 require.True(t, ok)
144 require.Equal(t, DirectionServer, serverEnvelope.Direction)
145 require.Equal(t, FrameControl, serverEnvelope.Class)
146 require.Nil(t, serverEnvelope.Client)
147 decodedServer, err := DecodeServer(serverEnvelope.Bytes, ceiling.MaxReceiveEnvelopeBytes, ceiling.StreamChunkLimit)
148 require.NoError(t, err)
149 require.Equal(t, Opened{Ref: testRef(ids[0])}, decodedServer)
150}
151
152// TestSchedulerTwoAndHundredStreamsFairness proves deterministic round-robin
153// over a small and a large concurrent set: every stream is served once per
154// round, in admission order, with its own frames in order.
155func TestSchedulerTwoAndHundredStreamsFairness(t *testing.T) {
156 for _, count := range []int{2, 100} {
157 t.Run(fmt.Sprintf("%d streams", count), func(t *testing.T) {
158 _, scheduler, ids := testScheduler(t, count)
159 const rounds = 2
160 for round := 0; round < rounds; round++ {
161 for _, id := range ids {
162 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte{byte(id), byte(round)}}))
163 }
164 }
165 require.Equal(t, count, scheduler.Active())
166
167 seen := make(map[PhysicalStreamID]int, count)
168 for served := 0; served < rounds*count; served++ {
169 envelope, ok := scheduler.Dequeue()
170 require.True(t, ok, "dequeue %d", served)
171 want := ids[served%count]
172 require.Equal(t, want, envelope.Physical, "dequeue %d", served)
173 require.Equal(t, FrameData, envelope.Class)
174 require.Equal(t, DirectionClient, envelope.Direction)
175 data, ok := envelope.Client.(Data)
176 require.True(t, ok)
177 require.Equal(t, []byte{byte(want), byte(seen[want])}, data.Data, "per-stream order")
178 seen[want]++
179 }
180 require.False(t, scheduler.Pending())
181 require.Zero(t, scheduler.AggregateBytes())
182 })
183 }
184}
185
186// TestSchedulerStalledAndHotStream proves a continuously backlogged hot stream
187// never starves a sibling, and a full stalled backlog never blocks another
188// stream: every stream is served once per round while all queues stay full.
189func TestSchedulerStalledAndHotStream(t *testing.T) {
190 _, scheduler, ids := testScheduler(t, 4)
191
192 // The hot stream keeps a maximal backlog; the siblings are refilled to full
193 // whenever the round drains a slot.
194 const backlog = 8
195 fill := func(id PhysicalStreamID, seed byte) {
196 for scheduler.QueuedFrames(id) < backlog {
197 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte{seed}}))
198 }
199 }
200
201 const iterations = 400
202 served := make(map[PhysicalStreamID]int, len(ids))
203 for i := 0; i < iterations; i++ {
204 fill(ids[0], byte(i))
205 for _, id := range ids[1:] {
206 fill(id, byte(i))
207 }
208 envelope, ok := scheduler.Dequeue()
209 require.True(t, ok, "iteration %d", i)
210 served[envelope.Physical]++
211 }
212
213 for _, id := range ids {
214 require.Greater(t, served[id], 0, "stream %d starved", id)
215 }
216 fair := iterations / (len(ids) + 1)
217 for _, id := range ids {
218 require.GreaterOrEqual(t, served[id], fair, "stream %d fell behind the hot stream", id)
219 }
220}
221
222// TestSchedulerCloseOrdering proves a Close never overtakes earlier accepted
223// data of its own stream while control metadata still gets bounded priority
224// over another stream's data.
225func TestSchedulerCloseOrdering(t *testing.T) {
226 _, scheduler, ids := testScheduler(t, 2)
227
228 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("a")}))
229 require.NoError(t, scheduler.EnqueueClient(Close{Physical: ids[1]}))
230
231 first, ok := scheduler.Dequeue()
232 require.True(t, ok)
233 require.Equal(t, ids[1], first.Physical, "control metadata takes bounded priority")
234 require.Equal(t, FrameControl, first.Class)
235 _, isClose := first.Client.(Close)
236 require.True(t, isClose)
237
238 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("b")}))
239 require.NoError(t, scheduler.EnqueueClient(Close{Physical: ids[0]}))
240
241 var order []string
242 for scheduler.Pending() {
243 envelope, ok := scheduler.Dequeue()
244 require.True(t, ok)
245 if envelope.Physical != ids[0] {
246 continue
247 }
248 switch message := envelope.Client.(type) {
249 case Data:
250 order = append(order, string(message.Data))
251 case Close:
252 order = append(order, "close")
253 default:
254 t.Fatalf("unexpected message %T", message)
255 }
256 }
257 require.Equal(t, []string{"a", "b", "close"}, order)
258
259 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("c")}), ErrSchedulerSettled)
260 require.ErrorIs(t, scheduler.EnqueueClient(Close{Physical: ids[0]}), ErrSchedulerSettled)
261}
262
263// TestSchedulerControlPriorityBounded proves control frames are prioritized
264// over data but only for a bounded burst, so control metadata can never starve
265// data.
266func TestSchedulerControlPriorityBounded(t *testing.T) {
267 _, scheduler, ids := testScheduler(t, 5)
268 const backlog = 8
269 for round := 0; round < backlog; round++ {
270 for _, id := range ids[:4] {
271 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte{byte(id), byte(round)}}))
272 }
273 require.NoError(t, scheduler.EnqueueServer(Opened{Ref: testRef(ids[4])}))
274 }
275
276 dataServed := make(map[PhysicalStreamID]int, 4)
277 controls := 0
278 run := 0
279 for scheduler.Pending() {
280 envelope, ok := scheduler.Dequeue()
281 require.True(t, ok)
282 if envelope.Class == FrameControl {
283 require.Equal(t, ids[4], envelope.Physical)
284 controls++
285 run++
286 require.LessOrEqual(t, run, MaxMuxSchedulerControlBurst, "control burst exceeded")
287 continue
288 }
289 run = 0
290 dataServed[envelope.Physical]++
291 }
292 require.Equal(t, backlog, controls)
293 for _, id := range ids[:4] {
294 require.Equal(t, backlog, dataServed[id], "stream %d starved by control", id)
295 }
296}
297
298// TestSchedulerResetDiscard proves a reset discards exactly its own queue,
299// frees its bytes immediately, is idempotent, and never affects a sibling.
300func TestSchedulerResetDiscard(t *testing.T) {
301 engine, scheduler, ids := testScheduler(t, 2)
302 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("one")}))
303 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("two")}))
304 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[1], Data: []byte("sibling")}))
305
306 siblingBytes := scheduler.QueuedBytes(ids[1])
307 require.Greater(t, scheduler.QueuedBytes(ids[0]), 0)
308
309 require.True(t, scheduler.Reset(ids[0]))
310 require.Zero(t, scheduler.QueuedFrames(ids[0]))
311 require.Zero(t, scheduler.QueuedBytes(ids[0]))
312 require.Equal(t, siblingBytes, scheduler.AggregateBytes(), "the sibling keeps its budget")
313 require.Equal(t, 1, scheduler.Active())
314
315 require.False(t, scheduler.Reset(ids[0]), "cancel is idempotent")
316 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: []byte("later")}), ErrSchedulerSettled)
317
318 envelope, ok := scheduler.Dequeue()
319 require.True(t, ok)
320 require.Equal(t, ids[1], envelope.Physical)
321 require.Zero(t, scheduler.AggregateBytes())
322
323 require.False(t, scheduler.Reset(99), "an unadmitted stream is a no-op")
324
325 openTestStream(t, engine, 3)
326 require.True(t, scheduler.Reset(3), "an admitted but idle stream settles")
327 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 3, Data: []byte("x")}), ErrSchedulerSettled)
328}
329
330// TestSchedulerAggregateCap proves the aggregate bound discards exactly the
331// offending stream and leaves its siblings' budget intact.
332func TestSchedulerAggregateCap(t *testing.T) {
333 t.Run("aggregate byte bound", func(t *testing.T) {
334 _, scheduler, ids := testScheduler(t, 3)
335 payload := make([]byte, 300)
336 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: payload}))
337 unit := scheduler.QueuedBytes(ids[0])
338 require.Positive(t, unit)
339
340 scheduler.maxAggregateBytes = unit * 3
341
342 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: payload}))
343 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[1], Data: payload}))
344 require.Equal(t, unit*3, scheduler.AggregateBytes())
345
346 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: ids[2], Data: payload}), ErrSchedulerFull)
347 require.Zero(t, scheduler.QueuedBytes(ids[2]))
348 require.Equal(t, unit*3, scheduler.AggregateBytes(), "siblings keep their budget")
349 require.Equal(t, 2, scheduler.Active())
350 })
351}
352
353// TestSchedulerConcurrentEnqueueDequeueReset exercises every scheduler method
354// from many goroutines so the race detector can prove the shared state is
355// serialized and the accounting stays consistent.
356func TestSchedulerConcurrentEnqueueDequeueReset(t *testing.T) {
357 const streams = 8
358 _, scheduler, ids := testScheduler(t, streams)
359
360 const (
361 producers = 4
362 perProducer = 300
363 consumers = 2
364 perConsumer = 600
365 resets = 50
366 )
367
368 var wg sync.WaitGroup
369 wg.Add(producers + consumers + 1)
370 for producer := 0; producer < producers; producer++ {
371 go func(producer int) {
372 defer wg.Done()
373 for i := 0; i < perProducer; i++ {
374 id := ids[(producer+i)%streams]
375 _ = scheduler.EnqueueClient(Data{Physical: id, Data: []byte{byte(producer), byte(i)}})
376 }
377 }(producer)
378 }
379 for consumer := 0; consumer < consumers; consumer++ {
380 go func() {
381 defer wg.Done()
382 for i := 0; i < perConsumer; i++ {
383 _, _ = scheduler.Dequeue()
384 }
385 }()
386 }
387 go func() {
388 defer wg.Done()
389 for i := 0; i < resets; i++ {
390 _ = scheduler.Reset(ids[i%streams])
391 }
392 }()
393 wg.Wait()
394
395 require.GreaterOrEqual(t, scheduler.AggregateBytes(), 0)
396 require.LessOrEqual(t, scheduler.AggregateBytes(), int(MaxMuxAggregateBytes))
397
398 total := 0
399 for _, id := range ids {
400 total += scheduler.QueuedBytes(id)
401 }
402 require.Equal(t, total, scheduler.AggregateBytes(), "aggregate accounting is exact")
403 require.Equal(t, countActive(scheduler, ids), scheduler.Active())
404}
405
406// BenchmarkSchedulerEnqueueDequeue measures the steady-state cost of encoding,
407// enqueuing, and dequeuing a single-stream data frame.
408func BenchmarkSchedulerEnqueueDequeue(b *testing.B) {
409 engine := NewStreamEngine()
410 if err := engine.Open(Open{Ref: testRef(1)}); err != nil {
411 b.Fatal(err)
412 }
413 if err := engine.Opened(Opened{Ref: testRef(1)}); err != nil {
414 b.Fatal(err)
415 }
416 scheduler := NewScheduler(engine)
417 payload := make([]byte, 256)
418
419 b.ReportAllocs()
420 b.ResetTimer()
421 for i := 0; i < b.N; i++ {
422 if err := scheduler.EnqueueClient(Data{Physical: 1, Data: payload}); err != nil {
423 b.Fatal(err)
424 }
425 if _, ok := scheduler.Dequeue(); !ok {
426 b.Fatal("scheduler empty")
427 }
428 }
429}
430
431// BenchmarkSchedulerEnqueueDequeueManyStreams measures the same path across a
432// full active ring.
433func BenchmarkSchedulerEnqueueDequeueManyStreams(b *testing.B) {
434 const streams = 32
435 engine := NewStreamEngine()
436 for i := 1; i <= streams; i++ {
437 ref := testRef(PhysicalStreamID(i))
438 if err := engine.Open(Open{Ref: ref}); err != nil {
439 b.Fatal(err)
440 }
441 if err := engine.Opened(Opened{Ref: ref}); err != nil {
442 b.Fatal(err)
443 }
444 }
445 scheduler := NewScheduler(engine)
446 payload := make([]byte, 256)
447
448 b.ReportAllocs()
449 b.ResetTimer()
450 for i := 0; i < b.N; i++ {
451 id := PhysicalStreamID(i%streams + 1)
452 if err := scheduler.EnqueueClient(Data{Physical: id, Data: payload}); err != nil {
453 b.Fatal(err)
454 }
455 if _, ok := scheduler.Dequeue(); !ok {
456 b.Fatal("scheduler empty")
457 }
458 }
459}
460
461// dequeueOrTimeout dequeues one frame, failing the test instead of hanging if
462// the DRR ring stalls on a frame larger than the quantum.
463func dequeueOrTimeout(t *testing.T, scheduler *Scheduler) (OutboundEnvelope, bool) {
464 t.Helper()
465 type result struct {
466 envelope OutboundEnvelope
467 ok bool
468 }
469 done := make(chan result, 1)
470 go func() {
471 envelope, ok := scheduler.Dequeue()
472 done <- result{envelope: envelope, ok: ok}
473 }()
474 select {
475 case r := <-done:
476 return r.envelope, r.ok
477 case <-time.After(5 * time.Second):
478 t.Fatal("Dequeue made no progress: the DRR ring stalled")
479 return OutboundEnvelope{}, false
480 }
481}
482
483// maxEnvOpen builds one Open whose environment fills the per-request bound, so
484// its serialized envelope is far larger than the 64 KiB quantum.
485func maxEnvOpen(t *testing.T, id PhysicalStreamID) Open {
486 t.Helper()
487 env := make([]string, ports.BrokerMaxEnvEntries)
488 for i := range env {
489 prefix := fmt.Sprintf("K%d=", i)
490 env[i] = prefix + strings.Repeat("v", ports.BrokerMaxEnvEntryBytes-len(prefix))
491 }
492 open := testOpen()
493 open.Ref = testRef(id)
494 open.Env = env
495 return open
496}
497
498// TestNewSchedulerWithCeilings proves the constructor validates the negotiated
499// advertisement, enforces the negotiated chunk and aggregate ceilings, and
500// copies the value so a caller cannot mutate live enforcement.
501func TestNewSchedulerWithCeilings(t *testing.T) {
502 t.Run("invalid advertisement refused", func(t *testing.T) {
503 bad := DefaultMuxCeilings()
504 bad.StreamChunkLimit = 0
505 scheduler, err := NewSchedulerWithCeilings(nil, bad)
506 require.ErrorIs(t, err, ErrInvalidCeilings)
507 require.Nil(t, scheduler)
508 _, err = NewSchedulerWithCeilings(nil, MuxCeilings{})
509 require.ErrorIs(t, err, ErrInvalidCeilings)
510 })
511
512 t.Run("negotiated chunk and aggregate limits enforced", func(t *testing.T) {
513 probeCeiling := MuxCeilings{
514 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
515 StreamChunkLimit: 8,
516 MaxStreams: 1,
517 MaxAggregateBytes: MinMuxAggregateBytes,
518 }
519 probe, err := NewSchedulerWithCeilings(nil, probeCeiling)
520 require.NoError(t, err)
521 require.NoError(t, probe.EnqueueClient(Data{Physical: 1, Data: make([]byte, 8)}))
522 unit := probe.QueuedBytes(1)
523 require.Positive(t, unit)
524
525 // The aggregate floor admits exactly one maximum chunk plus envelope
526 // overhead.
527 floorCeiling := probeCeiling
528 floorCeiling.StreamChunkLimit = MaxMuxChunkBytes
529 floor, err := NewSchedulerWithCeilings(nil, floorCeiling)
530 require.NoError(t, err)
531 require.NoError(t, floor.EnqueueClient(Data{Physical: 1, Data: make([]byte, MaxMuxChunkBytes)}))
532 require.Equal(t, 1, floor.Active())
533
534 // A scheduler tightened below the negotiated aggregate discards exactly
535 // the offending stream and frees its budget.
536 scheduler, err := NewSchedulerWithCeilings(nil, probeCeiling)
537 require.NoError(t, err)
538 scheduler.maxAggregateBytes = unit * 2
539 // The negotiated chunk ceiling is refused statelessly.
540 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: make([]byte, 9)}), ErrTooLarge)
541 require.Zero(t, scheduler.AggregateBytes())
542 require.NoError(t, scheduler.EnqueueClient(Data{Physical: 1, Data: make([]byte, 8)}))
543 require.NoError(t, scheduler.EnqueueClient(Data{Physical: 1, Data: make([]byte, 8)}))
544 require.Equal(t, unit*2, scheduler.AggregateBytes())
545 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: make([]byte, 8)}), ErrSchedulerFull)
546 require.Zero(t, scheduler.AggregateBytes())
547 })
548}
549
550// TestSchedulerTerminalBypass proves exactly one terminal Close or Reset is
551// accepted for a known identity the engine has already settled, that the
552// terminal frame bypasses normal admission, that a duplicate terminal is
553// refused, and that data stays refused.
554func TestSchedulerTerminalBypass(t *testing.T) {
555 t.Run("settled identity signals once", func(t *testing.T) {
556 engine := testEngine()
557 scheduler := NewScheduler(engine)
558 openTestStream(t, engine, 1)
559 _, err := engine.Reset(Reset{Physical: 1})
560 require.NoError(t, err)
561
562 // Data for the settled, unadmitted identity stays refused.
563 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: []byte("x")}), ErrSchedulerUnadmitted)
564 // Exactly one terminal frame is accepted so the peer can be signaled.
565 require.NoError(t, scheduler.EnqueueClient(Reset{Physical: 1}))
566 envelope, ok := dequeueOrTimeout(t, scheduler)
567 require.True(t, ok)
568 require.Equal(t, PhysicalStreamID(1), envelope.Physical)
569 require.Equal(t, FrameControl, envelope.Class)
570 _, isReset := envelope.Client.(Reset)
571 require.True(t, isReset)
572
573 // The duplicate terminal, and every later data frame, is refused.
574 require.ErrorIs(t, scheduler.EnqueueClient(Reset{Physical: 1}), ErrSchedulerSettled)
575 require.ErrorIs(t, scheduler.EnqueueClient(Close{Physical: 1}), ErrSchedulerSettled)
576 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: []byte("y")}), ErrSchedulerSettled)
577 // An identity the engine never admitted is still refused.
578 require.ErrorIs(t, scheduler.EnqueueClient(Close{Physical: 9}), ErrSchedulerUnadmitted)
579 })
580
581 t.Run("queue overflow signals the peer", func(t *testing.T) {
582 engine := testEngine()
583 scheduler := NewScheduler(engine)
584 openTestStream(t, engine, 1)
585 chunk, full := floodWindow(DefaultMuxCeilings())
586 for i := 0; i < full; i++ {
587 _, err := engine.Data(Data{Physical: 1, Data: chunk})
588 require.NoError(t, err)
589 }
590 _, err := engine.Data(Data{Physical: 1, Data: chunk})
591 require.ErrorIs(t, err, ErrStreamQueueFull)
592 require.Equal(t, StreamTerminal, mustStatus(t, engine, 1).State)
593
594 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 1, Data: []byte("x")}), ErrSchedulerUnadmitted)
595 require.NoError(t, scheduler.EnqueueClient(Reset{Physical: 1}))
596 _, ok := dequeueOrTimeout(t, scheduler)
597 require.True(t, ok)
598 require.ErrorIs(t, scheduler.EnqueueClient(Reset{Physical: 1}), ErrSchedulerSettled)
599 })
600
601 t.Run("nil gate trusts one terminal", func(t *testing.T) {
602 scheduler := NewScheduler(nil)
603 require.NoError(t, scheduler.EnqueueClient(Close{Physical: 7}))
604 require.ErrorIs(t, scheduler.EnqueueClient(Close{Physical: 7}), ErrSchedulerSettled)
605 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: 7, Data: []byte("x")}), ErrSchedulerSettled)
606 })
607}
608
609// TestSchedulerLargeFrameProgress is the regression for the stalled DRR ring:
610// the maximum 64 KiB chunk and a maximal-environment Open both make progress
611// without spinning.
612func TestSchedulerLargeFrameProgress(t *testing.T) {
613 t.Run("maximum chunk fits the quantum", func(t *testing.T) {
614 _, scheduler, ids := testScheduler(t, 1)
615 chunk := make([]byte, MaxMuxChunkBytes)
616 require.NoError(t, scheduler.EnqueueClient(Data{Physical: ids[0], Data: chunk}))
617 envelope, ok := dequeueOrTimeout(t, scheduler)
618 require.True(t, ok)
619 require.Equal(t, ids[0], envelope.Physical)
620 require.False(t, scheduler.Pending())
621 })
622
623 t.Run("maximum environment Open exceeds the quantum", func(t *testing.T) {
624 scheduler := NewScheduler(nil)
625 const streams = 5
626 ids := make([]PhysicalStreamID, streams)
627 for i := range ids {
628 ids[i] = PhysicalStreamID(i + 1)
629 require.NoError(t, scheduler.EnqueueClient(maxEnvOpen(t, ids[i])))
630 }
631 seen := make(map[PhysicalStreamID]bool, streams)
632 for i := 0; i < streams; i++ {
633 envelope, ok := dequeueOrTimeout(t, scheduler)
634 require.True(t, ok, "dequeue %d", i)
635 open, isOpen := envelope.Client.(Open)
636 require.True(t, isOpen, "dequeue %d carries %T", i, envelope.Client)
637 require.Len(t, open.Env, ports.BrokerMaxEnvEntries)
638 seen[envelope.Physical] = true
639 }
640 require.Len(t, seen, streams)
641 require.False(t, scheduler.Pending())
642 })
643}
644
645// TestSchedulerBoundedRetiredRecords proves a long-lived scheduler keeps a
646// bounded record table: the oldest settled records are evicted, while the
647// newest identity still refuses a duplicate terminal.
648func TestSchedulerBoundedRetiredRecords(t *testing.T) {
649 scheduler := NewScheduler(nil)
650 const churn = 4000
651 for i := 1; i <= churn; i++ {
652 id := PhysicalStreamID(i)
653 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte{byte(i)}}))
654 require.True(t, scheduler.Reset(id))
655 }
656 require.LessOrEqual(t, len(scheduler.streams), MaxRetiredSchedulerRecords)
657 require.Len(t, scheduler.retired, MaxRetiredSchedulerRecords)
658 require.Zero(t, scheduler.Active())
659 require.Zero(t, scheduler.AggregateBytes())
660
661 newest := PhysicalStreamID(churn)
662 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: newest, Data: []byte("x")}), ErrSchedulerSettled)
663}
664
665// FuzzSchedulerEngineSmoke drives a scheduler over an engine gate with
666// arbitrary operation bytes and proves the pair never panics, keeps its
667// accounting exact, and stays bounded.
668func FuzzSchedulerEngineSmoke(f *testing.F) {
669 f.Add([]byte{0, 1, 2, 3, 4, 5, 6, 7})
670 f.Add([]byte{1, 1, 1, 1, 1, 1, 1, 1})
671 f.Fuzz(func(t *testing.T, ops []byte) {
672 engine := NewStreamEngine()
673 scheduler := NewScheduler(engine)
674 limit := len(ops)
675 if limit > 256 {
676 limit = 256
677 }
678 for i := 0; i+1 < limit; i += 2 {
679 id := PhysicalStreamID(ops[i]%8 + 1)
680 switch ops[i+1] % 6 {
681 case 0:
682 _ = engine.Open(Open{Ref: testRef(id)})
683 case 1:
684 _ = engine.Opened(Opened{Ref: testRef(id)})
685 case 2:
686 _ = scheduler.EnqueueClient(Data{Physical: id, Data: []byte{ops[i]}})
687 case 3:
688 _ = scheduler.EnqueueClient(Close{Physical: id})
689 case 4:
690 _, _ = engine.Data(Data{Physical: id, Data: []byte{ops[i]}})
691 case 5:
692 _ = scheduler.Reset(id)
693 }
694 }
695 _, _ = scheduler.Dequeue()
696
697 total := 0
698 for id := PhysicalStreamID(1); id <= 8; id++ {
699 total += scheduler.QueuedBytes(id)
700 }
701 require.Equal(t, total, scheduler.AggregateBytes())
702 require.GreaterOrEqual(t, scheduler.AggregateBytes(), 0)
703 require.LessOrEqual(t, len(scheduler.streams), int(MaxMuxStreams+MaxRetiredSchedulerRecords))
704 })
705}
706
707// TestSchedulerOverflowAllowsOneTerminal proves an overflowed stream discards
708// its queue and refuses data, yet still accepts exactly one terminal Close or
709// Reset, sealing only after that terminal.
710func TestSchedulerOverflowAllowsOneTerminal(t *testing.T) {
711 terminals := []struct {
712 name string
713 enqueue func(*Scheduler, PhysicalStreamID) error
714 }{
715 {"reset", func(s *Scheduler, id PhysicalStreamID) error { return s.EnqueueClient(Reset{Physical: id}) }},
716 {"close", func(s *Scheduler, id PhysicalStreamID) error { return s.EnqueueClient(Close{Physical: id}) }},
717 }
718 for _, terminal := range terminals {
719 t.Run(terminal.name, func(t *testing.T) {
720 _, scheduler, ids := testScheduler(t, 1)
721 id := ids[0]
722 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("a")}))
723 scheduler.maxAggregateBytes = 2 * scheduler.AggregateBytes()
724 require.NoError(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("b")}))
725 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("c")}), ErrSchedulerFull)
726 require.Zero(t, scheduler.QueuedFrames(id))
727 require.Zero(t, scheduler.AggregateBytes())
728 require.False(t, scheduler.Pending())
729
730 // The record is overflowed but not sealed: data stays refused with
731 // the overflow sentinel.
732 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("d")}), ErrSchedulerFull)
733
734 require.NoError(t, terminal.enqueue(scheduler, id))
735 require.Equal(t, 1, scheduler.QueuedFrames(id))
736 envelope, ok := scheduler.Dequeue()
737 require.True(t, ok)
738 require.Equal(t, id, envelope.Physical)
739 require.Equal(t, FrameControl, envelope.Class)
740
741 // The one terminal sealed the record: everything afterwards,
742 // including a duplicate terminal, is refused.
743 require.ErrorIs(t, scheduler.EnqueueClient(Data{Physical: id, Data: []byte("e")}), ErrSchedulerSettled)
744 require.ErrorIs(t, terminal.enqueue(scheduler, id), ErrSchedulerSettled)
745 require.False(t, scheduler.Reset(id))
746 })
747 }
748}
749
750// TestSchedulerOverflowRecordsBounded proves overflow records that never take
751// their terminal stay bounded: they are evicted oldest-first past
752// MaxRetiredSchedulerRecords.
753func TestSchedulerOverflowRecordsBounded(t *testing.T) {
754 scheduler := NewScheduler(nil)
755 const churn = MaxRetiredSchedulerRecords * 4
756 for i := 1; i <= churn; i++ {
757 id := PhysicalStreamID(i)
758 overflowOne(t, scheduler, id)
759 }
760 require.LessOrEqual(t, len(scheduler.streams), MaxRetiredSchedulerRecords)
761 require.Zero(t, scheduler.AggregateBytes())
762 require.False(t, scheduler.Pending())
763}
764
765// TestSchedulerOverflowTerminalReclaimsRetiredRecords is the regression for a
766// record that is listed retired while empty, later accepts its one terminal
767// Reset while that terminal stays queued, and is then pushed past the
768// retention bound by later streams: eviction must not permanently orphan the
769// record. Every terminal stays deliverable, and draining them returns the
770// stream table to the configured bound.
771func TestSchedulerOverflowTerminalReclaimsRetiredRecords(t *testing.T) {
772 scheduler := NewScheduler(nil)
773 const churn = MaxRetiredSchedulerRecords + 64
774
775 for i := 1; i <= churn; i++ {
776 id := PhysicalStreamID(i)
777 // Overflow the queue so the record is retired while still empty.
778 overflowOne(t, scheduler, id)
779 // The one terminal the overflow leaves open lands while the record sits
780 // in the retired list; later streams force it past the bound.
781 require.NoError(t, scheduler.EnqueueClient(Reset{Physical: id}))
782 }
783
784 delivered := make(map[PhysicalStreamID]int, churn)
785 for scheduler.Pending() {
786 envelope, ok := scheduler.Dequeue()
787 require.True(t, ok)
788 _, isReset := envelope.Client.(Reset)
789 require.True(t, isReset, "terminal frame must stay deliverable, got %T", envelope.Client)
790 delivered[envelope.Physical]++
791 }
792 require.Len(t, delivered, churn, "every overflowed stream signals its peer exactly once")
793 for id, count := range delivered {
794 require.Equal(t, 1, count, "stream %d delivered %d terminals", id, count)
795 }
796
797 require.Zero(t, scheduler.AggregateBytes())
798 require.Zero(t, scheduler.Active())
799 require.LessOrEqual(t, len(scheduler.streams), MaxRetiredSchedulerRecords,
800 "drained overflow records must not be orphaned past the retention bound")
801 require.LessOrEqual(t, len(scheduler.retired), MaxRetiredSchedulerRecords)
802
803 // The newest identity still refuses a duplicate terminal, so the retained
804 // record set is live, not merely truncated.
805 require.ErrorIs(t, scheduler.EnqueueClient(Reset{Physical: PhysicalStreamID(churn)}), ErrSchedulerSettled)
806}
807
808// TestSchedulerRefuseAdmitsFreshRefusalOnce proves the fresh-refusal exception:
809// a Refused for an identity the engine never admitted is scheduled once and
810// seals the record, so a duplicate and every later frame are refused; an
811// unencodable refusal creates no record.
812func TestSchedulerRefuseAdmitsFreshRefusalOnce(t *testing.T) {
813 s := NewScheduler(nil)
814 detail := testErrorDetail()
815 detail.AdmissionCode = 1
816 refused := Refused{Ref: testRef(1), Error: detail}
817
818 require.NoError(t, s.Refuse(refused))
819 envelope, ok := s.Dequeue()
820 require.True(t, ok)
821 require.Equal(t, PhysicalStreamID(1), envelope.Physical)
822 require.Equal(t, DirectionServer, envelope.Direction)
823 require.Equal(t, FrameControl, envelope.Class)
824 require.Equal(t, refused, envelope.Server)
825
826 // The fresh refusal seals the identity even though the engine never
827 // admitted the stream: a duplicate and every later frame are refused.
828 require.ErrorIs(t, s.Refuse(refused), ErrSchedulerSettled)
829 require.ErrorIs(t, s.EnqueueServer(Refused{Ref: testRef(1), Error: detail}), ErrSchedulerSettled)
830 require.ErrorIs(t, s.EnqueueServer(Data{Physical: 1, Data: []byte("x")}), ErrSchedulerSettled)
831 _, ok = s.Dequeue()
832 require.False(t, ok)
833
834 // A refusal whose reference cannot be encoded is rejected without creating
835 // a record.
836 require.ErrorIs(t, s.Refuse(Refused{}), ErrInvalidMessage)
837 require.Zero(t, s.QueuedFrames(0))
838
839 var nilScheduler *Scheduler
840 require.ErrorIs(t, nilScheduler.Refuse(refused), ErrInvalidMessage)
841}