fix(dgram): serialize UDP health recovery

commit: 68a2ea203d920106d62233328797fdef18d46731
author: brice · 3mo ago · 2026-07-10 06:35 UTC
parent: aeaca5d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +211 -47

M internal/adapters/dgram/transport.go +78 -47


@@ -167,41 +167,47 @@ type Transport struct {
167	167		// afterHealthDecision is a test synchronization hook. It runs after a health
168	168		// snapshot is taken and before that decision is conditionally committed.
169	169		afterHealthDecision func()
170		-	observe DiagnosticObserver
171		-	diagnosticCh chan Diagnostic
172		-	heartbeat time.Duration
173		-	resendAfter time.Duration
174		-	maxResendAfter time.Duration
175		-	maxResendPerTick int
176		-	srtt time.Duration
177		-	rttvar time.Duration
178		-	rto time.Duration
179		-	writeTimeout time.Duration
180		-	degradedAfter time.Duration
181		-	probeAfter time.Duration
182		-	offlineAfter time.Duration
183		-	deadAfter time.Duration
184		-	maxPending int
185		-	maxPendingWait time.Duration
186		-	maxRecvBuffer int
187		-	clock ports.Clock
188		-	linkState ports.LinkState
189		-	linkEvents chan ports.LinkEvent
190		-	probeWait map[uint64]chan struct{}
191		-	sendWake chan struct{}
192		-	control chan controlRecord
193		-	ackWake chan struct{}
194		-	ackSend chan uint64
195		-	controlMu sync.Mutex
196		-	ackSeq uint64
197		-	ackQueued bool
198		-	rebind func(net.PacketConn) (net.PacketConn, error)
199		-	hoppedOffline bool
200		-	outputQueue []queuedSend
201		-	outputWake chan struct{}
202		-	outputNext time.Time
203		-	retransmitWork chan []retransmitRecord
204		-	writeDeadlines *writeDeadlineState
	170	+	// afterLinkStateCommit is a test synchronization hook. It runs after state
	171	+	// mutation and before ordered event publication.
	172	+	afterLinkStateCommit func(ports.LinkState)
	173	+	observe DiagnosticObserver
	174	+	diagnosticCh chan Diagnostic
	175	+	heartbeat time.Duration
	176	+	resendAfter time.Duration
	177	+	maxResendAfter time.Duration
	178	+	maxResendPerTick int
	179	+	srtt time.Duration
	180	+	rttvar time.Duration
	181	+	rto time.Duration
	182	+	writeTimeout time.Duration
	183	+	degradedAfter time.Duration
	184	+	probeAfter time.Duration
	185	+	offlineAfter time.Duration
	186	+	deadAfter time.Duration
	187	+	maxPending int
	188	+	maxPendingWait time.Duration
	189	+	maxRecvBuffer int
	190	+	clock ports.Clock
	191	+	linkState ports.LinkState
	192	+	linkStateGeneration uint64
	193	+	linkEvents chan ports.LinkEvent
	194	+	linkEventMu sync.Mutex
	195	+	probeWait map[uint64]chan struct{}
	196	+	sendWake chan struct{}
	197	+	control chan controlRecord
	198	+	ackWake chan struct{}
	199	+	ackSend chan uint64
	200	+	controlMu sync.Mutex
	201	+	ackSeq uint64
	202	+	ackQueued bool
	203	+	rebind func(net.PacketConn) (net.PacketConn, error)
	204	+	hoppedOffline bool
	205	+	hopGeneration uint64
	206	+	outputQueue []queuedSend
	207	+	outputWake chan struct{}
	208	+	outputNext time.Time
	209	+	retransmitWork chan []retransmitRecord
	210	+	writeDeadlines *writeDeadlineState
205	211
206	212		recvMu sync.Mutex
207	213		replay *pdgram.ReplayWindow

@@ -1189,18 +1195,18 @@ func (t *Transport) hopPacketConnOnce(generation uint64) {
1189	1195		old := t.pc
1190	1196		rebind := t.rebind
1191	1197		t.hoppedOffline = true
	1198	+	t.hopGeneration = generation
1192	1199		t.mu.Unlock()
1193	1200		pc, err := rebind(old)
1194	1201		if err != nil {
1195	1202		t.mu.Lock()
1196		-	if t.pc == old && t.health.generation == generation && t.linkState == ports.LinkStateProbing {
1197		-	t.hoppedOffline = false
1198		-	}
	1203	+	t.rollbackHopLocked(generation)
1199	1204		t.mu.Unlock()
1200	1205		return
1201	1206		}
1202	1207		t.mu.Lock()
1203	1208		if t.closed \|\| t.pc != old \|\| t.health.generation != generation \|\| t.linkState != ports.LinkStateProbing {
	1209	+	t.rollbackHopLocked(generation)
1204	1210		t.mu.Unlock()
1205	1211		_ = pc.Close()
1206	1212		return

@@ -1212,6 +1218,13 @@ func (t *Transport) hopPacketConnOnce(generation uint64) {
1212	1218		_ = old.Close()
1213	1219		}
1214	1220
	1221	+	func (t *Transport) rollbackHopLocked(generation uint64) {
	1222	+	if t.hoppedOffline && t.hopGeneration == generation {
	1223	+	t.hoppedOffline = false
	1224	+	t.hopGeneration = 0
	1225	+	}
	1226	+	}
	1227	+
1215	1228		func (t *Transport) checkSilence() {
1216	1229		runHook := true
1217	1230		for {

@@ -1242,7 +1255,8 @@ func (t *Transport) checkSilence() {
1242	1255		t.mu.Unlock()
1243	1256		continue
1244	1257		}
1245		-	changed := t.setLinkStateLocked(state)
	1258	+	changed, stateGeneration := t.setLinkStateLocked(state)
	1259	+	afterStateCommit := t.afterLinkStateCommit
1246	1260		closed := false
1247	1261		if dead {
1248	1262		closed = t.closeWithErrorLocked(ErrLinkDead)

@@ -1250,11 +1264,14 @@ func (t *Transport) checkSilence() {
1250	1264		t.mu.Unlock()
1251	1265
1252	1266		if changed {
	1267	+	if afterStateCommit != nil {
	1268	+	afterStateCommit(state)
	1269	+	}
1253	1270		var linkErr error
1254	1271		if dead {
1255	1272		linkErr = ErrLinkDead
1256	1273		}
1257		-	t.publishLinkState(state, now, linkErr)
	1274	+	t.publishLinkState(state, now, linkErr, stateGeneration)
1258	1275		}
1259	1276		if closed {
1260	1277		t.broadcastClosed()

@@ -1270,20 +1287,26 @@ func (t *Transport) checkSilence() {
1270	1287		func (t *Transport) setLinkState(state ports.LinkState, err error) {
1271	1288		now := t.clock.Now()
1272	1289		t.mu.Lock()
1273		-	changed := t.setLinkStateLocked(state)
	1290	+	changed, stateGeneration := t.setLinkStateLocked(state)
	1291	+	afterStateCommit := t.afterLinkStateCommit
1274	1292		t.mu.Unlock()
1275	1293		if changed {
1276		-	t.publishLinkState(state, now, err)
	1294	+	if afterStateCommit != nil {
	1295	+	afterStateCommit(state)
	1296	+	}
	1297	+	t.publishLinkState(state, now, err, stateGeneration)
1277	1298		}
1278	1299		}
1279	1300
1280		-	func (t *Transport) setLinkStateLocked(state ports.LinkState) bool {
	1301	+	func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
1281	1302		if t.linkState == state {
1282		-	return false
	1303	+	return false, t.linkStateGeneration
1283	1304		}
1284	1305		t.linkState = state
	1306	+	t.linkStateGeneration++
1285	1307		if state == ports.LinkStateConnected {
1286	1308		t.hoppedOffline = false
	1309	+	t.hopGeneration = 0
1287	1310		for _, p := range t.pending {
1288	1311		if p != nil {
1289	1312		p.attempts = 0

@@ -1291,16 +1314,24 @@ func (t *Transport) setLinkStateLocked(state ports.LinkState) bool {
1291	1314		}
1292	1315		}
1293	1316		t.notifySendWaitersLocked()
1294		-	return true
	1317	+	return true, t.linkStateGeneration
1295	1318		}
1296	1319
1297		-	func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error) {
1298		-	t.emitDiagnostic()
	1320	+	func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
	1321	+	t.linkEventMu.Lock()
	1322	+	defer t.linkEventMu.Unlock()
	1323	+	t.mu.Lock()
	1324	+	if t.linkState != state \|\| t.linkStateGeneration != generation {
	1325	+	t.mu.Unlock()
	1326	+	return
	1327	+	}
1299	1328		event := ports.LinkEvent{State: state, At: now, Err: err}
1300	1329		select {
1301	1330		case t.linkEvents <- event:
1302	1331		default:
1303	1332		}
	1333	+	t.mu.Unlock()
	1334	+	t.emitDiagnostic()
1304	1335		}
1305	1336
1306	1337		func (t *Transport) diagnosticLoop() {

 	// afterHealthDecision is a test synchronization hook. It runs after a health
 	// snapshot is taken and before that decision is conditionally committed.
 	afterHealthDecision func()
-	observe             DiagnosticObserver
+	// afterLinkStateCommit is a test synchronization hook. It runs after state
-	diagnosticCh        chan Diagnostic
+	// mutation and before ordered event publication.
-	heartbeat           time.Duration
+	afterLinkStateCommit func(ports.LinkState)
-	resendAfter         time.Duration
+	observe              DiagnosticObserver
-	maxResendAfter      time.Duration
+	diagnosticCh         chan Diagnostic
-	maxResendPerTick    int
+	heartbeat            time.Duration
-	srtt                time.Duration
+	resendAfter          time.Duration
-	rttvar              time.Duration
+	maxResendAfter       time.Duration
-	rto                 time.Duration
+	maxResendPerTick     int
-	writeTimeout        time.Duration
+	srtt                 time.Duration
-	degradedAfter       time.Duration
+	rttvar               time.Duration
-	probeAfter          time.Duration
+	rto                  time.Duration
-	offlineAfter        time.Duration
+	writeTimeout         time.Duration
-	deadAfter           time.Duration
+	degradedAfter        time.Duration
-	maxPending          int
+	probeAfter           time.Duration
-	maxPendingWait      time.Duration
+	offlineAfter         time.Duration
-	maxRecvBuffer       int
+	deadAfter            time.Duration
-	clock               ports.Clock
+	maxPending           int
-	linkState           ports.LinkState
+	maxPendingWait       time.Duration
-	linkEvents          chan ports.LinkEvent
+	maxRecvBuffer        int
-	probeWait           map[uint64]chan struct{}
+	clock                ports.Clock
-	sendWake            chan struct{}
+	linkState            ports.LinkState
-	control             chan controlRecord
+	linkStateGeneration  uint64
-	ackWake             chan struct{}
+	linkEvents           chan ports.LinkEvent
-	ackSend             chan uint64
+	linkEventMu          sync.Mutex
-	controlMu           sync.Mutex
+	probeWait            map[uint64]chan struct{}
-	ackSeq              uint64
+	sendWake             chan struct{}
-	ackQueued           bool
+	control              chan controlRecord
-	rebind              func(net.PacketConn) (net.PacketConn, error)
+	ackWake              chan struct{}
-	hoppedOffline       bool
+	ackSend              chan uint64
-	outputQueue         []queuedSend
+	controlMu            sync.Mutex
-	outputWake          chan struct{}
+	ackSeq               uint64
-	outputNext          time.Time
+	ackQueued            bool
-	retransmitWork      chan []retransmitRecord
+	rebind               func(net.PacketConn) (net.PacketConn, error)
-	writeDeadlines      *writeDeadlineState
+	hoppedOffline        bool
+	hopGeneration        uint64
+	outputQueue          []queuedSend
+	outputWake           chan struct{}
+	outputNext           time.Time
+	retransmitWork       chan []retransmitRecord
+	writeDeadlines       *writeDeadlineState
 	recvMu sync.Mutex
 	replay *pdgram.ReplayWindow
 	old := t.pc
 	rebind := t.rebind
 	t.hoppedOffline = true
+	t.hopGeneration = generation
 	t.mu.Unlock()
 	pc, err := rebind(old)
 	if err != nil {
 		t.mu.Lock()
-		if t.pc == old && t.health.generation == generation && t.linkState == ports.LinkStateProbing {
+		t.rollbackHopLocked(generation)
-			t.hoppedOffline = false
-		}
 		t.mu.Unlock()
 		return
 	}
 	t.mu.Lock()
 	if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
+		t.rollbackHopLocked(generation)
 		t.mu.Unlock()
 		_ = pc.Close()
 		return
 	_ = old.Close()
 }
+func (t *Transport) rollbackHopLocked(generation uint64) {
+	if t.hoppedOffline && t.hopGeneration == generation {
+		t.hoppedOffline = false
+		t.hopGeneration = 0
+	}
+}
 func (t *Transport) checkSilence() {
 	runHook := true
 	for {
 			t.mu.Unlock()
 			continue
 		}
-		changed := t.setLinkStateLocked(state)
+		changed, stateGeneration := t.setLinkStateLocked(state)
+		afterStateCommit := t.afterLinkStateCommit
 		closed := false
 		if dead {
 			closed = t.closeWithErrorLocked(ErrLinkDead)
 		t.mu.Unlock()
 		if changed {
+			if afterStateCommit != nil {
+				afterStateCommit(state)
+			}
 			var linkErr error
 			if dead {
 				linkErr = ErrLinkDead
 			}
-			t.publishLinkState(state, now, linkErr)
+			t.publishLinkState(state, now, linkErr, stateGeneration)
 		}
 		if closed {
 			t.broadcastClosed()
 func (t *Transport) setLinkState(state ports.LinkState, err error) {
 	now := t.clock.Now()
 	t.mu.Lock()
-	changed := t.setLinkStateLocked(state)
+	changed, stateGeneration := t.setLinkStateLocked(state)
+	afterStateCommit := t.afterLinkStateCommit
 	t.mu.Unlock()
 	if changed {
-		t.publishLinkState(state, now, err)
+		if afterStateCommit != nil {
+			afterStateCommit(state)
+		}
+		t.publishLinkState(state, now, err, stateGeneration)
 	}
 }
-func (t *Transport) setLinkStateLocked(state ports.LinkState) bool {
+func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
 	if t.linkState == state {
-		return false
+		return false, t.linkStateGeneration
 	}
 	t.linkState = state
+	t.linkStateGeneration++
 	if state == ports.LinkStateConnected {
 		t.hoppedOffline = false
+		t.hopGeneration = 0
 		for _, p := range t.pending {
 			if p != nil {
 				p.attempts = 0
 		}
 	}
 	t.notifySendWaitersLocked()
-	return true
+	return true, t.linkStateGeneration
 }
-func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error) {
+func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
-	t.emitDiagnostic()
+	t.linkEventMu.Lock()
+	defer t.linkEventMu.Unlock()
+	t.mu.Lock()
+	if t.linkState != state || t.linkStateGeneration != generation {
+		t.mu.Unlock()
+		return
+	}
 	event := ports.LinkEvent{State: state, At: now, Err: err}
 	select {
 	case t.linkEvents <- event:
 	default:
 	}
+	t.mu.Unlock()
+	t.emitDiagnostic()
 }
 func (t *Transport) diagnosticLoop() {

M internal/adapters/dgram/transport_test.go +133 -0


@@ -5,6 +5,7 @@ import (
5	5		"context"
6	6		"encoding/binary"
7	7		"errors"
	8	+	"fmt"
8	9		"net"
9	10		"sync"
10	11		"sync/atomic"

@@ -2025,6 +2026,138 @@ func TestLinkHealthFreshProgressInvalidatesStaleDecision(t *testing.T) {
2025	2026		}
2026	2027		}
2027	2028
	2029	+	func TestLinkHealthStaleRebindRestoresFutureHopAllowance(t *testing.T) {
	2030	+	aPC, bPC := newPair()
	2031	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	2032	+	rebindStarted := make(chan struct{})
	2033	+	resumeRebind := make(chan struct{})
	2034	+	var rebinds atomic.Int32
	2035	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	2036	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	2037	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	2038	+	attempt := rebinds.Add(1)
	2039	+	pc := &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
	2040	+	if attempt == 1 {
	2041	+	close(rebindStarted)
	2042	+	<-resumeRebind
	2043	+	}
	2044	+	return pc, nil
	2045	+	},
	2046	+	})
	2047	+	if err != nil {
	2048	+	t.Fatal(err)
	2049	+	}
	2050	+	defer func() { _ = tr.Close() }()
	2051	+	defer func() { _ = bPC.Close() }()
	2052	+
	2053	+	tr.mu.Lock()
	2054	+	staleAt := clk.Now().Add(-20 * time.Second)
	2055	+	tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
	2056	+	tr.mu.Unlock()
	2057	+	firstDone := make(chan struct{})
	2058	+	go func() {
	2059	+	tr.checkSilence()
	2060	+	close(firstDone)
	2061	+	}()
	2062	+	select {
	2063	+	case <-rebindStarted:
	2064	+	case <-time.After(time.Second):
	2065	+	t.Fatal("first rebind did not start")
	2066	+	}
	2067	+
	2068	+	tr.mu.Lock()
	2069	+	now := clk.Now()
	2070	+	tr.health.authenticatedPacket(now)
	2071	+	tr.lastAuthenticatedPacket = now
	2072	+	tr.mu.Unlock()
	2073	+	close(resumeRebind)
	2074	+	select {
	2075	+	case <-firstDone:
	2076	+	case <-time.After(time.Second):
	2077	+	t.Fatal("stale rebind did not finish")
	2078	+	}
	2079	+	tr.mu.Lock()
	2080	+	latched := tr.hoppedOffline
	2081	+	pcAfterStale := tr.pc
	2082	+	tr.mu.Unlock()
	2083	+	if latched {
	2084	+	t.Fatal("stale rebind consumed the future hop allowance")
	2085	+	}
	2086	+	if pcAfterStale != aPC {
	2087	+	t.Fatal("stale rebind replaced the active packet conn")
	2088	+	}
	2089	+
	2090	+	clk.advance(20 * time.Second)
	2091	+	tr.checkSilence()
	2092	+	tr.mu.Lock()
	2093	+	pcAfterRetry := tr.pc
	2094	+	tr.mu.Unlock()
	2095	+	if got := rebinds.Load(); got != 2 {
	2096	+	t.Fatalf("rebind attempts=%d, want a later eligible retry", got)
	2097	+	}
	2098	+	if pcAfterRetry == aPC {
	2099	+	t.Fatal("later stale interval did not hop the packet conn")
	2100	+	}
	2101	+	}
	2102	+
	2103	+	func TestLinkHealthStateEventsCannotPublishAfterNewerRecovery(t *testing.T) {
	2104	+	aPC, bPC := newPair()
	2105	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
	2106	+	if err != nil {
	2107	+	t.Fatal(err)
	2108	+	}
	2109	+	defer func() { _ = tr.Close() }()
	2110	+	defer func() { _ = bPC.Close() }()
	2111	+
	2112	+	probingCommitted := make(chan struct{})
	2113	+	resumeProbing := make(chan struct{})
	2114	+	tr.mu.Lock()
	2115	+	tr.linkState = ports.LinkStateDegraded
	2116	+	tr.afterLinkStateCommit = func(state ports.LinkState) {
	2117	+	if state != ports.LinkStateProbing {
	2118	+	return
	2119	+	}
	2120	+	close(probingCommitted)
	2121	+	<-resumeProbing
	2122	+	}
	2123	+	tr.mu.Unlock()
	2124	+
	2125	+	probingDone := make(chan struct{})
	2126	+	go func() {
	2127	+	tr.setLinkState(ports.LinkStateProbing, nil)
	2128	+	close(probingDone)
	2129	+	}()
	2130	+	select {
	2131	+	case <-probingCommitted:
	2132	+	case <-time.After(time.Second):
	2133	+	t.Fatal("probing commit hook was not reached")
	2134	+	}
	2135	+	tr.setLinkState(ports.LinkStateConnected, nil)
	2136	+	close(resumeProbing)
	2137	+	select {
	2138	+	case <-probingDone:
	2139	+	case <-time.After(time.Second):
	2140	+	t.Fatal("probing publisher did not finish")
	2141	+	}
	2142	+
	2143	+	if got := tr.LinkState(); got != ports.LinkStateConnected {
	2144	+	t.Fatalf("state=%v, want connected", got)
	2145	+	}
	2146	+	select {
	2147	+	case ev := <-tr.LinkEvents():
	2148	+	if ev.State != ports.LinkStateConnected {
	2149	+	t.Fatalf("first event=%v, want connected", ev.State)
	2150	+	}
	2151	+	default:
	2152	+	t.Fatal("missing connected recovery event")
	2153	+	}
	2154	+	select {
	2155	+	case ev := <-tr.LinkEvents():
	2156	+	t.Fatalf("stale event published after recovery: %v", ev.State)
	2157	+	default:
	2158	+	}
	2159	+	}
	2160	+
2028	2161		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
2029	2162		aPC, bPC := newPair()
2030	2163		var hopped atomic.Bool


@@ -5,6 +5,7 @@ import (
5	"context"	5		"context"
6	"encoding/binary"	6		"encoding/binary"
7	"errors"	7		"errors"
		8	+	"fmt"
8	"net"	9		"net"
9	"sync"	10		"sync"
10	"sync/atomic"	11		"sync/atomic"

@@ -2025,6 +2026,138 @@ func TestLinkHealthFreshProgressInvalidatesStaleDecision(t *testing.T) {
2025	}	2026		}
2026	}	2027		}
2027		2028
		2029	+	func TestLinkHealthStaleRebindRestoresFutureHopAllowance(t *testing.T) {
		2030	+	aPC, bPC := newPair()
		2031	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
		2032	+	rebindStarted := make(chan struct{})
		2033	+	resumeRebind := make(chan struct{})
		2034	+	var rebinds atomic.Int32
		2035	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
		2036	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
		2037	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
		2038	+	attempt := rebinds.Add(1)
		2039	+	pc := &fakePC{addr: testAddr(fmt.Sprintf("a%d", attempt+1)), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
		2040	+	if attempt == 1 {
		2041	+	close(rebindStarted)
		2042	+	<-resumeRebind
		2043	+	}
		2044	+	return pc, nil
		2045	+	},
		2046	+	})
		2047	+	if err != nil {
		2048	+	t.Fatal(err)
		2049	+	}
		2050	+	defer func() { _ = tr.Close() }()
		2051	+	defer func() { _ = bPC.Close() }()
		2052	+
		2053	+	tr.mu.Lock()
		2054	+	staleAt := clk.Now().Add(-20 * time.Second)
		2055	+	tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
		2056	+	tr.mu.Unlock()
		2057	+	firstDone := make(chan struct{})
		2058	+	go func() {
		2059	+	tr.checkSilence()
		2060	+	close(firstDone)
		2061	+	}()
		2062	+	select {
		2063	+	case <-rebindStarted:
		2064	+	case <-time.After(time.Second):
		2065	+	t.Fatal("first rebind did not start")
		2066	+	}
		2067	+
		2068	+	tr.mu.Lock()
		2069	+	now := clk.Now()
		2070	+	tr.health.authenticatedPacket(now)
		2071	+	tr.lastAuthenticatedPacket = now
		2072	+	tr.mu.Unlock()
		2073	+	close(resumeRebind)
		2074	+	select {
		2075	+	case <-firstDone:
		2076	+	case <-time.After(time.Second):
		2077	+	t.Fatal("stale rebind did not finish")
		2078	+	}
		2079	+	tr.mu.Lock()
		2080	+	latched := tr.hoppedOffline
		2081	+	pcAfterStale := tr.pc
		2082	+	tr.mu.Unlock()
		2083	+	if latched {
		2084	+	t.Fatal("stale rebind consumed the future hop allowance")
		2085	+	}
		2086	+	if pcAfterStale != aPC {
		2087	+	t.Fatal("stale rebind replaced the active packet conn")
		2088	+	}
		2089	+
		2090	+	clk.advance(20 * time.Second)
		2091	+	tr.checkSilence()
		2092	+	tr.mu.Lock()
		2093	+	pcAfterRetry := tr.pc
		2094	+	tr.mu.Unlock()
		2095	+	if got := rebinds.Load(); got != 2 {
		2096	+	t.Fatalf("rebind attempts=%d, want a later eligible retry", got)
		2097	+	}
		2098	+	if pcAfterRetry == aPC {
		2099	+	t.Fatal("later stale interval did not hop the packet conn")
		2100	+	}
		2101	+	}
		2102	+
		2103	+	func TestLinkHealthStateEventsCannotPublishAfterNewerRecovery(t *testing.T) {
		2104	+	aPC, bPC := newPair()
		2105	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
		2106	+	if err != nil {
		2107	+	t.Fatal(err)
		2108	+	}
		2109	+	defer func() { _ = tr.Close() }()
		2110	+	defer func() { _ = bPC.Close() }()
		2111	+
		2112	+	probingCommitted := make(chan struct{})
		2113	+	resumeProbing := make(chan struct{})
		2114	+	tr.mu.Lock()
		2115	+	tr.linkState = ports.LinkStateDegraded
		2116	+	tr.afterLinkStateCommit = func(state ports.LinkState) {
		2117	+	if state != ports.LinkStateProbing {
		2118	+	return
		2119	+	}
		2120	+	close(probingCommitted)
		2121	+	<-resumeProbing
		2122	+	}
		2123	+	tr.mu.Unlock()
		2124	+
		2125	+	probingDone := make(chan struct{})
		2126	+	go func() {
		2127	+	tr.setLinkState(ports.LinkStateProbing, nil)
		2128	+	close(probingDone)
		2129	+	}()
		2130	+	select {
		2131	+	case <-probingCommitted:
		2132	+	case <-time.After(time.Second):
		2133	+	t.Fatal("probing commit hook was not reached")
		2134	+	}
		2135	+	tr.setLinkState(ports.LinkStateConnected, nil)
		2136	+	close(resumeProbing)
		2137	+	select {
		2138	+	case <-probingDone:
		2139	+	case <-time.After(time.Second):
		2140	+	t.Fatal("probing publisher did not finish")
		2141	+	}
		2142	+
		2143	+	if got := tr.LinkState(); got != ports.LinkStateConnected {
		2144	+	t.Fatalf("state=%v, want connected", got)
		2145	+	}
		2146	+	select {
		2147	+	case ev := <-tr.LinkEvents():
		2148	+	if ev.State != ports.LinkStateConnected {
		2149	+	t.Fatalf("first event=%v, want connected", ev.State)
		2150	+	}
		2151	+	default:
		2152	+	t.Fatal("missing connected recovery event")
		2153	+	}
		2154	+	select {
		2155	+	case ev := <-tr.LinkEvents():
		2156	+	t.Fatalf("stale event published after recovery: %v", ev.State)
		2157	+	default:
		2158	+	}
		2159	+	}
		2160	+
2028	func TestHopFiresAtProbingNotDegraded(t *testing.T) {	2161		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
2029	aPC, bPC := newPair()	2162		aPC, bPC := newPair()
2030	var hopped atomic.Bool	2163		var hopped atomic.Bool