1package dgram
2
3import (
4 "sync"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11func TestDiagnosticSnapshotSeparatesProgressAges(t *testing.T) {
12 now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
13 got := make(chan Diagnostic, 1)
14 tr := &Transport{
15 clock: fixedClock{now: now},
16 linkState: ports.LinkStateDegraded,
17 lastAuthenticatedPacket: now.Add(-time.Second),
18 lastCompleteRecord: now.Add(-2 * time.Second),
19 lastACKProgress: now.Add(-3 * time.Second),
20 retransmits: 64,
21 reassemblyInflight: 2,
22 pending: map[uint64]*pending{
23 1: {frame: ports.Frame{Payload: make([]byte, 7)}},
24 2: {frame: ports.Frame{Payload: make([]byte, 11)}},
25 },
26 diagnosticCh: make(chan Diagnostic, diagnosticBufferSize),
27 done: make(chan struct{}),
28 }
29 tr.observe = func(d Diagnostic) {
30 // This would deadlock if the observer were invoked under Transport.mu.
31 _ = tr.LinkState()
32 got <- d
33 }
34 go tr.diagnosticLoop()
35 defer close(tr.done)
36
37 tr.emitDiagnostic()
38 select {
39 case d := <-got:
40 if d.At != now || d.State != ports.LinkStateDegraded {
41 t.Fatalf("identity = %+v", d)
42 }
43 if d.SinceAuthenticatedPacket != time.Second || d.SinceCompleteRecord != 2*time.Second || d.SinceACKProgress != 3*time.Second {
44 t.Fatalf("ages = packet %v, record %v, ack %v", d.SinceAuthenticatedPacket, d.SinceCompleteRecord, d.SinceACKProgress)
45 }
46 if d.PendingRecords != 2 || d.PendingBytes != 18 || d.Retransmits != 64 || d.ReassemblyInflight != 2 {
47 t.Fatalf("counters = %+v", d)
48 }
49 case <-time.After(time.Second):
50 t.Fatal("diagnostic observer was not called")
51 }
52}
53
54func TestDiagnosticSnapshotIsLockConsistentDuringConcurrentUpdate(t *testing.T) {
55 now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
56 clock := &snapshotClock{now: now, sampled: make(chan struct{}), release: make(chan struct{})}
57 tr := &Transport{
58 clock: clock,
59 lastAuthenticatedPacket: now.Add(-time.Second),
60 lastCompleteRecord: now.Add(-2 * time.Second),
61 lastACKProgress: now.Add(-3 * time.Second),
62 pending: make(map[uint64]*pending),
63 diagnosticCh: make(chan Diagnostic, 1),
64 }
65
66 go tr.emitDiagnostic()
67 select {
68 case <-clock.sampled:
69 case <-time.After(time.Second):
70 t.Fatal("diagnostic did not begin its clock snapshot")
71 }
72 updated := make(chan struct{})
73 go func() {
74 tr.mu.Lock()
75 tr.lastAuthenticatedPacket = now.Add(time.Second)
76 tr.lastCompleteRecord = now.Add(time.Second)
77 tr.lastACKProgress = now.Add(time.Second)
78 tr.retransmits = 1
79 tr.mu.Unlock()
80 close(updated)
81 }()
82 close(clock.release)
83
84 var d Diagnostic
85 select {
86 case d = <-tr.diagnosticCh:
87 case <-time.After(time.Second):
88 t.Fatal("diagnostic was not emitted")
89 }
90 select {
91 case <-updated:
92 case <-time.After(time.Second):
93 t.Fatal("concurrent update was blocked")
94 }
95 if d.At != now || d.Retransmits != 0 {
96 t.Fatalf("mixed diagnostic snapshot: %+v", d)
97 }
98 if d.SinceAuthenticatedPacket != time.Second || d.SinceCompleteRecord != 2*time.Second || d.SinceACKProgress != 3*time.Second {
99 t.Fatalf("inconsistent diagnostic ages: %+v", d)
100 }
101 if d.SinceAuthenticatedPacket < 0 || d.SinceCompleteRecord < 0 || d.SinceACKProgress < 0 {
102 t.Fatalf("negative diagnostic age: %+v", d)
103 }
104}
105
106type snapshotClock struct {
107 mu sync.Mutex
108 now time.Time
109 sampled chan struct{}
110 release chan struct{}
111}
112
113func (c *snapshotClock) Now() time.Time {
114 c.mu.Lock()
115 defer c.mu.Unlock()
116 select {
117 case c.sampled <- struct{}{}:
118 default:
119 }
120 <-c.release
121 return c.now
122}
123
124func (c *snapshotClock) NewTimer(time.Duration) ports.Timer { panic("not used") }
125
126func TestDiagnosticObserverIsBounded(t *testing.T) {
127 tr := &Transport{
128 clock: fixedClock{now: time.Now()},
129 pending: make(map[uint64]*pending),
130 diagnosticCh: make(chan Diagnostic, 1),
131 done: make(chan struct{}),
132 }
133 for range 100 {
134 tr.emitDiagnostic()
135 }
136 if got := len(tr.diagnosticCh); got != 1 {
137 t.Fatalf("queued diagnostics = %d, want bounded buffer", got)
138 }
139 close(tr.done)
140}