fix(dgram): revalidate UDP health decisions

commit: aeaca5d1e33799e2dff90f95574dd5cfe60535ca
author: brice · 3mo ago · 2026-07-10 06:29 UTC
parent: 246a824
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +236 -63

M internal/adapters/dgram/health.go +13 -3


@@ -12,17 +12,27 @@ type healthTracker struct {
12	12		lastPacket time.Time
13	13		lastRecord time.Time
14	14		lastProgress time.Time
	15	+	generation uint64
15	16		}
16	17
17	18		func newHealthTracker(now time.Time) healthTracker {
18	19		return healthTracker{lastPacket: now, lastRecord: now, lastProgress: now}
19	20		}
20	21
21		-	func (h *healthTracker) authenticatedPacket(now time.Time) { h.lastPacket = now }
	22	+	func (h *healthTracker) authenticatedPacket(now time.Time) {
	23	+	h.lastPacket = now
	24	+	h.generation++
	25	+	}
22	26
23		-	func (h *healthTracker) completeRecord(now time.Time) { h.lastRecord = now }
	27	+	func (h *healthTracker) completeRecord(now time.Time) {
	28	+	h.lastRecord = now
	29	+	h.generation++
	30	+	}
24	31
25		-	func (h *healthTracker) ackProgress(now time.Time) { h.lastProgress = now }
	32	+	func (h *healthTracker) ackProgress(now time.Time) {
	33	+	h.lastProgress = now
	34	+	h.generation++
	35	+	}
26	36
27	37		func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
28	38		packetAge := now.Sub(h.lastPacket)

 	lastPacket   time.Time
 	lastRecord   time.Time
 	lastProgress time.Time
+	generation   uint64
 }
 func newHealthTracker(now time.Time) healthTracker {
 	return healthTracker{lastPacket: now, lastRecord: now, lastProgress: now}
 }
-func (h *healthTracker) authenticatedPacket(now time.Time) { h.lastPacket = now }
+func (h *healthTracker) authenticatedPacket(now time.Time) {
+	h.lastPacket = now
+	h.generation++
+}
-func (h *healthTracker) completeRecord(now time.Time) { h.lastRecord = now }
+func (h *healthTracker) completeRecord(now time.Time) {
+	h.lastRecord = now
+	h.generation++
+}
-func (h *healthTracker) ackProgress(now time.Time) { h.lastProgress = now }
+func (h *healthTracker) ackProgress(now time.Time) {
+	h.lastProgress = now
+	h.generation++
+}
 func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
 	packetAge := now.Sub(h.lastPacket)

M internal/adapters/dgram/transport.go +118 -60


@@ -164,41 +164,44 @@ type Transport struct {
164	164		// afterMalformedFragment is a test synchronization hook. It runs after an
165	165		// authenticated datagram has been rejected as an invalid fragment.
166	166		afterMalformedFragment func()
167		-	observe DiagnosticObserver
168		-	diagnosticCh chan Diagnostic
169		-	heartbeat time.Duration
170		-	resendAfter time.Duration
171		-	maxResendAfter time.Duration
172		-	maxResendPerTick int
173		-	srtt time.Duration
174		-	rttvar time.Duration
175		-	rto time.Duration
176		-	writeTimeout time.Duration
177		-	degradedAfter time.Duration
178		-	probeAfter time.Duration
179		-	offlineAfter time.Duration
180		-	deadAfter time.Duration
181		-	maxPending int
182		-	maxPendingWait time.Duration
183		-	maxRecvBuffer int
184		-	clock ports.Clock
185		-	linkState ports.LinkState
186		-	linkEvents chan ports.LinkEvent
187		-	probeWait map[uint64]chan struct{}
188		-	sendWake chan struct{}
189		-	control chan controlRecord
190		-	ackWake chan struct{}
191		-	ackSend chan uint64
192		-	controlMu sync.Mutex
193		-	ackSeq uint64
194		-	ackQueued bool
195		-	rebind func(net.PacketConn) (net.PacketConn, error)
196		-	hoppedOffline bool
197		-	outputQueue []queuedSend
198		-	outputWake chan struct{}
199		-	outputNext time.Time
200		-	retransmitWork chan []retransmitRecord
201		-	writeDeadlines *writeDeadlineState
	167	+	// afterHealthDecision is a test synchronization hook. It runs after a health
	168	+	// snapshot is taken and before that decision is conditionally committed.
	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
202	205
203	206		recvMu sync.Mutex
204	207		replay *pdgram.ReplayWindow

@@ -1177,9 +1180,9 @@ func (t *Transport) updateRTTLocked(sample time.Duration) {
1177	1180		t.rto = rto
1178	1181		}
1179	1182
1180		-	func (t *Transport) hopPacketConnOnce() {
	1183	+	func (t *Transport) hopPacketConnOnce(generation uint64) {
1181	1184		t.mu.Lock()
1182		-	if t.rebind == nil \|\| t.hoppedOffline \|\| t.closed {
	1185	+	if t.rebind == nil \|\| t.hoppedOffline \|\| t.closed \|\| t.linkState != ports.LinkStateProbing \|\| t.health.generation != generation {
1183	1186		t.mu.Unlock()
1184	1187		return
1185	1188		}

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

@@ -1210,31 +1213,73 @@ func (t *Transport) hopPacketConnOnce() {
1210	1213		}
1211	1214
1212	1215		func (t *Transport) checkSilence() {
1213		-	t.mu.Lock()
1214		-	closed := t.closed
1215		-	now := t.clock.Now()
1216		-	state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
1217		-	t.mu.Unlock()
1218		-	if closed {
1219		-	return
1220		-	}
1221		-	if dead {
1222		-	t.setLinkState(state, ErrLinkDead)
1223		-	t.closeWithError(ErrLinkDead)
	1216	+	runHook := true
	1217	+	for {
	1218	+	t.mu.Lock()
	1219	+	if t.closed {
	1220	+	t.mu.Unlock()
	1221	+	return
	1222	+	}
	1223	+	now := t.clock.Now()
	1224	+	generation := t.health.generation
	1225	+	state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
	1226	+	afterDecision := t.afterHealthDecision
	1227	+	t.mu.Unlock()
	1228	+
	1229	+	if runHook {
	1230	+	runHook = false
	1231	+	if afterDecision != nil {
	1232	+	afterDecision()
	1233	+	}
	1234	+	}
	1235	+
	1236	+	t.mu.Lock()
	1237	+	if t.closed {
	1238	+	t.mu.Unlock()
	1239	+	return
	1240	+	}
	1241	+	if t.health.generation != generation {
	1242	+	t.mu.Unlock()
	1243	+	continue
	1244	+	}
	1245	+	changed := t.setLinkStateLocked(state)
	1246	+	closed := false
	1247	+	if dead {
	1248	+	closed = t.closeWithErrorLocked(ErrLinkDead)
	1249	+	}
	1250	+	t.mu.Unlock()
	1251	+
	1252	+	if changed {
	1253	+	var linkErr error
	1254	+	if dead {
	1255	+	linkErr = ErrLinkDead
	1256	+	}
	1257	+	t.publishLinkState(state, now, linkErr)
	1258	+	}
	1259	+	if closed {
	1260	+	t.broadcastClosed()
	1261	+	return
	1262	+	}
	1263	+	if hop {
	1264	+	t.hopPacketConnOnce(generation)
	1265	+	}
1224	1266		return
1225	1267		}
1226		-	t.setLinkState(state, nil)
1227		-	if hop {
1228		-	t.hopPacketConnOnce()
1229		-	}
1230	1268		}
1231	1269
1232	1270		func (t *Transport) setLinkState(state ports.LinkState, err error) {
1233	1271		now := t.clock.Now()
1234	1272		t.mu.Lock()
	1273	+	changed := t.setLinkStateLocked(state)
	1274	+	t.mu.Unlock()
	1275	+	if changed {
	1276	+	t.publishLinkState(state, now, err)
	1277	+	}
	1278	+	}
	1279	+
	1280	+	func (t *Transport) setLinkStateLocked(state ports.LinkState) bool {
1235	1281		if t.linkState == state {
1236		-	t.mu.Unlock()
1237		-	return
	1282	+	return false
1238	1283		}
1239	1284		t.linkState = state
1240	1285		if state == ports.LinkStateConnected {

@@ -1246,7 +1291,10 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
1246	1291		}
1247	1292		}
1248	1293		t.notifySendWaitersLocked()
1249		-	t.mu.Unlock()
	1294	+	return true
	1295	+	}
	1296	+
	1297	+	func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error) {
1250	1298		t.emitDiagnostic()
1251	1299		event := ports.LinkEvent{State: state, At: now, Err: err}
1252	1300		select {

@@ -1317,9 +1365,16 @@ func diagnosticAge(now, then time.Time) time.Duration {
1317	1365
1318	1366		func (t *Transport) closeWithError(err error) {
1319	1367		t.mu.Lock()
	1368	+	closed := t.closeWithErrorLocked(err)
	1369	+	t.mu.Unlock()
	1370	+	if closed {
	1371	+	t.broadcastClosed()
	1372	+	}
	1373	+	}
	1374	+
	1375	+	func (t *Transport) closeWithErrorLocked(err error) bool {
1320	1376		if t.closed {
1321		-	t.mu.Unlock()
1322		-	return
	1377	+	return false
1323	1378		}
1324	1379		t.closed = true
1325	1380		t.closeErr = err

@@ -1333,7 +1388,10 @@ func (t *Transport) closeWithError(err error) {
1333	1388		t.probeWait = make(map[uint64]chan struct{})
1334	1389		close(t.done)
1335	1390		t.notifySendWaitersLocked()
1336		-	t.mu.Unlock()
	1391	+	return true
	1392	+	}
	1393	+
	1394	+	func (t *Transport) broadcastClosed() {
1337	1395		t.deliverMu.Lock()
1338	1396		t.deliverCond.Broadcast()
1339	1397		t.deliverMu.Unlock()

 	// afterMalformedFragment is a test synchronization hook. It runs after an
 	// authenticated datagram has been rejected as an invalid fragment.
 	afterMalformedFragment func()
-	observe                DiagnosticObserver
+	// afterHealthDecision is a test synchronization hook. It runs after a health
-	diagnosticCh           chan Diagnostic
+	// snapshot is taken and before that decision is conditionally committed.
-	heartbeat              time.Duration
+	afterHealthDecision func()
-	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
+	linkEvents          chan ports.LinkEvent
-	ackWake                chan struct{}
+	probeWait           map[uint64]chan struct{}
-	ackSend                chan uint64
+	sendWake            chan struct{}
-	controlMu              sync.Mutex
+	control             chan controlRecord
-	ackSeq                 uint64
+	ackWake             chan struct{}
-	ackQueued              bool
+	ackSend             chan uint64
-	rebind                 func(net.PacketConn) (net.PacketConn, error)
+	controlMu           sync.Mutex
-	hoppedOffline          bool
+	ackSeq              uint64
-	outputQueue            []queuedSend
+	ackQueued           bool
-	outputWake             chan struct{}
+	rebind              func(net.PacketConn) (net.PacketConn, error)
-	outputNext             time.Time
+	hoppedOffline       bool
-	retransmitWork         chan []retransmitRecord
+	outputQueue         []queuedSend
-	writeDeadlines         *writeDeadlineState
+	outputWake          chan struct{}
+	outputNext          time.Time
+	retransmitWork      chan []retransmitRecord
+	writeDeadlines      *writeDeadlineState
 	recvMu sync.Mutex
 	replay *pdgram.ReplayWindow
 	t.rto = rto
 }
-func (t *Transport) hopPacketConnOnce() {
+func (t *Transport) hopPacketConnOnce(generation uint64) {
 	t.mu.Lock()
-	if t.rebind == nil || t.hoppedOffline || t.closed {
+	if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
 		t.mu.Unlock()
 		return
 	}
 	pc, err := rebind(old)
 	if err != nil {
 		t.mu.Lock()
-		if t.pc == old {
+		if t.pc == old && t.health.generation == generation && t.linkState == ports.LinkStateProbing {
 			t.hoppedOffline = false
 		}
 		t.mu.Unlock()
 		return
 	}
 	t.mu.Lock()
-	if t.closed || t.pc != old {
+	if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
 		t.mu.Unlock()
 		_ = pc.Close()
 		return
 }
 func (t *Transport) checkSilence() {
-	t.mu.Lock()
+	runHook := true
-	closed := t.closed
+	for {
-	now := t.clock.Now()
+		t.mu.Lock()
-	state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
+		if t.closed {
-	t.mu.Unlock()
+			t.mu.Unlock()
-	if closed {
+			return
-		return
+		}
-	}
+		now := t.clock.Now()
-	if dead {
+		generation := t.health.generation
-		t.setLinkState(state, ErrLinkDead)
+		state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
-		t.closeWithError(ErrLinkDead)
+		afterDecision := t.afterHealthDecision
+		t.mu.Unlock()
+		if runHook {
+			runHook = false
+			if afterDecision != nil {
+				afterDecision()
+			}
+		}
+		t.mu.Lock()
+		if t.closed {
+			t.mu.Unlock()
+			return
+		}
+		if t.health.generation != generation {
+			t.mu.Unlock()
+			continue
+		}
+		changed := t.setLinkStateLocked(state)
+		closed := false
+		if dead {
+			closed = t.closeWithErrorLocked(ErrLinkDead)
+		}
+		t.mu.Unlock()
+		if changed {
+			var linkErr error
+			if dead {
+				linkErr = ErrLinkDead
+			}
+			t.publishLinkState(state, now, linkErr)
+		}
+		if closed {
+			t.broadcastClosed()
+			return
+		}
+		if hop {
+			t.hopPacketConnOnce(generation)
+		}
 		return
 	}
-	t.setLinkState(state, nil)
-	if hop {
-		t.hopPacketConnOnce()
-	}
 }
 func (t *Transport) setLinkState(state ports.LinkState, err error) {
 	now := t.clock.Now()
 	t.mu.Lock()
+	changed := t.setLinkStateLocked(state)
+	t.mu.Unlock()
+	if changed {
+		t.publishLinkState(state, now, err)
+	}
+}
+func (t *Transport) setLinkStateLocked(state ports.LinkState) bool {
 	if t.linkState == state {
-		t.mu.Unlock()
+		return false
-		return
 	}
 	t.linkState = state
 	if state == ports.LinkStateConnected {
 		}
 	}
 	t.notifySendWaitersLocked()
-	t.mu.Unlock()
+	return true
+}
+func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error) {
 	t.emitDiagnostic()
 	event := ports.LinkEvent{State: state, At: now, Err: err}
 	select {
 func (t *Transport) closeWithError(err error) {
 	t.mu.Lock()
+	closed := t.closeWithErrorLocked(err)
+	t.mu.Unlock()
+	if closed {
+		t.broadcastClosed()
+	}
+}
+func (t *Transport) closeWithErrorLocked(err error) bool {
 	if t.closed {
-		t.mu.Unlock()
+		return false
-		return
 	}
 	t.closed = true
 	t.closeErr = err
 	t.probeWait = make(map[uint64]chan struct{})
 	close(t.done)
 	t.notifySendWaitersLocked()
-	t.mu.Unlock()
+	return true
+}
+func (t *Transport) broadcastClosed() {
 	t.deliverMu.Lock()
 	t.deliverCond.Broadcast()
 	t.deliverMu.Unlock()

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


@@ -1920,6 +1920,111 @@ func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing
1920	1920		}
1921	1921		}
1922	1922
	1923	+	func TestLinkHealthFreshProgressInvalidatesStaleDecision(t *testing.T) {
	1924	+	tests := []struct {
	1925	+	name string
	1926	+	staleAge time.Duration
	1927	+	freshen func(*Transport)
	1928	+	wantState ports.LinkState
	1929	+	}{
	1930	+	{
	1931	+	name: "fresh packet contact prevents stale dead close",
	1932	+	staleAge: time.Minute,
	1933	+	freshen: func(tr *Transport) {
	1934	+	tr.mu.Lock()
	1935	+	now := tr.clock.Now()
	1936	+	tr.health.authenticatedPacket(now)
	1937	+	tr.lastAuthenticatedPacket = now
	1938	+	tr.mu.Unlock()
	1939	+	},
	1940	+	wantState: ports.LinkStateDegraded,
	1941	+	},
	1942	+	{
	1943	+	name: "complete record prevents stale probing state and hop",
	1944	+	staleAge: 20 * time.Second,
	1945	+	freshen: func(tr *Transport) {
	1946	+	tr.mu.Lock()
	1947	+	now := tr.clock.Now()
	1948	+	tr.health.authenticatedPacket(now)
	1949	+	tr.health.completeRecord(now)
	1950	+	tr.lastAuthenticatedPacket = now
	1951	+	tr.lastCompleteRecord = now
	1952	+	tr.mu.Unlock()
	1953	+	tr.setLinkState(ports.LinkStateConnected, nil)
	1954	+	},
	1955	+	wantState: ports.LinkStateConnected,
	1956	+	},
	1957	+	}
	1958	+
	1959	+	for _, tt := range tests {
	1960	+	t.Run(tt.name, func(t *testing.T) {
	1961	+	aPC, bPC := newPair()
	1962	+	var hops atomic.Int32
	1963	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	1964	+	ResendAfter: time.Hour,
	1965	+	Heartbeat: time.Hour,
	1966	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	1967	+	hops.Add(1)
	1968	+	return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
	1969	+	},
	1970	+	})
	1971	+	if err != nil {
	1972	+	t.Fatal(err)
	1973	+	}
	1974	+	defer func() { _ = tr.Close() }()
	1975	+	defer func() { _ = bPC.Close() }()
	1976	+
	1977	+	now := tr.clock.Now()
	1978	+	tr.mu.Lock()
	1979	+	staleAt := now.Add(-tt.staleAge)
	1980	+	tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
	1981	+	decisionReady := make(chan struct{})
	1982	+	resume := make(chan struct{})
	1983	+	tr.afterHealthDecision = func() {
	1984	+	close(decisionReady)
	1985	+	<-resume
	1986	+	}
	1987	+	tr.mu.Unlock()
	1988	+
	1989	+	done := make(chan struct{})
	1990	+	go func() {
	1991	+	tr.checkSilence()
	1992	+	close(done)
	1993	+	}()
	1994	+	select {
	1995	+	case <-decisionReady:
	1996	+	case <-time.After(time.Second):
	1997	+	t.Fatal("health decision hook was not reached")
	1998	+	}
	1999	+	tt.freshen(tr)
	2000	+	close(resume)
	2001	+	select {
	2002	+	case <-done:
	2003	+	case <-time.After(time.Second):
	2004	+	t.Fatal("health check did not finish")
	2005	+	}
	2006	+
	2007	+	if got := tr.LinkState(); got != tt.wantState {
	2008	+	t.Fatalf("state=%v, want %v after fresh health evidence", got, tt.wantState)
	2009	+	}
	2010	+	if got := hops.Load(); got != 0 {
	2011	+	t.Fatalf("socket hops=%d, want none after recovery", got)
	2012	+	}
	2013	+	for len(tr.linkEvents) > 0 {
	2014	+	ev := <-tr.linkEvents
	2015	+	if ev.State == ports.LinkStateDead \|\| ev.State == ports.LinkStateProbing {
	2016	+	t.Fatalf("stale link event emitted after recovery: %v", ev.State)
	2017	+	}
	2018	+	}
	2019	+	select {
	2020	+	case <-tr.done:
	2021	+	t.Fatal("transport closed from stale health decision")
	2022	+	default:
	2023	+	}
	2024	+	})
	2025	+	}
	2026	+	}
	2027	+
1923	2028		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1924	2029		aPC, bPC := newPair()
1925	2030		var hopped atomic.Bool


@@ -1920,6 +1920,111 @@ func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing
1920	}	1920		}
1921	}	1921		}
1922		1922
		1923	+	func TestLinkHealthFreshProgressInvalidatesStaleDecision(t *testing.T) {
		1924	+	tests := []struct {
		1925	+	name string
		1926	+	staleAge time.Duration
		1927	+	freshen func(*Transport)
		1928	+	wantState ports.LinkState
		1929	+	}{
		1930	+	{
		1931	+	name: "fresh packet contact prevents stale dead close",
		1932	+	staleAge: time.Minute,
		1933	+	freshen: func(tr *Transport) {
		1934	+	tr.mu.Lock()
		1935	+	now := tr.clock.Now()
		1936	+	tr.health.authenticatedPacket(now)
		1937	+	tr.lastAuthenticatedPacket = now
		1938	+	tr.mu.Unlock()
		1939	+	},
		1940	+	wantState: ports.LinkStateDegraded,
		1941	+	},
		1942	+	{
		1943	+	name: "complete record prevents stale probing state and hop",
		1944	+	staleAge: 20 * time.Second,
		1945	+	freshen: func(tr *Transport) {
		1946	+	tr.mu.Lock()
		1947	+	now := tr.clock.Now()
		1948	+	tr.health.authenticatedPacket(now)
		1949	+	tr.health.completeRecord(now)
		1950	+	tr.lastAuthenticatedPacket = now
		1951	+	tr.lastCompleteRecord = now
		1952	+	tr.mu.Unlock()
		1953	+	tr.setLinkState(ports.LinkStateConnected, nil)
		1954	+	},
		1955	+	wantState: ports.LinkStateConnected,
		1956	+	},
		1957	+	}
		1958	+
		1959	+	for _, tt := range tests {
		1960	+	t.Run(tt.name, func(t *testing.T) {
		1961	+	aPC, bPC := newPair()
		1962	+	var hops atomic.Int32
		1963	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
		1964	+	ResendAfter: time.Hour,
		1965	+	Heartbeat: time.Hour,
		1966	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
		1967	+	hops.Add(1)
		1968	+	return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
		1969	+	},
		1970	+	})
		1971	+	if err != nil {
		1972	+	t.Fatal(err)
		1973	+	}
		1974	+	defer func() { _ = tr.Close() }()
		1975	+	defer func() { _ = bPC.Close() }()
		1976	+
		1977	+	now := tr.clock.Now()
		1978	+	tr.mu.Lock()
		1979	+	staleAt := now.Add(-tt.staleAge)
		1980	+	tr.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
		1981	+	decisionReady := make(chan struct{})
		1982	+	resume := make(chan struct{})
		1983	+	tr.afterHealthDecision = func() {
		1984	+	close(decisionReady)
		1985	+	<-resume
		1986	+	}
		1987	+	tr.mu.Unlock()
		1988	+
		1989	+	done := make(chan struct{})
		1990	+	go func() {
		1991	+	tr.checkSilence()
		1992	+	close(done)
		1993	+	}()
		1994	+	select {
		1995	+	case <-decisionReady:
		1996	+	case <-time.After(time.Second):
		1997	+	t.Fatal("health decision hook was not reached")
		1998	+	}
		1999	+	tt.freshen(tr)
		2000	+	close(resume)
		2001	+	select {
		2002	+	case <-done:
		2003	+	case <-time.After(time.Second):
		2004	+	t.Fatal("health check did not finish")
		2005	+	}
		2006	+
		2007	+	if got := tr.LinkState(); got != tt.wantState {
		2008	+	t.Fatalf("state=%v, want %v after fresh health evidence", got, tt.wantState)
		2009	+	}
		2010	+	if got := hops.Load(); got != 0 {
		2011	+	t.Fatalf("socket hops=%d, want none after recovery", got)
		2012	+	}
		2013	+	for len(tr.linkEvents) > 0 {
		2014	+	ev := <-tr.linkEvents
		2015	+	if ev.State == ports.LinkStateDead \|\| ev.State == ports.LinkStateProbing {
		2016	+	t.Fatalf("stale link event emitted after recovery: %v", ev.State)
		2017	+	}
		2018	+	}
		2019	+	select {
		2020	+	case <-tr.done:
		2021	+	t.Fatal("transport closed from stale health decision")
		2022	+	default:
		2023	+	}
		2024	+	})
		2025	+	}
		2026	+	}
		2027	+
1923	func TestHopFiresAtProbingNotDegraded(t *testing.T) {	2028		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1924	aPC, bPC := newPair()	2029		aPC, bPC := newPair()
1925	var hopped atomic.Bool	2030		var hopped atomic.Bool