1package dgram
2
3import (
4 "net"
5 "sync/atomic"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10)
11
12func TestResendSkipsQueuedOutputBeforeFirstWireAttempt(t *testing.T) {
13 tr, writes := newResendTestTransport(t, 10)
14 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}}
15
16 tr.resendPending()
17
18 if got := writes.Load(); got != 0 {
19 t.Fatalf("writes=%d, want queued output excluded from resend", got)
20 }
21 if !tr.pending[1].first.IsZero() || !tr.pending[1].last.IsZero() {
22 t.Fatal("resend scan marked queued output as sent")
23 }
24}
25
26func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
27 tr, writes := newResendTestTransport(t, 10)
28 now := tr.clock.Now()
29 tr.pending[1] = &pending{
30 frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("sending")},
31 first: now.Add(-time.Second),
32 last: now.Add(-time.Second),
33 initialInFlight: true,
34 }
35
36 tr.resendPending()
37 if got := writes.Load(); got != 0 {
38 t.Fatalf("writes=%d, want first wire attempt excluded from resend", got)
39 }
40 tr.pending[1].initialInFlight = false
41 tr.resendPending()
42 if got := writes.Load(); got != 1 {
43 t.Fatalf("writes=%d, want resend after first wire attempt completed", got)
44 }
45}
46
47func TestReliableInputRetransmitsUntilAck(t *testing.T) {
48 aPC, bPC := newPair()
49 var dropped atomic.Bool
50 aPC.drop = func(_ []byte, addr net.Addr) bool {
51 return addr.String() == "b" && dropped.CompareAndSwap(false, true)
52 }
53 a, err := NewTransport(aPC, testAddr("b"), key(), 1, 2)
54 if err != nil {
55 t.Fatal(err)
56 }
57 b, err := NewTransport(bPC, testAddr("a"), key(), 2, 1)
58 if err != nil {
59 t.Fatal(err)
60 }
61 defer func() {
62 if err := a.Close(); err != nil {
63 t.Errorf("close a: %v", err)
64 }
65 }()
66 defer func() {
67 if err := b.Close(); err != nil {
68 t.Errorf("close b: %v", err)
69 }
70 }()
71 if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}); err != nil {
72 t.Fatal(err)
73 }
74 got := recvWithin(t, b, time.Second)
75 if got.Type != ports.MsgInput || string(got.Payload) != "typed" || !dropped.Load() {
76 t.Fatalf("got=%+v dropped=%v", got, dropped.Load())
77 }
78}
79
80func TestOutputRetransmitsUntilAck(t *testing.T) {
81 aPC, bPC := newPair()
82 var dropped atomic.Bool
83 aPC.drop = func(_ []byte, addr net.Addr) bool {
84 return addr.String() == "b" && dropped.CompareAndSwap(false, true)
85 }
86 a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
87 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
88 defer func() {
89 if err := a.Close(); err != nil {
90 t.Errorf("close a: %v", err)
91 }
92 }()
93 defer func() {
94 if err := b.Close(); err != nil {
95 t.Errorf("close b: %v", err)
96 }
97 }()
98 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("state1")}); err != nil {
99 t.Fatal(err)
100 }
101 got := recvWithin(t, b, time.Second)
102 if got.Type != ports.MsgOutput || string(got.Payload) != "state1" || !dropped.Load() {
103 t.Fatalf("got=%+v dropped=%v", got, dropped.Load())
104 }
105}
106
107func TestImmediateLoopbackAckSamplesStampedFinalWrite(t *testing.T) {
108 aPC, bPC := newPair()
109 a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
110 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
111 defer func() { _ = a.Close() }()
112 defer func() { _ = b.Close() }()
113 aPC.afterWrite = func() {
114 deadline := time.Now().Add(time.Second)
115 for time.Now().Before(deadline) {
116 a.mu.Lock()
117 _, pending := a.pending[1]
118 a.mu.Unlock()
119 if !pending {
120 return
121 }
122 time.Sleep(time.Microsecond)
123 }
124 t.Error("ACK did not retire pending frame before WriteTo returned")
125 }
126
127 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("state")}); err != nil {
128 t.Fatal(err)
129 }
130 eventually(t, time.Second, func() bool {
131 a.mu.Lock()
132 defer a.mu.Unlock()
133 return a.srtt > 0
134 })
135 a.mu.Lock()
136 srtt, rto := a.srtt, a.rto
137 a.mu.Unlock()
138 if srtt <= 0 || srtt > time.Second {
139 t.Fatalf("srtt=%v, want finite immediate-loopback sample", srtt)
140 }
141 if rto < a.resendAfter || rto > a.maxResendAfter {
142 t.Fatalf("rto=%v outside [%v,%v]", rto, a.resendAfter, a.maxResendAfter)
143 }
144}
145
146func TestOutputDependencyChainRetransmitsDroppedPredecessors(t *testing.T) {
147 aPC, bPC := newPair()
148 var sendAttempts atomic.Int32
149 aPC.drop = func(_ []byte, addr net.Addr) bool {
150 return addr.String() == "b" && sendAttempts.Add(1) <= 2
151 }
152 a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
153 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
154 defer func() { _ = a.Close() }()
155 defer func() { _ = b.Close() }()
156
157 for _, payload := range []string{"state1", "state2", "state3"} {
158 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
159 t.Fatal(err)
160 }
161 }
162
163 for _, want := range []string{"state1", "state2", "state3"} {
164 got := recvWithin(t, b, time.Second)
165 if got.Type != ports.MsgOutput || string(got.Payload) != want {
166 t.Fatalf("got=%+v, want ordered prerequisite output %q", got, want)
167 }
168 }
169 if sendAttempts.Load() < 3 {
170 t.Fatalf("send attempts=%d, want first two output sends dropped before newest arrived", sendAttempts.Load())
171 }
172}
173
174func TestResendPendingCapsBurst(t *testing.T) {
175 tr, writes := newResendTestTransport(t, 2)
176 now := tr.clock.Now()
177 for i := range 5 {
178 tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: now.Add(-time.Second)}
179 }
180
181 tr.resendPending()
182 if got := writes.Load(); got != 2 {
183 t.Fatalf("resend writes=%d, want capped burst of 2", got)
184 }
185}
186
187func TestRetransmitBatchReducesCongestionWindowOnce(t *testing.T) {
188 tr, writes := newResendTestTransport(t, 10)
189 now := tr.clock.Now()
190 for seq := uint64(1); seq <= 3; seq++ {
191 tr.pending[seq] = &pending{
192 frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}},
193 last: now.Add(-time.Second),
194 }
195 }
196 initialCwnd := tr.congestion.cwndBytes
197
198 tr.resendPending()
199
200 if got := writes.Load(); got != 3 {
201 t.Fatalf("retransmit writes=%d, want 3", got)
202 }
203 if got, want := tr.congestion.cwndBytes, initialCwnd/2; got != want {
204 t.Fatalf("cwnd after one retransmit batch=%d, want %d", got, want)
205 }
206}
207
208func TestACKBeforeQueuedRetransmitDoesNotReduceCongestionWindow(t *testing.T) {
209 tr, writes := newResendTestTransport(t, 1)
210 now := tr.clock.Now()
211 tr.pending[1] = &pending{
212 frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("acked")},
213 first: now.Add(-time.Second),
214 last: now.Add(-time.Second),
215 wireBytes: tr.mtu,
216 }
217 initialCwnd := tr.congestion.cwndBytes
218 tr.mu.Lock()
219 batch, _ := tr.selectRetransmitsLocked(now)
220 tr.mu.Unlock()
221 if len(batch) != 1 {
222 t.Fatalf("selected retransmits=%d, want 1", len(batch))
223 }
224 tr.handleRecord(ackRecord(1))
225 cwndAfterACK := initialCwnd + max(1, tr.mtu*tr.mtu/initialCwnd)
226 counterAfterACK := tr.ctr
227
228 tr.runRetransmitBatch(&bytePacer{clk: tr.clock}, batch)
229
230 if got := tr.ctr; got != counterAfterACK {
231 t.Fatalf("packet counter after ACK=%d, want unchanged %d", got, counterAfterACK)
232 }
233 if got := writes.Load(); got != 0 {
234 t.Fatalf("writes after ACK=%d, want 0", got)
235 }
236 if got := tr.congestion.cwndBytes; got != cwndAfterACK {
237 t.Fatalf("cwnd after ACK race=%d, want ACK-only growth %d", got, cwndAfterACK)
238 }
239}
240
241func TestOversizedRetransmitHonorsPacketBudget(t *testing.T) {
242 tr, writes := newResendTestTransport(t, 1)
243 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
244 tr.clock = clk
245 tr.done = make(chan struct{})
246 tr.pending[1] = &pending{
247 frame: ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 24*tr.mtu)},
248 last: clk.now.Add(-time.Second),
249 }
250 done := make(chan struct{})
251 go func() {
252 tr.resendPending()
253 close(done)
254 }()
255
256 previous := int32(0)
257 for tick := 0; tick < 128; tick++ {
258 clk.advance(initialPacingRTT)
259 time.Sleep(time.Millisecond)
260 current := writes.Load()
261 if delta := current - previous; delta > dataBurstPackets {
262 t.Fatalf("tick %d retransmitted %d packets, burst %d", tick, delta, dataBurstPackets)
263 }
264 previous = current
265 select {
266 case <-done:
267 return
268 default:
269 }
270 }
271 t.Fatal("oversized retransmit did not complete within pacing ticks")
272}
273
274func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
275 tr, writes := newResendTestTransport(t, 10)
276 now := tr.clock.Now()
277 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}
278 tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}
279 tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}
280
281 tr.resendPending()
282 if got := writes.Load(); got != 2 {
283 t.Fatalf("resend writes=%d, want ready and capped frames only", got)
284 }
285 if attempts := tr.pending[1].attempts; attempts != 1 {
286 t.Fatalf("too-soon attempts=%d, want unchanged", attempts)
287 }
288 if attempts := tr.pending[2].attempts; attempts != 2 {
289 t.Fatalf("ready attempts=%d, want incremented", attempts)
290 }
291 if attempts := tr.pending[3].attempts; attempts != 11 {
292 t.Fatalf("capped attempts=%d, want incremented", attempts)
293 }
294}
295
296func TestCumulativeAckReleasesAllEarlierPendingFrames(t *testing.T) {
297 tr, _ := newResendTestTransport(t, 10)
298 now := tr.clock.Now()
299 for seq := uint64(1); seq <= 3; seq++ {
300 tr.pending[seq] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}}, first: now, last: now}
301 }
302
303 tr.handleRecord(ackRecord(3))
304
305 if got := len(tr.pending); got != 0 {
306 t.Fatalf("pending after cumulative ack=%d, want 0", got)
307 }
308}
309
310func TestRTTEstimatorUpdatesRTOFromAckSample(t *testing.T) {
311 clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
312 tr, writes := newResendTestTransport(t, 10)
313 tr.clock = clk
314 tr.resendAfter = 100 * time.Millisecond
315 tr.maxResendAfter = time.Second
316 tr.rto = tr.resendAfter
317 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-200 * time.Millisecond)}
318
319 tr.handleRecord(ackRecord(1))
320
321 if tr.rto <= tr.resendAfter {
322 t.Fatalf("rto=%v, want above initial resend after RTT sample", tr.rto)
323 }
324 tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now, last: clk.now.Add(-(tr.rto - time.Millisecond))}
325 tr.resendPending()
326 if got := writes.Load(); got != 0 {
327 t.Fatalf("writes before rto=%d, want 0", got)
328 }
329 tr.pending[2].last = clk.now.Add(-tr.rto)
330 tr.resendPending()
331 if got := writes.Load(); got != 1 {
332 t.Fatalf("writes at rto=%d, want 1", got)
333 }
334}
335
336func TestRTTEstimatorIgnoresRetransmittedFramesAfterRecovery(t *testing.T) {
337 clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
338 tr, _ := newResendTestTransport(t, 10)
339 tr.clock = clk
340 tr.resendAfter = 100 * time.Millisecond
341 tr.maxResendAfter = time.Second
342 tr.rto = tr.resendAfter
343 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-tr.resendAfter)}
344 tr.resendPending()
345 tr.linkState = ports.LinkStateOffline
346
347 tr.setLinkState(ports.LinkStateConnected, nil)
348 tr.handleRecord(ackRecord(1))
349
350 if tr.rto != tr.resendAfter || tr.srtt != 0 || tr.rttvar != 0 {
351 t.Fatalf("rtt sampled retransmitted frame after recovery: rto=%v srtt=%v rttvar=%v", tr.rto, tr.srtt, tr.rttvar)
352 }
353}