Merge pull request #61 from bnema/fix/udp-send-pacing-coalescing

improve(dgram): pace UDP output sends

commit: f392c58522036ba091cf1f95f944e90cbedd7645
author: bnema · 3mo ago · 2026-07-09 19:10 UTC
committer: GitHub · 3mo ago · 2026-07-09 19:10 UTC
merge of: c01d864 + b9d3800 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

3 files · +593 -68

A internal/adapters/dgram/output_pacer.go +139 -0

@@ -0,0 +1,139 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"time"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	)
	8	+
	9	+	type queuedSend struct {
	10	+	seq uint64
	11	+	reliable bool
	12	+	frame ports.Frame
	13	+	outputSkipThrough uint64
	14	+	}
	15	+
	16	+	func (t *Transport) flushQueuedOutputLocked() []queuedSend {
	17	+	if len(t.outputQueue) == 0 {
	18	+	return nil
	19	+	}
	20	+	queued := append([]queuedSend(nil), t.outputQueue...)
	21	+	t.outputQueue = nil
	22	+	return queued
	23	+	}
	24	+
	25	+	func (t *Transport) outputPaceLoop() {
	26	+	for {
	27	+	t.outboundMu.Lock()
	28	+	t.mu.Lock()
	29	+	if len(t.outputQueue) == 0 && !t.closed {
	30	+	wake := t.outputWake
	31	+	t.mu.Unlock()
	32	+	t.outboundMu.Unlock()
	33	+	select {
	34	+	case <-wake:
	35	+	case <-t.done:
	36	+	return
	37	+	}
	38	+	continue
	39	+	}
	40	+	if t.closed {
	41	+	t.mu.Unlock()
	42	+	t.outboundMu.Unlock()
	43	+	return
	44	+	}
	45	+	now := t.clock.Now()
	46	+	wait := t.outputNext.Sub(now)
	47	+	if wait > 0 {
	48	+	wake := t.outputWake
	49	+	timer := t.clock.NewTimer(wait)
	50	+	t.mu.Unlock()
	51	+	t.outboundMu.Unlock()
	52	+	select {
	53	+	case <-timer.C():
	54	+	case <-wake:
	55	+	case <-t.done:
	56	+	timer.Stop()
	57	+	return
	58	+	}
	59	+	timer.Stop()
	60	+	continue
	61	+	}
	62	+	limit := min(defaultOutputPaceBatch, len(t.outputQueue))
	63	+	batch := append([]queuedSend(nil), t.outputQueue[:limit]...)
	64	+	copy(t.outputQueue, t.outputQueue[limit:])
	65	+	t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
	66	+	t.outputNext = now.Add(t.outputPaceDelayLocked())
	67	+	t.mu.Unlock()
	68	+	for i, q := range batch {
	69	+	if err := t.sendQueuedData(q); err != nil {
	70	+	t.removeQueuedPending(batch[i+1:])
	71	+	t.closeWithError(err)
	72	+	t.outboundMu.Unlock()
	73	+	return
	74	+	}
	75	+	}
	76	+	t.outboundMu.Unlock()
	77	+	}
	78	+	}
	79	+
	80	+	func (t *Transport) notifyOutputPacerLocked() {
	81	+	if t.outputWake == nil {
	82	+	return
	83	+	}
	84	+	close(t.outputWake)
	85	+	t.outputWake = make(chan struct{})
	86	+	}
	87	+
	88	+	func (t *Transport) outputPaceDelayLocked() time.Duration {
	89	+	if t.srtt <= 0 {
	90	+	return defaultOutputPaceMinDelay
	91	+	}
	92	+	d := t.srtt / 2
	93	+	if d < defaultOutputPaceMinCadence {
	94	+	d = defaultOutputPaceMinCadence
	95	+	}
	96	+	if d > defaultOutputPaceMaxDelay {
	97	+	d = defaultOutputPaceMaxDelay
	98	+	}
	99	+	return d
	100	+	}
	101	+
	102	+	func shouldPaceOutput(f ports.Frame) bool {
	103	+	return f.Type == ports.MsgOutput
	104	+	}
	105	+
	106	+	func (t *Transport) sendQueuedData(q queuedSend) error {
	107	+	t.markPendingSent(q.seq, q.reliable)
	108	+	if err := t.sendData(q.seq, q.reliable, q.frame, q.outputSkipThrough); err != nil {
	109	+	t.removePending(q.seq, q.reliable)
	110	+	return err
	111	+	}
	112	+	t.markPendingReady(q.seq, q.reliable)
	113	+	return nil
	114	+	}
	115	+
	116	+	func (t *Transport) markPendingSent(seq uint64, reliable bool) {
	117	+	if !reliable {
	118	+	return
	119	+	}
	120	+	now := t.clock.Now()
	121	+	t.mu.Lock()
	122	+	if p := t.pending[seq]; p != nil && p.first.IsZero() {
	123	+	p.first = now
	124	+	p.last = now
	125	+	p.initialInFlight = true
	126	+	}
	127	+	t.mu.Unlock()
	128	+	}
	129	+
	130	+	func (t *Transport) markPendingReady(seq uint64, reliable bool) {
	131	+	if !reliable {
	132	+	return
	133	+	}
	134	+	t.mu.Lock()
	135	+	if p := t.pending[seq]; p != nil {
	136	+	p.initialInFlight = false
	137	+	}
	138	+	t.mu.Unlock()
	139	+	}

@@ -0,0 +1,139 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"time"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	)
	8	+
	9	+	type queuedSend struct {
	10	+	seq uint64
	11	+	reliable bool
	12	+	frame ports.Frame
	13	+	outputSkipThrough uint64
	14	+	}
	15	+
	16	+	func (t *Transport) flushQueuedOutputLocked() []queuedSend {
	17	+	if len(t.outputQueue) == 0 {
	18	+	return nil
	19	+	}
	20	+	queued := append([]queuedSend(nil), t.outputQueue...)
	21	+	t.outputQueue = nil
	22	+	return queued
	23	+	}
	24	+
	25	+	func (t *Transport) outputPaceLoop() {
	26	+	for {
	27	+	t.outboundMu.Lock()
	28	+	t.mu.Lock()
	29	+	if len(t.outputQueue) == 0 && !t.closed {
	30	+	wake := t.outputWake
	31	+	t.mu.Unlock()
	32	+	t.outboundMu.Unlock()
	33	+	select {
	34	+	case <-wake:
	35	+	case <-t.done:
	36	+	return
	37	+	}
	38	+	continue
	39	+	}
	40	+	if t.closed {
	41	+	t.mu.Unlock()
	42	+	t.outboundMu.Unlock()
	43	+	return
	44	+	}
	45	+	now := t.clock.Now()
	46	+	wait := t.outputNext.Sub(now)
	47	+	if wait > 0 {
	48	+	wake := t.outputWake
	49	+	timer := t.clock.NewTimer(wait)
	50	+	t.mu.Unlock()
	51	+	t.outboundMu.Unlock()
	52	+	select {
	53	+	case <-timer.C():
	54	+	case <-wake:
	55	+	case <-t.done:
	56	+	timer.Stop()
	57	+	return
	58	+	}
	59	+	timer.Stop()
	60	+	continue
	61	+	}
	62	+	limit := min(defaultOutputPaceBatch, len(t.outputQueue))
	63	+	batch := append([]queuedSend(nil), t.outputQueue[:limit]...)
	64	+	copy(t.outputQueue, t.outputQueue[limit:])
	65	+	t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
	66	+	t.outputNext = now.Add(t.outputPaceDelayLocked())
	67	+	t.mu.Unlock()
	68	+	for i, q := range batch {
	69	+	if err := t.sendQueuedData(q); err != nil {
	70	+	t.removeQueuedPending(batch[i+1:])
	71	+	t.closeWithError(err)
	72	+	t.outboundMu.Unlock()
	73	+	return
	74	+	}
	75	+	}
	76	+	t.outboundMu.Unlock()
	77	+	}
	78	+	}
	79	+
	80	+	func (t *Transport) notifyOutputPacerLocked() {
	81	+	if t.outputWake == nil {
	82	+	return
	83	+	}
	84	+	close(t.outputWake)
	85	+	t.outputWake = make(chan struct{})
	86	+	}
	87	+
	88	+	func (t *Transport) outputPaceDelayLocked() time.Duration {
	89	+	if t.srtt <= 0 {
	90	+	return defaultOutputPaceMinDelay
	91	+	}
	92	+	d := t.srtt / 2
	93	+	if d < defaultOutputPaceMinCadence {
	94	+	d = defaultOutputPaceMinCadence
	95	+	}
	96	+	if d > defaultOutputPaceMaxDelay {
	97	+	d = defaultOutputPaceMaxDelay
	98	+	}
	99	+	return d
	100	+	}
	101	+
	102	+	func shouldPaceOutput(f ports.Frame) bool {
	103	+	return f.Type == ports.MsgOutput
	104	+	}
	105	+
	106	+	func (t *Transport) sendQueuedData(q queuedSend) error {
	107	+	t.markPendingSent(q.seq, q.reliable)
	108	+	if err := t.sendData(q.seq, q.reliable, q.frame, q.outputSkipThrough); err != nil {
	109	+	t.removePending(q.seq, q.reliable)
	110	+	return err
	111	+	}
	112	+	t.markPendingReady(q.seq, q.reliable)
	113	+	return nil
	114	+	}
	115	+
	116	+	func (t *Transport) markPendingSent(seq uint64, reliable bool) {
	117	+	if !reliable {
	118	+	return
	119	+	}
	120	+	now := t.clock.Now()
	121	+	t.mu.Lock()
	122	+	if p := t.pending[seq]; p != nil && p.first.IsZero() {
	123	+	p.first = now
	124	+	p.last = now
	125	+	p.initialInFlight = true
	126	+	}
	127	+	t.mu.Unlock()
	128	+	}
	129	+
	130	+	func (t *Transport) markPendingReady(seq uint64, reliable bool) {
	131	+	if !reliable {
	132	+	return
	133	+	}
	134	+	t.mu.Lock()
	135	+	if p := t.pending[seq]; p != nil {
	136	+	p.initialInFlight = false
	137	+	}
	138	+	t.mu.Unlock()
	139	+	}

M internal/adapters/dgram/transport.go +141 -48


@@ -20,22 +20,26 @@ const (
20	20		recProbe byte = 3
21	21		recPong byte = 4
22	22
23		-	defaultMTU = pdgram.DefaultMTU
24		-	defaultResend = 250 * time.Millisecond
25		-	defaultMaxResendAfter = 2 * time.Second
26		-	defaultMaxResendPerTick = 64
27		-	defaultWriteTimeout = 250 * time.Millisecond
28		-	defaultHeartbeat = 3 * time.Second
29		-	defaultDegraded = 10 * time.Second
30		-	defaultProbe = 20 * time.Second
31		-	defaultOffline = 30 * time.Second
32		-	defaultDead = 60 * time.Second
33		-	defaultMaxPending = 1024
34		-	defaultMaxPendingWait = 50 * time.Millisecond
35		-	defaultMaxRecvBuf = 1024
36		-	maxRecvBuffer = defaultMaxRecvBuf
37		-	linkEventBufferSize = 16
38		-	probeReplyBufferSize = 16
	23	+	defaultMTU = pdgram.DefaultMTU
	24	+	defaultResend = 250 * time.Millisecond
	25	+	defaultMaxResendAfter = 2 * time.Second
	26	+	defaultMaxResendPerTick = 64
	27	+	defaultWriteTimeout = 250 * time.Millisecond
	28	+	defaultHeartbeat = 3 * time.Second
	29	+	defaultDegraded = 10 * time.Second
	30	+	defaultProbe = 20 * time.Second
	31	+	defaultOffline = 30 * time.Second
	32	+	defaultDead = 60 * time.Second
	33	+	defaultMaxPending = 1024
	34	+	defaultMaxPendingWait = 50 * time.Millisecond
	35	+	defaultMaxRecvBuf = 1024
	36	+	defaultOutputPaceMinDelay = 8 * time.Millisecond
	37	+	defaultOutputPaceMinCadence = 20 * time.Millisecond
	38	+	defaultOutputPaceMaxDelay = 250 * time.Millisecond
	39	+	defaultOutputPaceBatch = 16
	40	+	maxRecvBuffer = defaultMaxRecvBuf
	41	+	linkEventBufferSize = 16
	42	+	probeReplyBufferSize = 16
39	43		)
40	44
41	45		var (

@@ -163,6 +167,9 @@ type Transport struct {
163	167		probeReply chan uint64
164	168		rebind func(net.PacketConn) (net.PacketConn, error)
165	169		hoppedOffline bool
	170	+	outputQueue []queuedSend
	171	+	outputWake chan struct{}
	172	+	outputNext time.Time
166	173
167	174		recvMu sync.Mutex
168	175		replay *pdgram.ReplayWindow

@@ -171,6 +178,9 @@ type Transport struct {
171	178		done chan struct{}
172	179
173	180		writeMu sync.Mutex
	181	+	// outboundMu orders first transmissions and paced batches. Retransmits use
	182	+	// writeMu directly because they do not establish new sequence order.
	183	+	outboundMu sync.Mutex
174	184
175	185		deliverMu sync.Mutex
176	186		deliverCond *sync.Cond

@@ -183,6 +193,7 @@ type pending struct {
183	193		frame ports.Frame
184	194		first time.Time
185	195		last time.Time
	196	+	initialInFlight bool
186	197		attempts int
187	198		retransmitted bool
188	199		outputSkipThrough uint64

@@ -233,12 +244,14 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
233	244		recvBuf: make(map[uint64]ports.Frame),
234	245		}
235	246		t.sendWake = make(chan struct{})
	247	+	t.outputWake = make(chan struct{})
236	248		t.probeReply = make(chan uint64, probeReplyBufferSize)
237	249		t.deliverCond = sync.NewCond(&t.deliverMu)
238	250		go t.readLoop(pc)
239	251		go t.resendLoop()
240	252		go t.deliveryLoop()
241	253		go t.probeReplyLoop()
	254	+	go t.outputPaceLoop()
242	255		return t, nil
243	256		}
244	257

@@ -248,16 +261,81 @@ func (t *Transport) DatagramTransport() {}
248	261
249	262		func (t *Transport) Send(f ports.Frame) error {
250	263		reliable := true
251		-	t.mu.Lock()
252		-	outputSkipThrough := uint64(0)
253		-	if f.Type == ports.MsgOutput {
254		-	outputSkipThrough = t.supersedePendingOutputLocked()
	264	+	outputSkipThrough, err := t.lockOutboundSlot(f, reliable)
	265	+	if err != nil {
	266	+	return err
	267	+	}
	268	+	defer t.outboundMu.Unlock()
	269	+	// lockOutboundSlot returns with both outboundMu and mu held.
	270	+
	271	+	t.seq++
	272	+	seq := t.seq
	273	+	if reliable {
	274	+	t.pending[seq] = &pending{frame: f, outputSkipThrough: outputSkipThrough}
255	275		}
	276	+	if shouldPaceOutput(f) {
	277	+	now := t.clock.Now()
	278	+	if len(t.outputQueue) > 0 \|\| (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
	279	+	t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, outputSkipThrough: outputSkipThrough})
	280	+	t.notifyOutputPacerLocked()
	281	+	t.mu.Unlock()
	282	+	return nil
	283	+	}
	284	+	t.outputNext = now.Add(t.outputPaceDelayLocked())
	285	+	t.mu.Unlock()
	286	+	t.markPendingSent(seq, reliable)
	287	+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
	288	+	t.removePending(seq, reliable)
	289	+	return err
	290	+	}
	291	+	t.markPendingReady(seq, reliable)
	292	+	return nil
	293	+	}
	294	+	queued := t.flushQueuedOutputLocked()
	295	+	t.mu.Unlock()
	296	+	for i, q := range queued {
	297	+	if err := t.sendQueuedData(q); err != nil {
	298	+	t.removePending(seq, reliable)
	299	+	t.removeQueuedPending(queued[i+1:])
	300	+	return err
	301	+	}
	302	+	}
	303	+	t.markPendingSent(seq, reliable)
	304	+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
	305	+	t.removePending(seq, reliable)
	306	+	return err
	307	+	}
	308	+	t.markPendingReady(seq, reliable)
	309	+	return nil
	310	+	}
	311	+
	312	+	func (t *Transport) lockOutboundSlot(f ports.Frame, reliable bool) (uint64, error) {
256	313		var pendingWaitStarted time.Time
257		-	for reliable && len(t.pending) >= t.maxPending && !t.closed {
	314	+	var outputSkipThrough uint64
	315	+	for {
	316	+	t.outboundMu.Lock()
	317	+	t.mu.Lock()
	318	+	if t.closed {
	319	+	err := t.closeErr
	320	+	t.mu.Unlock()
	321	+	t.outboundMu.Unlock()
	322	+	if err != nil {
	323	+	return 0, err
	324	+	}
	325	+	return 0, errors.New("dgram: closed")
	326	+	}
	327	+	if f.Type == ports.MsgOutput {
	328	+	if skipped := t.supersedePendingOutputLocked(); skipped > outputSkipThrough {
	329	+	outputSkipThrough = skipped
	330	+	}
	331	+	}
	332	+	if !reliable \|\| len(t.pending) < t.maxPending {
	333	+	return outputSkipThrough, nil
	334	+	}
258	335		if t.linkState != ports.LinkStateConnected {
259	336		t.mu.Unlock()
260		-	return ErrPendingFull
	337	+	t.outboundMu.Unlock()
	338	+	return 0, ErrPendingFull
261	339		}
262	340		if pendingWaitStarted.IsZero() {
263	341		pendingWaitStarted = t.clock.Now()

@@ -265,46 +343,54 @@ func (t *Transport) Send(f ports.Frame) error {
265	343		remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
266	344		if remaining <= 0 {
267	345		t.mu.Unlock()
268		-	return ErrPendingFull
	346	+	t.outboundMu.Unlock()
	347	+	return 0, ErrPendingFull
269	348		}
270	349		wake := t.sendWake
271	350		timer := t.clock.NewTimer(remaining)
272	351		t.mu.Unlock()
	352	+	t.outboundMu.Unlock()
273	353		select {
274	354		case <-timer.C():
275	355		case <-wake:
276	356		case <-t.done:
277	357		}
278	358		timer.Stop()
279		-	t.mu.Lock()
280	359		}
281		-	if t.closed {
282		-	err := t.closeErr
283		-	t.mu.Unlock()
284		-	if err != nil {
285		-	return err
286		-	}
287		-	return errors.New("dgram: closed")
	360	+	}
	361	+
	362	+	func (t *Transport) removePending(seq uint64, reliable bool) {
	363	+	if !reliable {
	364	+	return
288	365		}
289		-	t.seq++
290		-	seq := t.seq
291		-	if reliable {
292		-	now := t.clock.Now()
293		-	t.pending[seq] = &pending{frame: f, first: now, last: now, outputSkipThrough: outputSkipThrough}
	366	+	t.mu.Lock()
	367	+	removed := t.removePendingLocked(seq)
	368	+	if removed {
	369	+	t.notifySendWaitersLocked()
294	370		}
295	371		t.mu.Unlock()
296		-	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
297		-	if reliable {
298		-	t.mu.Lock()
299		-	if _, ok := t.pending[seq]; ok {
300		-	delete(t.pending, seq)
301		-	t.notifySendWaitersLocked()
302		-	}
303		-	t.mu.Unlock()
	372	+	}
	373	+
	374	+	func (t *Transport) removeQueuedPending(queued []queuedSend) {
	375	+	t.mu.Lock()
	376	+	removed := false
	377	+	for _, q := range queued {
	378	+	if q.reliable {
	379	+	removed = t.removePendingLocked(q.seq) \|\| removed
304	380		}
305		-	return err
306	381		}
307		-	return nil
	382	+	if removed {
	383	+	t.notifySendWaitersLocked()
	384	+	}
	385	+	t.mu.Unlock()
	386	+	}
	387	+
	388	+	func (t *Transport) removePendingLocked(seq uint64) bool {
	389	+	if _, ok := t.pending[seq]; !ok {
	390	+	return false
	391	+	}
	392	+	delete(t.pending, seq)
	393	+	return true
308	394		}
309	395
310	396		func (t *Transport) supersedePendingOutputLocked() uint64 {

@@ -331,10 +417,17 @@ func (t *Transport) supersedePendingOutputLocked() uint64 {
331	417		delete(t.pending, seq)
332	418		}
333	419		}
	420	+	kept := t.outputQueue[:0]
	421	+	for _, q := range t.outputQueue {
	422	+	if q.seq > barrier && q.seq <= newest {
	423	+	continue
	424	+	}
	425	+	kept = append(kept, q)
	426	+	}
	427	+	t.outputQueue = kept
334	428		t.notifySendWaitersLocked()
335	429		return newest
336	430		}
337		-
338	431		func (t *Transport) Recv() (ports.Frame, error) {
339	432		select {
340	433		case f := <-t.in:

@@ -748,7 +841,7 @@ func (t *Transport) resendPending() {
748	841		break
749	842		}
750	843		p := t.pending[seq]
751		-	if p == nil \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
	844	+	if p == nil \|\| p.last.IsZero() \|\| p.initialInFlight \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
752	845		continue
753	846		}
754	847		p.last = now


@@ -20,22 +20,26 @@ const (
20		recProbe byte = 3	20		recProbe byte = 3
21		recPong byte = 4	21		recPong byte = 4
22			22
23	-	defaultMTU = pdgram.DefaultMTU	23	+	defaultMTU = pdgram.DefaultMTU
24	-	defaultResend = 250 * time.Millisecond	24	+	defaultResend = 250 * time.Millisecond
25	-	defaultMaxResendAfter = 2 * time.Second	25	+	defaultMaxResendAfter = 2 * time.Second
26	-	defaultMaxResendPerTick = 64	26	+	defaultMaxResendPerTick = 64
27	-	defaultWriteTimeout = 250 * time.Millisecond	27	+	defaultWriteTimeout = 250 * time.Millisecond
28	-	defaultHeartbeat = 3 * time.Second	28	+	defaultHeartbeat = 3 * time.Second
29	-	defaultDegraded = 10 * time.Second	29	+	defaultDegraded = 10 * time.Second
30	-	defaultProbe = 20 * time.Second	30	+	defaultProbe = 20 * time.Second
31	-	defaultOffline = 30 * time.Second	31	+	defaultOffline = 30 * time.Second
32	-	defaultDead = 60 * time.Second	32	+	defaultDead = 60 * time.Second
33	-	defaultMaxPending = 1024	33	+	defaultMaxPending = 1024
34	-	defaultMaxPendingWait = 50 * time.Millisecond	34	+	defaultMaxPendingWait = 50 * time.Millisecond
35	-	defaultMaxRecvBuf = 1024	35	+	defaultMaxRecvBuf = 1024
36	-	maxRecvBuffer = defaultMaxRecvBuf	36	+	defaultOutputPaceMinDelay = 8 * time.Millisecond
37	-	linkEventBufferSize = 16	37	+	defaultOutputPaceMinCadence = 20 * time.Millisecond
38	-	probeReplyBufferSize = 16	38	+	defaultOutputPaceMaxDelay = 250 * time.Millisecond
			39	+	defaultOutputPaceBatch = 16
			40	+	maxRecvBuffer = defaultMaxRecvBuf
			41	+	linkEventBufferSize = 16
			42	+	probeReplyBufferSize = 16
39		)	43		)
40			44
41		var (	45		var (

@@ -163,6 +167,9 @@ type Transport struct {
163		probeReply chan uint64	167		probeReply chan uint64
164		rebind func(net.PacketConn) (net.PacketConn, error)	168		rebind func(net.PacketConn) (net.PacketConn, error)
165		hoppedOffline bool	169		hoppedOffline bool
			170	+	outputQueue []queuedSend
			171	+	outputWake chan struct{}
			172	+	outputNext time.Time
166			173
167		recvMu sync.Mutex	174		recvMu sync.Mutex
168		replay *pdgram.ReplayWindow	175		replay *pdgram.ReplayWindow

@@ -171,6 +178,9 @@ type Transport struct {
171		done chan struct{}	178		done chan struct{}
172			179
173		writeMu sync.Mutex	180		writeMu sync.Mutex
			181	+	// outboundMu orders first transmissions and paced batches. Retransmits use
			182	+	// writeMu directly because they do not establish new sequence order.
			183	+	outboundMu sync.Mutex
174			184
175		deliverMu sync.Mutex	185		deliverMu sync.Mutex
176		deliverCond *sync.Cond	186		deliverCond *sync.Cond

@@ -183,6 +193,7 @@ type pending struct {
183		frame ports.Frame	193		frame ports.Frame
184		first time.Time	194		first time.Time
185		last time.Time	195		last time.Time
			196	+	initialInFlight bool
186		attempts int	197		attempts int
187		retransmitted bool	198		retransmitted bool
188		outputSkipThrough uint64	199		outputSkipThrough uint64

@@ -233,12 +244,14 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
233		recvBuf: make(map[uint64]ports.Frame),	244		recvBuf: make(map[uint64]ports.Frame),
234		}	245		}
235		t.sendWake = make(chan struct{})	246		t.sendWake = make(chan struct{})
			247	+	t.outputWake = make(chan struct{})
236		t.probeReply = make(chan uint64, probeReplyBufferSize)	248		t.probeReply = make(chan uint64, probeReplyBufferSize)
237		t.deliverCond = sync.NewCond(&t.deliverMu)	249		t.deliverCond = sync.NewCond(&t.deliverMu)
238		go t.readLoop(pc)	250		go t.readLoop(pc)
239		go t.resendLoop()	251		go t.resendLoop()
240		go t.deliveryLoop()	252		go t.deliveryLoop()
241		go t.probeReplyLoop()	253		go t.probeReplyLoop()
			254	+	go t.outputPaceLoop()
242		return t, nil	255		return t, nil
243		}	256		}
244			257

@@ -248,16 +261,81 @@ func (t *Transport) DatagramTransport() {}
248			261
249		func (t *Transport) Send(f ports.Frame) error {	262		func (t *Transport) Send(f ports.Frame) error {
250		reliable := true	263		reliable := true
251	-	t.mu.Lock()	264	+	outputSkipThrough, err := t.lockOutboundSlot(f, reliable)
252	-	outputSkipThrough := uint64(0)	265	+	if err != nil {
253	-	if f.Type == ports.MsgOutput {	266	+	return err
254	-	outputSkipThrough = t.supersedePendingOutputLocked()	267	+	}
			268	+	defer t.outboundMu.Unlock()
			269	+	// lockOutboundSlot returns with both outboundMu and mu held.
			270	+
			271	+	t.seq++
			272	+	seq := t.seq
			273	+	if reliable {
			274	+	t.pending[seq] = &pending{frame: f, outputSkipThrough: outputSkipThrough}
255		}	275		}
			276	+	if shouldPaceOutput(f) {
			277	+	now := t.clock.Now()
			278	+	if len(t.outputQueue) > 0 \|\| (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
			279	+	t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, outputSkipThrough: outputSkipThrough})
			280	+	t.notifyOutputPacerLocked()
			281	+	t.mu.Unlock()
			282	+	return nil
			283	+	}
			284	+	t.outputNext = now.Add(t.outputPaceDelayLocked())
			285	+	t.mu.Unlock()
			286	+	t.markPendingSent(seq, reliable)
			287	+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
			288	+	t.removePending(seq, reliable)
			289	+	return err
			290	+	}
			291	+	t.markPendingReady(seq, reliable)
			292	+	return nil
			293	+	}
			294	+	queued := t.flushQueuedOutputLocked()
			295	+	t.mu.Unlock()
			296	+	for i, q := range queued {
			297	+	if err := t.sendQueuedData(q); err != nil {
			298	+	t.removePending(seq, reliable)
			299	+	t.removeQueuedPending(queued[i+1:])
			300	+	return err
			301	+	}
			302	+	}
			303	+	t.markPendingSent(seq, reliable)
			304	+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
			305	+	t.removePending(seq, reliable)
			306	+	return err
			307	+	}
			308	+	t.markPendingReady(seq, reliable)
			309	+	return nil
			310	+	}
			311	+
			312	+	func (t *Transport) lockOutboundSlot(f ports.Frame, reliable bool) (uint64, error) {
256		var pendingWaitStarted time.Time	313		var pendingWaitStarted time.Time
257	-	for reliable && len(t.pending) >= t.maxPending && !t.closed {	314	+	var outputSkipThrough uint64
			315	+	for {
			316	+	t.outboundMu.Lock()
			317	+	t.mu.Lock()
			318	+	if t.closed {
			319	+	err := t.closeErr
			320	+	t.mu.Unlock()
			321	+	t.outboundMu.Unlock()
			322	+	if err != nil {
			323	+	return 0, err
			324	+	}
			325	+	return 0, errors.New("dgram: closed")
			326	+	}
			327	+	if f.Type == ports.MsgOutput {
			328	+	if skipped := t.supersedePendingOutputLocked(); skipped > outputSkipThrough {
			329	+	outputSkipThrough = skipped
			330	+	}
			331	+	}
			332	+	if !reliable \|\| len(t.pending) < t.maxPending {
			333	+	return outputSkipThrough, nil
			334	+	}
258		if t.linkState != ports.LinkStateConnected {	335		if t.linkState != ports.LinkStateConnected {
259		t.mu.Unlock()	336		t.mu.Unlock()
260	-	return ErrPendingFull	337	+	t.outboundMu.Unlock()
			338	+	return 0, ErrPendingFull
261		}	339		}
262		if pendingWaitStarted.IsZero() {	340		if pendingWaitStarted.IsZero() {
263		pendingWaitStarted = t.clock.Now()	341		pendingWaitStarted = t.clock.Now()

@@ -265,46 +343,54 @@ func (t *Transport) Send(f ports.Frame) error {
265		remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)	343		remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
266		if remaining <= 0 {	344		if remaining <= 0 {
267		t.mu.Unlock()	345		t.mu.Unlock()
268	-	return ErrPendingFull	346	+	t.outboundMu.Unlock()
			347	+	return 0, ErrPendingFull
269		}	348		}
270		wake := t.sendWake	349		wake := t.sendWake
271		timer := t.clock.NewTimer(remaining)	350		timer := t.clock.NewTimer(remaining)
272		t.mu.Unlock()	351		t.mu.Unlock()
			352	+	t.outboundMu.Unlock()
273		select {	353		select {
274		case <-timer.C():	354		case <-timer.C():
275		case <-wake:	355		case <-wake:
276		case <-t.done:	356		case <-t.done:
277		}	357		}
278		timer.Stop()	358		timer.Stop()
279	-	t.mu.Lock()
280		}	359		}
281	-	if t.closed {	360	+	}
282	-	err := t.closeErr	361	+
283	-	t.mu.Unlock()	362	+	func (t *Transport) removePending(seq uint64, reliable bool) {
284	-	if err != nil {	363	+	if !reliable {
285	-	return err	364	+	return
286	-	}
287	-	return errors.New("dgram: closed")
288		}	365		}
289	-	t.seq++	366	+	t.mu.Lock()
290	-	seq := t.seq	367	+	removed := t.removePendingLocked(seq)
291	-	if reliable {	368	+	if removed {
292	-	now := t.clock.Now()	369	+	t.notifySendWaitersLocked()
293	-	t.pending[seq] = &pending{frame: f, first: now, last: now, outputSkipThrough: outputSkipThrough}
294		}	370		}
295		t.mu.Unlock()	371		t.mu.Unlock()
296	-	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {	372	+	}
297	-	if reliable {	373	+
298	-	t.mu.Lock()	374	+	func (t *Transport) removeQueuedPending(queued []queuedSend) {
299	-	if _, ok := t.pending[seq]; ok {	375	+	t.mu.Lock()
300	-	delete(t.pending, seq)	376	+	removed := false
301	-	t.notifySendWaitersLocked()	377	+	for _, q := range queued {
302	-	}	378	+	if q.reliable {
303	-	t.mu.Unlock()	379	+	removed = t.removePendingLocked(q.seq) \|\| removed
304		}	380		}
305	-	return err
306		}	381		}
307	-	return nil	382	+	if removed {
			383	+	t.notifySendWaitersLocked()
			384	+	}
			385	+	t.mu.Unlock()
			386	+	}
			387	+
			388	+	func (t *Transport) removePendingLocked(seq uint64) bool {
			389	+	if _, ok := t.pending[seq]; !ok {
			390	+	return false
			391	+	}
			392	+	delete(t.pending, seq)
			393	+	return true
308		}	394		}
309			395
310		func (t *Transport) supersedePendingOutputLocked() uint64 {	396		func (t *Transport) supersedePendingOutputLocked() uint64 {

@@ -331,10 +417,17 @@ func (t *Transport) supersedePendingOutputLocked() uint64 {
331		delete(t.pending, seq)	417		delete(t.pending, seq)
332		}	418		}
333		}	419		}
			420	+	kept := t.outputQueue[:0]
			421	+	for _, q := range t.outputQueue {
			422	+	if q.seq > barrier && q.seq <= newest {
			423	+	continue
			424	+	}
			425	+	kept = append(kept, q)
			426	+	}
			427	+	t.outputQueue = kept
334		t.notifySendWaitersLocked()	428		t.notifySendWaitersLocked()
335		return newest	429		return newest
336		}	430		}
337	-
338		func (t *Transport) Recv() (ports.Frame, error) {	431		func (t *Transport) Recv() (ports.Frame, error) {
339		select {	432		select {
340		case f := <-t.in:	433		case f := <-t.in:

@@ -748,7 +841,7 @@ func (t *Transport) resendPending() {
748		break	841		break
749		}	842		}
750		p := t.pending[seq]	843		p := t.pending[seq]
751	-	if p == nil \|\| now.Sub(p.last) < t.resendDelayLocked(p) {	844	+	if p == nil \|\| p.last.IsZero() \|\| p.initialInFlight \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
752		continue	845		continue
753		}	846		}
754		p.last = now	847		p.last = now

M internal/adapters/dgram/transport_test.go +313 -20


@@ -171,11 +171,11 @@ func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
171	171		type manualClock struct {
172	172		mu sync.Mutex
173	173		now time.Time
174		-	timers chan *manualTimer
	174	+	timers map[*manualTimer]struct{}
175	175		}
176	176
177	177		func newManualClock(now time.Time) *manualClock {
178		-	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}
	178	+	return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}
179	179		}
180	180
181	181		func (c *manualClock) Now() time.Time {

@@ -185,8 +185,10 @@ func (c *manualClock) Now() time.Time {
185	185		}
186	186
187	187		func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
188		-	tm := &manualTimer{c: make(chan time.Time, 1), d: d}
189		-	c.timers <- tm
	188	+	c.mu.Lock()
	189	+	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
	190	+	c.timers[tm] = struct{}{}
	191	+	c.mu.Unlock()
190	192		return tm
191	193		}
192	194

@@ -194,45 +196,78 @@ func (c *manualClock) advance(d time.Duration) {
194	196		c.mu.Lock()
195	197		c.now = c.now.Add(d)
196	198		now := c.now
	199	+	timers := make([]*manualTimer, 0, len(c.timers))
	200	+	for tm := range c.timers {
	201	+	timers = append(timers, tm)
	202	+	}
197	203		c.mu.Unlock()
198		-	for {
199		-	select {
200		-	case tm := <-c.timers:
201		-	tm.fire(now)
202		-	default:
203		-	return
204		-	}
	204	+	for _, tm := range timers {
	205	+	tm.fireIfDue(now)
205	206		}
206	207		}
207	208
208	209		type manualTimer struct {
209		-	c chan time.Time
210		-	d time.Duration
	210	+	clock *manualClock
	211	+	c chan time.Time
	212	+	deadline time.Time
	213	+	active bool
211	214		}
212	215
213	216		func (t *manualTimer) C() <-chan time.Time { return t.c }
214	217		func (t *manualTimer) Reset(d time.Duration) bool {
215		-	t.d = d
216		-	return true
217		-	}
218		-	func (t *manualTimer) Stop() bool { return true }
219		-	func (t *manualTimer) fire(now time.Time) {
	218	+	t.clock.mu.Lock()
	219	+	wasActive := t.active
	220	+	t.deadline = t.clock.now.Add(d)
	221	+	t.active = true
	222	+	t.clock.timers[t] = struct{}{}
	223	+	t.clock.mu.Unlock()
	224	+	return wasActive
	225	+	}
	226	+	func (t *manualTimer) Stop() bool {
	227	+	t.clock.mu.Lock()
	228	+	defer t.clock.mu.Unlock()
	229	+	wasActive := t.active
	230	+	t.active = false
	231	+	delete(t.clock.timers, t)
	232	+	return wasActive
	233	+	}
	234	+	func (t *manualTimer) fireIfDue(now time.Time) bool {
	235	+	t.clock.mu.Lock()
	236	+	if !t.active {
	237	+	delete(t.clock.timers, t)
	238	+	t.clock.mu.Unlock()
	239	+	return false
	240	+	}
	241	+	if now.Before(t.deadline) {
	242	+	t.clock.mu.Unlock()
	243	+	return false
	244	+	}
	245	+	t.active = false
	246	+	delete(t.clock.timers, t)
	247	+	t.clock.mu.Unlock()
220	248		select {
221	249		case t.c <- now:
222	250		default:
223	251		}
	252	+	return true
224	253		}
225	254
226	255		func waitForManualTimers(t testing.T, clk manualClock, n int) {
227	256		t.Helper()
228	257		deadline := time.Now().Add(time.Second)
229	258		for time.Now().Before(deadline) {
230		-	if len(clk.timers) >= n {
	259	+	clk.mu.Lock()
	260	+	active := len(clk.timers)
	261	+	clk.mu.Unlock()
	262	+	if active >= n {
231	263		return
232	264		}
233	265		time.Sleep(time.Millisecond)
234	266		}
235		-	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
	267	+	clk.mu.Lock()
	268	+	active := len(clk.timers)
	269	+	clk.mu.Unlock()
	270	+	t.Fatalf("manual timers=%d, want at least %d", active, n)
236	271		}
237	272
238	273		type deadlineCapturePC struct {

@@ -264,6 +299,264 @@ func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
264	299		return nil
265	300		}
266	301
	302	+	func TestOutputFirstSendIsPacedAndQueuedOutputsAreCoalesced(t *testing.T) {
	303	+	aPC, bPC := newPair()
	304	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	305	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	306	+	if err != nil {
	307	+	t.Fatal(err)
	308	+	}
	309	+	defer func() { _ = a.Close() }()
	310	+
	311	+	for i := range 3 {
	312	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
	313	+	t.Fatal(err)
	314	+	}
	315	+	}
	316	+	if got := len(bPC.in); got != 1 {
	317	+	t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
	318	+	}
	319	+	waitForManualTimers(t, clk, 3)
	320	+	clk.advance(defaultOutputPaceMinDelay)
	321	+	eventuallyPacketCount(t, bPC, 2)
	322	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {
	323	+	t.Fatal(err)
	324	+	}
	325	+	if got := len(bPC.in); got != 2 {
	326	+	t.Fatalf("datagrams before second pace tick=%d, want 2", got)
	327	+	}
	328	+	clk.advance(defaultOutputPaceMinDelay)
	329	+	eventuallyPacketCount(t, bPC, 3)
	330	+	}
	331	+
	332	+	func TestResendSkipsQueuedOutputBeforeFirstWireAttempt(t *testing.T) {
	333	+	tr, writes := newResendTestTransport(t, 10)
	334	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}}
	335	+
	336	+	tr.resendPending()
	337	+
	338	+	if got := writes.Load(); got != 0 {
	339	+	t.Fatalf("writes=%d, want queued output excluded from resend", got)
	340	+	}
	341	+	if !tr.pending[1].first.IsZero() \|\| !tr.pending[1].last.IsZero() {
	342	+	t.Fatal("resend scan marked queued output as sent")
	343	+	}
	344	+	}
	345	+
	346	+	func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
	347	+	tr, writes := newResendTestTransport(t, 10)
	348	+	now := tr.clock.Now()
	349	+	tr.pending[1] = &pending{
	350	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("sending")},
	351	+	first: now.Add(-time.Second),
	352	+	last: now.Add(-time.Second),
	353	+	initialInFlight: true,
	354	+	}
	355	+
	356	+	tr.resendPending()
	357	+	if got := writes.Load(); got != 0 {
	358	+	t.Fatalf("writes=%d, want first wire attempt excluded from resend", got)
	359	+	}
	360	+	tr.pending[1].initialInFlight = false
	361	+	tr.resendPending()
	362	+	if got := writes.Load(); got != 1 {
	363	+	t.Fatalf("writes=%d, want resend after first wire attempt completed", got)
	364	+	}
	365	+	}
	366	+
	367	+	func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
	368	+	tr, _ := newResendTestTransport(t, 10)
	369	+	tests := []struct {
	370	+	name string
	371	+	srtt time.Duration
	372	+	want time.Duration
	373	+	}{
	374	+	{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
	375	+	{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
	376	+	{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
	377	+	{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
	378	+	}
	379	+	for _, tt := range tests {
	380	+	t.Run(tt.name, func(t *testing.T) {
	381	+	tr.mu.Lock()
	382	+	tr.srtt = tt.srtt
	383	+	got := tr.outputPaceDelayLocked()
	384	+	tr.mu.Unlock()
	385	+	if got != tt.want {
	386	+	t.Fatalf("output pace delay=%v, want %v", got, tt.want)
	387	+	}
	388	+	})
	389	+	}
	390	+	}
	391	+
	392	+	func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
	393	+	aPC, bPC := newPair()
	394	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	395	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	396	+	if err != nil {
	397	+	t.Fatal(err)
	398	+	}
	399	+	defer func() { _ = a.Close() }()
	400	+
	401	+	for _, payload := range []string{"first", "queued"} {
	402	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
	403	+	t.Fatal(err)
	404	+	}
	405	+	}
	406	+	waitForManualTimers(t, clk, 3)
	407	+	a.writeMu.Lock()
	408	+	clk.advance(defaultOutputPaceMinDelay)
	409	+	deadline := time.Now().Add(time.Second)
	410	+	for {
	411	+	a.mu.Lock()
	412	+	dequeued := len(a.outputQueue) == 0
	413	+	a.mu.Unlock()
	414	+	if dequeued {
	415	+	break
	416	+	}
	417	+	if time.Now().After(deadline) {
	418	+	a.writeMu.Unlock()
	419	+	t.Fatal("pacer did not dequeue output batch")
	420	+	}
	421	+	time.Sleep(time.Millisecond)
	422	+	}
	423	+
	424	+	done := make(chan error, 1)
	425	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
	426	+	select {
	427	+	case err := <-done:
	428	+	a.writeMu.Unlock()
	429	+	t.Fatalf("input completed before dequeued output batch: %v", err)
	430	+	case <-time.After(20 * time.Millisecond):
	431	+	}
	432	+	a.writeMu.Unlock()
	433	+	select {
	434	+	case err := <-done:
	435	+	if err != nil {
	436	+	t.Fatal(err)
	437	+	}
	438	+	case <-time.After(time.Second):
	439	+	t.Fatal("input did not complete after output batch was released")
	440	+	}
	441	+	if got := len(bPC.in); got != 3 {
	442	+	t.Fatalf("datagrams when input completed=%d, want output batch before input", got)
	443	+	}
	444	+	}
	445	+
	446	+	func TestInputFlushFailureRemovesUnsentQueuedPendingFrames(t *testing.T) {
	447	+	aPC, bPC := newPair()
	448	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	449	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	450	+	if err != nil {
	451	+	t.Fatal(err)
	452	+	}
	453	+	defer func() { _ = a.Close() }()
	454	+
	455	+	for i := range 3 {
	456	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
	457	+	t.Fatal(err)
	458	+	}
	459	+	}
	460	+	if got := len(bPC.in); got != 1 {
	461	+	t.Fatalf("immediate output datagrams=%d, want 1", got)
	462	+	}
	463	+	if err := bPC.Close(); err != nil {
	464	+	t.Fatal(err)
	465	+	}
	466	+
	467	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}); err == nil {
	468	+	t.Fatal("input send after peer close succeeded, want write error")
	469	+	}
	470	+
	471	+	a.mu.Lock()
	472	+	defer a.mu.Unlock()
	473	+	for _, seq := range []uint64{1, 2, 3, 4} {
	474	+	if _, ok := a.pending[seq]; ok {
	475	+	t.Fatalf("pending[%d] stranded after queued flush failure", seq)
	476	+	}
	477	+	}
	478	+	}
	479	+
	480	+	func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
	481	+	aPC, bPC := newPair()
	482	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	483	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	484	+	if err != nil {
	485	+	t.Fatal(err)
	486	+	}
	487	+	defer func() { _ = a.Close() }()
	488	+
	489	+	for i := range 2 {
	490	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
	491	+	t.Fatal(err)
	492	+	}
	493	+	}
	494	+	waitForManualTimers(t, clk, 3)
	495	+	if err := bPC.Close(); err != nil {
	496	+	t.Fatal(err)
	497	+	}
	498	+	clk.advance(defaultOutputPaceMinDelay)
	499	+
	500	+	select {
	501	+	case <-a.done:
	502	+	case <-time.After(time.Second):
	503	+	t.Fatal("transport remained open after paced output write failed")
	504	+	}
	505	+	if err := a.Send(ports.Frame{Type: ports.MsgInput}); err == nil \|\| err.Error() != "peer closed" {
	506	+	t.Fatalf("send after paced write failure err=%v, want peer closed", err)
	507	+	}
	508	+	}
	509	+
	510	+	func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
	511	+	aPC, bPC := newPair()
	512	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	513	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	514	+	if err != nil {
	515	+	t.Fatal(err)
	516	+	}
	517	+	defer func() { _ = a.Close() }()
	518	+
	519	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("output")}); err != nil {
	520	+	t.Fatal(err)
	521	+	}
	522	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
	523	+	t.Fatal(err)
	524	+	}
	525	+	if got := len(bPC.in); got != 1 {
	526	+	t.Fatalf("immediate output datagrams=%d, want 1", got)
	527	+	}
	528	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}); err != nil {
	529	+	t.Fatal(err)
	530	+	}
	531	+	if got := len(bPC.in); got != 3 {
	532	+	t.Fatalf("datagrams after input=%d, want queued output flushed plus input immediately", got)
	533	+	}
	534	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
	535	+	t.Fatal(err)
	536	+	}
	537	+	if got := len(bPC.in); got != 3 {
	538	+	t.Fatalf("datagrams before control=%d, want queued output still paced", got)
	539	+	}
	540	+	if err := a.Send(ports.Frame{Type: ports.MsgPing}); err != nil {
	541	+	t.Fatal(err)
	542	+	}
	543	+	if got := len(bPC.in); got != 5 {
	544	+	t.Fatalf("datagrams after control=%d, want queued output flushed plus control sent immediately", got)
	545	+	}
	546	+	}
	547	+
	548	+	func eventuallyPacketCount(t testing.T, pc fakePC, want int) {
	549	+	t.Helper()
	550	+	deadline := time.Now().Add(time.Second)
	551	+	for time.Now().Before(deadline) {
	552	+	if len(pc.in) >= want {
	553	+	return
	554	+	}
	555	+	time.Sleep(time.Millisecond)
	556	+	}
	557	+	t.Fatalf("datagrams=%d, want at least %d", len(pc.in), want)
	558	+	}
	559	+
267	560		func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
268	561		pc := newBlockingWritePC()
269	562		tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{


@@ -171,11 +171,11 @@ func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
171		type manualClock struct {	171		type manualClock struct {
172		mu sync.Mutex	172		mu sync.Mutex
173		now time.Time	173		now time.Time
174	-	timers chan *manualTimer	174	+	timers map[*manualTimer]struct{}
175		}	175		}
176			176
177		func newManualClock(now time.Time) *manualClock {	177		func newManualClock(now time.Time) *manualClock {
178	-	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}	178	+	return &manualClock{now: now, timers: make(map[*manualTimer]struct{})}
179		}	179		}
180			180
181		func (c *manualClock) Now() time.Time {	181		func (c *manualClock) Now() time.Time {

@@ -185,8 +185,10 @@ func (c *manualClock) Now() time.Time {
185		}	185		}
186			186
187		func (c *manualClock) NewTimer(d time.Duration) ports.Timer {	187		func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
188	-	tm := &manualTimer{c: make(chan time.Time, 1), d: d}	188	+	c.mu.Lock()
189	-	c.timers <- tm	189	+	tm := &manualTimer{clock: c, c: make(chan time.Time, 1), deadline: c.now.Add(d), active: true}
			190	+	c.timers[tm] = struct{}{}
			191	+	c.mu.Unlock()
190		return tm	192		return tm
191		}	193		}
192			194

@@ -194,45 +196,78 @@ func (c *manualClock) advance(d time.Duration) {
194		c.mu.Lock()	196		c.mu.Lock()
195		c.now = c.now.Add(d)	197		c.now = c.now.Add(d)
196		now := c.now	198		now := c.now
			199	+	timers := make([]*manualTimer, 0, len(c.timers))
			200	+	for tm := range c.timers {
			201	+	timers = append(timers, tm)
			202	+	}
197		c.mu.Unlock()	203		c.mu.Unlock()
198	-	for {	204	+	for _, tm := range timers {
199	-	select {	205	+	tm.fireIfDue(now)
200	-	case tm := <-c.timers:
201	-	tm.fire(now)
202	-	default:
203	-	return
204	-	}
205		}	206		}
206		}	207		}
207			208
208		type manualTimer struct {	209		type manualTimer struct {
209	-	c chan time.Time	210	+	clock *manualClock
210	-	d time.Duration	211	+	c chan time.Time
			212	+	deadline time.Time
			213	+	active bool
211		}	214		}
212			215
213		func (t *manualTimer) C() <-chan time.Time { return t.c }	216		func (t *manualTimer) C() <-chan time.Time { return t.c }
214		func (t *manualTimer) Reset(d time.Duration) bool {	217		func (t *manualTimer) Reset(d time.Duration) bool {
215	-	t.d = d	218	+	t.clock.mu.Lock()
216	-	return true	219	+	wasActive := t.active
217	-	}	220	+	t.deadline = t.clock.now.Add(d)
218	-	func (t *manualTimer) Stop() bool { return true }	221	+	t.active = true
219	-	func (t *manualTimer) fire(now time.Time) {	222	+	t.clock.timers[t] = struct{}{}
			223	+	t.clock.mu.Unlock()
			224	+	return wasActive
			225	+	}
			226	+	func (t *manualTimer) Stop() bool {
			227	+	t.clock.mu.Lock()
			228	+	defer t.clock.mu.Unlock()
			229	+	wasActive := t.active
			230	+	t.active = false
			231	+	delete(t.clock.timers, t)
			232	+	return wasActive
			233	+	}
			234	+	func (t *manualTimer) fireIfDue(now time.Time) bool {
			235	+	t.clock.mu.Lock()
			236	+	if !t.active {
			237	+	delete(t.clock.timers, t)
			238	+	t.clock.mu.Unlock()
			239	+	return false
			240	+	}
			241	+	if now.Before(t.deadline) {
			242	+	t.clock.mu.Unlock()
			243	+	return false
			244	+	}
			245	+	t.active = false
			246	+	delete(t.clock.timers, t)
			247	+	t.clock.mu.Unlock()
220		select {	248		select {
221		case t.c <- now:	249		case t.c <- now:
222		default:	250		default:
223		}	251		}
			252	+	return true
224		}	253		}
225			254
226		func waitForManualTimers(t testing.T, clk manualClock, n int) {	255		func waitForManualTimers(t testing.T, clk manualClock, n int) {
227		t.Helper()	256		t.Helper()
228		deadline := time.Now().Add(time.Second)	257		deadline := time.Now().Add(time.Second)
229		for time.Now().Before(deadline) {	258		for time.Now().Before(deadline) {
230	-	if len(clk.timers) >= n {	259	+	clk.mu.Lock()
			260	+	active := len(clk.timers)
			261	+	clk.mu.Unlock()
			262	+	if active >= n {
231		return	263		return
232		}	264		}
233		time.Sleep(time.Millisecond)	265		time.Sleep(time.Millisecond)
234		}	266		}
235	-	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)	267	+	clk.mu.Lock()
			268	+	active := len(clk.timers)
			269	+	clk.mu.Unlock()
			270	+	t.Fatalf("manual timers=%d, want at least %d", active, n)
236		}	271		}
237			272
238		type deadlineCapturePC struct {	273		type deadlineCapturePC struct {

@@ -264,6 +299,264 @@ func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
264		return nil	299		return nil
265		}	300		}
266			301
			302	+	func TestOutputFirstSendIsPacedAndQueuedOutputsAreCoalesced(t *testing.T) {
			303	+	aPC, bPC := newPair()
			304	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			305	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			306	+	if err != nil {
			307	+	t.Fatal(err)
			308	+	}
			309	+	defer func() { _ = a.Close() }()
			310	+
			311	+	for i := range 3 {
			312	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
			313	+	t.Fatal(err)
			314	+	}
			315	+	}
			316	+	if got := len(bPC.in); got != 1 {
			317	+	t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
			318	+	}
			319	+	waitForManualTimers(t, clk, 3)
			320	+	clk.advance(defaultOutputPaceMinDelay)
			321	+	eventuallyPacketCount(t, bPC, 2)
			322	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {
			323	+	t.Fatal(err)
			324	+	}
			325	+	if got := len(bPC.in); got != 2 {
			326	+	t.Fatalf("datagrams before second pace tick=%d, want 2", got)
			327	+	}
			328	+	clk.advance(defaultOutputPaceMinDelay)
			329	+	eventuallyPacketCount(t, bPC, 3)
			330	+	}
			331	+
			332	+	func TestResendSkipsQueuedOutputBeforeFirstWireAttempt(t *testing.T) {
			333	+	tr, writes := newResendTestTransport(t, 10)
			334	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}}
			335	+
			336	+	tr.resendPending()
			337	+
			338	+	if got := writes.Load(); got != 0 {
			339	+	t.Fatalf("writes=%d, want queued output excluded from resend", got)
			340	+	}
			341	+	if !tr.pending[1].first.IsZero() \|\| !tr.pending[1].last.IsZero() {
			342	+	t.Fatal("resend scan marked queued output as sent")
			343	+	}
			344	+	}
			345	+
			346	+	func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
			347	+	tr, writes := newResendTestTransport(t, 10)
			348	+	now := tr.clock.Now()
			349	+	tr.pending[1] = &pending{
			350	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("sending")},
			351	+	first: now.Add(-time.Second),
			352	+	last: now.Add(-time.Second),
			353	+	initialInFlight: true,
			354	+	}
			355	+
			356	+	tr.resendPending()
			357	+	if got := writes.Load(); got != 0 {
			358	+	t.Fatalf("writes=%d, want first wire attempt excluded from resend", got)
			359	+	}
			360	+	tr.pending[1].initialInFlight = false
			361	+	tr.resendPending()
			362	+	if got := writes.Load(); got != 1 {
			363	+	t.Fatalf("writes=%d, want resend after first wire attempt completed", got)
			364	+	}
			365	+	}
			366	+
			367	+	func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
			368	+	tr, _ := newResendTestTransport(t, 10)
			369	+	tests := []struct {
			370	+	name string
			371	+	srtt time.Duration
			372	+	want time.Duration
			373	+	}{
			374	+	{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
			375	+	{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
			376	+	{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
			377	+	{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
			378	+	}
			379	+	for _, tt := range tests {
			380	+	t.Run(tt.name, func(t *testing.T) {
			381	+	tr.mu.Lock()
			382	+	tr.srtt = tt.srtt
			383	+	got := tr.outputPaceDelayLocked()
			384	+	tr.mu.Unlock()
			385	+	if got != tt.want {
			386	+	t.Fatalf("output pace delay=%v, want %v", got, tt.want)
			387	+	}
			388	+	})
			389	+	}
			390	+	}
			391	+
			392	+	func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
			393	+	aPC, bPC := newPair()
			394	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			395	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			396	+	if err != nil {
			397	+	t.Fatal(err)
			398	+	}
			399	+	defer func() { _ = a.Close() }()
			400	+
			401	+	for _, payload := range []string{"first", "queued"} {
			402	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
			403	+	t.Fatal(err)
			404	+	}
			405	+	}
			406	+	waitForManualTimers(t, clk, 3)
			407	+	a.writeMu.Lock()
			408	+	clk.advance(defaultOutputPaceMinDelay)
			409	+	deadline := time.Now().Add(time.Second)
			410	+	for {
			411	+	a.mu.Lock()
			412	+	dequeued := len(a.outputQueue) == 0
			413	+	a.mu.Unlock()
			414	+	if dequeued {
			415	+	break
			416	+	}
			417	+	if time.Now().After(deadline) {
			418	+	a.writeMu.Unlock()
			419	+	t.Fatal("pacer did not dequeue output batch")
			420	+	}
			421	+	time.Sleep(time.Millisecond)
			422	+	}
			423	+
			424	+	done := make(chan error, 1)
			425	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
			426	+	select {
			427	+	case err := <-done:
			428	+	a.writeMu.Unlock()
			429	+	t.Fatalf("input completed before dequeued output batch: %v", err)
			430	+	case <-time.After(20 * time.Millisecond):
			431	+	}
			432	+	a.writeMu.Unlock()
			433	+	select {
			434	+	case err := <-done:
			435	+	if err != nil {
			436	+	t.Fatal(err)
			437	+	}
			438	+	case <-time.After(time.Second):
			439	+	t.Fatal("input did not complete after output batch was released")
			440	+	}
			441	+	if got := len(bPC.in); got != 3 {
			442	+	t.Fatalf("datagrams when input completed=%d, want output batch before input", got)
			443	+	}
			444	+	}
			445	+
			446	+	func TestInputFlushFailureRemovesUnsentQueuedPendingFrames(t *testing.T) {
			447	+	aPC, bPC := newPair()
			448	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			449	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			450	+	if err != nil {
			451	+	t.Fatal(err)
			452	+	}
			453	+	defer func() { _ = a.Close() }()
			454	+
			455	+	for i := range 3 {
			456	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
			457	+	t.Fatal(err)
			458	+	}
			459	+	}
			460	+	if got := len(bPC.in); got != 1 {
			461	+	t.Fatalf("immediate output datagrams=%d, want 1", got)
			462	+	}
			463	+	if err := bPC.Close(); err != nil {
			464	+	t.Fatal(err)
			465	+	}
			466	+
			467	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}); err == nil {
			468	+	t.Fatal("input send after peer close succeeded, want write error")
			469	+	}
			470	+
			471	+	a.mu.Lock()
			472	+	defer a.mu.Unlock()
			473	+	for _, seq := range []uint64{1, 2, 3, 4} {
			474	+	if _, ok := a.pending[seq]; ok {
			475	+	t.Fatalf("pending[%d] stranded after queued flush failure", seq)
			476	+	}
			477	+	}
			478	+	}
			479	+
			480	+	func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
			481	+	aPC, bPC := newPair()
			482	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			483	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			484	+	if err != nil {
			485	+	t.Fatal(err)
			486	+	}
			487	+	defer func() { _ = a.Close() }()
			488	+
			489	+	for i := range 2 {
			490	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
			491	+	t.Fatal(err)
			492	+	}
			493	+	}
			494	+	waitForManualTimers(t, clk, 3)
			495	+	if err := bPC.Close(); err != nil {
			496	+	t.Fatal(err)
			497	+	}
			498	+	clk.advance(defaultOutputPaceMinDelay)
			499	+
			500	+	select {
			501	+	case <-a.done:
			502	+	case <-time.After(time.Second):
			503	+	t.Fatal("transport remained open after paced output write failed")
			504	+	}
			505	+	if err := a.Send(ports.Frame{Type: ports.MsgInput}); err == nil \|\| err.Error() != "peer closed" {
			506	+	t.Fatalf("send after paced write failure err=%v, want peer closed", err)
			507	+	}
			508	+	}
			509	+
			510	+	func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
			511	+	aPC, bPC := newPair()
			512	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			513	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			514	+	if err != nil {
			515	+	t.Fatal(err)
			516	+	}
			517	+	defer func() { _ = a.Close() }()
			518	+
			519	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("output")}); err != nil {
			520	+	t.Fatal(err)
			521	+	}
			522	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
			523	+	t.Fatal(err)
			524	+	}
			525	+	if got := len(bPC.in); got != 1 {
			526	+	t.Fatalf("immediate output datagrams=%d, want 1", got)
			527	+	}
			528	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}); err != nil {
			529	+	t.Fatal(err)
			530	+	}
			531	+	if got := len(bPC.in); got != 3 {
			532	+	t.Fatalf("datagrams after input=%d, want queued output flushed plus input immediately", got)
			533	+	}
			534	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
			535	+	t.Fatal(err)
			536	+	}
			537	+	if got := len(bPC.in); got != 3 {
			538	+	t.Fatalf("datagrams before control=%d, want queued output still paced", got)
			539	+	}
			540	+	if err := a.Send(ports.Frame{Type: ports.MsgPing}); err != nil {
			541	+	t.Fatal(err)
			542	+	}
			543	+	if got := len(bPC.in); got != 5 {
			544	+	t.Fatalf("datagrams after control=%d, want queued output flushed plus control sent immediately", got)
			545	+	}
			546	+	}
			547	+
			548	+	func eventuallyPacketCount(t testing.T, pc fakePC, want int) {
			549	+	t.Helper()
			550	+	deadline := time.Now().Add(time.Second)
			551	+	for time.Now().Before(deadline) {
			552	+	if len(pc.in) >= want {
			553	+	return
			554	+	}
			555	+	time.Sleep(time.Millisecond)
			556	+	}
			557	+	t.Fatalf("datagrams=%d, want at least %d", len(pc.in), want)
			558	+	}
			559	+
267		func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {	560		func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
268		pc := newBlockingWritePC()	561		pc := newBlockingWritePC()
269		tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{	562		tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{