test(dgram): characterize fresh and retransmit contention

commit: 96bc22d0c7decabc50f4e32e4767d36028fa7b0d
author: brice · 3mo ago · 2026-07-10 00:13 UTC
parent: 56d79df
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +46 -7

M internal/adapters/dgram/network_sim_test.go +38 -7


@@ -266,7 +266,7 @@ func TestTransportFloodClassification(t *testing.T) {
266	266		}
267	267		})
268	268
269		-	t.Run("large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
	269	+	t.Run("fresh output and large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
270	270		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
271	271		aPC, bPC := newPair()
272	272		inspect, err := pdgram.NewCodec(key())

@@ -291,6 +291,9 @@ func TestTransportFloodClassification(t *testing.T) {
291	291		retransmitBlocked := make(chan struct{}, 1)
292	292		releaseRetransmit := make(chan struct{})
293	293		probeSent := make(chan struct{}, 1)
	294	+	freshAtPace := make(chan struct{})
	295	+	allowFreshPace := make(chan struct{})
	296	+	freshReturned := make(chan error, 1)
294	297		writes := make(map[uint64]int)
295	298		var writesMu sync.Mutex
296	299		aPC.drop = func(pkt []byte, _ net.Addr) bool {

@@ -360,22 +363,50 @@ func TestTransportFloodClassification(t *testing.T) {
360	363
361	364		clk.advance(10 * time.Millisecond)
362	365		<-retransmitBlocked // resendPending is synchronously inside writePacket.
	366	+	if got := floodState(a).retransmits; got < 2 {
	367	+	t.Fatalf("retransmits=%d, want both overdue large records selected", got)
	368	+	}
	369	+
	370	+	// Start a third large output at this same fake-clock instant. The pacing
	371	+	// barrier proves its real Send reaches the shared pacing gate while the
	372	+	// retransmit owns it inside WriteTo.
	373	+	a.mu.Lock()
	374	+	a.outputNext = time.Time{}
	375	+	a.beforeDataPace = func() {
	376	+	close(freshAtPace)
	377	+	<-allowFreshPace
	378	+	}
	379	+	a.mu.Unlock()
	380	+	go func() {
	381	+	freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})
	382	+	}()
	383	+	<-freshAtPace
	384	+	a.mu.Lock()
	385	+	a.beforeDataPace = nil
	386	+	a.mu.Unlock()
	387	+	close(allowFreshPace)
	388	+	select {
	389	+	case err := <-freshReturned:
	390	+	t.Fatalf("fresh output completed while retransmit held pacing: %v", err)
	391	+	default:
	392	+	}
	393	+
363	394		clk.advance(time.Millisecond)
364	395		select {
365	396		case <-probeSent:
366		-	t.Fatal("heartbeat probe ran while a large retransmission held the resend loop")
	397	+	t.Fatal("heartbeat probe ran while fresh and retransmitted large output contended")
367	398		default:
368	399		}
369		-	if got := floodState(a).retransmits; got < 2 {
370		-	t.Fatalf("retransmits=%d, want both overdue large records selected", got)
371		-	}
372	400
373	401		close(releaseRetransmit)
	402	+	if err := <-freshReturned; err != nil {
	403	+	t.Fatal(err)
	404	+	}
374	405		<-probeSent
375	406		writesMu.Lock()
376	407		defer writesMu.Unlock()
377		-	if writes[1] < 2 \|\| writes[2] < 2 {
378		-	t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both records", writes)
	408	+	if writes[1] < 2 \|\| writes[2] < 2 \|\| writes[3] == 0 {
	409	+	t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both overdue records", writes)
379	410		}
380	411		})
381	412		}

 		}
 	})
-	t.Run("large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
+	t.Run("fresh output and large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
 		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
 		aPC, bPC := newPair()
 		inspect, err := pdgram.NewCodec(key())
 		retransmitBlocked := make(chan struct{}, 1)
 		releaseRetransmit := make(chan struct{})
 		probeSent := make(chan struct{}, 1)
+		freshAtPace := make(chan struct{})
+		allowFreshPace := make(chan struct{})
+		freshReturned := make(chan error, 1)
 		writes := make(map[uint64]int)
 		var writesMu sync.Mutex
 		aPC.drop = func(pkt []byte, _ net.Addr) bool {
 		clk.advance(10 * time.Millisecond)
 		<-retransmitBlocked // resendPending is synchronously inside writePacket.
+		if got := floodState(a).retransmits; got < 2 {
+			t.Fatalf("retransmits=%d, want both overdue large records selected", got)
+		}
+		// Start a third large output at this same fake-clock instant. The pacing
+		// barrier proves its real Send reaches the shared pacing gate while the
+		// retransmit owns it inside WriteTo.
+		a.mu.Lock()
+		a.outputNext = time.Time{}
+		a.beforeDataPace = func() {
+			close(freshAtPace)
+			<-allowFreshPace
+		}
+		a.mu.Unlock()
+		go func() {
+			freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 500)})
+		}()
+		<-freshAtPace
+		a.mu.Lock()
+		a.beforeDataPace = nil
+		a.mu.Unlock()
+		close(allowFreshPace)
+		select {
+		case err := <-freshReturned:
+			t.Fatalf("fresh output completed while retransmit held pacing: %v", err)
+		default:
+		}
 		clk.advance(time.Millisecond)
 		select {
 		case <-probeSent:
-			t.Fatal("heartbeat probe ran while a large retransmission held the resend loop")
+			t.Fatal("heartbeat probe ran while fresh and retransmitted large output contended")
 		default:
 		}
-		if got := floodState(a).retransmits; got < 2 {
-			t.Fatalf("retransmits=%d, want both overdue large records selected", got)
-		}
 		close(releaseRetransmit)
+		if err := <-freshReturned; err != nil {
+			t.Fatal(err)
+		}
 		<-probeSent
 		writesMu.Lock()
 		defer writesMu.Unlock()
-		if writes[1] < 2 || writes[2] < 2 {
+		if writes[1] < 2 || writes[2] < 2 || writes[3] == 0 {
-			t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both records", writes)
+			t.Fatalf("writes=%v, want fresh large traffic and retransmissions for both overdue records", writes)
 		}
 	})
 }

M internal/adapters/dgram/transport.go +8 -0


@@ -192,6 +192,8 @@ type Transport struct {
192	192		dataPaceMu sync.Mutex
193	193		dataPaceRemaining int
194	194		dataPaceNext time.Time
	195	+	// beforeDataPace is a test synchronization hook. It is read under mu.
	196	+	beforeDataPace func()
195	197		// outboundMu orders first transmissions and paced batches. Retransmits use
196	198		// writeMu directly because they do not establish new sequence order.
197	199		outboundMu sync.Mutex

@@ -562,6 +564,12 @@ func (t *Transport) sendProbe(kind byte, id uint64) error {
562	564		}
563	565		func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
564	566		if paced {
	567	+	t.mu.Lock()
	568	+	beforeDataPace := t.beforeDataPace
	569	+	t.mu.Unlock()
	570	+	if beforeDataPace != nil {
	571	+	beforeDataPace()
	572	+	}
565	573		t.dataPaceMu.Lock()
566	574		defer t.dataPaceMu.Unlock()
567	575		}


@@ -192,6 +192,8 @@ type Transport struct {
192	dataPaceMu sync.Mutex	192		dataPaceMu sync.Mutex
193	dataPaceRemaining int	193		dataPaceRemaining int
194	dataPaceNext time.Time	194		dataPaceNext time.Time
		195	+	// beforeDataPace is a test synchronization hook. It is read under mu.
		196	+	beforeDataPace func()
195	// outboundMu orders first transmissions and paced batches. Retransmits use	197		// outboundMu orders first transmissions and paced batches. Retransmits use
196	// writeMu directly because they do not establish new sequence order.	198		// writeMu directly because they do not establish new sequence order.
197	outboundMu sync.Mutex	199		outboundMu sync.Mutex

@@ -562,6 +564,12 @@ func (t *Transport) sendProbe(kind byte, id uint64) error {
562	}	564		}
563	func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {	565		func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
564	if paced {	566		if paced {
		567	+	t.mu.Lock()
		568	+	beforeDataPace := t.beforeDataPace
		569	+	t.mu.Unlock()
		570	+	if beforeDataPace != nil {
		571	+	beforeDataPace()
		572	+	}
565	t.dataPaceMu.Lock()	573		t.dataPaceMu.Lock()
566	defer t.dataPaceMu.Unlock()	574		defer t.dataPaceMu.Unlock()
567	}	575		}