fix(dgram): isolate UDP health from bulk retransmits

commit: 7bc940975cef17ff7aebe731fd3c6d932a200ca9
author: brice · 3mo ago · 2026-07-10 05:46 UTC
parent: ba81cef
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +638 -86

M internal/adapters/dgram/control.go +99 -22


@@ -2,16 +2,23 @@ package dgram
2	2
3	3		import (
4	4		"encoding/binary"
5		-	"runtime"
	5	+	"errors"
	6	+	"net"
	7	+	"time"
6	8
	9	+	"github.com/bnema/vev/internal/ports"
7	10		pdgram "github.com/bnema/vev/pkg/dgram"
8	11		)
9	12
10		-	const controlQueueSize = 16
	13	+	const (
	14	+	controlQueueSize = 16
	15	+	maxACKDelay = 25 * time.Millisecond
	16	+	)
11	17
12	18		type controlRecord struct {
13		-	kind byte
14		-	id uint64
	19	+	kind byte
	20	+	id uint64
	21	+	result chan<- error
15	22		}
16	23
17	24		// queueACK retains the greatest cumulative acknowledgement without ever making

@@ -33,16 +40,17 @@ func (t *Transport) queueACK(seq uint64) {
33	40		case t.ackWake <- struct{}{}:
34	41		default:
35	42		}
36		-	// Let the dedicated writer claim queued control before delivery wakes a
37		-	// consumer; this remains a non-blocking receive-path operation.
38		-	runtime.Gosched()
39	43		}
40	44
41	45		// queueControl is deliberately lossy when the bounded control queue is full:
42	46		// probes and pongs are advisory and must not stall authenticated receipt.
43	47		func (t *Transport) queueControl(kind byte, id uint64) bool {
	48	+	return t.queueControlRecord(controlRecord{kind: kind, id: id})
	49	+	}
	50	+
	51	+	func (t *Transport) queueControlRecord(record controlRecord) bool {
44	52		select {
45		-	case t.control <- controlRecord{kind: kind, id: id}:
	53	+	case t.control <- record:
46	54		return true
47	55		default:
48	56		return false

@@ -59,54 +67,123 @@ func (t *Transport) takeACK() (uint64, bool) {
59	67		}
60	68
61	69		func (t *Transport) controlLoop() {
	70	+	var ackTimer ports.Timer
	71	+	var ackTimerC <-chan time.Time
	72	+	defer func() {
	73	+	if ackTimer != nil {
	74	+	ackTimer.Stop()
	75	+	}
	76	+	}()
	77	+	scheduleACK := func() {
	78	+	if ackTimerC != nil {
	79	+	return
	80	+	}
	81	+	if ackTimer == nil {
	82	+	ackTimer = t.clock.NewTimer(maxACKDelay)
	83	+	} else {
	84	+	ackTimer.Reset(maxACKDelay)
	85	+	}
	86	+	ackTimerC = ackTimer.C()
	87	+	}
	88	+	dispatchACK := func() {
	89	+	seq, ok := t.takeACK()
	90	+	if !ok {
	91	+	return
	92	+	}
	93	+	select {
	94	+	case t.ackSend <- seq:
	95	+	default:
	96	+	// Preserve the cumulative maximum until the bounded ACK sender has
	97	+	// capacity again.
	98	+	t.queueACK(seq)
	99	+	}
	100	+	}
	101	+
62	102		for {
63		-	// Probes and pongs always win over an ACK that is ready to send.
	103	+	// Check ACK scheduling explicitly so a sustained stream of advisory
	104	+	// controls cannot prevent the bounded coalescing deadline from starting
	105	+	// or expiring. Dispatch itself never writes to the socket.
64	106		select {
65		-	case record := <-t.control:
66		-	t.sendControl(record.kind, record.id)
	107	+	case <-ackTimerC:
	108	+	ackTimerC = nil
	109	+	dispatchACK()
	110	+	continue
	111	+	default:
	112	+	}
	113	+	select {
	114	+	case <-t.ackWake:
	115	+	scheduleACK()
67	116		continue
68	117		default:
69	118		}
70	119
71	120		select {
72	121		case record := <-t.control:
73		-	t.sendControl(record.kind, record.id)
74		-	case <-t.ackWake:
75		-	if seq, ok := t.takeACK(); ok {
76		-	t.sendControl(recAck, seq)
	122	+	err := t.sendControl(record.kind, record.id)
	123	+	if record.result != nil {
	124	+	select {
	125	+	case record.result <- err:
	126	+	default:
	127	+	}
77	128		}
	129	+	case <-t.ackWake:
	130	+	scheduleACK()
	131	+	case <-ackTimerC:
	132	+	ackTimerC = nil
	133	+	dispatchACK()
	134	+	case <-t.done:
	135	+	return
	136	+	}
	137	+	}
	138	+	}
	139	+
	140	+	func (t *Transport) ackSendLoop() {
	141	+	for {
	142	+	select {
	143	+	case seq := <-t.ackSend:
	144	+	_ = t.sendControl(recAck, seq)
78	145		case <-t.done:
79	146		return
80	147		}
81	148		}
82	149		}
83	150
84		-	func (t *Transport) sendControl(kind byte, id uint64) {
	151	+	func (t *Transport) sendControl(kind byte, id uint64) error {
85	152		var payload [9]byte
86	153		payload[0] = kind
87	154		binary.BigEndian.PutUint64(payload[1:], id)
88	155		frags, err := pdgram.FragmentPayload(t.nextCounter(), payload[:], t.mtu-pdgram.HeaderSize-t.codec.Overhead())
89	156		if err != nil {
90		-	return
	157	+	return err
91	158		}
92	159		for _, frag := range frags {
93	160		raw, err := pdgram.MarshalFragment(frag)
94	161		if err != nil {
95		-	return
	162	+	return err
96	163		}
97	164		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
98	165		t.mu.Lock()
99	166		pc := t.pc
100	167		peer := t.peer
101	168		closed := t.closed
	169	+	closeErr := t.closeErr
	170	+	writeTimeout := t.writeTimeout
	171	+	writeDeadlines := t.writeDeadlines
102	172		t.mu.Unlock()
103		-	if closed \|\| peer == nil {
104		-	return
	173	+	if closed {
	174	+	if closeErr != nil {
	175	+	return closeErr
	176	+	}
	177	+	return net.ErrClosed
	178	+	}
	179	+	if peer == nil {
	180	+	return errors.New("dgram: control peer unavailable")
105	181		}
106	182		// PacketConn permits concurrent callers. Control traffic deliberately
107	183		// bypasses data's write lock so a retransmit cannot starve heartbeats.
108		-	if _, err := pc.WriteTo(pkt, peer); err != nil {
109		-	return
	184	+	if err := t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout); err != nil {
	185	+	return err
110	186		}
111	187		}
	188	+	return nil
112	189		}

 import (
 	"encoding/binary"
-	"runtime"
+	"errors"
+	"net"
+	"time"
+	"github.com/bnema/vev/internal/ports"
 	pdgram "github.com/bnema/vev/pkg/dgram"
 )
-const controlQueueSize = 16
+const (
+	controlQueueSize = 16
+	maxACKDelay      = 25 * time.Millisecond
+)
 type controlRecord struct {
-	kind byte
+	kind   byte
-	id   uint64
+	id     uint64
+	result chan<- error
 }
 // queueACK retains the greatest cumulative acknowledgement without ever making
 	case t.ackWake <- struct{}{}:
 	default:
 	}
-	// Let the dedicated writer claim queued control before delivery wakes a
-	// consumer; this remains a non-blocking receive-path operation.
-	runtime.Gosched()
 }
 // queueControl is deliberately lossy when the bounded control queue is full:
 // probes and pongs are advisory and must not stall authenticated receipt.
 func (t *Transport) queueControl(kind byte, id uint64) bool {
+	return t.queueControlRecord(controlRecord{kind: kind, id: id})
+}
+func (t *Transport) queueControlRecord(record controlRecord) bool {
 	select {
-	case t.control <- controlRecord{kind: kind, id: id}:
+	case t.control <- record:
 		return true
 	default:
 		return false
 }
 func (t *Transport) controlLoop() {
+	var ackTimer ports.Timer
+	var ackTimerC <-chan time.Time
+	defer func() {
+		if ackTimer != nil {
+			ackTimer.Stop()
+		}
+	}()
+	scheduleACK := func() {
+		if ackTimerC != nil {
+			return
+		}
+		if ackTimer == nil {
+			ackTimer = t.clock.NewTimer(maxACKDelay)
+		} else {
+			ackTimer.Reset(maxACKDelay)
+		}
+		ackTimerC = ackTimer.C()
+	}
+	dispatchACK := func() {
+		seq, ok := t.takeACK()
+		if !ok {
+			return
+		}
+		select {
+		case t.ackSend <- seq:
+		default:
+			// Preserve the cumulative maximum until the bounded ACK sender has
+			// capacity again.
+			t.queueACK(seq)
+		}
+	}
 	for {
-		// Probes and pongs always win over an ACK that is ready to send.
+		// Check ACK scheduling explicitly so a sustained stream of advisory
+		// controls cannot prevent the bounded coalescing deadline from starting
+		// or expiring. Dispatch itself never writes to the socket.
 		select {
-		case record := <-t.control:
+		case <-ackTimerC:
-			t.sendControl(record.kind, record.id)
+			ackTimerC = nil
+			dispatchACK()
+			continue
+		default:
+		}
+		select {
+		case <-t.ackWake:
+			scheduleACK()
 			continue
 		default:
 		}
 		select {
 		case record := <-t.control:
-			t.sendControl(record.kind, record.id)
+			err := t.sendControl(record.kind, record.id)
-		case <-t.ackWake:
+			if record.result != nil {
-			if seq, ok := t.takeACK(); ok {
+				select {
-				t.sendControl(recAck, seq)
+				case record.result <- err:
+				default:
+				}
 			}
+		case <-t.ackWake:
+			scheduleACK()
+		case <-ackTimerC:
+			ackTimerC = nil
+			dispatchACK()
+		case <-t.done:
+			return
+		}
+	}
+}
+func (t *Transport) ackSendLoop() {
+	for {
+		select {
+		case seq := <-t.ackSend:
+			_ = t.sendControl(recAck, seq)
 		case <-t.done:
 			return
 		}
 	}
 }
-func (t *Transport) sendControl(kind byte, id uint64) {
+func (t *Transport) sendControl(kind byte, id uint64) error {
 	var payload [9]byte
 	payload[0] = kind
 	binary.BigEndian.PutUint64(payload[1:], id)
 	frags, err := pdgram.FragmentPayload(t.nextCounter(), payload[:], t.mtu-pdgram.HeaderSize-t.codec.Overhead())
 	if err != nil {
-		return
+		return err
 	}
 	for _, frag := range frags {
 		raw, err := pdgram.MarshalFragment(frag)
 		if err != nil {
-			return
+			return err
 		}
 		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
 		t.mu.Lock()
 		pc := t.pc
 		peer := t.peer
 		closed := t.closed
+		closeErr := t.closeErr
+		writeTimeout := t.writeTimeout
+		writeDeadlines := t.writeDeadlines
 		t.mu.Unlock()
-		if closed || peer == nil {
+		if closed {
-			return
+			if closeErr != nil {
+				return closeErr
+			}
+			return net.ErrClosed
+		}
+		if peer == nil {
+			return errors.New("dgram: control peer unavailable")
 		}
 		// PacketConn permits concurrent callers. Control traffic deliberately
 		// bypasses data's write lock so a retransmit cannot starve heartbeats.
-		if _, err := pc.WriteTo(pkt, peer); err != nil {
+		if err := t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout); err != nil {
-			return
+			return err
 		}
 	}
+	return nil
 }

M internal/adapters/dgram/control_plane_test.go +328 -2

@@ -1,7 +1,11 @@
1	1		package dgram
2	2
3	3		import (
	4	+	"context"
	5	+	"encoding/binary"
	6	+	"errors"
4	7		"net"
	8	+	"sync"
5	9		"testing"
6	10		"time"
7	11

@@ -9,6 +13,40 @@ import (
9	13		pdgram "github.com/bnema/vev/pkg/dgram"
10	14		)
11	15
	16	+	type stagedTimeoutPC struct {
	17	+	mu sync.Mutex
	18	+	deadline time.Time
	19	+	calls int
	20	+	started chan struct{}
	21	+	release chan struct{}
	22	+	}
	23	+
	24	+	func (p *stagedTimeoutPC) ReadFrom([]byte) (int, net.Addr, error) {
	25	+	return 0, nil, net.ErrClosed
	26	+	}
	27	+	func (p *stagedTimeoutPC) WriteTo(b []byte, _ net.Addr) (int, error) {
	28	+	p.mu.Lock()
	29	+	p.calls++
	30	+	call := p.calls
	31	+	p.mu.Unlock()
	32	+	if call <= 2 {
	33	+	p.started <- struct{}{}
	34	+	<-p.release
	35	+	return 0, timeoutErr{}
	36	+	}
	37	+	return len(b), nil
	38	+	}
	39	+	func (p *stagedTimeoutPC) Close() error { return nil }
	40	+	func (p *stagedTimeoutPC) LocalAddr() net.Addr { return testAddr("staged") }
	41	+	func (p *stagedTimeoutPC) SetDeadline(time.Time) error { return nil }
	42	+	func (p *stagedTimeoutPC) SetReadDeadline(time.Time) error { return nil }
	43	+	func (p *stagedTimeoutPC) SetWriteDeadline(d time.Time) error {
	44	+	p.mu.Lock()
	45	+	p.deadline = d
	46	+	p.mu.Unlock()
	47	+	return nil
	48	+	}
	49	+
12	50		func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
13	51		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
14	52		aPC, bPC := newPair()

@@ -47,6 +85,10 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
47	85		t.Fatal(err)
48	86		}
49	87		awaitSignal(t, contacts, "first authenticated record")
	88	+	for range 7 {
	89	+	awaitSignal(t, clk.timerCreated, "transport and ACK timer creation")
	90	+	}
	91	+	clk.advance(maxACKDelay)
50	92		awaitSignal(t, ackStarted, "blocked ACK")
51	93
52	94		clk.advance(time.Second)

@@ -54,6 +96,12 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
54	96		t.Fatal(err)
55	97		}
56	98		awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")
	99	+	wantComplete := clk.Now()
	100	+	eventually(t, time.Second, func() bool {
	101	+	a.mu.Lock()
	102	+	defer a.mu.Unlock()
	103	+	return a.lastCompleteRecord.Equal(wantComplete)
	104	+	})
57	105		a.emitDiagnostic()
58	106		diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")
59	107		if diagnostic.SinceCompleteRecord != 0 {

@@ -61,6 +109,237 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
61	109		}
62	110		}
63	111
	112	+	func TestCumulativeACKCoalescesForBoundedDelay(t *testing.T) {
	113	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	114	+	aPC, bPC := newPair()
	115	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	116	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	117	+	})
	118	+	if err != nil {
	119	+	t.Fatal(err)
	120	+	}
	121	+	defer func() { _ = a.Close() }()
	122	+	for range 3 {
	123	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
	124	+	}
	125	+
	126	+	acks := make(chan uint64, 2)
	127	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
	128	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
	129	+	if err != nil {
	130	+	t.Errorf("open outbound ACK: %v", err)
	131	+	return true
	132	+	}
	133	+	frag, err := pdgram.UnmarshalFragment(raw)
	134	+	if err != nil {
	135	+	t.Errorf("unmarshal outbound ACK: %v", err)
	136	+	return true
	137	+	}
	138	+	if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck {
	139	+	acks <- binary.BigEndian.Uint64(frag.Data[1:])
	140	+	}
	141	+	return true
	142	+	}
	143	+
	144	+	a.queueACK(1)
	145	+	a.queueACK(2)
	146	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
	147	+	clk.advance(maxACKDelay - time.Millisecond)
	148	+	select {
	149	+	case seq := <-acks:
	150	+	t.Fatalf("ACK %d sent before coalescing delay", seq)
	151	+	default:
	152	+	}
	153	+	clk.advance(time.Millisecond)
	154	+	if seq := awaitResult(t, acks, "coalesced ACK"); seq != 2 {
	155	+	t.Fatalf("ACK=%d, want cumulative maximum 2", seq)
	156	+	}
	157	+	select {
	158	+	case seq := <-acks:
	159	+	t.Fatalf("extra ACK %d sent for coalesced records", seq)
	160	+	default:
	161	+	}
	162	+	}
	163	+
	164	+	func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {
	165	+	aPC, bPC := newPair()
	166	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	167	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
	168	+	})
	169	+	if err != nil {
	170	+	t.Fatal(err)
	171	+	}
	172	+	b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
	173	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
	174	+	})
	175	+	if err != nil {
	176	+	_ = a.Close()
	177	+	t.Fatal(err)
	178	+	}
	179	+	defer closeFloodTransports(a, b)
	180	+
	181	+	a.writeMu.Lock()
	182	+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	183	+	defer cancel()
	184	+	done := make(chan error, 1)
	185	+	go func() { done <- a.Probe(ctx) }()
	186	+	select {
	187	+	case err := <-done:
	188	+	a.writeMu.Unlock()
	189	+	if err != nil {
	190	+	t.Fatalf("Probe failed: %v", err)
	191	+	}
	192	+	case <-time.After(50 * time.Millisecond):
	193	+	a.writeMu.Unlock()
	194	+	<-done
	195	+	t.Fatal("Probe blocked behind data writer")
	196	+	}
	197	+	}
	198	+
	199	+	func TestProbeDoesNotBlockBehindACKWrite(t *testing.T) {
	200	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	201	+	aPC, bPC := newPair()
	202	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	203	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	204	+	})
	205	+	if err != nil {
	206	+	t.Fatal(err)
	207	+	}
	208	+	defer func() { _ = a.Close() }()
	209	+	for range 3 {
	210	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
	211	+	}
	212	+
	213	+	ackStarted := make(chan struct{}, 1)
	214	+	probeStarted := make(chan struct{}, 1)
	215	+	releaseACK := make(chan struct{})
	216	+	defer close(releaseACK)
	217	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
	218	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
	219	+	if err != nil {
	220	+	return true
	221	+	}
	222	+	frag, err := pdgram.UnmarshalFragment(raw)
	223	+	if err != nil \|\| frag.Count != 1 \|\| len(frag.Data) != 9 {
	224	+	return true
	225	+	}
	226	+	switch frag.Data[0] {
	227	+	case recAck:
	228	+	ackStarted <- struct{}{}
	229	+	<-releaseACK
	230	+	case recProbe:
	231	+	probeStarted <- struct{}{}
	232	+	}
	233	+	return true
	234	+	}
	235	+
	236	+	a.queueACK(1)
	237	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
	238	+	clk.advance(maxACKDelay)
	239	+	awaitSignal(t, ackStarted, "blocked ACK")
	240	+	if !a.queueControl(recProbe, 1) {
	241	+	t.Fatal("probe control queue unexpectedly full")
	242	+	}
	243	+	awaitSignal(t, probeStarted, "probe while ACK write is blocked")
	244	+	}
	245	+
	246	+	func TestControlWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
	247	+	pc := newBlockingWritePC()
	248	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
	249	+	WriteTimeout: 10 * time.Millisecond,
	250	+	ResendAfter: time.Hour,
	251	+	Heartbeat: time.Hour,
	252	+	})
	253	+	if err != nil {
	254	+	t.Fatal(err)
	255	+	}
	256	+	defer func() { _ = tr.Close() }()
	257	+
	258	+	done := make(chan error, 1)
	259	+	go func() { done <- tr.sendControl(recAck, 1) }()
	260	+	select {
	261	+	case err := <-done:
	262	+	var timeout interface{ Timeout() bool }
	263	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
	264	+	t.Fatalf("control write err=%v, want timeout", err)
	265	+	}
	266	+	case <-time.After(250 * time.Millisecond):
	267	+	t.Fatal("control write blocked without honoring WriteTimeout")
	268	+	}
	269	+	}
	270	+
	271	+	func TestConcurrentWriterCannotClearEarlierDeadline(t *testing.T) {
	272	+	pc := &deadlineCapturePC{addr: testAddr("a"), done: make(chan struct{})}
	273	+	state := newWriteDeadlineState(pc)
	274	+	earlier := time.Now().Add(time.Second)
	275	+	later := earlier.Add(time.Second)
	276	+	finishEarlier, err := state.begin(earlier)
	277	+	if err != nil {
	278	+	t.Fatal(err)
	279	+	}
	280	+	defer finishEarlier()
	281	+	finishLater, err := state.begin(later)
	282	+	if err != nil {
	283	+	t.Fatal(err)
	284	+	}
	285	+	finishLater()
	286	+
	287	+	got, _ := pc.deadline.Load().(time.Time)
	288	+	if !got.Equal(earlier) {
	289	+	t.Fatalf("deadline after concurrent writer completed=%v, want earlier active deadline %v", got, earlier)
	290	+	}
	291	+	}
	292	+
	293	+	func TestFreshControlWriteRetriesAfterInheritedDeadline(t *testing.T) {
	294	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	295	+	pc := &stagedTimeoutPC{started: make(chan struct{}, 2), release: make(chan struct{})}
	296	+	tr := &Transport{clock: clk}
	297	+	deadlines := newWriteDeadlineState(pc)
	298	+	oldDone := make(chan error, 1)
	299	+	go func() {
	300	+	oldDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("old"), deadlines, 10*time.Millisecond)
	301	+	}()
	302	+	awaitSignal(t, pc.started, "older blocked write")
	303	+	clk.advance(5 * time.Millisecond)
	304	+	freshDone := make(chan error, 1)
	305	+	go func() {
	306	+	freshDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("fresh"), deadlines, 100*time.Millisecond)
	307	+	}()
	308	+	awaitSignal(t, pc.started, "fresh write sharing earlier deadline")
	309	+	clk.advance(5 * time.Millisecond)
	310	+	close(pc.release)
	311	+	if err := awaitResult(t, oldDone, "older write timeout"); err == nil {
	312	+	t.Fatal("older write succeeded, want timeout")
	313	+	}
	314	+	if err := awaitResult(t, freshDone, "fresh write retry"); err != nil {
	315	+	t.Fatalf("fresh write failed at inherited deadline: %v", err)
	316	+	}
	317	+	}
	318	+
	319	+	func TestProbeReturnsLocalWriteError(t *testing.T) {
	320	+	aPC, bPC := newPair()
	321	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	322	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
	323	+	})
	324	+	if err != nil {
	325	+	t.Fatal(err)
	326	+	}
	327	+	defer func() { _ = a.Close() }()
	328	+	if err := bPC.Close(); err != nil {
	329	+	t.Fatal(err)
	330	+	}
	331	+
	332	+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	333	+	defer cancel()
	334	+	start := time.Now()
	335	+	if err := a.Probe(ctx); err == nil \|\| errors.Is(err, context.DeadlineExceeded) {
	336	+	t.Fatalf("Probe err=%v, want immediate local write error", err)
	337	+	}
	338	+	if elapsed := time.Since(start); elapsed > 250*time.Millisecond {
	339	+	t.Fatalf("Probe returned local write error after %v", elapsed)
	340	+	}
	341	+	}
	342	+
64	343		func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
65	344		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
66	345		aPC, bPC := newPair()

@@ -72,8 +351,8 @@ func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
72	351		t.Fatal(err)
73	352		}
74	353		defer func() { _ = a.Close() }()
75		-	for range 2 {
76		-	awaitSignal(t, clk.timerCreated, "resend and heartbeat timer creation")
	354	+	for range 3 {
	355	+	awaitSignal(t, clk.timerCreated, "retransmit, health, and heartbeat timer creation")
77	356		}
78	357
79	358		retransmitStarted := make(chan struct{}, 1)

@@ -120,3 +399,50 @@ func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
120	399		awaitSignal(t, heartbeats, "heartbeat during retransmit storm")
121	400		}
122	401		}
	402	+
	403	+	func TestHealthChecksContinueDuringRetransmitStorm(t *testing.T) {
	404	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	405	+	aPC, bPC := newPair()
	406	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
	407	+	Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond,
	408	+	Heartbeat: time.Hour, DegradedAfter: 10 * time.Millisecond,
	409	+	})
	410	+	if err != nil {
	411	+	t.Fatal(err)
	412	+	}
	413	+	defer func() { _ = a.Close() }()
	414	+	for range 3 {
	415	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
	416	+	}
	417	+
	418	+	retransmitStarted := make(chan struct{}, 1)
	419	+	releaseRetransmit := make(chan struct{})
	420	+	defer close(releaseRetransmit)
	421	+	aPC.drop = func(_ []byte, _ net.Addr) bool {
	422	+	select {
	423	+	case retransmitStarted <- struct{}{}:
	424	+	<-releaseRetransmit
	425	+	default:
	426	+	}
	427	+	return true
	428	+	}
	429	+
	430	+	now := clk.Now()
	431	+	a.mu.Lock()
	432	+	a.pending[1] = &pending{
	433	+	frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
	434	+	enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
	435	+	}
	436	+	a.mu.Unlock()
	437	+
	438	+	clk.advance(10 * time.Millisecond)
	439	+	awaitSignal(t, retransmitStarted, "blocked retransmit")
	440	+	select {
	441	+	case event := <-a.LinkEvents():
	442	+	if event.State != ports.LinkStateDegraded {
	443	+	t.Fatalf("link state=%s, want degraded", event.State)
	444	+	}
	445	+	case <-time.After(time.Second):
	446	+	t.Fatal("health check did not run while retransmit sender was blocked")
	447	+	}
	448	+	}

@@ -1,7 +1,11 @@
1		package dgram	1		package dgram
2			2
3		import (	3		import (
			4	+	"context"
			5	+	"encoding/binary"
			6	+	"errors"
4		"net"	7		"net"
			8	+	"sync"
5		"testing"	9		"testing"
6		"time"	10		"time"
7			11

@@ -9,6 +13,40 @@ import (
9		pdgram "github.com/bnema/vev/pkg/dgram"	13		pdgram "github.com/bnema/vev/pkg/dgram"
10		)	14		)
11			15
			16	+	type stagedTimeoutPC struct {
			17	+	mu sync.Mutex
			18	+	deadline time.Time
			19	+	calls int
			20	+	started chan struct{}
			21	+	release chan struct{}
			22	+	}
			23	+
			24	+	func (p *stagedTimeoutPC) ReadFrom([]byte) (int, net.Addr, error) {
			25	+	return 0, nil, net.ErrClosed
			26	+	}
			27	+	func (p *stagedTimeoutPC) WriteTo(b []byte, _ net.Addr) (int, error) {
			28	+	p.mu.Lock()
			29	+	p.calls++
			30	+	call := p.calls
			31	+	p.mu.Unlock()
			32	+	if call <= 2 {
			33	+	p.started <- struct{}{}
			34	+	<-p.release
			35	+	return 0, timeoutErr{}
			36	+	}
			37	+	return len(b), nil
			38	+	}
			39	+	func (p *stagedTimeoutPC) Close() error { return nil }
			40	+	func (p *stagedTimeoutPC) LocalAddr() net.Addr { return testAddr("staged") }
			41	+	func (p *stagedTimeoutPC) SetDeadline(time.Time) error { return nil }
			42	+	func (p *stagedTimeoutPC) SetReadDeadline(time.Time) error { return nil }
			43	+	func (p *stagedTimeoutPC) SetWriteDeadline(d time.Time) error {
			44	+	p.mu.Lock()
			45	+	p.deadline = d
			46	+	p.mu.Unlock()
			47	+	return nil
			48	+	}
			49	+
12		func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {	50		func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
13		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	51		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
14		aPC, bPC := newPair()	52		aPC, bPC := newPair()

@@ -47,6 +85,10 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
47		t.Fatal(err)	85		t.Fatal(err)
48		}	86		}
49		awaitSignal(t, contacts, "first authenticated record")	87		awaitSignal(t, contacts, "first authenticated record")
			88	+	for range 7 {
			89	+	awaitSignal(t, clk.timerCreated, "transport and ACK timer creation")
			90	+	}
			91	+	clk.advance(maxACKDelay)
50		awaitSignal(t, ackStarted, "blocked ACK")	92		awaitSignal(t, ackStarted, "blocked ACK")
51			93
52		clk.advance(time.Second)	94		clk.advance(time.Second)

@@ -54,6 +96,12 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
54		t.Fatal(err)	96		t.Fatal(err)
55		}	97		}
56		awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")	98		awaitSignal(t, contacts, "second authenticated record while ACK write is blocked")
			99	+	wantComplete := clk.Now()
			100	+	eventually(t, time.Second, func() bool {
			101	+	a.mu.Lock()
			102	+	defer a.mu.Unlock()
			103	+	return a.lastCompleteRecord.Equal(wantComplete)
			104	+	})
57		a.emitDiagnostic()	105		a.emitDiagnostic()
58		diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")	106		diagnostic := awaitResult(t, a.diagnosticCh, "second-record diagnostic")
59		if diagnostic.SinceCompleteRecord != 0 {	107		if diagnostic.SinceCompleteRecord != 0 {

@@ -61,6 +109,237 @@ func TestReadLoopDoesNotWriteACKSynchronously(t *testing.T) {
61		}	109		}
62		}	110		}
63			111
			112	+	func TestCumulativeACKCoalescesForBoundedDelay(t *testing.T) {
			113	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			114	+	aPC, bPC := newPair()
			115	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
			116	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
			117	+	})
			118	+	if err != nil {
			119	+	t.Fatal(err)
			120	+	}
			121	+	defer func() { _ = a.Close() }()
			122	+	for range 3 {
			123	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
			124	+	}
			125	+
			126	+	acks := make(chan uint64, 2)
			127	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
			128	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
			129	+	if err != nil {
			130	+	t.Errorf("open outbound ACK: %v", err)
			131	+	return true
			132	+	}
			133	+	frag, err := pdgram.UnmarshalFragment(raw)
			134	+	if err != nil {
			135	+	t.Errorf("unmarshal outbound ACK: %v", err)
			136	+	return true
			137	+	}
			138	+	if frag.Count == 1 && len(frag.Data) == 9 && frag.Data[0] == recAck {
			139	+	acks <- binary.BigEndian.Uint64(frag.Data[1:])
			140	+	}
			141	+	return true
			142	+	}
			143	+
			144	+	a.queueACK(1)
			145	+	a.queueACK(2)
			146	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
			147	+	clk.advance(maxACKDelay - time.Millisecond)
			148	+	select {
			149	+	case seq := <-acks:
			150	+	t.Fatalf("ACK %d sent before coalescing delay", seq)
			151	+	default:
			152	+	}
			153	+	clk.advance(time.Millisecond)
			154	+	if seq := awaitResult(t, acks, "coalesced ACK"); seq != 2 {
			155	+	t.Fatalf("ACK=%d, want cumulative maximum 2", seq)
			156	+	}
			157	+	select {
			158	+	case seq := <-acks:
			159	+	t.Fatalf("extra ACK %d sent for coalesced records", seq)
			160	+	default:
			161	+	}
			162	+	}
			163	+
			164	+	func TestProbeDoesNotBlockOnDataWriter(t *testing.T) {
			165	+	aPC, bPC := newPair()
			166	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
			167	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
			168	+	})
			169	+	if err != nil {
			170	+	t.Fatal(err)
			171	+	}
			172	+	b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, Options{
			173	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
			174	+	})
			175	+	if err != nil {
			176	+	_ = a.Close()
			177	+	t.Fatal(err)
			178	+	}
			179	+	defer closeFloodTransports(a, b)
			180	+
			181	+	a.writeMu.Lock()
			182	+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			183	+	defer cancel()
			184	+	done := make(chan error, 1)
			185	+	go func() { done <- a.Probe(ctx) }()
			186	+	select {
			187	+	case err := <-done:
			188	+	a.writeMu.Unlock()
			189	+	if err != nil {
			190	+	t.Fatalf("Probe failed: %v", err)
			191	+	}
			192	+	case <-time.After(50 * time.Millisecond):
			193	+	a.writeMu.Unlock()
			194	+	<-done
			195	+	t.Fatal("Probe blocked behind data writer")
			196	+	}
			197	+	}
			198	+
			199	+	func TestProbeDoesNotBlockBehindACKWrite(t *testing.T) {
			200	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			201	+	aPC, bPC := newPair()
			202	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
			203	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
			204	+	})
			205	+	if err != nil {
			206	+	t.Fatal(err)
			207	+	}
			208	+	defer func() { _ = a.Close() }()
			209	+	for range 3 {
			210	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
			211	+	}
			212	+
			213	+	ackStarted := make(chan struct{}, 1)
			214	+	probeStarted := make(chan struct{}, 1)
			215	+	releaseACK := make(chan struct{})
			216	+	defer close(releaseACK)
			217	+	aPC.drop = func(pkt []byte, _ net.Addr) bool {
			218	+	_, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil)
			219	+	if err != nil {
			220	+	return true
			221	+	}
			222	+	frag, err := pdgram.UnmarshalFragment(raw)
			223	+	if err != nil \|\| frag.Count != 1 \|\| len(frag.Data) != 9 {
			224	+	return true
			225	+	}
			226	+	switch frag.Data[0] {
			227	+	case recAck:
			228	+	ackStarted <- struct{}{}
			229	+	<-releaseACK
			230	+	case recProbe:
			231	+	probeStarted <- struct{}{}
			232	+	}
			233	+	return true
			234	+	}
			235	+
			236	+	a.queueACK(1)
			237	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
			238	+	clk.advance(maxACKDelay)
			239	+	awaitSignal(t, ackStarted, "blocked ACK")
			240	+	if !a.queueControl(recProbe, 1) {
			241	+	t.Fatal("probe control queue unexpectedly full")
			242	+	}
			243	+	awaitSignal(t, probeStarted, "probe while ACK write is blocked")
			244	+	}
			245	+
			246	+	func TestControlWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
			247	+	pc := newBlockingWritePC()
			248	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
			249	+	WriteTimeout: 10 * time.Millisecond,
			250	+	ResendAfter: time.Hour,
			251	+	Heartbeat: time.Hour,
			252	+	})
			253	+	if err != nil {
			254	+	t.Fatal(err)
			255	+	}
			256	+	defer func() { _ = tr.Close() }()
			257	+
			258	+	done := make(chan error, 1)
			259	+	go func() { done <- tr.sendControl(recAck, 1) }()
			260	+	select {
			261	+	case err := <-done:
			262	+	var timeout interface{ Timeout() bool }
			263	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
			264	+	t.Fatalf("control write err=%v, want timeout", err)
			265	+	}
			266	+	case <-time.After(250 * time.Millisecond):
			267	+	t.Fatal("control write blocked without honoring WriteTimeout")
			268	+	}
			269	+	}
			270	+
			271	+	func TestConcurrentWriterCannotClearEarlierDeadline(t *testing.T) {
			272	+	pc := &deadlineCapturePC{addr: testAddr("a"), done: make(chan struct{})}
			273	+	state := newWriteDeadlineState(pc)
			274	+	earlier := time.Now().Add(time.Second)
			275	+	later := earlier.Add(time.Second)
			276	+	finishEarlier, err := state.begin(earlier)
			277	+	if err != nil {
			278	+	t.Fatal(err)
			279	+	}
			280	+	defer finishEarlier()
			281	+	finishLater, err := state.begin(later)
			282	+	if err != nil {
			283	+	t.Fatal(err)
			284	+	}
			285	+	finishLater()
			286	+
			287	+	got, _ := pc.deadline.Load().(time.Time)
			288	+	if !got.Equal(earlier) {
			289	+	t.Fatalf("deadline after concurrent writer completed=%v, want earlier active deadline %v", got, earlier)
			290	+	}
			291	+	}
			292	+
			293	+	func TestFreshControlWriteRetriesAfterInheritedDeadline(t *testing.T) {
			294	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			295	+	pc := &stagedTimeoutPC{started: make(chan struct{}, 2), release: make(chan struct{})}
			296	+	tr := &Transport{clock: clk}
			297	+	deadlines := newWriteDeadlineState(pc)
			298	+	oldDone := make(chan error, 1)
			299	+	go func() {
			300	+	oldDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("old"), deadlines, 10*time.Millisecond)
			301	+	}()
			302	+	awaitSignal(t, pc.started, "older blocked write")
			303	+	clk.advance(5 * time.Millisecond)
			304	+	freshDone := make(chan error, 1)
			305	+	go func() {
			306	+	freshDone <- tr.writeDatagram(pc, testAddr("peer"), []byte("fresh"), deadlines, 100*time.Millisecond)
			307	+	}()
			308	+	awaitSignal(t, pc.started, "fresh write sharing earlier deadline")
			309	+	clk.advance(5 * time.Millisecond)
			310	+	close(pc.release)
			311	+	if err := awaitResult(t, oldDone, "older write timeout"); err == nil {
			312	+	t.Fatal("older write succeeded, want timeout")
			313	+	}
			314	+	if err := awaitResult(t, freshDone, "fresh write retry"); err != nil {
			315	+	t.Fatalf("fresh write failed at inherited deadline: %v", err)
			316	+	}
			317	+	}
			318	+
			319	+	func TestProbeReturnsLocalWriteError(t *testing.T) {
			320	+	aPC, bPC := newPair()
			321	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
			322	+	ResendAfter: time.Hour, Heartbeat: time.Hour,
			323	+	})
			324	+	if err != nil {
			325	+	t.Fatal(err)
			326	+	}
			327	+	defer func() { _ = a.Close() }()
			328	+	if err := bPC.Close(); err != nil {
			329	+	t.Fatal(err)
			330	+	}
			331	+
			332	+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			333	+	defer cancel()
			334	+	start := time.Now()
			335	+	if err := a.Probe(ctx); err == nil \|\| errors.Is(err, context.DeadlineExceeded) {
			336	+	t.Fatalf("Probe err=%v, want immediate local write error", err)
			337	+	}
			338	+	if elapsed := time.Since(start); elapsed > 250*time.Millisecond {
			339	+	t.Fatalf("Probe returned local write error after %v", elapsed)
			340	+	}
			341	+	}
			342	+
64		func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {	343		func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
65		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	344		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
66		aPC, bPC := newPair()	345		aPC, bPC := newPair()

@@ -72,8 +351,8 @@ func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
72		t.Fatal(err)	351		t.Fatal(err)
73		}	352		}
74		defer func() { _ = a.Close() }()	353		defer func() { _ = a.Close() }()
75	-	for range 2 {	354	+	for range 3 {
76	-	awaitSignal(t, clk.timerCreated, "resend and heartbeat timer creation")	355	+	awaitSignal(t, clk.timerCreated, "retransmit, health, and heartbeat timer creation")
77		}	356		}
78			357
79		retransmitStarted := make(chan struct{}, 1)	358		retransmitStarted := make(chan struct{}, 1)

@@ -120,3 +399,50 @@ func TestHeartbeatContinuesDuringRetransmitStorm(t *testing.T) {
120		awaitSignal(t, heartbeats, "heartbeat during retransmit storm")	399		awaitSignal(t, heartbeats, "heartbeat during retransmit storm")
121		}	400		}
122		}	401		}
			402	+
			403	+	func TestHealthChecksContinueDuringRetransmitStorm(t *testing.T) {
			404	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			405	+	aPC, bPC := newPair()
			406	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{
			407	+	Clock: clk, MTU: 128, ResendAfter: 10 * time.Millisecond,
			408	+	Heartbeat: time.Hour, DegradedAfter: 10 * time.Millisecond,
			409	+	})
			410	+	if err != nil {
			411	+	t.Fatal(err)
			412	+	}
			413	+	defer func() { _ = a.Close() }()
			414	+	for range 3 {
			415	+	awaitSignal(t, clk.timerCreated, "transport timer creation")
			416	+	}
			417	+
			418	+	retransmitStarted := make(chan struct{}, 1)
			419	+	releaseRetransmit := make(chan struct{})
			420	+	defer close(releaseRetransmit)
			421	+	aPC.drop = func(_ []byte, _ net.Addr) bool {
			422	+	select {
			423	+	case retransmitStarted <- struct{}{}:
			424	+	<-releaseRetransmit
			425	+	default:
			426	+	}
			427	+	return true
			428	+	}
			429	+
			430	+	now := clk.Now()
			431	+	a.mu.Lock()
			432	+	a.pending[1] = &pending{
			433	+	frame: ports.Frame{Type: ports.MsgInput, Payload: make([]byte, 24*a.mtu)},
			434	+	enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
			435	+	}
			436	+	a.mu.Unlock()
			437	+
			438	+	clk.advance(10 * time.Millisecond)
			439	+	awaitSignal(t, retransmitStarted, "blocked retransmit")
			440	+	select {
			441	+	case event := <-a.LinkEvents():
			442	+	if event.State != ports.LinkStateDegraded {
			443	+	t.Fatalf("link state=%s, want degraded", event.State)
			444	+	}
			445	+	case <-time.After(time.Second):
			446	+	t.Fatal("health check did not run while retransmit sender was blocked")
			447	+	}
			448	+	}

M internal/adapters/dgram/network_sim_test.go +12 -14


@@ -308,7 +308,7 @@ func TestTransportFloodClassification(t *testing.T) {
308	308		}
309	309		})
310	310
311		-	t.Run("fresh output and large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
	311	+	t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {
312	312		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
313	313		aPC, bPC := newPair()
314	314		inspect, err := pdgram.NewCodec(key())

@@ -380,7 +380,7 @@ func TestTransportFloodClassification(t *testing.T) {
380	380		t.Fatal(err)
381	381		}
382	382		defer closeFloodTransports(a, b)
383		-	for range 4 { // Both transports install resend and heartbeat timers.
	383	+	for range 6 { // Both transports install retransmit, health, and heartbeat timers.
384	384		awaitSignal(t, clk.timerCreated, "transport timer creation")
385	385		}
386	386

@@ -399,12 +399,11 @@ func TestTransportFloodClassification(t *testing.T) {
399	399		t.Fatal(err)
400	400		}
401	401		}
402		-	for range 2 {
403		-	awaitSignal(t, acksDropped, "dropped cumulative ACK")
404		-	}
	402	+	awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
	403	+	clk.advance(maxACKDelay)
	404	+	awaitSignal(t, acksDropped, "dropped cumulative ACK")
405	405
406		-	clk.advance(10 * time.Millisecond)
407		-	awaitSignal(t, retransmitBlocked, "retransmit pacing barrier") // resendPending is synchronously inside writePacket.
	406	+	awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
408	407		if got := floodState(a).retransmits; got < 2 {
409	408		t.Fatalf("retransmits=%d, want both overdue large records selected", got)
410	409		}

@@ -433,18 +432,17 @@ func TestTransportFloodClassification(t *testing.T) {
433	432		default:
434	433		}
435	434
436		-	clk.advance(time.Millisecond)
437		-	select {
438		-	case <-probeSent:
439		-	t.Fatal("heartbeat probe ran while fresh and retransmitted large output contended")
440		-	default:
441		-	}
	435	+	awaitSignal(t, probeSent, "heartbeat probe during output contention")
442	436
443	437		close(releaseRetransmit)
444	438		if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
445	439		t.Fatal(err)
446	440		}
447		-	awaitSignal(t, probeSent, "heartbeat probe")
	441	+	eventually(t, time.Second, func() bool {
	442	+	writesMu.Lock()
	443	+	defer writesMu.Unlock()
	444	+	return writes[1] >= 2 && writes[2] >= 2
	445	+	})
448	446		writesMu.Lock()
449	447		defer writesMu.Unlock()
450	448		if writes[1] < 2 \|\| writes[2] < 2 \|\| writes[3] == 0 {

 		}
 	})
-	t.Run("fresh output and large retransmits synchronously delay heartbeat health work", func(t *testing.T) {
+	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()
 		inspect, err := pdgram.NewCodec(key())
 			t.Fatal(err)
 		}
 		defer closeFloodTransports(a, b)
-		for range 4 { // Both transports install resend and heartbeat timers.
+		for range 6 { // Both transports install retransmit, health, and heartbeat timers.
 			awaitSignal(t, clk.timerCreated, "transport timer creation")
 		}
 				t.Fatal(err)
 			}
 		}
-		for range 2 {
+		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
-			awaitSignal(t, acksDropped, "dropped cumulative ACK")
+		clk.advance(maxACKDelay)
-		}
+		awaitSignal(t, acksDropped, "dropped cumulative ACK")
-		clk.advance(10 * time.Millisecond)
+		awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
-		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)
 		}
 		default:
 		}
-		clk.advance(time.Millisecond)
+		awaitSignal(t, probeSent, "heartbeat probe during output contention")
-		select {
-		case <-probeSent:
-			t.Fatal("heartbeat probe ran while fresh and retransmitted large output contended")
-		default:
-		}
 		close(releaseRetransmit)
 		if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
 			t.Fatal(err)
 		}
-		awaitSignal(t, probeSent, "heartbeat probe")
+		eventually(t, time.Second, func() bool {
+			writesMu.Lock()
+			defer writesMu.Unlock()
+			return writes[1] >= 2 && writes[2] >= 2
+		})
 		writesMu.Lock()
 		defer writesMu.Unlock()
 		if writes[1] < 2 || writes[2] < 2 || writes[3] == 0 {

M internal/adapters/dgram/transport.go +192 -41


@@ -47,6 +47,7 @@ const (
47	47		var (
48	48		ErrPendingFull = errors.New("dgram: pending reliable queue full")
49	49		ErrLinkDead = errors.New("dgram: link dead")
	50	+	errControlFull = errors.New("dgram: control queue full")
50	51		)
51	52
52	53		type Options struct {

@@ -183,6 +184,7 @@ type Transport struct {
183	184		sendWake chan struct{}
184	185		control chan controlRecord
185	186		ackWake chan struct{}
	187	+	ackSend chan uint64
186	188		controlMu sync.Mutex
187	189		ackSeq uint64
188	190		ackQueued bool

@@ -191,6 +193,8 @@ type Transport struct {
191	193		outputQueue []queuedSend
192	194		outputWake chan struct{}
193	195		outputNext time.Time
	196	+	retransmitWork chan []retransmitRecord
	197	+	writeDeadlines *writeDeadlineState
194	198
195	199		recvMu sync.Mutex
196	200		replay *pdgram.ReplayWindow

@@ -224,6 +228,62 @@ type pending struct {
224	228		retransmitted bool
225	229		}
226	230
	231	+	type retransmitRecord struct {
	232	+	seq uint64
	233	+	f ports.Frame
	234	+	}
	235	+
	236	+	type writeDeadlineState struct {
	237	+	mu sync.Mutex
	238	+	pc net.PacketConn
	239	+	nextID uint64
	240	+	deadlines map[uint64]time.Time
	241	+	}
	242	+
	243	+	func newWriteDeadlineState(pc net.PacketConn) *writeDeadlineState {
	244	+	return &writeDeadlineState{pc: pc, deadlines: make(map[uint64]time.Time)}
	245	+	}
	246	+
	247	+	func (s *writeDeadlineState) begin(deadline time.Time) (func(), error) {
	248	+	s.mu.Lock()
	249	+	s.nextID++
	250	+	id := s.nextID
	251	+	s.deadlines[id] = deadline
	252	+	if err := s.applyLocked(); err != nil {
	253	+	delete(s.deadlines, id)
	254	+	s.mu.Unlock()
	255	+	return nil, err
	256	+	}
	257	+	s.mu.Unlock()
	258	+	return func() {
	259	+	s.mu.Lock()
	260	+	delete(s.deadlines, id)
	261	+	_ = s.applyLocked()
	262	+	s.mu.Unlock()
	263	+	}, nil
	264	+	}
	265	+
	266	+	func (s *writeDeadlineState) applyLocked() error {
	267	+	var earliest time.Time
	268	+	for _, deadline := range s.deadlines {
	269	+	if earliest.IsZero() \|\| deadline.Before(earliest) {
	270	+	earliest = deadline
	271	+	}
	272	+	}
	273	+	return s.pc.SetWriteDeadline(earliest)
	274	+	}
	275	+
	276	+	func (s *writeDeadlineState) expire(now time.Time) {
	277	+	s.mu.Lock()
	278	+	for id, deadline := range s.deadlines {
	279	+	if !deadline.After(now) {
	280	+	delete(s.deadlines, id)
	281	+	}
	282	+	}
	283	+	_ = s.applyLocked()
	284	+	s.mu.Unlock()
	285	+	}
	286	+
227	287		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
228	288		return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{})
229	289		}

@@ -275,17 +335,24 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
275	335		t.sendWake = make(chan struct{})
276	336		t.control = make(chan controlRecord, controlQueueSize)
277	337		t.ackWake = make(chan struct{}, 1)
	338	+	t.ackSend = make(chan uint64, 1)
278	339		t.outputWake = make(chan struct{})
	340	+	t.writeDeadlines = newWriteDeadlineState(pc)
	341	+	// One queued batch bounds retransmit work while its sole sender is paced.
	342	+	t.retransmitWork = make(chan []retransmitRecord, 1)
279	343		if t.observe != nil {
280	344		t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
281	345		go t.diagnosticLoop()
282	346		}
283	347		t.deliverCond = sync.NewCond(&t.deliverMu)
284	348		go t.readLoop(pc)
285		-	go t.resendLoop()
	349	+	go t.retransmitLoop()
286	350		go t.heartbeatLoop()
	351	+	go t.healthLoop()
	352	+	go t.retransmitSenderLoop()
287	353		go t.deliveryLoop()
288	354		go t.controlLoop()
	355	+	go t.ackSendLoop()
289	356		go t.outputPaceLoop()
290	357		t.emitDiagnostic()
291	358		return t, nil

@@ -517,22 +584,30 @@ func (t *Transport) Probe(ctx context.Context) error {
517	584		return err
518	585		}
519	586		defer t.unregisterProbe(id)
520		-	if err := t.sendProbe(recProbe, id); err != nil {
521		-	return err
	587	+	sendResult := make(chan error, 1)
	588	+	if !t.queueControlRecord(controlRecord{kind: recProbe, id: id, result: sendResult}) {
	589	+	return errControlFull
522	590		}
523		-	select {
524		-	case <-ch:
525		-	return nil
526		-	case <-ctx.Done():
527		-	return ctx.Err()
528		-	case <-t.done:
529		-	t.mu.Lock()
530		-	err := t.closeErr
531		-	t.mu.Unlock()
532		-	if err != nil {
533		-	return err
	591	+	for {
	592	+	select {
	593	+	case err := <-sendResult:
	594	+	if err != nil {
	595	+	return err
	596	+	}
	597	+	sendResult = nil
	598	+	case <-ch:
	599	+	return nil
	600	+	case <-ctx.Done():
	601	+	return ctx.Err()
	602	+	case <-t.done:
	603	+	t.mu.Lock()
	604	+	err := t.closeErr
	605	+	t.mu.Unlock()
	606	+	if err != nil {
	607	+	return err
	608	+	}
	609	+	return errors.New("dgram: closed")
534	610		}
535		-	return errors.New("dgram: closed")
536	611		}
537	612		}
538	613

@@ -568,12 +643,6 @@ func (t *Transport) sendAck(seq uint64) {
568	643		binary.BigEndian.PutUint64(b[1:], seq)
569	644		_ = t.sendPayload(b[:], false, 0)
570	645		}
571		-	func (t *Transport) sendProbe(kind byte, id uint64) error {
572		-	var b [9]byte
573		-	b[0] = kind
574		-	binary.BigEndian.PutUint64(b[1:], id)
575		-	return t.sendPayload(b[:], false, 0)
576		-	}
577	646		func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
578	647		if paced {
579	648		t.mu.Lock()

@@ -665,16 +734,35 @@ func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
665	734		t.mu.Lock()
666	735		pc := t.pc
667	736		writeTimeout := t.writeTimeout
	737	+	writeDeadlines := t.writeDeadlines
668	738		t.mu.Unlock()
669	739
670		-	if writeTimeout > 0 {
671		-	if err := pc.SetWriteDeadline(t.clock.Now().Add(writeTimeout)); err != nil {
	740	+	return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
	741	+	}
	742	+
	743	+	func (t Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines writeDeadlineState, timeout time.Duration) error {
	744	+	if timeout <= 0 {
	745	+	_, err := pc.WriteTo(pkt, peer)
	746	+	return err
	747	+	}
	748	+	ownDeadline := t.clock.Now().Add(timeout)
	749	+	finish, err := deadlines.begin(ownDeadline)
	750	+	if err != nil {
	751	+	return err
	752	+	}
	753	+	defer finish()
	754	+	for {
	755	+	_, err = pc.WriteTo(pkt, peer)
	756	+	if err == nil {
	757	+	return nil
	758	+	}
	759	+	var timeoutErr interface{ Timeout() bool }
	760	+	now := t.clock.Now()
	761	+	if !errors.As(err, &timeoutErr) \|\| !timeoutErr.Timeout() \|\| !now.Before(ownDeadline) {
672	762		return err
673	763		}
674		-	defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()
	764	+	deadlines.expire(now)
675	765		}
676		-	_, err := pc.WriteTo(pkt, peer)
677		-	return err
678	766		}
679	767
680	768		func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }

@@ -889,16 +977,52 @@ func (t *Transport) notifySendWaitersLocked() {
889	977		t.sendWake = make(chan struct{})
890	978		}
891	979
892		-	func (t *Transport) resendLoop() {
	980	+	func (t *Transport) retransmitLoop() {
893	981		resend := t.clock.NewTimer(t.resendAfter)
894	982		defer resend.Stop()
895	983		for {
896	984		select {
897	985		case <-resend.C():
898		-	t.resendPending()
	986	+	t.queueRetransmits()
	987	+	resend.Reset(t.resendAfter)
	988	+	case <-t.done:
	989	+	return
	990	+	}
	991	+	}
	992	+	}
	993	+
	994	+	// healthLoop owns link-health timing so bulk retransmission cannot delay state
	995	+	// transitions or socket hopping.
	996	+	func (t *Transport) healthLoop() {
	997	+	health := t.clock.NewTimer(t.resendAfter)
	998	+	defer health.Stop()
	999	+	for {
	1000	+	select {
	1001	+	case <-health.C():
899	1002		t.checkSilence()
900	1003		t.checkSendStall()
901		-	resend.Reset(t.resendAfter)
	1004	+	health.Reset(t.resendAfter)
	1005	+	case <-t.done:
	1006	+	return
	1007	+	}
	1008	+	}
	1009	+	}
	1010	+
	1011	+	func (t *Transport) retransmitSenderLoop() {
	1012	+	for {
	1013	+	select {
	1014	+	case batch := <-t.retransmitWork:
	1015	+	for _, r := range batch {
	1016	+	if err := t.sendData(r.seq, true, r.f); err == nil {
	1017	+	t.markPendingReady(r.seq, true)
	1018	+	} else {
	1019	+	t.mu.Lock()
	1020	+	if p := t.pending[r.seq]; p != nil {
	1021	+	p.initialInFlight = false
	1022	+	}
	1023	+	t.mu.Unlock()
	1024	+	}
	1025	+	}
902	1026		case <-t.done:
903	1027		return
904	1028		}

@@ -919,13 +1043,36 @@ func (t *Transport) heartbeatLoop() {
919	1043		}
920	1044		}
921	1045
922		-	func (t *Transport) resendPending() {
923		-	now := t.clock.Now()
924		-	var resend []struct {
925		-	seq uint64
926		-	f ports.Frame
	1046	+	func (t *Transport) queueRetransmits() {
	1047	+	// Do not mark records in-flight while the one-slot sender queue is full.
	1048	+	// There is one producer and the sender only consumes, so capacity cannot be
	1049	+	// lost after this check.
	1050	+	if len(t.retransmitWork) == cap(t.retransmitWork) {
	1051	+	return
927	1052		}
	1053	+
	1054	+	now := t.clock.Now()
928	1055		t.mu.Lock()
	1056	+	resend, emitDiagnostic := t.selectRetransmitsLocked(now)
	1057	+	t.mu.Unlock()
	1058	+
	1059	+	if len(resend) > 0 {
	1060	+	select {
	1061	+	case t.retransmitWork <- resend:
	1062	+	case <-t.done:
	1063	+	return
	1064	+	}
	1065	+	}
	1066	+	if emitDiagnostic {
	1067	+	t.emitDiagnostic()
	1068	+	}
	1069	+	}
	1070	+
	1071	+	// selectRetransmitsLocked deterministically reserves one bounded retransmit
	1072	+	// turn. The caller must hold t.mu and performs all pacing and writes after
	1073	+	// releasing it.
	1074	+	func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
	1075	+	resend := make([]retransmitRecord, 0, t.maxResendPerTick)
929	1076		seqs := make([]uint64, 0, len(t.pending))
930	1077		for seq := range t.pending {
931	1078		seqs = append(seqs, seq)

@@ -946,16 +1093,19 @@ func (t *Transport) resendPending() {
946	1093		p.retransmitted = true
947	1094		t.retransmits++
948	1095		emitDiagnostic = emitDiagnostic \|\| t.retransmits%64 == 0
949		-	resend = append(resend, struct {
950		-	seq uint64
951		-	f ports.Frame
952		-	}{seq, p.frame})
	1096	+	resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
953	1097		limit--
954	1098		}
	1099	+	return resend, emitDiagnostic
	1100	+	}
	1101	+
	1102	+	// resendPending remains a synchronous test hook for retransmission selection
	1103	+	// unit tests. Production timer paths use queueRetransmits and the sole sender.
	1104	+	func (t *Transport) resendPending() {
	1105	+	now := t.clock.Now()
	1106	+	t.mu.Lock()
	1107	+	resend, _ := t.selectRetransmitsLocked(now)
955	1108		t.mu.Unlock()
956		-	if emitDiagnostic {
957		-	t.emitDiagnostic()
958		-	}
959	1109		for _, r := range resend {
960	1110		if err := t.sendData(r.seq, true, r.f); err == nil {
961	1111		t.markPendingReady(r.seq, true)

@@ -1038,6 +1188,7 @@ func (t *Transport) hopPacketConnOnce() {
1038	1188		return
1039	1189		}
1040	1190		t.pc = pc
	1191	+	t.writeDeadlines = newWriteDeadlineState(pc)
1041	1192		t.mu.Unlock()
1042	1193		go t.readLoop(pc)
1043	1194		_ = old.Close()


@@ -47,6 +47,7 @@ const (
47		var (	47		var (
48		ErrPendingFull = errors.New("dgram: pending reliable queue full")	48		ErrPendingFull = errors.New("dgram: pending reliable queue full")
49		ErrLinkDead = errors.New("dgram: link dead")	49		ErrLinkDead = errors.New("dgram: link dead")
			50	+	errControlFull = errors.New("dgram: control queue full")
50		)	51		)
51			52
52		type Options struct {	53		type Options struct {

@@ -183,6 +184,7 @@ type Transport struct {
183		sendWake chan struct{}	184		sendWake chan struct{}
184		control chan controlRecord	185		control chan controlRecord
185		ackWake chan struct{}	186		ackWake chan struct{}
			187	+	ackSend chan uint64
186		controlMu sync.Mutex	188		controlMu sync.Mutex
187		ackSeq uint64	189		ackSeq uint64
188		ackQueued bool	190		ackQueued bool

@@ -191,6 +193,8 @@ type Transport struct {
191		outputQueue []queuedSend	193		outputQueue []queuedSend
192		outputWake chan struct{}	194		outputWake chan struct{}
193		outputNext time.Time	195		outputNext time.Time
			196	+	retransmitWork chan []retransmitRecord
			197	+	writeDeadlines *writeDeadlineState
194			198
195		recvMu sync.Mutex	199		recvMu sync.Mutex
196		replay *pdgram.ReplayWindow	200		replay *pdgram.ReplayWindow

@@ -224,6 +228,62 @@ type pending struct {
224		retransmitted bool	228		retransmitted bool
225		}	229		}
226			230
			231	+	type retransmitRecord struct {
			232	+	seq uint64
			233	+	f ports.Frame
			234	+	}
			235	+
			236	+	type writeDeadlineState struct {
			237	+	mu sync.Mutex
			238	+	pc net.PacketConn
			239	+	nextID uint64
			240	+	deadlines map[uint64]time.Time
			241	+	}
			242	+
			243	+	func newWriteDeadlineState(pc net.PacketConn) *writeDeadlineState {
			244	+	return &writeDeadlineState{pc: pc, deadlines: make(map[uint64]time.Time)}
			245	+	}
			246	+
			247	+	func (s *writeDeadlineState) begin(deadline time.Time) (func(), error) {
			248	+	s.mu.Lock()
			249	+	s.nextID++
			250	+	id := s.nextID
			251	+	s.deadlines[id] = deadline
			252	+	if err := s.applyLocked(); err != nil {
			253	+	delete(s.deadlines, id)
			254	+	s.mu.Unlock()
			255	+	return nil, err
			256	+	}
			257	+	s.mu.Unlock()
			258	+	return func() {
			259	+	s.mu.Lock()
			260	+	delete(s.deadlines, id)
			261	+	_ = s.applyLocked()
			262	+	s.mu.Unlock()
			263	+	}, nil
			264	+	}
			265	+
			266	+	func (s *writeDeadlineState) applyLocked() error {
			267	+	var earliest time.Time
			268	+	for _, deadline := range s.deadlines {
			269	+	if earliest.IsZero() \|\| deadline.Before(earliest) {
			270	+	earliest = deadline
			271	+	}
			272	+	}
			273	+	return s.pc.SetWriteDeadline(earliest)
			274	+	}
			275	+
			276	+	func (s *writeDeadlineState) expire(now time.Time) {
			277	+	s.mu.Lock()
			278	+	for id, deadline := range s.deadlines {
			279	+	if !deadline.After(now) {
			280	+	delete(s.deadlines, id)
			281	+	}
			282	+	}
			283	+	_ = s.applyLocked()
			284	+	s.mu.Unlock()
			285	+	}
			286	+
227		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {	287		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
228		return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{})	288		return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{})
229		}	289		}

@@ -275,17 +335,24 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
275		t.sendWake = make(chan struct{})	335		t.sendWake = make(chan struct{})
276		t.control = make(chan controlRecord, controlQueueSize)	336		t.control = make(chan controlRecord, controlQueueSize)
277		t.ackWake = make(chan struct{}, 1)	337		t.ackWake = make(chan struct{}, 1)
			338	+	t.ackSend = make(chan uint64, 1)
278		t.outputWake = make(chan struct{})	339		t.outputWake = make(chan struct{})
			340	+	t.writeDeadlines = newWriteDeadlineState(pc)
			341	+	// One queued batch bounds retransmit work while its sole sender is paced.
			342	+	t.retransmitWork = make(chan []retransmitRecord, 1)
279		if t.observe != nil {	343		if t.observe != nil {
280		t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)	344		t.diagnosticCh = make(chan Diagnostic, diagnosticBufferSize)
281		go t.diagnosticLoop()	345		go t.diagnosticLoop()
282		}	346		}
283		t.deliverCond = sync.NewCond(&t.deliverMu)	347		t.deliverCond = sync.NewCond(&t.deliverMu)
284		go t.readLoop(pc)	348		go t.readLoop(pc)
285	-	go t.resendLoop()	349	+	go t.retransmitLoop()
286		go t.heartbeatLoop()	350		go t.heartbeatLoop()
			351	+	go t.healthLoop()
			352	+	go t.retransmitSenderLoop()
287		go t.deliveryLoop()	353		go t.deliveryLoop()
288		go t.controlLoop()	354		go t.controlLoop()
			355	+	go t.ackSendLoop()
289		go t.outputPaceLoop()	356		go t.outputPaceLoop()
290		t.emitDiagnostic()	357		t.emitDiagnostic()
291		return t, nil	358		return t, nil

@@ -517,22 +584,30 @@ func (t *Transport) Probe(ctx context.Context) error {
517		return err	584		return err
518		}	585		}
519		defer t.unregisterProbe(id)	586		defer t.unregisterProbe(id)
520	-	if err := t.sendProbe(recProbe, id); err != nil {	587	+	sendResult := make(chan error, 1)
521	-	return err	588	+	if !t.queueControlRecord(controlRecord{kind: recProbe, id: id, result: sendResult}) {
			589	+	return errControlFull
522		}	590		}
523	-	select {	591	+	for {
524	-	case <-ch:	592	+	select {
525	-	return nil	593	+	case err := <-sendResult:
526	-	case <-ctx.Done():	594	+	if err != nil {
527	-	return ctx.Err()	595	+	return err
528	-	case <-t.done:	596	+	}
529	-	t.mu.Lock()	597	+	sendResult = nil
530	-	err := t.closeErr	598	+	case <-ch:
531	-	t.mu.Unlock()	599	+	return nil
532	-	if err != nil {	600	+	case <-ctx.Done():
533	-	return err	601	+	return ctx.Err()
			602	+	case <-t.done:
			603	+	t.mu.Lock()
			604	+	err := t.closeErr
			605	+	t.mu.Unlock()
			606	+	if err != nil {
			607	+	return err
			608	+	}
			609	+	return errors.New("dgram: closed")
534		}	610		}
535	-	return errors.New("dgram: closed")
536		}	611		}
537		}	612		}
538			613

@@ -568,12 +643,6 @@ func (t *Transport) sendAck(seq uint64) {
568		binary.BigEndian.PutUint64(b[1:], seq)	643		binary.BigEndian.PutUint64(b[1:], seq)
569		_ = t.sendPayload(b[:], false, 0)	644		_ = t.sendPayload(b[:], false, 0)
570		}	645		}
571	-	func (t *Transport) sendProbe(kind byte, id uint64) error {
572	-	var b [9]byte
573	-	b[0] = kind
574	-	binary.BigEndian.PutUint64(b[1:], id)
575	-	return t.sendPayload(b[:], false, 0)
576	-	}
577		func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {	646		func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
578		if paced {	647		if paced {
579		t.mu.Lock()	648		t.mu.Lock()

@@ -665,16 +734,35 @@ func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
665		t.mu.Lock()	734		t.mu.Lock()
666		pc := t.pc	735		pc := t.pc
667		writeTimeout := t.writeTimeout	736		writeTimeout := t.writeTimeout
			737	+	writeDeadlines := t.writeDeadlines
668		t.mu.Unlock()	738		t.mu.Unlock()
669			739
670	-	if writeTimeout > 0 {	740	+	return t.writeDatagram(pc, peer, pkt, writeDeadlines, writeTimeout)
671	-	if err := pc.SetWriteDeadline(t.clock.Now().Add(writeTimeout)); err != nil {	741	+	}
			742	+
			743	+	func (t Transport) writeDatagram(pc net.PacketConn, peer net.Addr, pkt []byte, deadlines writeDeadlineState, timeout time.Duration) error {
			744	+	if timeout <= 0 {
			745	+	_, err := pc.WriteTo(pkt, peer)
			746	+	return err
			747	+	}
			748	+	ownDeadline := t.clock.Now().Add(timeout)
			749	+	finish, err := deadlines.begin(ownDeadline)
			750	+	if err != nil {
			751	+	return err
			752	+	}
			753	+	defer finish()
			754	+	for {
			755	+	_, err = pc.WriteTo(pkt, peer)
			756	+	if err == nil {
			757	+	return nil
			758	+	}
			759	+	var timeoutErr interface{ Timeout() bool }
			760	+	now := t.clock.Now()
			761	+	if !errors.As(err, &timeoutErr) \|\| !timeoutErr.Timeout() \|\| !now.Before(ownDeadline) {
672		return err	762		return err
673		}	763		}
674	-	defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()	764	+	deadlines.expire(now)
675		}	765		}
676	-	_, err := pc.WriteTo(pkt, peer)
677	-	return err
678		}	766		}
679			767
680		func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }	768		func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }

@@ -889,16 +977,52 @@ func (t *Transport) notifySendWaitersLocked() {
889		t.sendWake = make(chan struct{})	977		t.sendWake = make(chan struct{})
890		}	978		}
891			979
892	-	func (t *Transport) resendLoop() {	980	+	func (t *Transport) retransmitLoop() {
893		resend := t.clock.NewTimer(t.resendAfter)	981		resend := t.clock.NewTimer(t.resendAfter)
894		defer resend.Stop()	982		defer resend.Stop()
895		for {	983		for {
896		select {	984		select {
897		case <-resend.C():	985		case <-resend.C():
898	-	t.resendPending()	986	+	t.queueRetransmits()
			987	+	resend.Reset(t.resendAfter)
			988	+	case <-t.done:
			989	+	return
			990	+	}
			991	+	}
			992	+	}
			993	+
			994	+	// healthLoop owns link-health timing so bulk retransmission cannot delay state
			995	+	// transitions or socket hopping.
			996	+	func (t *Transport) healthLoop() {
			997	+	health := t.clock.NewTimer(t.resendAfter)
			998	+	defer health.Stop()
			999	+	for {
			1000	+	select {
			1001	+	case <-health.C():
899		t.checkSilence()	1002		t.checkSilence()
900		t.checkSendStall()	1003		t.checkSendStall()
901	-	resend.Reset(t.resendAfter)	1004	+	health.Reset(t.resendAfter)
			1005	+	case <-t.done:
			1006	+	return
			1007	+	}
			1008	+	}
			1009	+	}
			1010	+
			1011	+	func (t *Transport) retransmitSenderLoop() {
			1012	+	for {
			1013	+	select {
			1014	+	case batch := <-t.retransmitWork:
			1015	+	for _, r := range batch {
			1016	+	if err := t.sendData(r.seq, true, r.f); err == nil {
			1017	+	t.markPendingReady(r.seq, true)
			1018	+	} else {
			1019	+	t.mu.Lock()
			1020	+	if p := t.pending[r.seq]; p != nil {
			1021	+	p.initialInFlight = false
			1022	+	}
			1023	+	t.mu.Unlock()
			1024	+	}
			1025	+	}
902		case <-t.done:	1026		case <-t.done:
903		return	1027		return
904		}	1028		}

@@ -919,13 +1043,36 @@ func (t *Transport) heartbeatLoop() {
919		}	1043		}
920		}	1044		}
921			1045
922	-	func (t *Transport) resendPending() {	1046	+	func (t *Transport) queueRetransmits() {
923	-	now := t.clock.Now()	1047	+	// Do not mark records in-flight while the one-slot sender queue is full.
924	-	var resend []struct {	1048	+	// There is one producer and the sender only consumes, so capacity cannot be
925	-	seq uint64	1049	+	// lost after this check.
926	-	f ports.Frame	1050	+	if len(t.retransmitWork) == cap(t.retransmitWork) {
			1051	+	return
927		}	1052		}
			1053	+
			1054	+	now := t.clock.Now()
928		t.mu.Lock()	1055		t.mu.Lock()
			1056	+	resend, emitDiagnostic := t.selectRetransmitsLocked(now)
			1057	+	t.mu.Unlock()
			1058	+
			1059	+	if len(resend) > 0 {
			1060	+	select {
			1061	+	case t.retransmitWork <- resend:
			1062	+	case <-t.done:
			1063	+	return
			1064	+	}
			1065	+	}
			1066	+	if emitDiagnostic {
			1067	+	t.emitDiagnostic()
			1068	+	}
			1069	+	}
			1070	+
			1071	+	// selectRetransmitsLocked deterministically reserves one bounded retransmit
			1072	+	// turn. The caller must hold t.mu and performs all pacing and writes after
			1073	+	// releasing it.
			1074	+	func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord, bool) {
			1075	+	resend := make([]retransmitRecord, 0, t.maxResendPerTick)
929		seqs := make([]uint64, 0, len(t.pending))	1076		seqs := make([]uint64, 0, len(t.pending))
930		for seq := range t.pending {	1077		for seq := range t.pending {
931		seqs = append(seqs, seq)	1078		seqs = append(seqs, seq)

@@ -946,16 +1093,19 @@ func (t *Transport) resendPending() {
946		p.retransmitted = true	1093		p.retransmitted = true
947		t.retransmits++	1094		t.retransmits++
948		emitDiagnostic = emitDiagnostic \|\| t.retransmits%64 == 0	1095		emitDiagnostic = emitDiagnostic \|\| t.retransmits%64 == 0
949	-	resend = append(resend, struct {	1096	+	resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
950	-	seq uint64
951	-	f ports.Frame
952	-	}{seq, p.frame})
953		limit--	1097		limit--
954		}	1098		}
			1099	+	return resend, emitDiagnostic
			1100	+	}
			1101	+
			1102	+	// resendPending remains a synchronous test hook for retransmission selection
			1103	+	// unit tests. Production timer paths use queueRetransmits and the sole sender.
			1104	+	func (t *Transport) resendPending() {
			1105	+	now := t.clock.Now()
			1106	+	t.mu.Lock()
			1107	+	resend, _ := t.selectRetransmitsLocked(now)
955		t.mu.Unlock()	1108		t.mu.Unlock()
956	-	if emitDiagnostic {
957	-	t.emitDiagnostic()
958	-	}
959		for _, r := range resend {	1109		for _, r := range resend {
960		if err := t.sendData(r.seq, true, r.f); err == nil {	1110		if err := t.sendData(r.seq, true, r.f); err == nil {
961		t.markPendingReady(r.seq, true)	1111		t.markPendingReady(r.seq, true)

@@ -1038,6 +1188,7 @@ func (t *Transport) hopPacketConnOnce() {
1038		return	1188		return
1039		}	1189		}
1040		t.pc = pc	1190		t.pc = pc
			1191	+	t.writeDeadlines = newWriteDeadlineState(pc)
1041		t.mu.Unlock()	1192		t.mu.Unlock()
1042		go t.readLoop(pc)	1193		go t.readLoop(pc)
1043		_ = old.Close()	1194		_ = old.Close()

M internal/adapters/dgram/transport_test.go +7 -7


@@ -258,7 +258,7 @@ func TestOutputFirstSendIsPacedAndQueuedOutputsAreBatched(t *testing.T) {
258	258		if got := len(bPC.in); got != 1 {
259	259		t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
260	260		}
261		-	waitForManualTimers(t, clk, 3)
	261	+	waitForManualTimers(t, clk, 4)
262	262		clk.advance(defaultOutputPaceMinDelay)
263	263		eventuallyPacketCount(t, bPC, 3)
264	264		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {

@@ -289,7 +289,7 @@ func TestOutputPacingSendsAtMostOneOversizedFramePerTick(t *testing.T) {
289	289		t.Fatal(err)
290	290		}
291	291		}
292		-	waitForManualTimers(t, clk, 3)
	292	+	waitForManualTimers(t, clk, 4)
293	293		previousPackets := len(aPC.in)
294	294		for tick := 0; tick < 128; tick++ {
295	295		clk.advance(defaultOutputPaceMaxDelay)

@@ -486,7 +486,7 @@ func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
486	486		t.Fatal(err)
487	487		}
488	488		}
489		-	waitForManualTimers(t, clk, 3)
	489	+	waitForManualTimers(t, clk, 4)
490	490		a.writeMu.Lock()
491	491		clk.advance(defaultOutputPaceMinDelay)
492	492		deadline := time.Now().Add(time.Second)

@@ -554,7 +554,7 @@ func TestPacedControlQueueFailureClosesTransport(t *testing.T) {
554	554		defer a.mu.Unlock()
555	555		return len(a.outputQueue) == 2
556	556		})
557		-	waitForManualTimers(t, clk, 3)
	557	+	waitForManualTimers(t, clk, 4)
558	558		clk.advance(defaultOutputPaceMinDelay)
559	559		if err := <-inputDone; err == nil {
560	560		t.Fatal("queued control send succeeded after paced wire failure")

@@ -636,7 +636,7 @@ func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
636	636		t.Fatal(err)
637	637		}
638	638		}
639		-	waitForManualTimers(t, clk, 3)
	639	+	waitForManualTimers(t, clk, 4)
640	640		if err := bPC.Close(); err != nil {
641	641		t.Fatal(err)
642	642		}

@@ -675,7 +675,7 @@ func TestInputAndControlRemainOrderedInBoundedPacedQueue(t *testing.T) {
675	675		if got := len(bPC.in); got != 1 {
676	676		t.Fatalf("datagrams after input=%d, want prerequisites and input still paced", got)
677	677		}
678		-	waitForManualTimers(t, clk, 3)
	678	+	waitForManualTimers(t, clk, 4)
679	679		clk.advance(defaultOutputPaceMinDelay)
680	680		eventuallyPacketCount(t, bPC, 3)
681	681		if err := <-inputDone; err != nil {

@@ -1419,7 +1419,7 @@ func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
1419	1419		}
1420	1420		done := make(chan error, 1)
1421	1421		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
1422		-	waitForManualTimers(t, clk, 3)
	1422	+	waitForManualTimers(t, clk, 4)
1423	1423
1424	1424		clk.advance(50 * time.Millisecond)
1425	1425		select {


@@ -258,7 +258,7 @@ func TestOutputFirstSendIsPacedAndQueuedOutputsAreBatched(t *testing.T) {
258		if got := len(bPC.in); got != 1 {	258		if got := len(bPC.in); got != 1 {
259		t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)	259		t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
260		}	260		}
261	-	waitForManualTimers(t, clk, 3)	261	+	waitForManualTimers(t, clk, 4)
262		clk.advance(defaultOutputPaceMinDelay)	262		clk.advance(defaultOutputPaceMinDelay)
263		eventuallyPacketCount(t, bPC, 3)	263		eventuallyPacketCount(t, bPC, 3)
264		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {	264		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {

@@ -289,7 +289,7 @@ func TestOutputPacingSendsAtMostOneOversizedFramePerTick(t *testing.T) {
289		t.Fatal(err)	289		t.Fatal(err)
290		}	290		}
291		}	291		}
292	-	waitForManualTimers(t, clk, 3)	292	+	waitForManualTimers(t, clk, 4)
293		previousPackets := len(aPC.in)	293		previousPackets := len(aPC.in)
294		for tick := 0; tick < 128; tick++ {	294		for tick := 0; tick < 128; tick++ {
295		clk.advance(defaultOutputPaceMaxDelay)	295		clk.advance(defaultOutputPaceMaxDelay)

@@ -486,7 +486,7 @@ func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
486		t.Fatal(err)	486		t.Fatal(err)
487		}	487		}
488		}	488		}
489	-	waitForManualTimers(t, clk, 3)	489	+	waitForManualTimers(t, clk, 4)
490		a.writeMu.Lock()	490		a.writeMu.Lock()
491		clk.advance(defaultOutputPaceMinDelay)	491		clk.advance(defaultOutputPaceMinDelay)
492		deadline := time.Now().Add(time.Second)	492		deadline := time.Now().Add(time.Second)

@@ -554,7 +554,7 @@ func TestPacedControlQueueFailureClosesTransport(t *testing.T) {
554		defer a.mu.Unlock()	554		defer a.mu.Unlock()
555		return len(a.outputQueue) == 2	555		return len(a.outputQueue) == 2
556		})	556		})
557	-	waitForManualTimers(t, clk, 3)	557	+	waitForManualTimers(t, clk, 4)
558		clk.advance(defaultOutputPaceMinDelay)	558		clk.advance(defaultOutputPaceMinDelay)
559		if err := <-inputDone; err == nil {	559		if err := <-inputDone; err == nil {
560		t.Fatal("queued control send succeeded after paced wire failure")	560		t.Fatal("queued control send succeeded after paced wire failure")

@@ -636,7 +636,7 @@ func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
636		t.Fatal(err)	636		t.Fatal(err)
637		}	637		}
638		}	638		}
639	-	waitForManualTimers(t, clk, 3)	639	+	waitForManualTimers(t, clk, 4)
640		if err := bPC.Close(); err != nil {	640		if err := bPC.Close(); err != nil {
641		t.Fatal(err)	641		t.Fatal(err)
642		}	642		}

@@ -675,7 +675,7 @@ func TestInputAndControlRemainOrderedInBoundedPacedQueue(t *testing.T) {
675		if got := len(bPC.in); got != 1 {	675		if got := len(bPC.in); got != 1 {
676		t.Fatalf("datagrams after input=%d, want prerequisites and input still paced", got)	676		t.Fatalf("datagrams after input=%d, want prerequisites and input still paced", got)
677		}	677		}
678	-	waitForManualTimers(t, clk, 3)	678	+	waitForManualTimers(t, clk, 4)
679		clk.advance(defaultOutputPaceMinDelay)	679		clk.advance(defaultOutputPaceMinDelay)
680		eventuallyPacketCount(t, bPC, 3)	680		eventuallyPacketCount(t, bPC, 3)
681		if err := <-inputDone; err != nil {	681		if err := <-inputDone; err != nil {

@@ -1419,7 +1419,7 @@ func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
1419		}	1419		}
1420		done := make(chan error, 1)	1420		done := make(chan error, 1)
1421		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()	1421		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
1422	-	waitForManualTimers(t, clk, 3)	1422	+	waitForManualTimers(t, clk, 4)
1423			1423
1424		clk.advance(50 * time.Millisecond)	1424		clk.advance(50 * time.Millisecond)
1425		select {	1425		select {