fix(dgram): keep reassembly diagnostics current

commit: b4de71286b30d4bc422d0702c6532de5bcf5ba55
author: brice · 3mo ago · 2026-07-10 06:05 UTC
parent: 5f61847
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +32 -11

M internal/adapters/dgram/network_sim_test.go +25 -8


@@ -134,10 +134,13 @@ type fakePC struct {
134	134		link *simulatedLink
135	135		}
136	136
137		-	func newPair() (fakePC, fakePC) { return newPairWithLink(nil) }
	137	+	func newPair() (fakePC, fakePC) { return newPairWithCapacity(nil, 100) }
138	138
139		-	func newPairWithLink(link simulatedLink) (fakePC, *fakePC) {
140		-	const queuePackets = floodRecordCount * floodRecordMTUs * 2
	139	+	func newFloodPair(link simulatedLink) (fakePC, *fakePC) {
	140	+	return newPairWithCapacity(link, floodRecordCountfloodRecordMTUs2)
	141	+	}
	142	+
	143	+	func newPairWithCapacity(link simulatedLink, queuePackets int) (fakePC, *fakePC) {
141	144		a := &fakePC{addr: testAddr("a"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
142	145		b := &fakePC{addr: testAddr("b"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
143	146		a.peers["b"] = b

@@ -240,7 +243,7 @@ func awaitResult[T any](t *testing.T, ch <-chan T, context string) T {
240	243		func TestTransportFloodClassification(t *testing.T) {
241	244		t.Run("one real fragment per large record loss contacts without completing", func(t *testing.T) {
242	245		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
243		-	aPC, bPC := newPair()
	246	+	aPC, bPC := newFloodPair(nil)
244	247		inspect, err := pdgram.NewCodec(key())
245	248		if err != nil {
246	249		t.Fatal(err)

@@ -290,7 +293,7 @@ func TestTransportFloodClassification(t *testing.T) {
290	293		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
291	294		limit := 2 * 128
292	295		link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
293		-	aPC, bPC := newPairWithLink(link)
	296	+	aPC, bPC := newFloodPair(link)
294	297		const mtu = 128
295	298		a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
296	299		defer closeFloodTransports(a, b)

@@ -320,7 +323,7 @@ func TestTransportFloodClassification(t *testing.T) {
320	323
321	324		t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {
322	325		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
323		-	aPC, bPC := newPair()
	326	+	aPC, bPC := newFloodPair(nil)
324	327		inspect, err := pdgram.NewCodec(key())
325	328		if err != nil {
326	329		t.Fatal(err)

@@ -405,8 +408,8 @@ func TestTransportFloodClassification(t *testing.T) {
405	408		}
406	409
407	410		// Fresh, fragmented output records become overdue while their real
408		-	// cumulative ACKs are lost. Disable only test pacing so the fixture can
409		-	// isolate resend-loop serialization rather than pacing delay.
	411	+	// cumulative ACKs are lost. Disabling pacing intentionally forces overload
	412	+	// to classify contention; it does not claim production pacing creates it.
410	413		sendFloodOutputs(t, a, floodRecordCount, 128)
411	414		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
412	415		clk.advance(maxACKDelay)

@@ -600,15 +603,29 @@ func TestInconsistentAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
600	603		awaitSignal(t, accepted, "accepted incomplete fragment")
601	604		b.mu.Lock()
602	605		lastContact := b.lastAuthenticatedPacket
	606	+	acceptedInflight := b.reassemblyInflight
603	607		b.mu.Unlock()
	608	+	if acceptedInflight != 1 {
	609	+	t.Fatalf("reassembly inflight after accepted fragment=%d, want 1", acceptedInflight)
	610	+	}
604	611
605	612		clk.advance(time.Second)
606	613		writeFragment(1000, pdgram.Fragment{Seq: 42, Index: 1, Count: 3, Data: []byte("inconsistent")})
607	614		awaitSignal(t, rejected, "inconsistent fragment rejection")
608	615		b.mu.Lock()
609	616		gotContact := b.lastAuthenticatedPacket
	617	+	gotInflight := b.reassemblyInflight
610	618		b.mu.Unlock()
611	619		if gotContact != lastContact {
612	620		t.Fatalf("rejected inconsistent fragment refreshed contact from %v to %v", lastContact, gotContact)
613	621		}
	622	+	if gotInflight != 0 {
	623	+	t.Fatalf("reassembly inflight after rejected inconsistent fragment=%d, want 0", gotInflight)
	624	+	}
	625	+	b.diagnosticCh = make(chan Diagnostic, 1)
	626	+	b.emitDiagnostic()
	627	+	diagnostic := awaitResult(t, b.diagnosticCh, "diagnostic snapshot after inconsistent fragment")
	628	+	if diagnostic.ReassemblyInflight != 0 {
	629	+	t.Fatalf("diagnostic reassembly inflight after rejected inconsistent fragment=%d, want 0", diagnostic.ReassemblyInflight)
	630	+	}
614	631		}

 	link       *simulatedLink
 }
-func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
+func newPair() (*fakePC, *fakePC) { return newPairWithCapacity(nil, 100) }
-func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
+func newFloodPair(link *simulatedLink) (*fakePC, *fakePC) {
-	const queuePackets = floodRecordCount * floodRecordMTUs * 2
+	return newPairWithCapacity(link, floodRecordCount*floodRecordMTUs*2)
+}
+func newPairWithCapacity(link *simulatedLink, queuePackets int) (*fakePC, *fakePC) {
 	a := &fakePC{addr: testAddr("a"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
 	b := &fakePC{addr: testAddr("b"), in: make(chan packet, queuePackets), read: make(chan struct{}, queuePackets), peers: map[string]*fakePC{}, link: link}
 	a.peers["b"] = b
 func TestTransportFloodClassification(t *testing.T) {
 	t.Run("one real fragment per large record loss contacts without completing", func(t *testing.T) {
 		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
-		aPC, bPC := newPair()
+		aPC, bPC := newFloodPair(nil)
 		inspect, err := pdgram.NewCodec(key())
 		if err != nil {
 			t.Fatal(err)
 		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
 		limit := 2 * 128
 		link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
-		aPC, bPC := newPairWithLink(link)
+		aPC, bPC := newFloodPair(link)
 		const mtu = 128
 		a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
 		defer closeFloodTransports(a, b)
 	t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {
 		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
-		aPC, bPC := newPair()
+		aPC, bPC := newFloodPair(nil)
 		inspect, err := pdgram.NewCodec(key())
 		if err != nil {
 			t.Fatal(err)
 		}
 		// Fresh, fragmented output records become overdue while their real
-		// cumulative ACKs are lost. Disable only test pacing so the fixture can
+		// cumulative ACKs are lost. Disabling pacing intentionally forces overload
-		// isolate resend-loop serialization rather than pacing delay.
+		// to classify contention; it does not claim production pacing creates it.
 		sendFloodOutputs(t, a, floodRecordCount, 128)
 		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
 		clk.advance(maxACKDelay)
 	awaitSignal(t, accepted, "accepted incomplete fragment")
 	b.mu.Lock()
 	lastContact := b.lastAuthenticatedPacket
+	acceptedInflight := b.reassemblyInflight
 	b.mu.Unlock()
+	if acceptedInflight != 1 {
+		t.Fatalf("reassembly inflight after accepted fragment=%d, want 1", acceptedInflight)
+	}
 	clk.advance(time.Second)
 	writeFragment(1000, pdgram.Fragment{Seq: 42, Index: 1, Count: 3, Data: []byte("inconsistent")})
 	awaitSignal(t, rejected, "inconsistent fragment rejection")
 	b.mu.Lock()
 	gotContact := b.lastAuthenticatedPacket
+	gotInflight := b.reassemblyInflight
 	b.mu.Unlock()
 	if gotContact != lastContact {
 		t.Fatalf("rejected inconsistent fragment refreshed contact from %v to %v", lastContact, gotContact)
 	}
+	if gotInflight != 0 {
+		t.Fatalf("reassembly inflight after rejected inconsistent fragment=%d, want 0", gotInflight)
+	}
+	b.diagnosticCh = make(chan Diagnostic, 1)
+	b.emitDiagnostic()
+	diagnostic := awaitResult(t, b.diagnosticCh, "diagnostic snapshot after inconsistent fragment")
+	if diagnostic.ReassemblyInflight != 0 {
+		t.Fatalf("diagnostic reassembly inflight after rejected inconsistent fragment=%d, want 0", diagnostic.ReassemblyInflight)
+	}
 }

M internal/adapters/dgram/transport.go +7 -3


@@ -796,11 +796,11 @@ func (t *Transport) readLoop(pc net.PacketConn) {
796	796		payload, complete, err := t.reasm.Add(frag)
797	797		inflight := t.reasm.Inflight()
798	798		t.recvMu.Unlock()
	799	+	t.recordFragmentResult(inflight, err == nil)
799	800		if err != nil {
800	801		t.notifyMalformedFragment()
801	802		continue
802	803		}
803		-	t.recordAuthenticatedFragment(inflight)
804	804		if !complete {
805	805		continue
806	806		}

@@ -819,10 +819,14 @@ func (t *Transport) notifyMalformedFragment() {
819	819		}
820	820		}
821	821
822		-	func (t *Transport) recordAuthenticatedFragment(inflight int) {
	822	+	func (t *Transport) recordFragmentResult(inflight int, accepted bool) {
823	823		t.mu.Lock()
824		-	t.lastAuthenticatedPacket = t.clock.Now()
825	824		t.reassemblyInflight = inflight
	825	+	if !accepted {
	826	+	t.mu.Unlock()
	827	+	return
	828	+	}
	829	+	t.lastAuthenticatedPacket = t.clock.Now()
826	830		after := t.afterAuthenticatedPacket
827	831		t.mu.Unlock()
828	832		if after != nil {

 		payload, complete, err := t.reasm.Add(frag)
 		inflight := t.reasm.Inflight()
 		t.recvMu.Unlock()
+		t.recordFragmentResult(inflight, err == nil)
 		if err != nil {
 			t.notifyMalformedFragment()
 			continue
 		}
-		t.recordAuthenticatedFragment(inflight)
 		if !complete {
 			continue
 		}
 	}
 }
-func (t *Transport) recordAuthenticatedFragment(inflight int) {
+func (t *Transport) recordFragmentResult(inflight int, accepted bool) {
 	t.mu.Lock()
-	t.lastAuthenticatedPacket = t.clock.Now()
 	t.reassemblyInflight = inflight
+	if !accepted {
+		t.mu.Unlock()
+		return
+	}
+	t.lastAuthenticatedPacket = t.clock.Now()
 	after := t.afterAuthenticatedPacket
 	t.mu.Unlock()
 	if after != nil {