fix(dgram): make UDP diagnostics and fixtures deterministic

commit: b9e591f053e0d9b9b2f3d8bf2b04ae60f56f1b74
author: brice · 3mo ago · 2026-07-10 00:19 UTC
parent: 96bc22d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +188 -63

M internal/adapters/dgram/diagnostics_test.go +73 -0

@@ -1,6 +1,7 @@
1	1		package dgram
2	2
3	3		import (
	4	+	"sync"
4	5		"testing"
5	6		"time"
6	7

@@ -50,6 +51,78 @@ func TestDiagnosticSnapshotSeparatesProgressAges(t *testing.T) {
50	51		}
51	52		}
52	53
	54	+	func TestDiagnosticSnapshotIsLockConsistentDuringConcurrentUpdate(t *testing.T) {
	55	+	now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
	56	+	clock := &snapshotClock{now: now, sampled: make(chan struct{}), release: make(chan struct{})}
	57	+	tr := &Transport{
	58	+	clock: clock,
	59	+	lastAuthenticatedPacket: now.Add(-time.Second),
	60	+	lastCompleteRecord: now.Add(-2 * time.Second),
	61	+	lastACKProgress: now.Add(-3 * time.Second),
	62	+	pending: make(map[uint64]*pending),
	63	+	diagnosticCh: make(chan Diagnostic, 1),
	64	+	}
	65	+
	66	+	go tr.emitDiagnostic()
	67	+	select {
	68	+	case <-clock.sampled:
	69	+	case <-time.After(time.Second):
	70	+	t.Fatal("diagnostic did not begin its clock snapshot")
	71	+	}
	72	+	updated := make(chan struct{})
	73	+	go func() {
	74	+	tr.mu.Lock()
	75	+	tr.lastAuthenticatedPacket = now.Add(time.Second)
	76	+	tr.lastCompleteRecord = now.Add(time.Second)
	77	+	tr.lastACKProgress = now.Add(time.Second)
	78	+	tr.retransmits = 1
	79	+	tr.mu.Unlock()
	80	+	close(updated)
	81	+	}()
	82	+	close(clock.release)
	83	+
	84	+	var d Diagnostic
	85	+	select {
	86	+	case d = <-tr.diagnosticCh:
	87	+	case <-time.After(time.Second):
	88	+	t.Fatal("diagnostic was not emitted")
	89	+	}
	90	+	select {
	91	+	case <-updated:
	92	+	case <-time.After(time.Second):
	93	+	t.Fatal("concurrent update was blocked")
	94	+	}
	95	+	if d.At != now \|\| d.Retransmits != 0 {
	96	+	t.Fatalf("mixed diagnostic snapshot: %+v", d)
	97	+	}
	98	+	if d.SinceAuthenticatedPacket != time.Second \|\| d.SinceCompleteRecord != 2time.Second \|\| d.SinceACKProgress != 3time.Second {
	99	+	t.Fatalf("inconsistent diagnostic ages: %+v", d)
	100	+	}
	101	+	if d.SinceAuthenticatedPacket < 0 \|\| d.SinceCompleteRecord < 0 \|\| d.SinceACKProgress < 0 {
	102	+	t.Fatalf("negative diagnostic age: %+v", d)
	103	+	}
	104	+	}
	105	+
	106	+	type snapshotClock struct {
	107	+	mu sync.Mutex
	108	+	now time.Time
	109	+	sampled chan struct{}
	110	+	release chan struct{}
	111	+	}
	112	+
	113	+	func (c *snapshotClock) Now() time.Time {
	114	+	c.mu.Lock()
	115	+	defer c.mu.Unlock()
	116	+	select {
	117	+	case c.sampled <- struct{}{}:
	118	+	default:
	119	+	}
	120	+	<-c.release
	121	+	return c.now
	122	+	}
	123	+
	124	+	func (c *snapshotClock) NewTimer(time.Duration) ports.Timer { panic("not used") }
	125	+
53	126		func TestDiagnosticObserverIsBounded(t *testing.T) {
54	127		tr := &Transport{
55	128		clock: fixedClock{now: time.Now()},

@@ -1,6 +1,7 @@
1	package dgram	1		package dgram
2		2
3	import (	3		import (
		4	+	"sync"
4	"testing"	5		"testing"
5	"time"	6		"time"
6		7

@@ -50,6 +51,78 @@ func TestDiagnosticSnapshotSeparatesProgressAges(t *testing.T) {
50	}	51		}
51	}	52		}
52		53
		54	+	func TestDiagnosticSnapshotIsLockConsistentDuringConcurrentUpdate(t *testing.T) {
		55	+	now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
		56	+	clock := &snapshotClock{now: now, sampled: make(chan struct{}), release: make(chan struct{})}
		57	+	tr := &Transport{
		58	+	clock: clock,
		59	+	lastAuthenticatedPacket: now.Add(-time.Second),
		60	+	lastCompleteRecord: now.Add(-2 * time.Second),
		61	+	lastACKProgress: now.Add(-3 * time.Second),
		62	+	pending: make(map[uint64]*pending),
		63	+	diagnosticCh: make(chan Diagnostic, 1),
		64	+	}
		65	+
		66	+	go tr.emitDiagnostic()
		67	+	select {
		68	+	case <-clock.sampled:
		69	+	case <-time.After(time.Second):
		70	+	t.Fatal("diagnostic did not begin its clock snapshot")
		71	+	}
		72	+	updated := make(chan struct{})
		73	+	go func() {
		74	+	tr.mu.Lock()
		75	+	tr.lastAuthenticatedPacket = now.Add(time.Second)
		76	+	tr.lastCompleteRecord = now.Add(time.Second)
		77	+	tr.lastACKProgress = now.Add(time.Second)
		78	+	tr.retransmits = 1
		79	+	tr.mu.Unlock()
		80	+	close(updated)
		81	+	}()
		82	+	close(clock.release)
		83	+
		84	+	var d Diagnostic
		85	+	select {
		86	+	case d = <-tr.diagnosticCh:
		87	+	case <-time.After(time.Second):
		88	+	t.Fatal("diagnostic was not emitted")
		89	+	}
		90	+	select {
		91	+	case <-updated:
		92	+	case <-time.After(time.Second):
		93	+	t.Fatal("concurrent update was blocked")
		94	+	}
		95	+	if d.At != now \|\| d.Retransmits != 0 {
		96	+	t.Fatalf("mixed diagnostic snapshot: %+v", d)
		97	+	}
		98	+	if d.SinceAuthenticatedPacket != time.Second \|\| d.SinceCompleteRecord != 2time.Second \|\| d.SinceACKProgress != 3time.Second {
		99	+	t.Fatalf("inconsistent diagnostic ages: %+v", d)
		100	+	}
		101	+	if d.SinceAuthenticatedPacket < 0 \|\| d.SinceCompleteRecord < 0 \|\| d.SinceACKProgress < 0 {
		102	+	t.Fatalf("negative diagnostic age: %+v", d)
		103	+	}
		104	+	}
		105	+
		106	+	type snapshotClock struct {
		107	+	mu sync.Mutex
		108	+	now time.Time
		109	+	sampled chan struct{}
		110	+	release chan struct{}
		111	+	}
		112	+
		113	+	func (c *snapshotClock) Now() time.Time {
		114	+	c.mu.Lock()
		115	+	defer c.mu.Unlock()
		116	+	select {
		117	+	case c.sampled <- struct{}{}:
		118	+	default:
		119	+	}
		120	+	<-c.release
		121	+	return c.now
		122	+	}
		123	+
		124	+	func (c *snapshotClock) NewTimer(time.Duration) ports.Timer { panic("not used") }
		125	+
53	func TestDiagnosticObserverIsBounded(t *testing.T) {	126		func TestDiagnosticObserverIsBounded(t *testing.T) {
54	tr := &Transport{	127		tr := &Transport{
55	clock: fixedClock{now: time.Now()},	128		clock: fixedClock{now: time.Now()},

M internal/adapters/dgram/network_sim_test.go +56 -26


@@ -127,6 +127,7 @@ type fakePC struct {
127	127		mu sync.Mutex
128	128		closed bool
129	129		in chan packet
	130	+	read chan struct{}
130	131		peers map[string]*fakePC
131	132		drop func([]byte, net.Addr) bool
132	133		afterWrite func()

@@ -136,8 +137,8 @@ type fakePC struct {
136	137		func newPair() (fakePC, fakePC) { return newPairWithLink(nil) }
137	138
138	139		func newPairWithLink(link simulatedLink) (fakePC, *fakePC) {
139		-	a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
140		-	b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
	140	+	a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
	141	+	b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
141	142		a.peers["b"] = b
142	143		b.peers["a"] = a
143	144		return a, b

@@ -148,6 +149,10 @@ func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
148	149		return 0, nil, errors.New("closed")
149	150		}
150	151		copy(b, q.b)
	152	+	select {
	153	+	case p.read <- struct{}{}:
	154	+	default:
	155	+	}
151	156		return len(q.b), q.addr, nil
152	157		}
153	158		func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {

@@ -172,7 +177,10 @@ func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
172	177		peer.mu.Unlock()
173	178		return 0, errors.New("peer closed")
174	179		}
175		-	peer.in <- packet{append([]byte(nil), b...), p.addr}
	180	+	select {
	181	+	case peer.in <- packet{append([]byte(nil), b...), p.addr}:
	182	+	default:
	183	+	}
176	184		peer.mu.Unlock()
177	185		}
178	186		if p.afterWrite != nil {

@@ -196,6 +204,31 @@ func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
196	204
197	205		func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
198	206
	207	+	const fixtureWaitTimeout = time.Second
	208	+
	209	+	// awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a
	210	+	// hang guard only; tests advance behavior with the manual clock and barriers.
	211	+	func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {
	212	+	t.Helper()
	213	+	select {
	214	+	case <-ch:
	215	+	case <-time.After(fixtureWaitTimeout):
	216	+	t.Fatalf("timed out waiting for %s", context)
	217	+	}
	218	+	}
	219	+
	220	+	func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
	221	+	t.Helper()
	222	+	select {
	223	+	case result := <-ch:
	224	+	return result
	225	+	case <-time.After(fixtureWaitTimeout):
	226	+	t.Fatalf("timed out waiting for %s", context)
	227	+	var zero T
	228	+	return zero
	229	+	}
	230	+	}
	231	+
199	232		// TestTransportFloodClassification drives authenticated datagrams through real
200	233		// transports. The metrics are transport state, not simulator bookkeeping.
201	234		func TestTransportFloodClassification(t *testing.T) {

@@ -221,6 +254,10 @@ func TestTransportFloodClassification(t *testing.T) {
221	254		}
222	255		a, b := newFloodTransports(t, aPC, bPC, clk, 128)
223	256		defer closeFloodTransports(a, b)
	257	+	contacted := make(chan struct{}, 2)
	258	+	b.mu.Lock()
	259	+	b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
	260	+	b.mu.Unlock()
224	261		start := clk.Now()
225	262		clk.advance(time.Nanosecond)
226	263		for range 2 {

@@ -228,7 +265,8 @@ func TestTransportFloodClassification(t *testing.T) {
228	265		t.Fatal(err)
229	266		}
230	267		}
231		-	floodEventually(t, func() bool { return floodState(b).packet && len(dropped) == 2 })
	268	+	awaitSignal(t, contacted, "first authenticated fragment")
	269	+	awaitSignal(t, contacted, "second authenticated fragment")
232	270		state := floodState(b)
233	271		if !state.packet \|\| state.record \|\| floodState(a).ack {
234	272		t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack)

@@ -245,6 +283,10 @@ func TestTransportFloodClassification(t *testing.T) {
245	283		aPC, bPC := newPairWithLink(link)
246	284		a, b := newFloodTransports(t, aPC, bPC, clk, 128)
247	285		defer closeFloodTransports(a, b)
	286	+	contacted := make(chan struct{}, 1)
	287	+	b.mu.Lock()
	288	+	b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
	289	+	b.mu.Unlock()
248	290		a.mu.Lock()
249	291		a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.
250	292		a.mu.Unlock()

@@ -257,7 +299,7 @@ func TestTransportFloodClassification(t *testing.T) {
257	299		}
258	300		clk.advance(10 * time.Second)
259	301		link.flush(bPC)
260		-	floodEventually(t, func() bool { return floodState(b).packet })
	302	+	awaitSignal(t, contacted, "queued authenticated fragment")
261	303		if got := link.queuedBytes(); got != 0 {
262	304		t.Fatalf("queued bytes after fake-clock release=%d, want 0", got)
263	305		}

@@ -322,7 +364,7 @@ func TestTransportFloodClassification(t *testing.T) {
322	364		writesMu.Unlock()
323	365		if secondWrite {
324	366		retransmitBlocked <- struct{}{}
325		-	<-releaseRetransmit
	367	+	awaitSignal(t, releaseRetransmit, "retransmit release")
326	368		}
327	369		return false
328	370		}

@@ -339,7 +381,7 @@ func TestTransportFloodClassification(t *testing.T) {
339	381		}
340	382		defer closeFloodTransports(a, b)
341	383		for range 4 { // Both transports install resend and heartbeat timers.
342		-	<-clk.timerCreated
	384	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
343	385		}
344	386
345	387		// Two fresh, fragmented output records become overdue while their real

@@ -358,11 +400,11 @@ func TestTransportFloodClassification(t *testing.T) {
358	400		}
359	401		}
360	402		for range 2 {
361		-	<-acksDropped
	403	+	awaitSignal(t, acksDropped, "dropped cumulative ACK")
362	404		}
363	405
364	406		clk.advance(10 * time.Millisecond)
365		-	<-retransmitBlocked // resendPending is synchronously inside writePacket.
	407	+	awaitSignal(t, retransmitBlocked, "retransmit pacing barrier") // resendPending is synchronously inside writePacket.
366	408		if got := floodState(a).retransmits; got < 2 {
367	409		t.Fatalf("retransmits=%d, want both overdue large records selected", got)
368	410		}

@@ -374,13 +416,13 @@ func TestTransportFloodClassification(t *testing.T) {
374	416		a.outputNext = time.Time{}
375	417		a.beforeDataPace = func() {
376	418		close(freshAtPace)
377		-	<-allowFreshPace
	419	+	awaitSignal(t, allowFreshPace, "fresh pacing release")
378	420		}
379	421		a.mu.Unlock()
380	422		go func() {
381	423		freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})
382	424		}()
383		-	<-freshAtPace
	425	+	awaitSignal(t, freshAtPace, "fresh output pacing barrier")
384	426		a.mu.Lock()
385	427		a.beforeDataPace = nil
386	428		a.mu.Unlock()

@@ -399,10 +441,10 @@ func TestTransportFloodClassification(t *testing.T) {
399	441		}
400	442
401	443		close(releaseRetransmit)
402		-	if err := <-freshReturned; err != nil {
	444	+	if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
403	445		t.Fatal(err)
404	446		}
405		-	<-probeSent
	447	+	awaitSignal(t, probeSent, "heartbeat probe")
406	448		writesMu.Lock()
407	449		defer writesMu.Unlock()
408	450		if writes[1] < 2 \|\| writes[2] < 2 \|\| writes[3] == 0 {

@@ -448,18 +490,6 @@ func closeFloodTransports(a, b *Transport) {
448	490		_ = b.Close()
449	491		}
450	492
451		-	func floodEventually(t *testing.T, condition func() bool) {
452		-	t.Helper()
453		-	deadline := time.Now().Add(time.Second)
454		-	for time.Now().Before(deadline) {
455		-	if condition() {
456		-	return
457		-	}
458		-	time.Sleep(time.Millisecond)
459		-	}
460		-	t.Fatal("condition was not met")
461		-	}
462		-
463	493		func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
464	494		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
465	495		aPC, bPC := newPair()

@@ -476,7 +506,7 @@ func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
476	506		if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil {
477	507		t.Fatal(err)
478	508		}
479		-	time.Sleep(10 * time.Millisecond)
	509	+	awaitSignal(t, bPC.read, "malformed datagram read")
480	510		b.mu.Lock()
481	511		gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
482	512		b.mu.Unlock()

 	mu         sync.Mutex
 	closed     bool
 	in         chan packet
+	read       chan struct{}
 	peers      map[string]*fakePC
 	drop       func([]byte, net.Addr) bool
 	afterWrite func()
 func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
 func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
-	a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
+	a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
-	b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
+	b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), read: make(chan struct{}, 100), peers: map[string]*fakePC{}, link: link}
 	a.peers["b"] = b
 	b.peers["a"] = a
 	return a, b
 		return 0, nil, errors.New("closed")
 	}
 	copy(b, q.b)
+	select {
+	case p.read <- struct{}{}:
+	default:
+	}
 	return len(q.b), q.addr, nil
 }
 func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
 			peer.mu.Unlock()
 			return 0, errors.New("peer closed")
 		}
-		peer.in <- packet{append([]byte(nil), b...), p.addr}
+		select {
+		case peer.in <- packet{append([]byte(nil), b...), p.addr}:
+		default:
+		}
 		peer.mu.Unlock()
 	}
 	if p.afterWrite != nil {
 func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
+const fixtureWaitTimeout = time.Second
+// awaitSignal gives fixture barriers a diagnostic deadline. The deadline is a
+// hang guard only; tests advance behavior with the manual clock and barriers.
+func awaitSignal(t *testing.T, ch <-chan struct{}, context string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(fixtureWaitTimeout):
+		t.Fatalf("timed out waiting for %s", context)
+	}
+}
+func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
+	t.Helper()
+	select {
+	case result := <-ch:
+		return result
+	case <-time.After(fixtureWaitTimeout):
+		t.Fatalf("timed out waiting for %s", context)
+		var zero T
+		return zero
+	}
+}
 // TestTransportFloodClassification drives authenticated datagrams through real
 // transports. The metrics are transport state, not simulator bookkeeping.
 func TestTransportFloodClassification(t *testing.T) {
 		}
 		a, b := newFloodTransports(t, aPC, bPC, clk, 128)
 		defer closeFloodTransports(a, b)
+		contacted := make(chan struct{}, 2)
+		b.mu.Lock()
+		b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
+		b.mu.Unlock()
 		start := clk.Now()
 		clk.advance(time.Nanosecond)
 		for range 2 {
 				t.Fatal(err)
 			}
 		}
-		floodEventually(t, func() bool { return floodState(b).packet && len(dropped) == 2 })
+		awaitSignal(t, contacted, "first authenticated fragment")
+		awaitSignal(t, contacted, "second authenticated fragment")
 		state := floodState(b)
 		if !state.packet || state.record || floodState(a).ack {
 			t.Fatalf("contact=%v complete=%v ack=%v, want contact only", state.packet, state.record, floodState(a).ack)
 		aPC, bPC := newPairWithLink(link)
 		a, b := newFloodTransports(t, aPC, bPC, clk, 128)
 		defer closeFloodTransports(a, b)
+		contacted := make(chan struct{}, 1)
+		b.mu.Lock()
+		b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
+		b.mu.Unlock()
 		a.mu.Lock()
 		a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.
 		a.mu.Unlock()
 		}
 		clk.advance(10 * time.Second)
 		link.flush(bPC)
-		floodEventually(t, func() bool { return floodState(b).packet })
+		awaitSignal(t, contacted, "queued authenticated fragment")
 		if got := link.queuedBytes(); got != 0 {
 			t.Fatalf("queued bytes after fake-clock release=%d, want 0", got)
 		}
 			writesMu.Unlock()
 			if secondWrite {
 				retransmitBlocked <- struct{}{}
-				<-releaseRetransmit
+				awaitSignal(t, releaseRetransmit, "retransmit release")
 			}
 			return false
 		}
 		}
 		defer closeFloodTransports(a, b)
 		for range 4 { // Both transports install resend and heartbeat timers.
-			<-clk.timerCreated
+			awaitSignal(t, clk.timerCreated, "transport timer creation")
 		}
 		// Two fresh, fragmented output records become overdue while their real
 			}
 		}
 		for range 2 {
-			<-acksDropped
+			awaitSignal(t, acksDropped, "dropped cumulative ACK")
 		}
 		clk.advance(10 * time.Millisecond)
-		<-retransmitBlocked // resendPending is synchronously inside writePacket.
+		awaitSignal(t, retransmitBlocked, "retransmit pacing barrier") // resendPending is synchronously inside writePacket.
 		if got := floodState(a).retransmits; got < 2 {
 			t.Fatalf("retransmits=%d, want both overdue large records selected", got)
 		}
 		a.outputNext = time.Time{}
 		a.beforeDataPace = func() {
 			close(freshAtPace)
-			<-allowFreshPace
+			awaitSignal(t, allowFreshPace, "fresh pacing release")
 		}
 		a.mu.Unlock()
 		go func() {
 			freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})
 		}()
-		<-freshAtPace
+		awaitSignal(t, freshAtPace, "fresh output pacing barrier")
 		a.mu.Lock()
 		a.beforeDataPace = nil
 		a.mu.Unlock()
 		}
 		close(releaseRetransmit)
-		if err := <-freshReturned; err != nil {
+		if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
 			t.Fatal(err)
 		}
-		<-probeSent
+		awaitSignal(t, probeSent, "heartbeat probe")
 		writesMu.Lock()
 		defer writesMu.Unlock()
 		if writes[1] < 2 || writes[2] < 2 || writes[3] == 0 {
 	_ = b.Close()
 }
-func floodEventually(t *testing.T, condition func() bool) {
-	t.Helper()
-	deadline := time.Now().Add(time.Second)
-	for time.Now().Before(deadline) {
-		if condition() {
-			return
-		}
-		time.Sleep(time.Millisecond)
-	}
-	t.Fatal("condition was not met")
-}
 func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
 	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
 	aPC, bPC := newPair()
 	if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(10 * time.Millisecond)
+	awaitSignal(t, bPC.read, "malformed datagram read")
 	b.mu.Lock()
 	gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
 	b.mu.Unlock()

M internal/adapters/dgram/transport.go +59 -37


@@ -153,34 +153,37 @@ type Transport struct {
153	153		lastACKProgress time.Time
154	154		retransmits uint64
155	155		reassemblyInflight int
156		-	observe DiagnosticObserver
157		-	diagnosticCh chan Diagnostic
158		-	heartbeat time.Duration
159		-	resendAfter time.Duration
160		-	maxResendAfter time.Duration
161		-	maxResendPerTick int
162		-	srtt time.Duration
163		-	rttvar time.Duration
164		-	rto time.Duration
165		-	writeTimeout time.Duration
166		-	degradedAfter time.Duration
167		-	probeAfter time.Duration
168		-	offlineAfter time.Duration
169		-	deadAfter time.Duration
170		-	maxPending int
171		-	maxPendingWait time.Duration
172		-	maxRecvBuffer int
173		-	clock ports.Clock
174		-	linkState ports.LinkState
175		-	linkEvents chan ports.LinkEvent
176		-	probeWait map[uint64]chan struct{}
177		-	sendWake chan struct{}
178		-	probeReply chan uint64
179		-	rebind func(net.PacketConn) (net.PacketConn, error)
180		-	hoppedOffline bool
181		-	outputQueue []queuedSend
182		-	outputWake chan struct{}
183		-	outputNext time.Time
	156	+	// afterAuthenticatedPacket is a test synchronization hook. It runs after
	157	+	// authenticated-contact state has been committed.
	158	+	afterAuthenticatedPacket func()
	159	+	observe DiagnosticObserver
	160	+	diagnosticCh chan Diagnostic
	161	+	heartbeat time.Duration
	162	+	resendAfter time.Duration
	163	+	maxResendAfter time.Duration
	164	+	maxResendPerTick int
	165	+	srtt time.Duration
	166	+	rttvar time.Duration
	167	+	rto time.Duration
	168	+	writeTimeout time.Duration
	169	+	degradedAfter time.Duration
	170	+	probeAfter time.Duration
	171	+	offlineAfter time.Duration
	172	+	deadAfter time.Duration
	173	+	maxPending int
	174	+	maxPendingWait time.Duration
	175	+	maxRecvBuffer int
	176	+	clock ports.Clock
	177	+	linkState ports.LinkState
	178	+	linkEvents chan ports.LinkEvent
	179	+	probeWait map[uint64]chan struct{}
	180	+	sendWake chan struct{}
	181	+	probeReply chan uint64
	182	+	rebind func(net.PacketConn) (net.PacketConn, error)
	183	+	hoppedOffline bool
	184	+	outputQueue []queuedSend
	185	+	outputWake chan struct{}
	186	+	outputNext time.Time
184	187
185	188		recvMu sync.Mutex
186	189		replay *pdgram.ReplayWindow

@@ -709,7 +712,11 @@ func (t *Transport) recordAuthenticatedFragment(inflight int) {
709	712		t.mu.Lock()
710	713		t.lastAuthenticatedPacket = t.clock.Now()
711	714		t.reassemblyInflight = inflight
	715	+	after := t.afterAuthenticatedPacket
712	716		t.mu.Unlock()
	717	+	if after != nil {
	718	+	after()
	719	+	}
713	720		}
714	721
715	722		func (t *Transport) recordCompleteRecord() {

@@ -1132,32 +1139,47 @@ func (t *Transport) emitDiagnostic() {
1132	1139		if t.diagnosticCh == nil {
1133	1140		return
1134	1141		}
1135		-	now := t.clock.Now()
1136	1142		t.mu.Lock()
	1143	+	now := t.clock.Now()
	1144	+	lastAuthenticatedPacket := t.lastAuthenticatedPacket
	1145	+	lastCompleteRecord := t.lastCompleteRecord
	1146	+	lastACKProgress := t.lastACKProgress
1137	1147		pendingBytes := 0
1138	1148		for _, p := range t.pending {
1139	1149		if p != nil {
1140	1150		pendingBytes += len(p.frame.Payload)
1141	1151		}
1142	1152		}
	1153	+	state := t.linkState
	1154	+	pendingRecords := len(t.pending)
	1155	+	retransmits := t.retransmits
	1156	+	reassemblyInflight := t.reassemblyInflight
	1157	+	t.mu.Unlock()
	1158	+
1143	1159		d := Diagnostic{
1144	1160		At: now,
1145		-	State: t.linkState,
1146		-	SinceAuthenticatedPacket: now.Sub(t.lastAuthenticatedPacket),
1147		-	SinceCompleteRecord: now.Sub(t.lastCompleteRecord),
1148		-	SinceACKProgress: now.Sub(t.lastACKProgress),
1149		-	PendingRecords: len(t.pending),
	1161	+	State: state,
	1162	+	SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
	1163	+	SinceCompleteRecord: diagnosticAge(now, lastCompleteRecord),
	1164	+	SinceACKProgress: diagnosticAge(now, lastACKProgress),
	1165	+	PendingRecords: pendingRecords,
1150	1166		PendingBytes: pendingBytes,
1151		-	Retransmits: t.retransmits,
1152		-	ReassemblyInflight: t.reassemblyInflight,
	1167	+	Retransmits: retransmits,
	1168	+	ReassemblyInflight: reassemblyInflight,
1153	1169		}
1154		-	t.mu.Unlock()
1155	1170		select {
1156	1171		case t.diagnosticCh <- d:
1157	1172		default:
1158	1173		}
1159	1174		}
1160	1175
	1176	+	func diagnosticAge(now, then time.Time) time.Duration {
	1177	+	if then.After(now) {
	1178	+	return 0
	1179	+	}
	1180	+	return now.Sub(then)
	1181	+	}
	1182	+
1161	1183		func (t *Transport) closeWithError(err error) {
1162	1184		t.mu.Lock()
1163	1185		if t.closed {

 	lastACKProgress         time.Time
 	retransmits             uint64
 	reassemblyInflight      int
-	observe                 DiagnosticObserver
+	// afterAuthenticatedPacket is a test synchronization hook. It runs after
-	diagnosticCh            chan Diagnostic
+	// authenticated-contact state has been committed.
-	heartbeat               time.Duration
+	afterAuthenticatedPacket 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
-	probeReply              chan uint64
+	linkEvents               chan ports.LinkEvent
-	rebind                  func(net.PacketConn) (net.PacketConn, error)
+	probeWait                map[uint64]chan struct{}
-	hoppedOffline           bool
+	sendWake                 chan struct{}
-	outputQueue             []queuedSend
+	probeReply               chan uint64
-	outputWake              chan struct{}
+	rebind                   func(net.PacketConn) (net.PacketConn, error)
-	outputNext              time.Time
+	hoppedOffline            bool
+	outputQueue              []queuedSend
+	outputWake               chan struct{}
+	outputNext               time.Time
 	recvMu sync.Mutex
 	replay *pdgram.ReplayWindow
 	t.mu.Lock()
 	t.lastAuthenticatedPacket = t.clock.Now()
 	t.reassemblyInflight = inflight
+	after := t.afterAuthenticatedPacket
 	t.mu.Unlock()
+	if after != nil {
+		after()
+	}
 }
 func (t *Transport) recordCompleteRecord() {
 	if t.diagnosticCh == nil {
 		return
 	}
-	now := t.clock.Now()
 	t.mu.Lock()
+	now := t.clock.Now()
+	lastAuthenticatedPacket := t.lastAuthenticatedPacket
+	lastCompleteRecord := t.lastCompleteRecord
+	lastACKProgress := t.lastACKProgress
 	pendingBytes := 0
 	for _, p := range t.pending {
 		if p != nil {
 			pendingBytes += len(p.frame.Payload)
 		}
 	}
+	state := t.linkState
+	pendingRecords := len(t.pending)
+	retransmits := t.retransmits
+	reassemblyInflight := t.reassemblyInflight
+	t.mu.Unlock()
 	d := Diagnostic{
 		At:                       now,
-		State:                    t.linkState,
+		State:                    state,
-		SinceAuthenticatedPacket: now.Sub(t.lastAuthenticatedPacket),
+		SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
-		SinceCompleteRecord:      now.Sub(t.lastCompleteRecord),
+		SinceCompleteRecord:      diagnosticAge(now, lastCompleteRecord),
-		SinceACKProgress:         now.Sub(t.lastACKProgress),
+		SinceACKProgress:         diagnosticAge(now, lastACKProgress),
-		PendingRecords:           len(t.pending),
+		PendingRecords:           pendingRecords,
 		PendingBytes:             pendingBytes,
-		Retransmits:              t.retransmits,
+		Retransmits:              retransmits,
-		ReassemblyInflight:       t.reassemblyInflight,
+		ReassemblyInflight:       reassemblyInflight,
 	}
-	t.mu.Unlock()
 	select {
 	case t.diagnosticCh <- d:
 	default:
 	}
 }
+func diagnosticAge(now, then time.Time) time.Duration {
+	if then.After(now) {
+		return 0
+	}
+	return now.Sub(then)
+}
 func (t *Transport) closeWithError(err error) {
 	t.mu.Lock()
 	if t.closed {