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