1package dgram
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "net"
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
16func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
17 aPC, bPC := newPair()
18 var hopped atomic.Bool
19 aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
20 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
21 ResendAfter: 20 * time.Millisecond,
22 ProbeAfter: time.Millisecond,
23 DeadAfter: time.Hour,
24 RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
25 newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
26 bPC.peers["a2"] = newPC
27 hopped.Store(true)
28 return newPC, nil
29 },
30 })
31 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
32 defer func() { _ = a.Close() }()
33 defer func() { _ = b.Close() }()
34
35 if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("survive")}); err != nil {
36 t.Fatal(err)
37 }
38 a.mu.Lock()
39 a.health.lastPacket = time.Now().Add(-time.Second)
40 a.mu.Unlock()
41 a.checkSilence()
42 if !hopped.Load() {
43 t.Fatal("expected packet conn rebind")
44 }
45 aPC.drop = nil
46 got := recvWithin(t, b, time.Second)
47 if got.Type != ports.MsgInput || string(got.Payload) != "survive" {
48 t.Fatalf("got=%+v, want pending reliable message", got)
49 }
50 select {
51 case <-a.done:
52 t.Fatal("transport closed during port hop")
53 default:
54 }
55}
56
57func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
58 aPC, bPC := newPair()
59 var attempts atomic.Int32
60 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
61 ResendAfter: time.Hour,
62 Heartbeat: time.Hour,
63 ProbeAfter: time.Millisecond,
64 DeadAfter: time.Hour,
65 RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
66 if attempts.Add(1) == 1 {
67 return nil, errors.New("temporary rebind failure")
68 }
69 newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
70 bPC.peers["a2"] = newPC
71 return newPC, nil
72 },
73 })
74 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
75 defer func() { _ = a.Close() }()
76 defer func() { _ = b.Close() }()
77
78 a.mu.Lock()
79 a.health.lastPacket = time.Now().Add(-time.Second)
80 a.mu.Unlock()
81 a.checkSilence()
82 a.mu.Lock()
83 firstPC := a.pc
84 a.mu.Unlock()
85 if attempts.Load() != 1 || firstPC != aPC {
86 t.Fatalf("first failed hop attempts=%d pc=%v, want one attempt and original pc", attempts.Load(), firstPC.LocalAddr())
87 }
88
89 a.checkSilence()
90 a.mu.Lock()
91 secondPC := a.pc
92 a.mu.Unlock()
93 if attempts.Load() != 2 || secondPC == aPC {
94 t.Fatalf("second hop attempts=%d pc=%v, want retry with new pc", attempts.Load(), secondPC.LocalAddr())
95 }
96}
97
98func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
99 aPC, bPC := newPair()
100 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ProbeAfter: time.Millisecond, DeadAfter: time.Hour})
101 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
102 defer func() { _ = a.Close() }()
103 defer func() { _ = b.Close() }()
104 a.mu.Lock()
105 a.health.lastPacket = time.Now().Add(-time.Second)
106 a.mu.Unlock()
107 a.checkSilence()
108 if a.pc != aPC {
109 t.Fatal("server/proxy transport hopped without rebind hook")
110 }
111}
112
113func TestProbeTimeoutAndSuccess(t *testing.T) {
114 aPC, bPC := newPair()
115 a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
116 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
117 defer func() { _ = a.Close() }()
118 defer func() { _ = b.Close() }()
119
120 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
121 if err := a.Probe(ctx); err != nil {
122 cancel()
123 t.Fatalf("probe success: %v", err)
124 }
125 cancel()
126
127 bPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "a" }
128 ctx, cancel = context.WithTimeout(context.Background(), 50*time.Millisecond)
129 defer cancel()
130 if err := a.Probe(ctx); !errors.Is(err, context.DeadlineExceeded) {
131 t.Fatalf("probe timeout err=%v, want context deadline", err)
132 }
133}
134
135func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
136 aPC, bPC := newPair()
137 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: 25 * time.Millisecond, DeadAfter: 75 * time.Millisecond})
138 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
139 defer func() { _ = a.Close() }()
140 defer func() { _ = b.Close() }()
141 a.mu.Lock()
142 a.health.lastPacket = time.Now().Add(-time.Second)
143 a.mu.Unlock()
144
145 _, err := recvErrWithin(t, a, time.Second)
146 if !errors.Is(err, ErrLinkDead) {
147 t.Fatalf("Recv err=%v, want ErrLinkDead", err)
148 }
149}
150
151func TestPendingAttemptsResetWhenLinkRecovers(t *testing.T) {
152 tr, _ := newResendTestTransport(t, 10)
153 tr.linkState = ports.LinkStateDegraded
154 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 4}
155 tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 7}
156
157 tr.setLinkState(ports.LinkStateConnected, nil)
158
159 for seq, p := range tr.pending {
160 if p.attempts != 0 {
161 t.Fatalf("pending[%d] attempts=%d, want reset on connected recovery", seq, p.attempts)
162 }
163 }
164}
165
166func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
167 aPC, bPC := newPair()
168 tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
169 if err != nil {
170 t.Fatal(err)
171 }
172 defer func() { _ = tr.Close() }()
173 defer func() { _ = bPC.Close() }()
174
175 tr.writeMu.Lock()
176 done := make(chan struct{})
177 go func() {
178 for id := uint64(1); id <= probeReplyBufferSize*4; id++ {
179 tr.handleRecord(probeRecord(recProbe, id))
180 }
181 close(done)
182 }()
183
184 select {
185 case <-done:
186 case <-time.After(time.Second):
187 tr.writeMu.Unlock()
188 <-done
189 t.Fatal("probe reply burst blocked behind writeMu")
190 }
191 tr.writeMu.Unlock()
192}
193
194func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
195 aPC, bPC := newPair()
196 aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
197 clk := newManualClock(time.Now())
198 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: time.Hour, Clock: clk})
199 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
200 defer func() { _ = a.Close() }()
201 defer func() { _ = b.Close() }()
202
203 if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
204 t.Fatal(err)
205 }
206 done := make(chan error, 2)
207 go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
208 go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("third")}) }()
209 waitForManualTimers(t, clk, 2)
210 a.setLinkState(ports.LinkStateDegraded, nil)
211 for i := range 2 {
212 select {
213 case err := <-done:
214 if !errors.Is(err, ErrPendingFull) {
215 t.Fatalf("send %d err=%v, want ErrPendingFull", i, err)
216 }
217 case <-time.After(time.Second):
218 t.Fatalf("send %d did not unblock after link degraded", i)
219 }
220 }
221}
222
223func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
224 aPC, bPC := newPair()
225 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: 4 * time.Second})
226 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
227 defer func() { _ = a.Close() }()
228 defer func() { _ = b.Close() }()
229
230 tests := []struct {
231 name string
232 age time.Duration
233 want ports.LinkState
234 }{
235 {"degraded", time.Second, ports.LinkStateDegraded},
236 {"probing", 2 * time.Second, ports.LinkStateProbing},
237 {"offline", 3 * time.Second, ports.LinkStateOffline},
238 }
239 for _, tt := range tests {
240 t.Run(tt.name, func(t *testing.T) {
241 a.mu.Lock()
242 staleAt := time.Now().Add(-tt.age)
243 a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
244 a.mu.Unlock()
245 a.checkSilence()
246 if got := a.LinkState(); got != tt.want {
247 t.Fatalf("LinkState()=%v, want %v", got, tt.want)
248 }
249 select {
250 case ev := <-a.LinkEvents():
251 if ev.State != tt.want {
252 t.Fatalf("event state=%v, want %v", ev.State, tt.want)
253 }
254 default:
255 t.Fatalf("missing link event for %v", tt.want)
256 }
257 })
258 }
259}
260
261func TestLinkHealthSeparatesPacketContactFromUsefulProgress(t *testing.T) {
262 now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
263 threshold := func(seconds int) time.Duration { return time.Duration(seconds) * time.Second }
264 tests := []struct {
265 name string
266 packetAge time.Duration
267 recordAge time.Duration
268 ackAge time.Duration
269 hasPending bool
270 wantState ports.LinkState
271 wantHop bool
272 wantDead bool
273 }{
274 {name: "fresh contact without useful progress degrades", packetAge: 0, recordAge: threshold(10), ackAge: threshold(10), wantState: ports.LinkStateDegraded},
275 {name: "pending round trip stall probes and hops", packetAge: 0, recordAge: threshold(20), ackAge: threshold(20), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
276 {name: "fragment-only contact never becomes offline", packetAge: 0, recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateDegraded},
277 {name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},
278 {name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},
279 {name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},
280 {name: "complete record cannot mask pending stall", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
281 {name: "complete record restores connected without pending data", packetAge: 0, recordAge: 0, ackAge: threshold(30), wantState: ports.LinkStateConnected},
282 {name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},
283 }
284
285 for _, tt := range tests {
286 t.Run(tt.name, func(t *testing.T) {
287 health := healthTracker{
288 lastPacket: now.Add(-tt.packetAge),
289 lastRecord: now.Add(-tt.recordAge),
290 lastProgress: now.Add(-tt.ackAge),
291 }
292 state, hop, dead := health.decide(now, tt.hasPending, threshold(10), threshold(20), threshold(30), threshold(60))
293 if state != tt.wantState || hop != tt.wantHop || dead != tt.wantDead {
294 t.Fatalf("decide()=(%v,%v,%v), want (%v,%v,%v)", state, hop, dead, tt.wantState, tt.wantHop, tt.wantDead)
295 }
296 })
297 }
298}
299
300func TestLinkHealthPendingStallWithFreshContactProbesOnceWithoutDying(t *testing.T) {
301 aPC, bPC := newPair()
302 var hops atomic.Int32
303 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
304 ResendAfter: time.Hour, Heartbeat: time.Hour, ProbeAfter: 20 * time.Second, DeadAfter: 60 * time.Second,
305 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
306 hops.Add(1)
307 return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
308 },
309 })
310 defer func() { _ = a.Close() }()
311 defer func() { _ = bPC.Close() }()
312
313 now := time.Now()
314 a.mu.Lock()
315 a.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
316 a.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, enqueued: now.Add(-time.Minute), attempts: 3}
317 a.mu.Unlock()
318
319 a.checkSilence()
320 a.checkSilence()
321 if got := a.LinkState(); got != ports.LinkStateProbing {
322 t.Fatalf("state=%v, want probing", got)
323 }
324 if got := hops.Load(); got != 1 {
325 t.Fatalf("socket hops=%d, want one", got)
326 }
327 select {
328 case <-a.done:
329 t.Fatal("fresh authenticated contact must prevent pending-stall death")
330 default:
331 }
332}
333
334func TestLinkHealthHeartbeatControlCannotMaskPendingACKStall(t *testing.T) {
335 aPC, bPC := newPair()
336 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
337 var hops atomic.Int32
338 tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
339 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
340 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
341 attempt := hops.Add(1)
342 return &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
343 },
344 })
345 if err != nil {
346 t.Fatal(err)
347 }
348 defer func() { _ = tr.Close() }()
349 defer func() { _ = bPC.Close() }()
350 if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("unacked")}); err != nil {
351 t.Fatal(err)
352 }
353
354 codec, err := pdgram.NewCodec(key())
355 if err != nil {
356 t.Fatal(err)
357 }
358 authenticated := make(chan struct{}, 32)
359 tr.mu.Lock()
360 tr.afterAuthenticatedPacket = func() { authenticated <- struct{}{} }
361 tr.mu.Unlock()
362 var counter uint64
363 sendHeartbeatPong := func() {
364 t.Helper()
365 counter++
366 frag := pdgram.Fragment{Seq: counter, Index: 0, Count: 1, Data: probeRecord(recPong, counter)}
367 raw, marshalErr := pdgram.MarshalFragment(frag)
368 if marshalErr != nil {
369 t.Fatal(marshalErr)
370 }
371 tr.mu.Lock()
372 actual := tr.pc
373 pc, ok := actual.(*fakePC)
374 tr.mu.Unlock()
375 if !ok {
376 t.Fatalf("transport packet conn = %T, want *fakePC", actual)
377 }
378 pc.in <- packet{b: codec.Seal(2, counter, raw, nil), addr: testAddr("b")}
379 select {
380 case <-authenticated:
381 case <-time.After(time.Second):
382 t.Fatal("heartbeat pong was not authenticated")
383 }
384 }
385
386 for elapsed := 3 * time.Second; elapsed <= 66*time.Second; elapsed += 3 * time.Second {
387 clk.advance(3 * time.Second)
388 sendHeartbeatPong()
389 eventually(t, time.Second, func() bool {
390 tr.mu.Lock()
391 defer tr.mu.Unlock()
392 return tr.health.lastRecord.Equal(clk.Now())
393 })
394 if elapsed >= 20*time.Second {
395 eventually(t, time.Second, func() bool {
396 tr.mu.Lock()
397 defer tr.mu.Unlock()
398 return tr.pc != aPC
399 })
400 }
401 }
402 if got := tr.LinkState(); got != ports.LinkStateProbing {
403 t.Fatalf("state=%v, want probing despite fresh heartbeat control traffic", got)
404 }
405 if got := hops.Load(); got != 1 {
406 t.Fatalf("socket hops=%d, want one pending round-trip recovery hop", got)
407 }
408 select {
409 case <-tr.done:
410 t.Fatal("fresh authenticated control contact must prevent offline/dead closure")
411 default:
412 }
413}
414
415func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {
416 tr, _ := newResendTestTransport(t, 10)
417 now := tr.clock.Now()
418 tr.linkState = ports.LinkStateDegraded
419 tr.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
420 tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: now, last: now}
421
422 tr.handleRecord(ackRecord(1))
423 if got := tr.LinkState(); got != ports.LinkStateConnected {
424 t.Fatalf("state after advancing ACK=%v, want connected", got)
425 }
426 tr.mu.Lock()
427 if tr.health.lastProgress.Before(now) {
428 tr.mu.Unlock()
429 t.Fatalf("last progress=%v, want at or after %v", tr.health.lastProgress, now)
430 }
431 tr.mu.Unlock()
432
433 tr.setLinkState(ports.LinkStateDegraded, nil)
434 tr.handleRecord(ackRecord(1))
435 if got := tr.LinkState(); got != ports.LinkStateDegraded {
436 t.Fatalf("duplicate ACK state=%v, want degraded", got)
437 }
438}
439
440func TestLinkHealthFreshProgressInvalidatesStaleDecision(t *testing.T) {
441 tests := []struct {
442 name string
443 staleAge time.Duration
444 freshen func(*Transport)
445 wantState ports.LinkState
446 }{
447 {
448 name: "fresh packet contact prevents stale dead close",
449 staleAge: time.Minute,
450 freshen: func(tr *Transport) {
451 tr.mu.Lock()
452 now := tr.clock.Now()
453 tr.health.authenticatedPacket(now)
454 tr.lastAuthenticatedPacket = now
455 tr.mu.Unlock()
456 },
457 wantState: ports.LinkStateDegraded,
458 },
459 {
460 name: "complete record prevents stale probing state and hop",
461 staleAge: 20 * time.Second,
462 freshen: func(tr *Transport) {
463 tr.mu.Lock()
464 now := tr.clock.Now()
465 tr.health.authenticatedPacket(now)
466 tr.health.completeRecord(now)
467 tr.lastAuthenticatedPacket = now
468 tr.lastCompleteRecord = now
469 tr.mu.Unlock()
470 tr.setLinkState(ports.LinkStateConnected, nil)
471 },
472 wantState: ports.LinkStateConnected,
473 },
474 }
475
476 for _, tt := range tests {
477 t.Run(tt.name, func(t *testing.T) {
478 aPC, bPC := newPair()
479 var hops atomic.Int32
480 tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
481 ResendAfter: time.Hour,
482 Heartbeat: time.Hour,
483 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
484 hops.Add(1)
485 return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
486 },
487 })
488 if err != nil {
489 t.Fatal(err)
490 }
491 defer func() { _ = tr.Close() }()
492 defer func() { _ = bPC.Close() }()
493
494 now := tr.clock.Now()
495 tr.mu.Lock()
496 staleAt := now.Add(-tt.staleAge)
497 tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
498 decisionReady := make(chan struct{})
499 resume := make(chan struct{})
500 tr.afterHealthDecision = func() {
501 close(decisionReady)
502 <-resume
503 }
504 tr.mu.Unlock()
505
506 done := make(chan struct{})
507 go func() {
508 tr.checkSilence()
509 close(done)
510 }()
511 select {
512 case <-decisionReady:
513 case <-time.After(time.Second):
514 t.Fatal("health decision hook was not reached")
515 }
516 tt.freshen(tr)
517 close(resume)
518 select {
519 case <-done:
520 case <-time.After(time.Second):
521 t.Fatal("health check did not finish")
522 }
523
524 if got := tr.LinkState(); got != tt.wantState {
525 t.Fatalf("state=%v, want %v after fresh health evidence", got, tt.wantState)
526 }
527 if got := hops.Load(); got != 0 {
528 t.Fatalf("socket hops=%d, want none after recovery", got)
529 }
530 for len(tr.linkEvents) > 0 {
531 ev := <-tr.linkEvents
532 if ev.State == ports.LinkStateDead || ev.State == ports.LinkStateProbing {
533 t.Fatalf("stale link event emitted after recovery: %v", ev.State)
534 }
535 }
536 select {
537 case <-tr.done:
538 t.Fatal("transport closed from stale health decision")
539 default:
540 }
541 })
542 }
543}
544
545func TestLinkHealthStaleRebindRestoresFutureHopAllowance(t *testing.T) {
546 aPC, bPC := newPair()
547 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
548 rebindStarted := make(chan struct{})
549 resumeRebind := make(chan struct{})
550 var rebinds atomic.Int32
551 tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
552 Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
553 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
554 attempt := rebinds.Add(1)
555 pc := &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
556 if attempt == 1 {
557 close(rebindStarted)
558 <-resumeRebind
559 }
560 return pc, nil
561 },
562 })
563 if err != nil {
564 t.Fatal(err)
565 }
566 defer func() { _ = tr.Close() }()
567 defer func() { _ = bPC.Close() }()
568
569 tr.mu.Lock()
570 staleAt := clk.Now().Add(-20 * time.Second)
571 tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
572 tr.mu.Unlock()
573 firstDone := make(chan struct{})
574 go func() {
575 tr.checkSilence()
576 close(firstDone)
577 }()
578 select {
579 case <-rebindStarted:
580 case <-time.After(time.Second):
581 t.Fatal("first rebind did not start")
582 }
583
584 tr.mu.Lock()
585 now := clk.Now()
586 tr.health.authenticatedPacket(now)
587 tr.lastAuthenticatedPacket = now
588 tr.mu.Unlock()
589 close(resumeRebind)
590 select {
591 case <-firstDone:
592 case <-time.After(time.Second):
593 t.Fatal("stale rebind did not finish")
594 }
595 tr.mu.Lock()
596 latched := tr.hoppedOffline
597 pcAfterStale := tr.pc
598 tr.mu.Unlock()
599 if latched {
600 t.Fatal("stale rebind consumed the future hop allowance")
601 }
602 if pcAfterStale != aPC {
603 t.Fatal("stale rebind replaced the active packet conn")
604 }
605
606 clk.advance(20 * time.Second)
607 tr.checkSilence()
608 tr.mu.Lock()
609 pcAfterRetry := tr.pc
610 tr.mu.Unlock()
611 if got := rebinds.Load(); got != 2 {
612 t.Fatalf("rebind attempts=%d, want a later eligible retry", got)
613 }
614 if pcAfterRetry == aPC {
615 t.Fatal("later stale interval did not hop the packet conn")
616 }
617}
618
619func TestLinkHealthStateEventsCannotPublishAfterNewerRecovery(t *testing.T) {
620 aPC, bPC := newPair()
621 tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
622 if err != nil {
623 t.Fatal(err)
624 }
625 defer func() { _ = tr.Close() }()
626 defer func() { _ = bPC.Close() }()
627
628 probingCommitted := make(chan struct{})
629 resumeProbing := make(chan struct{})
630 tr.mu.Lock()
631 tr.linkState = ports.LinkStateDegraded
632 tr.afterLinkStateCommit = func(state ports.LinkState) {
633 if state != ports.LinkStateProbing {
634 return
635 }
636 close(probingCommitted)
637 <-resumeProbing
638 }
639 tr.mu.Unlock()
640
641 probingDone := make(chan struct{})
642 go func() {
643 tr.setLinkState(ports.LinkStateProbing, nil)
644 close(probingDone)
645 }()
646 select {
647 case <-probingCommitted:
648 case <-time.After(time.Second):
649 t.Fatal("probing commit hook was not reached")
650 }
651 tr.setLinkState(ports.LinkStateConnected, nil)
652 close(resumeProbing)
653 select {
654 case <-probingDone:
655 case <-time.After(time.Second):
656 t.Fatal("probing publisher did not finish")
657 }
658
659 if got := tr.LinkState(); got != ports.LinkStateConnected {
660 t.Fatalf("state=%v, want connected", got)
661 }
662 select {
663 case ev := <-tr.LinkEvents():
664 if ev.State != ports.LinkStateConnected {
665 t.Fatalf("first event=%v, want connected", ev.State)
666 }
667 default:
668 t.Fatal("missing connected recovery event")
669 }
670 select {
671 case ev := <-tr.LinkEvents():
672 t.Fatalf("stale event published after recovery: %v", ev.State)
673 default:
674 }
675}
676
677func TestHopFiresAtProbingNotDegraded(t *testing.T) {
678 aPC, bPC := newPair()
679 var hopped atomic.Bool
680 aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
681 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
682 DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: time.Hour,
683 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
684 newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
685 bPC.peers["a2"] = newPC
686 hopped.Store(true)
687 return newPC, nil
688 },
689 })
690 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
691 defer func() { _ = a.Close() }()
692 defer func() { _ = b.Close() }()
693
694 a.mu.Lock()
695 staleAt := time.Now().Add(-time.Second)
696 a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
697 a.mu.Unlock()
698 a.checkSilence()
699 if hopped.Load() {
700 t.Fatal("socket hopped at degraded silence")
701 }
702
703 a.mu.Lock()
704 a.health.lastPacket = time.Now().Add(-2 * time.Second)
705 a.mu.Unlock()
706 a.checkSilence()
707 if !hopped.Load() {
708 t.Fatal("socket did not hop at probing silence")
709 }
710}
711
712func TestShortSilenceDoesNotKillTransport(t *testing.T) {
713 aPC, bPC := newPair()
714 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour})
715 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
716 defer func() { _ = a.Close() }()
717 defer func() { _ = b.Close() }()
718 a.mu.Lock()
719 staleAt := time.Now().Add(-time.Second)
720 a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
721 a.mu.Unlock()
722 a.checkSilence()
723 if got := a.LinkState(); got != ports.LinkStateDegraded {
724 t.Fatalf("LinkState()=%v, want degraded", got)
725 }
726 select {
727 case <-a.done:
728 t.Fatal("transport closed before DeadAfter")
729 default:
730 }
731}