1package dgram
2
3import (
4 "context"
5 "encoding/binary"
6 "errors"
7 "net"
8 "sync"
9 "testing"
10 "time"
11
12 "github.com/bnema/vev/internal/ports"
13 pdgram "github.com/bnema/vev/pkg/dgram"
14)
15
16type stagedTimeoutPC struct {
17 mu sync.Mutex
18 deadline time.Time
19 calls int
20 started chan struct{}
21 release chan struct{}
22}
23
24func (p *stagedTimeoutPC) ReadFrom([]byte) (int, net.Addr, error) {
25 return 0, nil, net.ErrClosed
26}
27func (p *stagedTimeoutPC) WriteTo(b []byte, _ net.Addr) (int, error) {
28 p.mu.Lock()
29 p.calls++
30 call := p.calls
31 p.mu.Unlock()
32 if call <= 2 {
33 p.started <- struct{}{}
34 <-p.release
35 return 0, timeoutErr{}
36 }
37 return len(b), nil
38}
39func (p *stagedTimeoutPC) Close() error { return nil }
40func (p *stagedTimeoutPC) LocalAddr() net.Addr { return testAddr("staged") }
41func (p *stagedTimeoutPC) SetDeadline(time.Time) error { return nil }
42func (p *stagedTimeoutPC) SetReadDeadline(time.Time) error { return nil }
43func (p *stagedTimeoutPC) SetWriteDeadline(d time.Time) error {
44 p.mu.Lock()
45 p.deadline = d
46 p.mu.Unlock()
47 return nil
48}
49
50func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
51 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
52 aPC, bPC := newPair()
53 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
54 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
55 })
56 if err != nil {
57 t.Fatal(err)
58 }
59 b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
60 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
61 })
62 if err != nil {
63 _ = a.Close()
64 t.Fatal(err)
65 }
66 releaseACK := make(chan struct{})
67 defer close(releaseACK)
68 defer closeFloodTransports(a, b)
69
70 ackStarted := make(chan struct{}, 1)
71 aPC.drop = func(_ []byte, _ net.Addr) bool {
72 ackStarted <- struct{}{}
73 <-releaseACK
74 return true
75 }
76 contacts := make(chan struct{}, 2)
77 a.mu.Lock()
78 a.afterAuthenticatedPacket = func() { contacts <- struct{}{} }
79 // Keep diagnostics local to this test so each snapshot has an explicit
80 // causative event instead of depending on observer scheduling.
81 a.diagnosticCh = make(chan Diagnostic, 1)
82 a.mu.Unlock()
83
84 if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
85 t.Fatal(err)
86 }
87 awaitSignal(t, contacts, "first authenticated record")
88 for range 7 {
89 awaitSignal(t, clk.timerCreated, "transport and ACK timer creation")
90 }
91 clk.advance(maxACKDelay)
92 awaitSignal(t, ackStarted, "blocked ACK")
93
94 clk.advance(time.Second)
95 if err := b.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}); err != nil {
96 t.Fatal(err)
97 }
98 awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")
99 wantComplete := clk.Now()
100 eventually(t, time.Second, func() bool {
101 a.mu.Lock()
102 defer a.mu.Unlock()
103 return a.lastCompleteRecord.Equal(wantComplete)
104 })
105 a.emitDiagnostic()
106 diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")
107 if diagnostic.SinceCompleteRecord != 0 {
108 t.Fatalf("second record age=%v, want 0 before ACK release", diagnostic.SinceCompleteRecord)
109 }
110}
111
112func TestCumulativeACKCoalescesForBoundedDelay(t *testing.T) {
113 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
114 aPC, bPC := newPair()
115 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
116 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
117 })
118 if err != nil {
119 t.Fatal(err)
120 }
121 defer func() { _ = a.Close() }()
122 for range 3 {
123 awaitSignal(t, clk.timerCreated, "transport timer creation")
124 }
125
126 acks := make(chan uint64, 2)
127 aPC.drop = func(pkt []byte, _ net.Addr) bool {
128 _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
129 if err != nil {
130 t.Errorf("open outbound ACK: %v", err)
131 return true
132 }
133 frag, err := pdgram.UnmarshalFragment(raw)
134 if err != nil {
135 t.Errorf("unmarshal outbound ACK: %v", err)
136 return true
137 }
138 if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck {
139 acks <- binary.BigEndian.Uint64(frag.Data[1:])
140 }
141 return true
142 }
143
144 a.queueACK(1)
145 a.queueACK(2)
146 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
147 clk.advance(maxACKDelay - time.Millisecond)
148 select {
149 case seq := <-acks:
150 t.Fatalf("ACK %d sent before coalescing delay", seq)
151 default:
152 }
153 clk.advance(time.Millisecond)
154 if seq := awaitResult(t, acks, "coalesced ACK"); seq != 2 {
155 t.Fatalf("ACK=%d, want cumulative maximum 2", seq)
156 }
157 select {
158 case seq := <-acks:
159 t.Fatalf("extra ACK %d sent for coalesced records", seq)
160 default:
161 }
162}
163
164func TestACKDeadlineDispatchesWhileAdvisoryWriteIsBlocked(t *testing.T) {
165 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
166 aPC, bPC := newPair()
167 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
168 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
169 })
170 if err != nil {
171 t.Fatal(err)
172 }
173 defer func() { _ = a.Close() }()
174 for range 3 {
175 awaitSignal(t, clk.timerCreated, "transport timer creation")
176 }
177
178 probeStarted := make(chan struct{}, 1)
179 ackStarted := make(chan struct{}, 1)
180 releaseProbe := make(chan struct{})
181 defer close(releaseProbe)
182 aPC.drop = func(pkt []byte, _ net.Addr) bool {
183 _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
184 if err != nil {
185 return true
186 }
187 frag, err := pdgram.UnmarshalFragment(raw)
188 if err != nil || frag.Count != 1 || len(frag.Data) != 9 {
189 return true
190 }
191 switch frag.Data[0] {
192 case recProbe:
193 probeStarted <- struct{}{}
194 <-releaseProbe
195 case recAck:
196 ackStarted <- struct{}{}
197 }
198 return true
199 }
200
201 if !a.queueControl(recProbe, 1) {
202 t.Fatal("probe control queue unexpectedly full")
203 }
204 awaitSignal(t, probeStarted, "blocked advisory probe write")
205 a.queueACK(1)
206 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
207 clk.advance(maxACKDelay)
208 awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked")
209}
210
211func TestACKBypassesDataWriterAndRemainsDecodable(t *testing.T) {
212 aPC, bPC := newPair()
213 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
214 ResendAfter: time.Hour, Heartbeat: time.Hour,
215 })
216 if err != nil {
217 t.Fatal(err)
218 }
219 defer func() { _ = a.Close() }()
220
221 acks := make(chan uint64, 1)
222 aPC.drop = func(pkt []byte, _ net.Addr) bool {
223 _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
224 if err != nil {
225 t.Errorf("open ACK: %v", err)
226 return true
227 }
228 frag, err := pdgram.UnmarshalFragment(raw)
229 if err != nil {
230 t.Errorf("unmarshal ACK: %v", err)
231 return true
232 }
233 if frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck {
234 t.Errorf("control record=%x, want encoded ACK", frag.Data)
235 return true
236 }
237 acks <- binary.BigEndian.Uint64(frag.Data[1:])
238 return true
239 }
240
241 a.writeMu.Lock()
242 defer a.writeMu.Unlock()
243 go a.sendAck(42)
244 if seq := awaitResult(t, acks, "ACK while data writer is blocked"); seq != 42 {
245 t.Fatalf("ACK sequence=%d, want 42", seq)
246 }
247}
248
249func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {
250 aPC, bPC := newPair()
251 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
252 ResendAfter: time.Hour, Heartbeat: time.Hour,
253 })
254 if err != nil {
255 t.Fatal(err)
256 }
257 b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
258 ResendAfter: time.Hour, Heartbeat: time.Hour,
259 })
260 if err != nil {
261 _ = a.Close()
262 t.Fatal(err)
263 }
264 defer closeFloodTransports(a, b)
265
266 a.writeMu.Lock()
267 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
268 defer cancel()
269 done := make(chan error, 1)
270 go func() { done <- a.Probe(ctx) }()
271 select {
272 case err := <-done:
273 a.writeMu.Unlock()
274 if err != nil {
275 t.Fatalf("Probe failed: %v", err)
276 }
277 case <-time.After(50 * time.Millisecond):
278 a.writeMu.Unlock()
279 <-done
280 t.Fatal("Probe blocked behind data writer")
281 }
282}
283
284func TestProbeDoesNotBlockBehindACKWrite(t *testing.T) {
285 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
286 aPC, bPC := newPair()
287 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
288 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
289 })
290 if err != nil {
291 t.Fatal(err)
292 }
293 defer func() { _ = a.Close() }()
294 for range 3 {
295 awaitSignal(t, clk.timerCreated, "transport timer creation")
296 }
297
298 ackStarted := make(chan struct{}, 1)
299 probeStarted := make(chan struct{}, 1)
300 releaseACK := make(chan struct{})
301 defer close(releaseACK)
302 aPC.drop = func(pkt []byte, _ net.Addr) bool {
303 _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
304 if err != nil {
305 return true
306 }
307 frag, err := pdgram.UnmarshalFragment(raw)
308 if err != nil || frag.Count != 1 || len(frag.Data) != 9 {
309 return true
310 }
311 switch frag.Data[0] {
312 case recAck:
313 ackStarted <- struct{}{}
314 <-releaseACK
315 case recProbe:
316 probeStarted <- struct{}{}
317 }
318 return true
319 }
320
321 a.queueACK(1)
322 awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
323 clk.advance(maxACKDelay)
324 awaitSignal(t, ackStarted, "blocked ACK")
325 if !a.queueControl(recProbe, 1) {
326 t.Fatal("probe control queue unexpectedly full")
327 }
328 awaitSignal(t, probeStarted, "probe while ACK write is blocked")
329}
330
331func TestControlWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
332 pc := newBlockingWritePC()
333 tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
334 WriteTimeout: 10 * time.Millisecond,
335 ResendAfter: time.Hour,
336 Heartbeat: time.Hour,
337 })
338 if err != nil {
339 t.Fatal(err)
340 }
341 defer func() { _ = tr.Close() }()
342
343 done := make(chan error, 1)
344 go func() { done <- tr.sendControl(recAck, 1) }()
345 select {
346 case err := <-done:
347 var timeout interface{ Timeout() bool }
348 if !errors.As(err, &timeout) || !timeout.Timeout() {
349 t.Fatalf("control write err=%v, want timeout", err)
350 }
351 case <-time.After(250 * time.Millisecond):
352 t.Fatal("control write blocked without honoring WriteTimeout")
353 }
354}
355
356func TestConcurrentWriterCannotClearEarlierDeadline(t *testing.T) {
357 pc := &deadlineCapturePC{addr: testAddr("a"), done: make(chan struct{})}
358 state := newWriteDeadlineState(pc)
359 earlier := time.Now().Add(time.Second)
360 later := earlier.Add(time.Second)
361 finishEarlier, err := state.begin(earlier)
362 if err != nil {
363 t.Fatal(err)
364 }
365 defer finishEarlier()
366 finishLater, err := state.begin(later)
367 if err != nil {
368 t.Fatal(err)
369 }
370 finishLater()
371
372 got, _ := pc.deadline.Load().(time.Time)
373 if !got.Equal(earlier) {
374 t.Fatalf("deadline after concurrent writer completed=%v, want earlier active deadline %v", got, earlier)
375 }
376}
377
378func TestFreshControlWriteRetriesAfterInheritedDeadline(t *testing.T) {
379 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
380 pc := &stagedTimeoutPC{started: make(chan struct{}, 2), release: make(chan struct{})}
381 tr := &Transport{clock: clk}
382 deadlines := newWriteDeadlineState(pc)
383 oldDone := make(chan error, 1)
384 go func() {
385 oldDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("old"), deadlines, 10*time.Millisecond)
386 }()
387 awaitSignal(t, pc.started, "older blocked write")
388 clk.advance(5 * time.Millisecond)
389 freshDone := make(chan error, 1)
390 go func() {
391 freshDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("fresh"), deadlines, 100*time.Millisecond)
392 }()
393 awaitSignal(t, pc.started, "fresh write sharing earlier deadline")
394 clk.advance(5 * time.Millisecond)
395 close(pc.release)
396 if err := awaitResult(t, oldDone, "older write timeout"); err == nil {
397 t.Fatal("older write succeeded, want timeout")
398 }
399 if err := awaitResult(t, freshDone, "fresh write retry"); err != nil {
400 t.Fatalf("fresh write failed at inherited deadline: %v", err)
401 }
402}
403
404func TestProbeReturnsLocalWriteError(t *testing.T) {
405 aPC, bPC := newPair()
406 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
407 ResendAfter: time.Hour, Heartbeat: time.Hour,
408 })
409 if err != nil {
410 t.Fatal(err)
411 }
412 defer func() { _ = a.Close() }()
413 if err := bPC.Close(); err != nil {
414 t.Fatal(err)
415 }
416
417 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
418 defer cancel()
419 start := time.Now()
420 if err := a.Probe(ctx); err == nil || errors.Is(err, context.DeadlineExceeded) {
421 t.Fatalf("Probe err=%v, want immediate local write error", err)
422 }
423 if elapsed := time.Since(start); elapsed > 250*time.Millisecond {
424 t.Fatalf("Probe returned local write error after %v", elapsed)
425 }
426}
427
428func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
429 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
430 aPC, bPC := newPair()
431 const heartbeat = 11 * time.Millisecond
432 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
433 Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond, Heartbeat: heartbeat,
434 })
435 if err != nil {
436 t.Fatal(err)
437 }
438 defer func() { _ = a.Close() }()
439 for range 3 {
440 awaitSignal(t, clk.timerCreated, "retransmit, health, and heartbeat timer creation")
441 }
442
443 retransmitStarted := make(chan struct{}, 1)
444 releaseRetransmit := make(chan struct{})
445 defer close(releaseRetransmit)
446 heartbeats := make(chan struct{}, 2)
447 aPC.drop = func(pkt []byte, _ net.Addr) bool {
448 _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
449 if err != nil {
450 t.Errorf("open outbound packet: %v", err)
451 return true
452 }
453 frag, err := pdgram.UnmarshalFragment(raw)
454 if err != nil {
455 t.Errorf("unmarshal outbound fragment: %v", err)
456 return true
457 }
458 if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recProbe {
459 heartbeats <- struct{}{}
460 return true
461 }
462 select {
463 case retransmitStarted <- struct{}{}:
464 <-releaseRetransmit
465 default:
466 }
467 return true
468 }
469
470 now := clk.Now()
471 a.mu.Lock()
472 for seq := uint64(1); seq <= 64; seq++ {
473 a.pending[seq] = &pending{
474 frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
475 enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
476 }
477 }
478 a.mu.Unlock()
479
480 clk.advance(10 * time.Millisecond)
481 awaitSignal(t, retransmitStarted, "first retransmit in 64-record storm")
482 for interval := 1; interval <= 2; interval++ {
483 clk.advance(heartbeat)
484 awaitSignal(t, heartbeats, "heartbeat during retransmit storm")
485 }
486}
487
488func TestHealthChecksContinueDuringRetransmitStorm(t *testing.T) {
489 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
490 aPC, bPC := newPair()
491 a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
492 Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond,
493 Heartbeat: time.Hour, DegradedAfter: 10 * time.Millisecond,
494 })
495 if err != nil {
496 t.Fatal(err)
497 }
498 defer func() { _ = a.Close() }()
499 for range 3 {
500 awaitSignal(t, clk.timerCreated, "transport timer creation")
501 }
502
503 retransmitStarted := make(chan struct{}, 1)
504 releaseRetransmit := make(chan struct{})
505 defer close(releaseRetransmit)
506 aPC.drop = func(_ []byte, _ net.Addr) bool {
507 select {
508 case retransmitStarted <- struct{}{}:
509 <-releaseRetransmit
510 default:
511 }
512 return true
513 }
514
515 now := clk.Now()
516 a.mu.Lock()
517 a.pending[1] = &pending{
518 frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
519 enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
520 }
521 a.mu.Unlock()
522
523 clk.advance(10 * time.Millisecond)
524 awaitSignal(t, retransmitStarted, "blocked retransmit")
525 select {
526 case event := <-a.LinkEvents():
527 if event.State != ports.LinkStateDegraded {
528 t.Fatalf("link state=%s, want degraded", event.State)
529 }
530 case <-time.After(time.Second):
531 t.Fatal("health check did not run while retransmit sender was blocked")
532 }
533}