test(dgram): synchronize malformed fragment rejection

commit: 2479821c091ae9e0930baff9f47315a25b858c22
author: brice · 3mo ago · 2026-07-10 00:23 UTC
parent: b9e591f
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +42 -29

M internal/adapters/dgram/network_sim_test.go +5 -1


@@ -502,11 +502,15 @@ func TestMalformedAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
502	502		if err != nil {
503	503		t.Fatal(err)
504	504		}
	505	+	malformedRejected := make(chan struct{}, 1)
	506	+	b.mu.Lock()
	507	+	b.afterMalformedFragment = func() { malformedRejected <- struct{}{} }
	508	+	b.mu.Unlock()
505	509		clk.advance(time.Second)
506	510		if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil {
507	511		t.Fatal(err)
508	512		}
509		-	awaitSignal(t, bPC.read, "malformed datagram read")
	513	+	awaitSignal(t, malformedRejected, "malformed fragment rejection")
510	514		b.mu.Lock()
511	515		gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
512	516		b.mu.Unlock()

 	if err != nil {
 		t.Fatal(err)
 	}
+	malformedRejected := make(chan struct{}, 1)
+	b.mu.Lock()
+	b.afterMalformedFragment = func() { malformedRejected <- struct{}{} }
+	b.mu.Unlock()
 	clk.advance(time.Second)
 	if _, err := aPC.WriteTo(codec.Seal(1, 999, []byte("not a fragment"), nil), bPC.addr); err != nil {
 		t.Fatal(err)
 	}
-	awaitSignal(t, bPC.read, "malformed datagram read")
+	awaitSignal(t, malformedRejected, "malformed fragment rejection")
 	b.mu.Lock()
 	gotPacket, gotRecord := b.lastAuthenticatedPacket, b.lastCompleteRecord
 	b.mu.Unlock()

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


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

@@ -693,6 +696,12 @@ func (t *Transport) readLoop(pc net.PacketConn) {
693	696		frag, err := pdgram.UnmarshalFragment(pt)
694	697		if err != nil {
695	698		t.recvMu.Unlock()
	699	+	t.mu.Lock()
	700	+	afterMalformed := t.afterMalformedFragment
	701	+	t.mu.Unlock()
	702	+	if afterMalformed != nil {
	703	+	afterMalformed()
	704	+	}
696	705		continue
697	706		}
698	707		payload, complete, err := t.reasm.Add(frag)


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

@@ -693,6 +696,12 @@ func (t *Transport) readLoop(pc net.PacketConn) {
693		frag, err := pdgram.UnmarshalFragment(pt)	696		frag, err := pdgram.UnmarshalFragment(pt)
694		if err != nil {	697		if err != nil {
695		t.recvMu.Unlock()	698		t.recvMu.Unlock()
			699	+	t.mu.Lock()
			700	+	afterMalformed := t.afterMalformedFragment
			701	+	t.mu.Unlock()
			702	+	if afterMalformed != nil {
			703	+	afterMalformed()
			704	+	}
696		continue	705		continue
697		}	706		}
698		payload, complete, err := t.reasm.Add(frag)	707		payload, complete, err := t.reasm.Add(frag)