fix(dgram): detect send-stall dead links and hop earlier
Add a per-frame first-send timestamp and a checkSendStall pass on the resend/heartbeat ticks: when the oldest unacked reliable frame stays unacked for deadAfter despite a retransmit, declare the link dead. This catches one-way-dead links whose receive side stays fresh from peer heartbeats, which the receive-silence check misses. Also move the one-shot socket hop from the Offline (30s) transition to Probing (20s) so the client rebind runs before the link is abandoned.
- commit
- ce08837a24ef18b9482346ea5374e5514c55fa3a
- author
- brice · · 2026-07-08 16:51 UTC
- parent
- 91b82aa
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +137 -10
M internal/adapters/dgram/transport.go +42 -2
| @@ -177,6 +177,7 @@ type Transport struct { | |||
| 177 | 177 | ||
| 178 | 178 | type pending struct { | |
| 179 | 179 | frame ports.Frame | |
| 180 | + | first time.Time | |
| 180 | 181 | last time.Time | |
| 181 | 182 | attempts int | |
| 182 | 183 | } | |
| @@ -275,7 +276,8 @@ func (t *Transport) Send(f ports.Frame) error { | |||
| 275 | 276 | t.seq++ | |
| 276 | 277 | seq := t.seq | |
| 277 | 278 | if reliable { | |
| 278 | - | t.pending[seq] = &pending{frame: f, last: t.clock.Now()} | |
| 279 | + | now := t.clock.Now() | |
| 280 | + | t.pending[seq] = &pending{frame: f, first: now, last: now} | |
| 279 | 281 | } | |
| 280 | 282 | t.mu.Unlock() | |
| 281 | 283 | if err := t.sendData(seq, reliable, f); err != nil { | |
| @@ -624,10 +626,12 @@ func (t *Transport) resendLoop() { | |||
| 624 | 626 | case <-resend.C(): | |
| 625 | 627 | t.resendPending() | |
| 626 | 628 | t.checkSilence() | |
| 629 | + | t.checkSendStall() | |
| 627 | 630 | resend.Reset(t.resendAfter) | |
| 628 | 631 | case <-heartbeat.C(): | |
| 629 | 632 | _ = t.sendProbe(recProbe, 0) | |
| 630 | 633 | t.checkSilence() | |
| 634 | + | t.checkSendStall() | |
| 631 | 635 | heartbeat.Reset(t.heartbeat) | |
| 632 | 636 | case <-t.done: | |
| 633 | 637 | return | |
| @@ -732,14 +736,50 @@ func (t *Transport) checkSilence() { | |||
| 732 | 736 | t.closeWithError(ErrLinkDead) | |
| 733 | 737 | case offlineAfter > 0 && silentFor >= offlineAfter: | |
| 734 | 738 | t.setLinkState(ports.LinkStateOffline, nil) | |
| 735 | - | t.hopPacketConnOnce() | |
| 736 | 739 | case probeAfter > 0 && silentFor >= probeAfter: | |
| 737 | 740 | t.setLinkState(ports.LinkStateProbing, nil) | |
| 741 | + | t.hopPacketConnOnce() | |
| 738 | 742 | case degradedAfter > 0 && silentFor >= degradedAfter: | |
| 739 | 743 | t.setLinkState(ports.LinkStateDegraded, nil) | |
| 740 | 744 | } | |
| 741 | 745 | } | |
| 742 | 746 | ||
| 747 | + | // checkSendStall declares the link dead when the oldest unacked reliable frame | |
| 748 | + | // has gone unacknowledged for deadAfter despite at least one retransmit. The | |
| 749 | + | // receive-silence check in checkSilence misses one-way-dead links whose receive | |
| 750 | + | // side stays fresh from peer heartbeats; this catches them at the layer that | |
| 751 | + | // owns death. first is set once when a frame is enqueued and never refreshed, | |
| 752 | + | // so acks (which delete pending entries) keep the oldest age bounded on a | |
| 753 | + | // progressing link — only a frame that is never acked crosses the threshold. | |
| 754 | + | func (t *Transport) checkSendStall() { | |
| 755 | + | t.mu.Lock() | |
| 756 | + | if t.closed || t.deadAfter <= 0 { | |
| 757 | + | t.mu.Unlock() | |
| 758 | + | return | |
| 759 | + | } | |
| 760 | + | deadAfter := t.deadAfter | |
| 761 | + | now := t.clock.Now() | |
| 762 | + | var oldest time.Time | |
| 763 | + | var attempts int | |
| 764 | + | found := false | |
| 765 | + | for _, p := range t.pending { | |
| 766 | + | if p == nil || p.first.IsZero() { | |
| 767 | + | continue | |
| 768 | + | } | |
| 769 | + | if !found || p.first.Before(oldest) { | |
| 770 | + | oldest = p.first | |
| 771 | + | attempts = p.attempts | |
| 772 | + | found = true | |
| 773 | + | } | |
| 774 | + | } | |
| 775 | + | t.mu.Unlock() | |
| 776 | + | if !found || attempts < 1 || now.Sub(oldest) < deadAfter { | |
| 777 | + | return | |
| 778 | + | } | |
| 779 | + | t.setLinkState(ports.LinkStateDead, ErrLinkDead) | |
| 780 | + | t.closeWithError(ErrLinkDead) | |
| 781 | + | } | |
| 782 | + | ||
| 743 | 783 | func (t *Transport) setLinkState(state ports.LinkState, err error) { | |
| 744 | 784 | now := t.clock.Now() | |
| 745 | 785 | t.mu.Lock() | |
| @@ -177,6 +177,7 @@ type Transport struct { | |||||
| 177 | 177 | ||||
| 178 | type pending struct { | 178 | type pending struct { | ||
| 179 | frame ports.Frame | 179 | frame ports.Frame | ||
| 180 | + | first time.Time | |||
| 180 | last time.Time | 181 | last time.Time | ||
| 181 | attempts int | 182 | attempts int | ||
| 182 | } | 183 | } | ||
| @@ -275,7 +276,8 @@ func (t *Transport) Send(f ports.Frame) error { | |||||
| 275 | t.seq++ | 276 | t.seq++ | ||
| 276 | seq := t.seq | 277 | seq := t.seq | ||
| 277 | if reliable { | 278 | if reliable { | ||
| 278 | - | t.pending[seq] = &pending{frame: f, last: t.clock.Now()} | 279 | + | now := t.clock.Now() |
| 280 | + | t.pending[seq] = &pending{frame: f, first: now, last: now} | |||
| 279 | } | 281 | } | ||
| 280 | t.mu.Unlock() | 282 | t.mu.Unlock() | ||
| 281 | if err := t.sendData(seq, reliable, f); err != nil { | 283 | if err := t.sendData(seq, reliable, f); err != nil { | ||
| @@ -624,10 +626,12 @@ func (t *Transport) resendLoop() { | |||||
| 624 | case <-resend.C(): | 626 | case <-resend.C(): | ||
| 625 | t.resendPending() | 627 | t.resendPending() | ||
| 626 | t.checkSilence() | 628 | t.checkSilence() | ||
| 629 | + | t.checkSendStall() | |||
| 627 | resend.Reset(t.resendAfter) | 630 | resend.Reset(t.resendAfter) | ||
| 628 | case <-heartbeat.C(): | 631 | case <-heartbeat.C(): | ||
| 629 | _ = t.sendProbe(recProbe, 0) | 632 | _ = t.sendProbe(recProbe, 0) | ||
| 630 | t.checkSilence() | 633 | t.checkSilence() | ||
| 634 | + | t.checkSendStall() | |||
| 631 | heartbeat.Reset(t.heartbeat) | 635 | heartbeat.Reset(t.heartbeat) | ||
| 632 | case <-t.done: | 636 | case <-t.done: | ||
| 633 | return | 637 | return | ||
| @@ -732,14 +736,50 @@ func (t *Transport) checkSilence() { | |||||
| 732 | t.closeWithError(ErrLinkDead) | 736 | t.closeWithError(ErrLinkDead) | ||
| 733 | case offlineAfter > 0 && silentFor >= offlineAfter: | 737 | case offlineAfter > 0 && silentFor >= offlineAfter: | ||
| 734 | t.setLinkState(ports.LinkStateOffline, nil) | 738 | t.setLinkState(ports.LinkStateOffline, nil) | ||
| 735 | - | t.hopPacketConnOnce() | |||
| 736 | case probeAfter > 0 && silentFor >= probeAfter: | 739 | case probeAfter > 0 && silentFor >= probeAfter: | ||
| 737 | t.setLinkState(ports.LinkStateProbing, nil) | 740 | t.setLinkState(ports.LinkStateProbing, nil) | ||
| 741 | + | t.hopPacketConnOnce() | |||
| 738 | case degradedAfter > 0 && silentFor >= degradedAfter: | 742 | case degradedAfter > 0 && silentFor >= degradedAfter: | ||
| 739 | t.setLinkState(ports.LinkStateDegraded, nil) | 743 | t.setLinkState(ports.LinkStateDegraded, nil) | ||
| 740 | } | 744 | } | ||
| 741 | } | 745 | } | ||
| 742 | 746 | ||||
| 747 | + | // checkSendStall declares the link dead when the oldest unacked reliable frame | |||
| 748 | + | // has gone unacknowledged for deadAfter despite at least one retransmit. The | |||
| 749 | + | // receive-silence check in checkSilence misses one-way-dead links whose receive | |||
| 750 | + | // side stays fresh from peer heartbeats; this catches them at the layer that | |||
| 751 | + | // owns death. first is set once when a frame is enqueued and never refreshed, | |||
| 752 | + | // so acks (which delete pending entries) keep the oldest age bounded on a | |||
| 753 | + | // progressing link — only a frame that is never acked crosses the threshold. | |||
| 754 | + | func (t *Transport) checkSendStall() { | |||
| 755 | + | t.mu.Lock() | |||
| 756 | + | if t.closed || t.deadAfter <= 0 { | |||
| 757 | + | t.mu.Unlock() | |||
| 758 | + | return | |||
| 759 | + | } | |||
| 760 | + | deadAfter := t.deadAfter | |||
| 761 | + | now := t.clock.Now() | |||
| 762 | + | var oldest time.Time | |||
| 763 | + | var attempts int | |||
| 764 | + | found := false | |||
| 765 | + | for _, p := range t.pending { | |||
| 766 | + | if p == nil || p.first.IsZero() { | |||
| 767 | + | continue | |||
| 768 | + | } | |||
| 769 | + | if !found || p.first.Before(oldest) { | |||
| 770 | + | oldest = p.first | |||
| 771 | + | attempts = p.attempts | |||
| 772 | + | found = true | |||
| 773 | + | } | |||
| 774 | + | } | |||
| 775 | + | t.mu.Unlock() | |||
| 776 | + | if !found || attempts < 1 || now.Sub(oldest) < deadAfter { | |||
| 777 | + | return | |||
| 778 | + | } | |||
| 779 | + | t.setLinkState(ports.LinkStateDead, ErrLinkDead) | |||
| 780 | + | t.closeWithError(ErrLinkDead) | |||
| 781 | + | } | |||
| 782 | + | ||||
| 743 | func (t *Transport) setLinkState(state ports.LinkState, err error) { | 783 | func (t *Transport) setLinkState(state ports.LinkState, err error) { | ||
| 744 | now := t.clock.Now() | 784 | now := t.clock.Now() | ||
| 745 | t.mu.Lock() | 785 | t.mu.Lock() | ||
M internal/adapters/dgram/transport_test.go +95 -8
| @@ -408,9 +408,9 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) { | |||
| 408 | 408 | var hopped atomic.Bool | |
| 409 | 409 | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | |
| 410 | 410 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |
| 411 | - | ResendAfter: 20 * time.Millisecond, | |
| 412 | - | OfflineAfter: time.Millisecond, | |
| 413 | - | DeadAfter: time.Hour, | |
| 411 | + | ResendAfter: 20 * time.Millisecond, | |
| 412 | + | ProbeAfter: time.Millisecond, | |
| 413 | + | DeadAfter: time.Hour, | |
| 414 | 414 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | |
| 415 | 415 | newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}} | |
| 416 | 416 | bPC.peers["a2"] = newPC | |
| @@ -448,10 +448,10 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) { | |||
| 448 | 448 | aPC, bPC := newPair() | |
| 449 | 449 | var attempts atomic.Int32 | |
| 450 | 450 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |
| 451 | - | ResendAfter: time.Hour, | |
| 452 | - | Heartbeat: time.Hour, | |
| 453 | - | OfflineAfter: time.Millisecond, | |
| 454 | - | DeadAfter: time.Hour, | |
| 451 | + | ResendAfter: time.Hour, | |
| 452 | + | Heartbeat: time.Hour, | |
| 453 | + | ProbeAfter: time.Millisecond, | |
| 454 | + | DeadAfter: time.Hour, | |
| 455 | 455 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | |
| 456 | 456 | if attempts.Add(1) == 1 { | |
| 457 | 457 | return nil, errors.New("temporary rebind failure") | |
| @@ -487,7 +487,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) { | |||
| 487 | 487 | ||
| 488 | 488 | func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) { | |
| 489 | 489 | aPC, bPC := newPair() | |
| 490 | - | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{OfflineAfter: time.Millisecond, DeadAfter: time.Hour}) | |
| 490 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ProbeAfter: time.Millisecond, DeadAfter: time.Hour}) | |
| 491 | 491 | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | |
| 492 | 492 | defer func() { _ = a.Close() }() | |
| 493 | 493 | defer func() { _ = b.Close() }() | |
| @@ -1005,6 +1005,93 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) { | |||
| 1005 | 1005 | } | |
| 1006 | 1006 | } | |
| 1007 | 1007 | ||
| 1008 | + | func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) { | |
| 1009 | + | tests := []struct { | |
| 1010 | + | name string | |
| 1011 | + | age time.Duration | |
| 1012 | + | attempts int | |
| 1013 | + | wantDead bool | |
| 1014 | + | }{ | |
| 1015 | + | {name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false}, | |
| 1016 | + | {name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false}, | |
| 1017 | + | {name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true}, | |
| 1018 | + | } | |
| 1019 | + | for _, tt := range tests { | |
| 1020 | + | t.Run(tt.name, func(t *testing.T) { | |
| 1021 | + | aPC, bPC := newPair() | |
| 1022 | + | // Drop everything a sends so the frame never gets acked, and keep the | |
| 1023 | + | // receive side fresh so only the send-stall check can declare death. | |
| 1024 | + | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | |
| 1025 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |
| 1026 | + | ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second, | |
| 1027 | + | }) | |
| 1028 | + | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | |
| 1029 | + | defer func() { _ = a.Close() }() | |
| 1030 | + | defer func() { _ = b.Close() }() | |
| 1031 | + | ||
| 1032 | + | now := time.Now() | |
| 1033 | + | a.mu.Lock() | |
| 1034 | + | a.lastHeard = now | |
| 1035 | + | a.pending[1] = &pending{ | |
| 1036 | + | frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")}, | |
| 1037 | + | first: now.Add(-tt.age), | |
| 1038 | + | last: now, | |
| 1039 | + | attempts: tt.attempts, | |
| 1040 | + | } | |
| 1041 | + | a.mu.Unlock() | |
| 1042 | + | ||
| 1043 | + | a.checkSendStall() | |
| 1044 | + | if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead { | |
| 1045 | + | t.Fatalf("dead=%v, want %v", dead, tt.wantDead) | |
| 1046 | + | } | |
| 1047 | + | select { | |
| 1048 | + | case <-a.done: | |
| 1049 | + | if !tt.wantDead { | |
| 1050 | + | t.Fatal("transport closed without a send stall") | |
| 1051 | + | } | |
| 1052 | + | default: | |
| 1053 | + | if tt.wantDead { | |
| 1054 | + | t.Fatal("transport not closed after send stall") | |
| 1055 | + | } | |
| 1056 | + | } | |
| 1057 | + | }) | |
| 1058 | + | } | |
| 1059 | + | } | |
| 1060 | + | ||
| 1061 | + | func TestHopFiresAtProbingNotDegraded(t *testing.T) { | |
| 1062 | + | aPC, bPC := newPair() | |
| 1063 | + | var hopped atomic.Bool | |
| 1064 | + | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | |
| 1065 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |
| 1066 | + | DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: time.Hour, | |
| 1067 | + | RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) { | |
| 1068 | + | newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}} | |
| 1069 | + | bPC.peers["a2"] = newPC | |
| 1070 | + | hopped.Store(true) | |
| 1071 | + | return newPC, nil | |
| 1072 | + | }, | |
| 1073 | + | }) | |
| 1074 | + | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | |
| 1075 | + | defer func() { _ = a.Close() }() | |
| 1076 | + | defer func() { _ = b.Close() }() | |
| 1077 | + | ||
| 1078 | + | a.mu.Lock() | |
| 1079 | + | a.lastHeard = time.Now().Add(-time.Second) | |
| 1080 | + | a.mu.Unlock() | |
| 1081 | + | a.checkSilence() | |
| 1082 | + | if hopped.Load() { | |
| 1083 | + | t.Fatal("socket hopped at degraded silence") | |
| 1084 | + | } | |
| 1085 | + | ||
| 1086 | + | a.mu.Lock() | |
| 1087 | + | a.lastHeard = time.Now().Add(-2 * time.Second) | |
| 1088 | + | a.mu.Unlock() | |
| 1089 | + | a.checkSilence() | |
| 1090 | + | if !hopped.Load() { | |
| 1091 | + | t.Fatal("socket did not hop at probing silence") | |
| 1092 | + | } | |
| 1093 | + | } | |
| 1094 | + | ||
| 1008 | 1095 | func TestShortSilenceDoesNotKillTransport(t *testing.T) { | |
| 1009 | 1096 | aPC, bPC := newPair() | |
| 1010 | 1097 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour}) | |
| @@ -408,9 +408,9 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) { | |||||
| 408 | var hopped atomic.Bool | 408 | var hopped atomic.Bool | ||
| 409 | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | 409 | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | ||
| 410 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | 410 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | ||
| 411 | - | ResendAfter: 20 * time.Millisecond, | 411 | + | ResendAfter: 20 * time.Millisecond, |
| 412 | - | OfflineAfter: time.Millisecond, | 412 | + | ProbeAfter: time.Millisecond, |
| 413 | - | DeadAfter: time.Hour, | 413 | + | DeadAfter: time.Hour, |
| 414 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | 414 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | ||
| 415 | newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}} | 415 | newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}} | ||
| 416 | bPC.peers["a2"] = newPC | 416 | bPC.peers["a2"] = newPC | ||
| @@ -448,10 +448,10 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) { | |||||
| 448 | aPC, bPC := newPair() | 448 | aPC, bPC := newPair() | ||
| 449 | var attempts atomic.Int32 | 449 | var attempts atomic.Int32 | ||
| 450 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | 450 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | ||
| 451 | - | ResendAfter: time.Hour, | 451 | + | ResendAfter: time.Hour, |
| 452 | - | Heartbeat: time.Hour, | 452 | + | Heartbeat: time.Hour, |
| 453 | - | OfflineAfter: time.Millisecond, | 453 | + | ProbeAfter: time.Millisecond, |
| 454 | - | DeadAfter: time.Hour, | 454 | + | DeadAfter: time.Hour, |
| 455 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | 455 | RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) { | ||
| 456 | if attempts.Add(1) == 1 { | 456 | if attempts.Add(1) == 1 { | ||
| 457 | return nil, errors.New("temporary rebind failure") | 457 | return nil, errors.New("temporary rebind failure") | ||
| @@ -487,7 +487,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) { | |||||
| 487 | 487 | ||||
| 488 | func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) { | 488 | func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) { | ||
| 489 | aPC, bPC := newPair() | 489 | aPC, bPC := newPair() | ||
| 490 | - | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{OfflineAfter: time.Millisecond, DeadAfter: time.Hour}) | 490 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ProbeAfter: time.Millisecond, DeadAfter: time.Hour}) |
| 491 | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | 491 | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | ||
| 492 | defer func() { _ = a.Close() }() | 492 | defer func() { _ = a.Close() }() | ||
| 493 | defer func() { _ = b.Close() }() | 493 | defer func() { _ = b.Close() }() | ||
| @@ -1005,6 +1005,93 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) { | |||||
| 1005 | } | 1005 | } | ||
| 1006 | } | 1006 | } | ||
| 1007 | 1007 | ||||
| 1008 | + | func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) { | |||
| 1009 | + | tests := []struct { | |||
| 1010 | + | name string | |||
| 1011 | + | age time.Duration | |||
| 1012 | + | attempts int | |||
| 1013 | + | wantDead bool | |||
| 1014 | + | }{ | |||
| 1015 | + | {name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false}, | |||
| 1016 | + | {name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false}, | |||
| 1017 | + | {name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true}, | |||
| 1018 | + | } | |||
| 1019 | + | for _, tt := range tests { | |||
| 1020 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 1021 | + | aPC, bPC := newPair() | |||
| 1022 | + | // Drop everything a sends so the frame never gets acked, and keep the | |||
| 1023 | + | // receive side fresh so only the send-stall check can declare death. | |||
| 1024 | + | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | |||
| 1025 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |||
| 1026 | + | ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second, | |||
| 1027 | + | }) | |||
| 1028 | + | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | |||
| 1029 | + | defer func() { _ = a.Close() }() | |||
| 1030 | + | defer func() { _ = b.Close() }() | |||
| 1031 | + | ||||
| 1032 | + | now := time.Now() | |||
| 1033 | + | a.mu.Lock() | |||
| 1034 | + | a.lastHeard = now | |||
| 1035 | + | a.pending[1] = &pending{ | |||
| 1036 | + | frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")}, | |||
| 1037 | + | first: now.Add(-tt.age), | |||
| 1038 | + | last: now, | |||
| 1039 | + | attempts: tt.attempts, | |||
| 1040 | + | } | |||
| 1041 | + | a.mu.Unlock() | |||
| 1042 | + | ||||
| 1043 | + | a.checkSendStall() | |||
| 1044 | + | if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead { | |||
| 1045 | + | t.Fatalf("dead=%v, want %v", dead, tt.wantDead) | |||
| 1046 | + | } | |||
| 1047 | + | select { | |||
| 1048 | + | case <-a.done: | |||
| 1049 | + | if !tt.wantDead { | |||
| 1050 | + | t.Fatal("transport closed without a send stall") | |||
| 1051 | + | } | |||
| 1052 | + | default: | |||
| 1053 | + | if tt.wantDead { | |||
| 1054 | + | t.Fatal("transport not closed after send stall") | |||
| 1055 | + | } | |||
| 1056 | + | } | |||
| 1057 | + | }) | |||
| 1058 | + | } | |||
| 1059 | + | } | |||
| 1060 | + | ||||
| 1061 | + | func TestHopFiresAtProbingNotDegraded(t *testing.T) { | |||
| 1062 | + | aPC, bPC := newPair() | |||
| 1063 | + | var hopped atomic.Bool | |||
| 1064 | + | aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" } | |||
| 1065 | + | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ | |||
| 1066 | + | DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: time.Hour, | |||
| 1067 | + | RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) { | |||
| 1068 | + | newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}} | |||
| 1069 | + | bPC.peers["a2"] = newPC | |||
| 1070 | + | hopped.Store(true) | |||
| 1071 | + | return newPC, nil | |||
| 1072 | + | }, | |||
| 1073 | + | }) | |||
| 1074 | + | b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1) | |||
| 1075 | + | defer func() { _ = a.Close() }() | |||
| 1076 | + | defer func() { _ = b.Close() }() | |||
| 1077 | + | ||||
| 1078 | + | a.mu.Lock() | |||
| 1079 | + | a.lastHeard = time.Now().Add(-time.Second) | |||
| 1080 | + | a.mu.Unlock() | |||
| 1081 | + | a.checkSilence() | |||
| 1082 | + | if hopped.Load() { | |||
| 1083 | + | t.Fatal("socket hopped at degraded silence") | |||
| 1084 | + | } | |||
| 1085 | + | ||||
| 1086 | + | a.mu.Lock() | |||
| 1087 | + | a.lastHeard = time.Now().Add(-2 * time.Second) | |||
| 1088 | + | a.mu.Unlock() | |||
| 1089 | + | a.checkSilence() | |||
| 1090 | + | if !hopped.Load() { | |||
| 1091 | + | t.Fatal("socket did not hop at probing silence") | |||
| 1092 | + | } | |||
| 1093 | + | } | |||
| 1094 | + | ||||
| 1008 | func TestShortSilenceDoesNotKillTransport(t *testing.T) { | 1095 | func TestShortSilenceDoesNotKillTransport(t *testing.T) { | ||
| 1009 | aPC, bPC := newPair() | 1096 | aPC, bPC := newPair() | ||
| 1010 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour}) | 1097 | a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour}) | ||