fix(dgram): order reassembly diagnostic commits

commit: 993f038ce4ddb3be3e40c6708eeabd4e1455f731
author: brice · 3mo ago · 2026-07-10 06:10 UTC
parent: b4de712
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +23 -24

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


@@ -622,9 +622,7 @@ func TestInconsistentAuthenticatedFragmentDoesNotCountAsContact(t *testing.T) {
622	622		if gotInflight != 0 {
623	623		t.Fatalf("reassembly inflight after rejected inconsistent fragment=%d, want 0", gotInflight)
624	624		}
625		-	b.diagnosticCh = make(chan Diagnostic, 1)
626		-	b.emitDiagnostic()
627		-	diagnostic := awaitResult(t, b.diagnosticCh, "diagnostic snapshot after inconsistent fragment")
	625	+	diagnostic := b.diagnosticSnapshot()
628	626		if diagnostic.ReassemblyInflight != 0 {
629	627		t.Fatalf("diagnostic reassembly inflight after rejected inconsistent fragment=%d, want 0", diagnostic.ReassemblyInflight)
630	628		}

 	if gotInflight != 0 {
 		t.Fatalf("reassembly inflight after rejected inconsistent fragment=%d, want 0", gotInflight)
 	}
-	b.diagnosticCh = make(chan Diagnostic, 1)
+	diagnostic := b.diagnosticSnapshot()
-	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 +22 -21


@@ -795,8 +795,20 @@ func (t *Transport) readLoop(pc net.PacketConn) {
795	795		}
796	796		payload, complete, err := t.reasm.Add(frag)
797	797		inflight := t.reasm.Inflight()
	798	+	// Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
	799	+	// overlap old and new read loops, so recvMu must remain held through commit.
	800	+	t.mu.Lock()
	801	+	t.reassemblyInflight = inflight
	802	+	var afterAuthenticated func()
	803	+	if err == nil {
	804	+	t.lastAuthenticatedPacket = t.clock.Now()
	805	+	afterAuthenticated = t.afterAuthenticatedPacket
	806	+	}
	807	+	t.mu.Unlock()
798	808		t.recvMu.Unlock()
799		-	t.recordFragmentResult(inflight, err == nil)
	809	+	if afterAuthenticated != nil {
	810	+	afterAuthenticated()
	811	+	}
800	812		if err != nil {
801	813		t.notifyMalformedFragment()
802	814		continue

@@ -819,21 +831,6 @@ func (t *Transport) notifyMalformedFragment() {
819	831		}
820	832		}
821	833
822		-	func (t *Transport) recordFragmentResult(inflight int, accepted bool) {
823		-	t.mu.Lock()
824		-	t.reassemblyInflight = inflight
825		-	if !accepted {
826		-	t.mu.Unlock()
827		-	return
828		-	}
829		-	t.lastAuthenticatedPacket = t.clock.Now()
830		-	after := t.afterAuthenticatedPacket
831		-	t.mu.Unlock()
832		-	if after != nil {
833		-	after()
834		-	}
835		-	}
836		-
837	834		func (t *Transport) recordCompleteRecord() {
838	835		t.mu.Lock()
839	836		t.lastCompleteRecord = t.clock.Now()

@@ -1310,6 +1307,14 @@ func (t *Transport) emitDiagnostic() {
1310	1307		if t.diagnosticCh == nil {
1311	1308		return
1312	1309		}
	1310	+	d := t.diagnosticSnapshot()
	1311	+	select {
	1312	+	case t.diagnosticCh <- d:
	1313	+	default:
	1314	+	}
	1315	+	}
	1316	+
	1317	+	func (t *Transport) diagnosticSnapshot() Diagnostic {
1313	1318		t.mu.Lock()
1314	1319		now := t.clock.Now()
1315	1320		lastAuthenticatedPacket := t.lastAuthenticatedPacket

@@ -1327,7 +1332,7 @@ func (t *Transport) emitDiagnostic() {
1327	1332		reassemblyInflight := t.reassemblyInflight
1328	1333		t.mu.Unlock()
1329	1334
1330		-	d := Diagnostic{
	1335	+	return Diagnostic{
1331	1336		At: now,
1332	1337		State: state,
1333	1338		SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),

@@ -1338,10 +1343,6 @@ func (t *Transport) emitDiagnostic() {
1338	1343		Retransmits: retransmits,
1339	1344		ReassemblyInflight: reassemblyInflight,
1340	1345		}
1341		-	select {
1342		-	case t.diagnosticCh <- d:
1343		-	default:
1344		-	}
1345	1346		}
1346	1347
1347	1348		func diagnosticAge(now, then time.Time) time.Duration {

 		}
 		payload, complete, err := t.reasm.Add(frag)
 		inflight := t.reasm.Inflight()
+		// Keep the diagnostic mirror ordered with reassembly mutations. Rebind can
+		// overlap old and new read loops, so recvMu must remain held through commit.
+		t.mu.Lock()
+		t.reassemblyInflight = inflight
+		var afterAuthenticated func()
+		if err == nil {
+			t.lastAuthenticatedPacket = t.clock.Now()
+			afterAuthenticated = t.afterAuthenticatedPacket
+		}
+		t.mu.Unlock()
 		t.recvMu.Unlock()
-		t.recordFragmentResult(inflight, err == nil)
+		if afterAuthenticated != nil {
+			afterAuthenticated()
+		}
 		if err != nil {
 			t.notifyMalformedFragment()
 			continue
 	}
 }
-func (t *Transport) recordFragmentResult(inflight int, accepted bool) {
-	t.mu.Lock()
-	t.reassemblyInflight = inflight
-	if !accepted {
-		t.mu.Unlock()
-		return
-	}
-	t.lastAuthenticatedPacket = t.clock.Now()
-	after := t.afterAuthenticatedPacket
-	t.mu.Unlock()
-	if after != nil {
-		after()
-	}
-}
 func (t *Transport) recordCompleteRecord() {
 	t.mu.Lock()
 	t.lastCompleteRecord = t.clock.Now()
 	if t.diagnosticCh == nil {
 		return
 	}
+	d := t.diagnosticSnapshot()
+	select {
+	case t.diagnosticCh <- d:
+	default:
+	}
+}
+func (t *Transport) diagnosticSnapshot() Diagnostic {
 	t.mu.Lock()
 	now := t.clock.Now()
 	lastAuthenticatedPacket := t.lastAuthenticatedPacket
 	reassemblyInflight := t.reassemblyInflight
 	t.mu.Unlock()
-	d := Diagnostic{
+	return Diagnostic{
 		At:                       now,
 		State:                    state,
 		SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
 		Retransmits:              retransmits,
 		ReassemblyInflight:       reassemblyInflight,
 	}
-	select {
-	case t.diagnosticCh <- d:
-	default:
-	}
 }
 func diagnosticAge(now, then time.Time) time.Duration {