1package dgram
2
3import (
4 "bytes"
5 "errors"
6 "net"
7 "sync"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 pdgram "github.com/bnema/vev/pkg/dgram"
15)
16
17type testAddr string
18
19func (a testAddr) Network() string { return "mem" }
20func (a testAddr) String() string { return string(a) }
21
22type packet struct {
23 b []byte
24 addr net.Addr
25}
26
27// packetMeta is the deterministic simulator's view of a datagram before it is
28// accepted by the link. It deliberately contains only link-level information.
29type packetMeta struct {
30 From, To net.Addr
31 Bytes int
32 At time.Time
33 QueueBytes int
34}
35
36// packetPolicy makes loss and constrained links explicit in transport tests.
37// A zero value delivers immediately, preserving newPair's historical behavior.
38type packetPolicy struct {
39 Drop func(packetMeta) bool
40 Delay time.Duration
41 MaxQueueBytes int
42 BytesPerSecond int
43}
44
45type scheduledPacket struct {
46 packet
47 due time.Time
48}
49
50// simulatedLink has a bounded byte queue. Tests advance its clock explicitly;
51// it never starts a goroutine or relies on wall-clock time.
52type simulatedLink struct {
53 mu sync.Mutex
54 clock interface{ Now() time.Time }
55 policy packetPolicy
56 queue []scheduledPacket
57 queueBytes int
58 next time.Time
59 dropped int
60 maxDelay time.Duration
61}
62
63func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {
64 return &simulatedLink{clock: clock, policy: policy}
65}
66
67func (l *simulatedLink) enqueue(p *fakePC, peer *fakePC, b []byte) bool {
68 l.mu.Lock()
69 defer l.mu.Unlock()
70 meta := packetMeta{From: p.addr, To: peer.addr, Bytes: len(b), At: l.clock.Now(), QueueBytes: l.queueBytes}
71 if l.policy.Drop != nil && l.policy.Drop(meta) {
72 return false
73 }
74 if l.policy.Delay == 0 && l.policy.BytesPerSecond == 0 {
75 return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})
76 }
77 if l.policy.MaxQueueBytes <= 0 || l.queueBytes+len(b) > l.policy.MaxQueueBytes {
78 l.dropped++
79 return false
80 }
81 due := meta.At.Add(l.policy.Delay)
82 if l.policy.BytesPerSecond > 0 {
83 if due.Before(l.next) {
84 due = l.next
85 }
86 ns := int64(time.Second) * int64(len(b)) / int64(l.policy.BytesPerSecond)
87 if ns == 0 {
88 ns = 1
89 }
90 l.next = due.Add(time.Duration(ns))
91 }
92 l.maxDelay = max(l.maxDelay, due.Sub(meta.At))
93 l.queue = append(l.queue, scheduledPacket{packet: packet{b: append([]byte(nil), b...), addr: p.addr}, due: due})
94 l.queueBytes += len(b)
95 return true
96}
97
98func (l *simulatedLink) flush(peer *fakePC) int {
99 l.mu.Lock()
100 defer l.mu.Unlock()
101 now := l.clock.Now()
102 delivered := 0
103 for len(l.queue) > 0 && !l.queue[0].due.After(now) {
104 q := l.queue[0]
105 l.queue = l.queue[1:]
106 l.queueBytes -= len(q.b)
107 if l.deliverLocked(peer, q.packet) {
108 delivered++
109 }
110 }
111 return delivered
112}
113
114func (l *simulatedLink) queuedBytes() int {
115 l.mu.Lock()
116 defer l.mu.Unlock()
117 return l.queueBytes
118}
119
120func (l *simulatedLink) droppedPackets() int {
121 l.mu.Lock()
122 defer l.mu.Unlock()
123 return l.dropped
124}
125
126func (l *simulatedLink) maximumQueueDelay() time.Duration {
127 l.mu.Lock()
128 defer l.mu.Unlock()
129 return l.maxDelay
130}
131
132func (l *simulatedLink) deliverLocked(peer *fakePC, q packet) bool {
133 peer.mu.Lock()
134 defer peer.mu.Unlock()
135 if peer.closed {
136 return false
137 }
138 select {
139 case peer.in <- q:
140 return true
141 default:
142 // The endpoint queue is intentionally bounded just like the link queue.
143 return false
144 }
145}
146
147type fakePC struct {
148 addr net.Addr
149 mu sync.Mutex
150 closed bool
151 in chan packet
152 read chan struct{}
153 peers map[string]*fakePC
154 drop func([]byte, net.Addr) bool
155 afterWrite func()
156 link *simulatedLink
157}
158
159func newPair() (*fakePC, *fakePC) { return newPairWithCapacity(nil, 100) }
160
161func newFloodPair(link *simulatedLink) (*fakePC, *fakePC) {
162 return newPairWithCapacity(link, floodRecordCount*floodRecordMTUs*2)
163}
164
165func newPairWithCapacity(link *simulatedLink, queuePackets int) (*fakePC, *fakePC) {
166 a := &fakePC{addr: testAddr("a"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
167 b := &fakePC{addr: testAddr("b"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
168 a.peers["b"] = b
169 b.peers["a"] = a
170 return a, b
171}
172func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
173 q, ok := <-p.in
174 if !ok {
175 return 0, nil, errors.New("closed")
176 }
177 copy(b, q.b)
178 select {
179 case p.read <- struct{}{}:
180 default:
181 }
182 return len(q.b), q.addr, nil
183}
184func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
185 p.mu.Lock()
186 closed := p.closed
187 afterWrite := p.afterWrite
188 p.mu.Unlock()
189 if closed {
190 return 0, errors.New("closed")
191 }
192 if p.drop != nil && p.drop(b, addr) {
193 return len(b), nil
194 }
195 peer := p.peers[addr.String()]
196 if peer == nil {
197 return 0, errors.New("unknown peer")
198 }
199 if p.link != nil {
200 p.link.enqueue(p, peer, b)
201 } else {
202 peer.mu.Lock()
203 if peer.closed {
204 peer.mu.Unlock()
205 return 0, errors.New("peer closed")
206 }
207 select {
208 case peer.in <- packet{append([]byte(nil), b...), p.addr}:
209 default:
210 }
211 peer.mu.Unlock()
212 }
213 if afterWrite != nil {
214 afterWrite()
215 }
216 return len(b), nil
217}
218func (p *fakePC) replaceAfterWrite(afterWrite func()) {
219 p.mu.Lock()
220 defer p.mu.Unlock()
221 p.afterWrite = afterWrite
222}
223
224func (p *fakePC) addAfterWrite(afterWrite func()) {
225 p.mu.Lock()
226 defer p.mu.Unlock()
227 previous := p.afterWrite
228 p.afterWrite = func() {
229 if previous != nil {
230 previous()
231 }
232 afterWrite()
233 }
234}
235
236func (p *fakePC) Close() error {
237 p.mu.Lock()
238 defer p.mu.Unlock()
239 if !p.closed {
240 p.closed = true
241 close(p.in)
242 }
243 return nil
244}
245func (p *fakePC) LocalAddr() net.Addr { return p.addr }
246func (p *fakePC) SetDeadline(time.Time) error { return nil }
247func (p *fakePC) SetReadDeadline(time.Time) error { return nil }
248func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
249
250func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
251
252const fixtureWaitTimeout = time.Second
253
254const (
255 floodRecordCount = 32
256 floodRecordMTUs = 24
257)
258
259// awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a
260// hang guard only; tests advance behavior with the manual clock and barriers.
261func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {
262 t.Helper()
263 select {
264 case <-ch:
265 case <-time.After(fixtureWaitTimeout):
266 t.Fatalf("timed out waiting for %s", context)
267 }
268}
269
270func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
271 t.Helper()
272 select {
273 case result := <-ch:
274 return result
275 case <-time.After(fixtureWaitTimeout):
276 t.Fatalf("timed out waiting for %s", context)
277 var zero T
278 return zero
279 }
280}
281
282func flushAndAwait(t *testing.T, link *simulatedLink, peer *fakePC, processed <-chan struct{}) int {
283 t.Helper()
284 delivered := link.flush(peer)
285 for range delivered {
286 awaitSignal(t, processed, "flushed packet processing")
287 }
288 return delivered
289}
290
291// TestTransportFloodClassification drives authenticated datagrams through real
292// transports. The metrics are transport state, not simulator bookkeeping.
293func TestTransportFloodClassification(t *testing.T) {
294 t.Run("one real fragment per large record loss contacts without completing", func(t *testing.T) {
295 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
296 aPC, bPC := newFloodPair(nil)
297 inspect, err := pdgram.NewCodec(key())
298 if err != nil {
299 t.Fatal(err)
300 }
301 dropped := map[uint64]bool{}
302 aPC.drop = func(pkt []byte, _ net.Addr) bool {
303 _, raw, err := inspect.Open(pkt, 1, nil, nil)
304 if err != nil {
305 return false
306 }
307 frag, err := pdgram.UnmarshalFragment(raw)
308 if err != nil || frag.Index != frag.Count-1 || dropped[frag.Seq] {
309 return false
310 }
311 dropped[frag.Seq] = true
312 return true
313 }
314 const mtu = 128
315 a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
316 defer closeFloodTransports(a, b)
317 contacted := make(chan struct{}, 1)
318 b.mu.Lock()
319 b.afterAuthenticatedPacket = func() {
320 select {
321 case contacted <- struct{}{}:
322 default:
323 }
324 }
325 b.mu.Unlock()
326 start := clk.Now()
327 clk.advance(time.Nanosecond)
328 sendFloodOutputs(t, a, floodRecordCount, mtu)
329 awaitSignal(t, contacted, "first authenticated fragment")
330 state := floodState(a, b, nil)
331 if len(dropped) != floodRecordCount {
332 t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)
333 }
334 if state.packetAge > initialPacingRTT || state.recordAge <= 0 || state.ackAge <= 0 {
335 t.Fatalf("flood state = %+v, want fresh packet contact without record or ACK progress", state)
336 }
337 if !clk.Now().After(start) {
338 t.Fatal("real authenticated fragments did not update contact")
339 }
340 })
341
342 t.Run("bandwidth_queue", func(t *testing.T) {
343 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
344 const (
345 mtu = 128
346 oneWayDelay = 50 * time.Millisecond
347 bandwidth = 16 * mtu
348 queueLimit = 4 * mtu
349 outputCount = 3
350 outputDataBytes = 6 * mtu
351 )
352 var attempts, deterministicDrops atomic.Int64
353 forward := newSimulatedLink(clk, packetPolicy{
354 Delay: oneWayDelay,
355 MaxQueueBytes: queueLimit,
356 BytesPerSecond: bandwidth,
357 Drop: func(packetMeta) bool {
358 attempt := attempts.Add(1)
359 if attempt%20 != 0 {
360 return false
361 }
362 deterministicDrops.Add(1)
363 return true
364 },
365 })
366 reverse := newSimulatedLink(clk, packetPolicy{
367 Delay: oneWayDelay,
368 MaxQueueBytes: queueLimit,
369 BytesPerSecond: bandwidth,
370 })
371 aPC, bPC := newFloodPair(nil)
372 aPC.link = forward
373 bPC.link = reverse
374 const probeAfter = defaultProbe
375 opts := Options{
376 Clock: clk, MTU: mtu, ResendAfter: 200 * time.Millisecond, MaxResendAfter: 800 * time.Millisecond,
377 Heartbeat: 500 * time.Millisecond, DegradedAfter: defaultDegraded,
378 ProbeAfter: probeAfter, OfflineAfter: defaultOffline, DeadAfter: defaultDead,
379 }
380 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
381 if err != nil {
382 t.Fatal(err)
383 }
384 b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
385 if err != nil {
386 _ = a.Close()
387 t.Fatal(err)
388 }
389 defer closeFloodTransports(a, b)
390 waitForManualTimers(t, clk, 6)
391 for range 6 {
392 awaitSignal(t, clk.timerCreated, "initial transport timer creation")
393 }
394 var contacted atomic.Bool
395 // All fixture hooks are lossy notifications: transport read/control loops
396 // must never wait for the test goroutine to be scheduled.
397 processed := make(chan struct{}, 256)
398 primerWritten := make(chan struct{}, 1)
399 primerAcknowledged := make(chan struct{}, 1)
400 ackWakeAccepted := make(chan struct{}, 128)
401 ackScheduled := make(chan struct{}, 128)
402 ackDispatched := make(chan struct{}, 128)
403 ackWritten := make(chan struct{}, 128)
404 aPC.replaceAfterWrite(func() {
405 select {
406 case primerWritten <- struct{}{}:
407 default:
408 }
409 })
410 bPC.replaceAfterWrite(func() {
411 select {
412 case primerAcknowledged <- struct{}{}:
413 default:
414 }
415 select {
416 case ackWritten <- struct{}{}:
417 default:
418 }
419 })
420 a.mu.Lock()
421 a.afterPacketProcessed = func() {
422 select {
423 case processed <- struct{}{}:
424 default:
425 }
426 }
427 a.mu.Unlock()
428 b.mu.Lock()
429 b.afterAuthenticatedPacket = func() { contacted.Store(true) }
430 b.afterPacketProcessed = func() {
431 select {
432 case processed <- struct{}{}:
433 default:
434 }
435 }
436 b.afterACKWakeAccepted = func() {
437 select {
438 case ackWakeAccepted <- struct{}{}:
439 default:
440 }
441 }
442 b.afterACKScheduled = func() {
443 select {
444 case ackScheduled <- struct{}{}:
445 default:
446 }
447 }
448 b.afterACKDispatched = func() {
449 select {
450 case ackDispatched <- struct{}{}:
451 default:
452 }
453 }
454 b.mu.Unlock()
455 primer := mustMarshalOutput(ports.Output{Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("primer")})
456 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {
457 t.Fatal(err)
458 }
459 awaitSignal(t, primerWritten, "primer write")
460 primerDrained := false
461 awaitingPrimerACK := false
462 for range 40 {
463 clk.advance(25 * time.Millisecond)
464 if awaitingPrimerACK {
465 awaitSignal(t, primerAcknowledged, "primer ACK write")
466 awaitingPrimerACK = false
467 }
468 if flushAndAwait(t, forward, bPC, processed) > 0 {
469 awaitSignal(t, ackWakeAccepted, "primer accepted ACK wake")
470 awaitSignal(t, ackScheduled, "primer ACK scheduling")
471 awaitingPrimerACK = true
472 }
473 flushAndAwait(t, reverse, aPC, processed)
474 a.mu.Lock()
475 primerDrained = len(a.pending) == 0
476 a.mu.Unlock()
477 if primerDrained {
478 break
479 }
480 }
481 if !primerDrained {
482 t.Fatal("clean RTT primer did not drain")
483 }
484 a.mu.Lock()
485 initialCwnd := a.congestion.cwndBytes
486 a.mu.Unlock()
487 timerWork := make(chan struct{}, 128)
488 dataPaceStarted := make(chan struct{}, 128)
489 dataPaceWait := make(chan time.Time, 128)
490 dataJobDone := make(chan struct{}, 128)
491 dataWritten := make(chan struct{}, 128)
492 notifyTimerWork := func() {
493 select {
494 case timerWork <- struct{}{}:
495 default:
496 }
497 }
498 for _, tr := range []*Transport{a, b} {
499 tr.mu.Lock()
500 tr.afterRetransmitTimer = notifyTimerWork
501 tr.afterHealthTimer = notifyTimerWork
502 tr.afterHeartbeatTimer = notifyTimerWork
503 tr.mu.Unlock()
504 }
505 a.mu.Lock()
506 a.beforeDataPace = func() {
507 select {
508 case dataPaceStarted <- struct{}{}:
509 default:
510 }
511 }
512 a.afterDataPaceWait = func(deadline time.Time) {
513 select {
514 case dataPaceWait <- deadline:
515 default:
516 }
517 }
518 a.afterDataJob = func() {
519 select {
520 case dataJobDone <- struct{}{}:
521 default:
522 }
523 }
524 a.mu.Unlock()
525 aPC.addAfterWrite(func() {
526 select {
527 case dataWritten <- struct{}{}:
528 default:
529 }
530 })
531 for state := 1; state < outputCount; state++ {
532 payload := mustMarshalOutput(ports.Output{
533 Epoch: 1,
534 Base: uint64(state),
535 New: uint64(state + 1),
536 Size: domain.Size{Cols: 1, Rows: 1},
537 Data: bytes.Repeat([]byte{byte(state + 1)}, outputDataBytes),
538 })
539 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: payload}); err != nil {
540 t.Fatal(err)
541 }
542 }
543 awaitSignal(t, dataPaceStarted, "output pacing")
544 awaitSignal(t, dataWritten, "first paced output write")
545 nextPace := awaitResult(t, dataPaceWait, "data pacing deadline")
546
547 var maxContactAge time.Duration
548 awaitingACKWrite := false
549 lossReducedCwnd := false
550 latestState := uint64(0)
551 probed := false
552 var probeAt time.Duration
553 probeWho := ""
554 for range 2000 {
555 clk.advance(25 * time.Millisecond)
556 elapsed := clk.Now().Sub(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
557 if elapsed%opts.ResendAfter == 0 {
558 // Both peers complete retransmit and health work before this
559 // advance may be followed by another manual-clock step.
560 for range 4 {
561 awaitSignal(t, timerWork, "retransmit or health timer work")
562 }
563 }
564 if elapsed%opts.Heartbeat == 0 {
565 for range 2 {
566 awaitSignal(t, timerWork, "heartbeat timer work")
567 }
568 }
569 // A manual-clock advance is not a scheduler yield. When it fires the
570 // data pacer, wait until the write reaches its next deterministic
571 // deadline (or its job completes) before advancing again.
572 if !nextPace.IsZero() && !nextPace.After(clk.Now()) {
573 awaitSignal(t, dataWritten, "paced data write")
574 select {
575 case nextPace = <-dataPaceWait:
576 case <-dataJobDone:
577 // A completed job may be awaiting an ACK or retransmit timer;
578 // neither has a data-pacer deadline to await yet.
579 nextPace = time.Time{}
580 }
581 }
582 if awaitingACKWrite {
583 awaitSignal(t, ackDispatched, "cumulative ACK dispatch")
584 awaitSignal(t, ackWritten, "cumulative ACK write")
585 awaitingACKWrite = false
586 }
587 if flushAndAwait(t, forward, bPC, processed) > 0 {
588 select {
589 case <-ackWakeAccepted:
590 awaitSignal(t, ackScheduled, "cumulative ACK scheduling")
591 awaitingACKWrite = true
592 default:
593 }
594 }
595 flushAndAwait(t, reverse, aPC, processed)
596 a.mu.Lock()
597 queuedOutput := len(a.outputQueue) > 0
598 a.mu.Unlock()
599 if queuedOutput {
600 awaitSignal(t, dataPaceStarted, "output pacing")
601 }
602
603 if forward.queuedBytes() > queueLimit || reverse.queuedBytes() > queueLimit {
604 t.Fatalf("queue exceeded bound: forward=%d reverse=%d limit=%d", forward.queuedBytes(), reverse.queuedBytes(), queueLimit)
605 }
606 if contacted.Load() {
607 b.mu.Lock()
608 age := clk.Now().Sub(b.lastAuthenticatedPacket)
609 b.mu.Unlock()
610 maxContactAge = max(maxContactAge, age)
611 }
612 for i, tr := range []*Transport{a, b} {
613 if tr.LinkState() == ports.LinkStateProbing {
614 if !probed {
615 probeAt = clk.Now().Sub(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
616 probeWho = []string{"sender", "receiver"}[i]
617 }
618 probed = true
619 }
620 select {
621 case event := <-tr.LinkEvents():
622 if event.State == ports.LinkStateProbing && !probed {
623 probeAt = event.At.Sub(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
624 probeWho = []string{"sender", "receiver"}[i]
625 }
626 probed = probed || event.State == ports.LinkStateProbing
627 default:
628 }
629 }
630 for {
631 select {
632 case frame := <-b.in:
633 output, err := ports.UnmarshalOutput(frame.Payload)
634 if err != nil {
635 t.Fatal(err)
636 }
637 latestState = max(latestState, output.New)
638 default:
639 goto drainedOutput
640 }
641 }
642 drainedOutput:
643 a.mu.Lock()
644 pending := len(a.pending)
645 lossReducedCwnd = lossReducedCwnd || a.congestion.cwndBytes < initialCwnd
646 a.mu.Unlock()
647 if pending == 0 && latestState == outputCount && forward.queuedBytes() == 0 && reverse.queuedBytes() == 0 {
648 break
649 }
650 }
651
652 a.mu.Lock()
653 pending := len(a.pending)
654 srtt := a.srtt
655 a.mu.Unlock()
656 if got, want := deterministicDrops.Load(), attempts.Load()/20; got == 0 || got != want {
657 t.Fatalf("deterministic loss=%d/%d, want exactly one in twenty attempts (%d)", got, attempts.Load(), want)
658 }
659 if forward.droppedPackets() == 0 {
660 t.Fatal("bounded bandwidth queue never overflowed")
661 }
662 if delay := max(forward.maximumQueueDelay(), reverse.maximumQueueDelay()); delay < 2*oneWayDelay || delay >= probeAfter {
663 t.Fatalf("maximum queue delay=%v, want in [%v,%v)", delay, 2*oneWayDelay, probeAfter)
664 }
665 if srtt < 2*oneWayDelay || srtt >= probeAfter {
666 t.Fatalf("smoothed RTT=%v, want in [%v,%v)", srtt, 2*oneWayDelay, probeAfter)
667 }
668 if !contacted.Load() || maxContactAge >= probeAfter {
669 t.Fatalf("authenticated contact=%v maximum age=%v, want fresh below %v", contacted.Load(), maxContactAge, probeAfter)
670 }
671 if probed {
672 t.Fatalf("healthy lossy transfer entered probing: %s at %v pending=%d latest=%d attempts=%d loss=%d overflow=%d", probeWho, probeAt, pending, latestState, attempts.Load(), deterministicDrops.Load(), forward.droppedPackets())
673 }
674 if !lossReducedCwnd {
675 t.Fatal("deterministic loss did not reduce congestion window")
676 }
677 if pending != 0 || latestState != outputCount {
678 t.Fatalf("convergence pending=%d latest state=%d, want 0 and %d", pending, latestState, outputCount)
679 }
680 })
681
682 t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {
683 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
684 aPC, bPC := newFloodPair(nil)
685 inspect, err := pdgram.NewCodec(key())
686 if err != nil {
687 t.Fatal(err)
688 }
689
690 acksDropped := make(chan struct{}, floodRecordCount)
691 bPC.drop = func(pkt []byte, _ net.Addr) bool {
692 _, raw, err := inspect.Open(pkt, 2, nil, nil)
693 if err != nil {
694 return false
695 }
696 frag, err := pdgram.UnmarshalFragment(raw)
697 if err != nil || frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
698 return false
699 }
700 select {
701 case acksDropped <- struct{}{}:
702 default:
703 }
704 return true
705 }
706
707 retransmitBlocked := make(chan struct{}, 1)
708 releaseRetransmit := make(chan struct{})
709 lastRetransmitted := make(chan struct{}, 1)
710 probeSent := make(chan struct{}, 1)
711 freshReturned := make(chan error, 1)
712 writes := make(map[uint64]int)
713 var writesMu sync.Mutex
714 aPC.drop = func(pkt []byte, _ net.Addr) bool {
715 _, raw, err := inspect.Open(pkt, 1, nil, nil)
716 if err != nil {
717 return false
718 }
719 frag, err := pdgram.UnmarshalFragment(raw)
720 if err != nil {
721 return false
722 }
723 if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
724 probeSent <- struct{}{}
725 return false
726 }
727 if frag.Index != 0 {
728 return false
729 }
730 seq, _, _, ok := decodeData(frag.Data)
731 if !ok {
732 return false
733 }
734 writesMu.Lock()
735 writes[seq]++
736 secondWrite := writes[seq] == 2
737 writesMu.Unlock()
738 if secondWrite {
739 select {
740 case retransmitBlocked <- struct{}{}:
741 default:
742 }
743 awaitSignal(t, releaseRetransmit, "retransmit release")
744 if seq == floodRecordCount {
745 lastRetransmitted <- struct{}{}
746 }
747 }
748 return false
749 }
750
751 opts := Options{Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: 11 * time.Millisecond}
752 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
753 if err != nil {
754 t.Fatal(err)
755 }
756 b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
757 if err != nil {
758 _ = a.Close()
759 t.Fatal(err)
760 }
761 defer closeFloodTransports(a, b)
762 for range 6 { // Both transports install retransmit, health, and heartbeat timers.
763 awaitSignal(t, clk.timerCreated, "transport timer creation")
764 }
765 a.mu.Lock()
766 a.rto = time.Hour
767 a.mu.Unlock()
768
769 // Fresh, fragmented output records become overdue while their real
770 // cumulative ACKs are lost. Disabling pacing intentionally forces overload
771 // to classify contention; it does not claim production pacing creates it.
772 sendFloodOutputs(t, a, floodRecordCount, 128)
773 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
774 clk.advance(maxACKDelay)
775 awaitSignal(t, acksDropped, "dropped cumulative ACK")
776 a.mu.Lock()
777 a.rto = 10 * time.Millisecond
778 a.mu.Unlock()
779 a.queueRetransmits()
780 a.mu.Lock()
781 a.rto = time.Hour
782 a.mu.Unlock()
783 clk.advance(initialPacingRTT)
784
785 awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
786 if state := floodState(a, b, nil); state.retransmits < floodRecordCount {
787 t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)
788 }
789 select {
790 case <-probeSent:
791 default:
792 }
793
794 // A fresh record queues behind the sole data sender while control traffic
795 // remains independent of congestion tokens and the blocked data write.
796 go func() {
797 freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})
798 }()
799 select {
800 case err := <-freshReturned:
801 t.Fatalf("fresh output completed while retransmit held pacing: %v", err)
802 default:
803 }
804
805 clk.advance(11 * time.Millisecond)
806 awaitSignal(t, probeSent, "heartbeat probe during output contention")
807
808 close(releaseRetransmit)
809 for range 1024 {
810 clk.advance(initialPacingRTT)
811 time.Sleep(time.Millisecond)
812 select {
813 case err := <-freshReturned:
814 if err != nil {
815 t.Fatal(err)
816 }
817 awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
818 writesMu.Lock()
819 defer writesMu.Unlock()
820 if writes[1] < 2 || writes[floodRecordCount] < 2 || writes[floodRecordCount+1] == 0 {
821 t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
822 }
823 return
824 default:
825 }
826 }
827 writesMu.Lock()
828 defer writesMu.Unlock()
829 t.Fatalf("fresh output did not complete after paced retransmissions: writes=%v state=%+v", writes, floodState(a, b, nil))
830 })
831}
832
833type floodTransportState struct {
834 packetAge, recordAge, ackAge time.Duration
835 queuedBytes int
836 retransmits uint64
837}
838
839func floodState(sender, receiver *Transport, link *simulatedLink) floodTransportState {
840 now := sender.clock.Now()
841 receiver.mu.Lock()
842 packetAge := now.Sub(receiver.lastAuthenticatedPacket)
843 recordAge := now.Sub(receiver.lastCompleteRecord)
844 receiver.mu.Unlock()
845 sender.mu.Lock()
846 ackAge := now.Sub(sender.lastACKProgress)
847 retransmits := sender.retransmits
848 sender.mu.Unlock()
849 queuedBytes := 0
850 if link != nil {
851 queuedBytes = link.queuedBytes()
852 }
853 return floodTransportState{packetAge: packetAge, recordAge: recordAge, ackAge: ackAge, queuedBytes: queuedBytes, retransmits: retransmits}
854}
855
856func floodOutputPayload(baseState, mtu int) []byte {
857 return mustMarshalOutput(ports.Output{
858 Epoch: 1,
859 Base: uint64(baseState),
860 New: uint64(baseState + 1),
861 Size: domain.Size{Cols: 1, Rows: 1},
862 Full: baseState == 0,
863 Data: make([]byte, floodRecordMTUs*mtu),
864 })
865}
866
867func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
868 const mtu = 128
869 for state := range floodRecordCount {
870 payload := floodOutputPayload(state, mtu)
871 output, err := ports.UnmarshalOutput(payload)
872 if err != nil {
873 t.Fatalf("state %d: %v", state, err)
874 }
875 if output.Base != uint64(state) || output.New != uint64(state+1) {
876 t.Fatalf("output state = %d -> %d, want %d -> %d", output.Base, output.New, state, state+1)
877 }
878 minimumBytes := floodRecordMTUs * mtu
879 if len(output.Data) < minimumBytes || len(payload) < minimumBytes {
880 t.Fatalf("state %d: data bytes=%d encoded bytes=%d, want each at least %d", state, len(output.Data), len(payload), minimumBytes)
881 }
882 }
883}
884
885func sendFloodOutputs(t *testing.T, sender *Transport, count, mtu int) {
886 t.Helper()
887 for state := range count {
888 if err := sender.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
889 t.Fatal(err)
890 }
891 }
892 clk, ok := sender.clock.(*manualClock)
893 if !ok {
894 return
895 }
896 for range count * floodRecordMTUs * 2 {
897 clk.advance(initialPacingRTT)
898 time.Sleep(time.Millisecond)
899 sender.mu.Lock()
900 complete := len(sender.outputQueue) == 0 && len(sender.dataSend) == 0
901 for _, p := range sender.pending {
902 complete = complete && !p.initialInFlight && !p.last.IsZero()
903 }
904 sender.mu.Unlock()
905 if complete {
906 return
907 }
908 }
909 t.Fatal("flood output pacing did not drain")
910}
911
912func newFloodTransports(t *testing.T, aPC, bPC *fakePC, clk *manualClock, mtu int) (*Transport, *Transport) {
913 t.Helper()
914 opts := Options{
915 Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
916 DegradedAfter: time.Hour, ProbeAfter: time.Hour, OfflineAfter: time.Hour, DeadAfter: time.Hour,
917 }
918 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
919 if err != nil {
920 t.Fatal(err)
921 }
922 b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
923 if err != nil {
924 _ = a.Close()
925 t.Fatal(err)
926 }
927 return a, b
928}
929
930func closeFloodTransports(a, b *Transport) {
931 _ = a.Close()
932 _ = b.Close()
933}
934
935func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
936 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
937 aPC, bPC := newPair()
938 a, b := newFloodTransports(t, aPC, bPC, clk, 128)
939 defer closeFloodTransports(a, b)
940 b.mu.Lock()
941 startPacket, startRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
942 b.mu.Unlock()
943 codec, err := pdgram.NewCodec(key())
944 if err != nil {
945 t.Fatal(err)
946 }
947 malformedRejected := make(chan struct{}, 1)
948 b.mu.Lock()
949 b.afterMalformedFragment = func() { malformedRejected <- struct{}{} }
950 b.mu.Unlock()
951 clk.advance(time.Second)
952 if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil {
953 t.Fatal(err)
954 }
955 awaitSignal(t, malformedRejected, "malformed fragment rejection")
956 b.mu.Lock()
957 gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
958 b.mu.Unlock()
959 if gotPacket != startPacket || gotRecord != startRecord {
960 t.Fatal("malformed authenticated plaintext counted as packet contact")
961 }
962}
963
964func TestInconsistentAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
965 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
966 aPC, bPC := newPair()
967 a, b := newFloodTransports(t, aPC, bPC, clk, 128)
968 defer closeFloodTransports(a, b)
969 codec, err := pdgram.NewCodec(key())
970 if err != nil {
971 t.Fatal(err)
972 }
973 accepted := make(chan struct{}, 1)
974 rejected := make(chan struct{}, 1)
975 b.mu.Lock()
976 b.afterAuthenticatedPacket = func() { accepted <- struct{}{} }
977 b.afterMalformedFragment = func() { rejected <- struct{}{} }
978 b.mu.Unlock()
979
980 writeFragment := func(counter uint64, frag pdgram.Fragment) {
981 t.Helper()
982 raw, err := pdgram.MarshalFragment(frag)
983 if err != nil {
984 t.Fatal(err)
985 }
986 if _, err := aPC.WriteTo(codec.Seal(1, counter, raw, nil), bPC.addr); err != nil {
987 t.Fatal(err)
988 }
989 }
990 writeFragment(999, pdgram.Fragment{Seq: 42, Index: 0, Count: 2, Data: []byte("first")})
991 awaitSignal(t, accepted, "accepted incomplete fragment")
992 b.mu.Lock()
993 lastContact := b.lastAuthenticatedPacket
994 acceptedInflight := b.reassemblyInflight
995 b.mu.Unlock()
996 if acceptedInflight != 1 {
997 t.Fatalf("reassembly inflight after accepted fragment=%d, want 1", acceptedInflight)
998 }
999
1000 clk.advance(time.Second)
1001 writeFragment(1000, pdgram.Fragment{Seq: 42, Index: 1, Count: 3, Data: []byte("inconsistent")})
1002 awaitSignal(t, rejected, "inconsistent fragment rejection")
1003 b.mu.Lock()
1004 gotContact := b.lastAuthenticatedPacket
1005 gotInflight := b.reassemblyInflight
1006 b.mu.Unlock()
1007 if gotContact != lastContact {
1008 t.Fatalf("rejected inconsistent fragment refreshed contact from %v to %v", lastContact, gotContact)
1009 }
1010 if gotInflight != 0 {
1011 t.Fatalf("reassembly inflight after rejected inconsistent fragment=%d, want 0", gotInflight)
1012 }
1013 diagnostic := b.diagnosticSnapshot()
1014 if diagnostic.ReassemblyInflight != 0 {
1015 t.Fatalf("diagnostic reassembly inflight after rejected inconsistent fragment=%d, want 0", diagnostic.ReassemblyInflight)
1016 }
1017}