fix(dgram): preserve paced output delivery

commit: 0377fb513b65cbf463e65058bf74148ba01c2107
author: brice · 3mo ago · 2026-07-09 18:12 UTC
parent: f180475
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +143 -44

M internal/adapters/dgram/transport.go +13 -20


@@ -300,15 +300,15 @@ func (t *Transport) Send(f ports.Frame) error {
300	300		t.pending[seq] = &pending{frame: f, first: now, last: now}
301	301		}
302	302		if shouldPaceOutput(f) {
303		-	sendNow := t.outputNext.IsZero() \|\| !t.clock.Now().Before(t.outputNext)
304		-	if sendNow {
305		-	t.outputNext = t.clock.Now().Add(t.outputPaceDelayLocked())
306		-	}
307		-	t.mu.Unlock()
308		-	if !sendNow {
309		-	t.enqueueOutputSend(queuedSend{seq: seq, reliable: reliable, frame: f})
	303	+	now := t.clock.Now()
	304	+	if len(t.outputQueue) > 0 \|\| (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
	305	+	t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f})
	306	+	t.notifyOutputPacerLocked()
	307	+	t.mu.Unlock()
310	308		return nil
311	309		}
	310	+	t.outputNext = now.Add(t.outputPaceDelayLocked())
	311	+	t.mu.Unlock()
312	312		if err := t.sendData(seq, reliable, f); err != nil {
313	313		t.removePending(seq, reliable)
314	314		return err

@@ -331,17 +331,6 @@ func (t *Transport) Send(f ports.Frame) error {
331	331		return nil
332	332		}
333	333
334		-	func (t *Transport) enqueueOutputSend(q queuedSend) {
335		-	t.mu.Lock()
336		-	if t.closed {
337		-	t.mu.Unlock()
338		-	return
339		-	}
340		-	t.outputQueue = append(t.outputQueue, q)
341		-	t.notifyOutputPacerLocked()
342		-	t.mu.Unlock()
343		-	}
344		-
345	334		func (t *Transport) flushQueuedOutputLocked() []queuedSend {
346	335		if len(t.outputQueue) == 0 {
347	336		return nil

@@ -390,8 +379,12 @@ func (t *Transport) outputPaceLoop() {
390	379		t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
391	380		t.outputNext = now.Add(t.outputPaceDelayLocked())
392	381		t.mu.Unlock()
393		-	for _, q := range batch {
394		-	_ = t.sendQueuedData(q)
	382	+	for i, q := range batch {
	383	+	if err := t.sendQueuedData(q); err != nil {
	384	+	t.removeQueuedPending(batch[i+1:])
	385	+	t.closeWithError(err)
	386	+	return
	387	+	}
395	388		}
396	389		}
397	390		}

 		t.pending[seq] = &pending{frame: f, first: now, last: now}
 	}
 	if shouldPaceOutput(f) {
-		sendNow := t.outputNext.IsZero() || !t.clock.Now().Before(t.outputNext)
+		now := t.clock.Now()
-		if sendNow {
+		if len(t.outputQueue) > 0 || (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
-			t.outputNext = t.clock.Now().Add(t.outputPaceDelayLocked())
+			t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f})
-		}
+			t.notifyOutputPacerLocked()
-		t.mu.Unlock()
+			t.mu.Unlock()
-		if !sendNow {
-			t.enqueueOutputSend(queuedSend{seq: seq, reliable: reliable, frame: f})
 			return nil
 		}
+		t.outputNext = now.Add(t.outputPaceDelayLocked())
+		t.mu.Unlock()
 		if err := t.sendData(seq, reliable, f); err != nil {
 			t.removePending(seq, reliable)
 			return err
 	return nil
 }
-func (t *Transport) enqueueOutputSend(q queuedSend) {
-	t.mu.Lock()
-	if t.closed {
-		t.mu.Unlock()
-		return
-	}
-	t.outputQueue = append(t.outputQueue, q)
-	t.notifyOutputPacerLocked()
-	t.mu.Unlock()
-}
 func (t *Transport) flushQueuedOutputLocked() []queuedSend {
 	if len(t.outputQueue) == 0 {
 		return nil
 		t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
 		t.outputNext = now.Add(t.outputPaceDelayLocked())
 		t.mu.Unlock()
-		for _, q := range batch {
+		for i, q := range batch {
-			_ = t.sendQueuedData(q)
+			if err := t.sendQueuedData(q); err != nil {
+				t.removeQueuedPending(batch[i+1:])
+				t.closeWithError(err)
+				return
+			}
 		}
 	}
 }

M internal/adapters/dgram/transport_test.go +130 -24


@@ -172,6 +172,7 @@ type manualClock struct {
172	172		mu sync.Mutex
173	173		now time.Time
174	174		timers chan *manualTimer
	175	+	active int
175	176		}
176	177
177	178		func newManualClock(now time.Time) *manualClock {

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

@@ -201,7 +203,7 @@ func (c *manualClock) advance(d time.Duration) {
201	203		n := len(c.timers)
202	204		for range n {
203	205		tm := <-c.timers
204		-	if tm.fireIfDue(now) {
	206	+	if fired, keep := tm.fireIfDue(now); fired \|\| !keep {
205	207		continue
206	208		}
207	209		c.timers <- tm

@@ -218,39 +220,64 @@ type manualTimer struct {
218	220		func (t *manualTimer) C() <-chan time.Time { return t.c }
219	221		func (t *manualTimer) Reset(d time.Duration) bool {
220	222		t.clock.mu.Lock()
	223	+	wasActive := t.active
221	224		t.deadline = t.clock.now.Add(d)
222		-	t.clock.mu.Unlock()
223	225		t.active = true
224		-	t.clock.timers <- t
225		-	return true
	226	+	if !wasActive {
	227	+	t.clock.active++
	228	+	}
	229	+	t.clock.mu.Unlock()
	230	+	if !wasActive {
	231	+	t.clock.timers <- t
	232	+	}
	233	+	return wasActive
226	234		}
227	235		func (t *manualTimer) Stop() bool {
	236	+	t.clock.mu.Lock()
	237	+	defer t.clock.mu.Unlock()
228	238		wasActive := t.active
229	239		t.active = false
	240	+	if wasActive {
	241	+	t.clock.active--
	242	+	}
230	243		return wasActive
231	244		}
232		-	func (t *manualTimer) fireIfDue(now time.Time) bool {
233		-	if !t.active \|\| now.Before(t.deadline) {
234		-	return false
	245	+	func (t *manualTimer) fireIfDue(now time.Time) (fired bool, keep bool) {
	246	+	t.clock.mu.Lock()
	247	+	if !t.active {
	248	+	t.clock.mu.Unlock()
	249	+	return false, false
	250	+	}
	251	+	if now.Before(t.deadline) {
	252	+	t.clock.mu.Unlock()
	253	+	return false, true
235	254		}
236	255		t.active = false
	256	+	t.clock.active--
	257	+	t.clock.mu.Unlock()
237	258		select {
238	259		case t.c <- now:
239	260		default:
240	261		}
241		-	return true
	262	+	return true, false
242	263		}
243	264
244	265		func waitForManualTimers(t testing.T, clk manualClock, n int) {
245	266		t.Helper()
246	267		deadline := time.Now().Add(time.Second)
247	268		for time.Now().Before(deadline) {
248		-	if len(clk.timers) >= n {
	269	+	clk.mu.Lock()
	270	+	active := clk.active
	271	+	clk.mu.Unlock()
	272	+	if active >= n {
249	273		return
250	274		}
251	275		time.Sleep(time.Millisecond)
252	276		}
253		-	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
	277	+	clk.mu.Lock()
	278	+	active := clk.active
	279	+	clk.mu.Unlock()
	280	+	t.Fatalf("manual timers=%d, want at least %d", active, n)
254	281		}
255	282
256	283		type deadlineCapturePC struct {

@@ -315,21 +342,64 @@ func TestOutputFirstSendsArePacedAfterInitialDatagram(t *testing.T) {
315	342
316	343		func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
317	344		tr, _ := newResendTestTransport(t, 10)
318		-	tr.srtt = time.Second
319		-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMaxDelay {
320		-	t.Fatalf("high-rtt output pace delay=%v, want max clamp %v", got, defaultOutputPaceMaxDelay)
	345	+	tests := []struct {
	346	+	name string
	347	+	srtt time.Duration
	348	+	want time.Duration
	349	+	}{
	350	+	{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
	351	+	{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
	352	+	{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
	353	+	{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
	354	+	}
	355	+	for _, tt := range tests {
	356	+	t.Run(tt.name, func(t *testing.T) {
	357	+	tr.mu.Lock()
	358	+	tr.srtt = tt.srtt
	359	+	got := tr.outputPaceDelayLocked()
	360	+	tr.mu.Unlock()
	361	+	if got != tt.want {
	362	+	t.Fatalf("output pace delay=%v, want %v", got, tt.want)
	363	+	}
	364	+	})
	365	+	}
	366	+	}
	367	+
	368	+	func TestOutputDoesNotOvertakeQueuedOutputWhenPaceWindowElapsed(t *testing.T) {
	369	+	aPC, bPC := newPair()
	370	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	371	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	372	+	if err != nil {
	373	+	t.Fatal(err)
	374	+	}
	375	+	defer func() { _ = a.Close() }()
	376	+
	377	+	for _, payload := range []string{"first", "second"} {
	378	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
	379	+	t.Fatal(err)
	380	+	}
	381	+	}
	382	+	waitForManualTimers(t, clk, 3)
	383	+	a.mu.Lock()
	384	+	a.outputNext = clk.Now().Add(-time.Millisecond)
	385	+	// Leave the pacer waiting on its captured wake channel so Send's ordering
	386	+	// decision can be inspected before the queue is drained.
	387	+	a.outputWake = make(chan struct{})
	388	+	a.mu.Unlock()
	389	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")}); err != nil {
	390	+	t.Fatal(err)
321	391		}
322		-	tr.srtt = 30 * time.Millisecond
323		-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMinCadence {
324		-	t.Fatalf("low-rtt output pace delay=%v, want min cadence %v", got, defaultOutputPaceMinCadence)
	392	+
	393	+	if got := len(bPC.in); got != 1 {
	394	+	t.Fatalf("immediate datagrams=%d, want only first output sent", got)
325	395		}
326		-	tr.srtt = 80 * time.Millisecond
327		-	if got := tr.outputPaceDelayLocked(); got != 40*time.Millisecond {
328		-	t.Fatalf("mid-rtt output pace delay=%v, want srtt/2", got)
	396	+	a.mu.Lock()
	397	+	defer a.mu.Unlock()
	398	+	if got := string(a.outputQueue[0].frame.Payload); got != "second" {
	399	+	t.Fatalf("first queued output=%q, want second", got)
329	400		}
330		-	tr.srtt = 0
331		-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMinDelay {
332		-	t.Fatalf("initial output pace delay=%v, want min delay %v", got, defaultOutputPaceMinDelay)
	401	+	if got := string(a.outputQueue[1].frame.Payload); got != "third" {
	402	+	t.Fatalf("second queued output=%q, want third", got)
333	403		}
334	404		}
335	405

@@ -370,6 +440,36 @@ func TestInputFlushFailureRemovesUnsentQueuedPendingFrames(t *testing.T) {
370	440		}
371	441		}
372	442
	443	+	func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
	444	+	aPC, bPC := newPair()
	445	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	446	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
	447	+	if err != nil {
	448	+	t.Fatal(err)
	449	+	}
	450	+	defer func() { _ = a.Close() }()
	451	+
	452	+	for i := range 2 {
	453	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
	454	+	t.Fatal(err)
	455	+	}
	456	+	}
	457	+	waitForManualTimers(t, clk, 3)
	458	+	if err := bPC.Close(); err != nil {
	459	+	t.Fatal(err)
	460	+	}
	461	+	clk.advance(defaultOutputPaceMinDelay)
	462	+
	463	+	select {
	464	+	case <-a.done:
	465	+	case <-time.After(time.Second):
	466	+	t.Fatal("transport remained open after paced output write failed")
	467	+	}
	468	+	if err := a.Send(ports.Frame{Type: ports.MsgInput}); err == nil \|\| err.Error() != "peer closed" {
	469	+	t.Fatalf("send after paced write failure err=%v, want peer closed", err)
	470	+	}
	471	+	}
	472	+
373	473		func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
374	474		aPC, bPC := newPair()
375	475		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))

@@ -394,11 +494,17 @@ func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
394	494		if got := len(bPC.in); got != 3 {
395	495		t.Fatalf("datagrams after input=%d, want queued output flushed plus input immediately", got)
396	496		}
	497	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
	498	+	t.Fatal(err)
	499	+	}
	500	+	if got := len(bPC.in); got != 3 {
	501	+	t.Fatalf("datagrams before control=%d, want queued output still paced", got)
	502	+	}
397	503		if err := a.Send(ports.Frame{Type: ports.MsgPing}); err != nil {
398	504		t.Fatal(err)
399	505		}
400		-	if got := len(bPC.in); got != 4 {
401		-	t.Fatalf("datagrams after control=%d, want control sent immediately", got)
	506	+	if got := len(bPC.in); got != 5 {
	507	+	t.Fatalf("datagrams after control=%d, want queued output flushed plus control sent immediately", got)
402	508		}
403	509		}
404	510


@@ -172,6 +172,7 @@ 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 chan *manualTimer
			175	+	active int
175		}	176		}
176			177
177		func newManualClock(now time.Time) *manualClock {	178		func newManualClock(now time.Time) *manualClock {

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

@@ -201,7 +203,7 @@ func (c *manualClock) advance(d time.Duration) {
201		n := len(c.timers)	203		n := len(c.timers)
202		for range n {	204		for range n {
203		tm := <-c.timers	205		tm := <-c.timers
204	-	if tm.fireIfDue(now) {	206	+	if fired, keep := tm.fireIfDue(now); fired \|\| !keep {
205		continue	207		continue
206		}	208		}
207		c.timers <- tm	209		c.timers <- tm

@@ -218,39 +220,64 @@ type manualTimer struct {
218		func (t *manualTimer) C() <-chan time.Time { return t.c }	220		func (t *manualTimer) C() <-chan time.Time { return t.c }
219		func (t *manualTimer) Reset(d time.Duration) bool {	221		func (t *manualTimer) Reset(d time.Duration) bool {
220		t.clock.mu.Lock()	222		t.clock.mu.Lock()
			223	+	wasActive := t.active
221		t.deadline = t.clock.now.Add(d)	224		t.deadline = t.clock.now.Add(d)
222	-	t.clock.mu.Unlock()
223		t.active = true	225		t.active = true
224	-	t.clock.timers <- t	226	+	if !wasActive {
225	-	return true	227	+	t.clock.active++
			228	+	}
			229	+	t.clock.mu.Unlock()
			230	+	if !wasActive {
			231	+	t.clock.timers <- t
			232	+	}
			233	+	return wasActive
226		}	234		}
227		func (t *manualTimer) Stop() bool {	235		func (t *manualTimer) Stop() bool {
			236	+	t.clock.mu.Lock()
			237	+	defer t.clock.mu.Unlock()
228		wasActive := t.active	238		wasActive := t.active
229		t.active = false	239		t.active = false
			240	+	if wasActive {
			241	+	t.clock.active--
			242	+	}
230		return wasActive	243		return wasActive
231		}	244		}
232	-	func (t *manualTimer) fireIfDue(now time.Time) bool {	245	+	func (t *manualTimer) fireIfDue(now time.Time) (fired bool, keep bool) {
233	-	if !t.active \|\| now.Before(t.deadline) {	246	+	t.clock.mu.Lock()
234	-	return false	247	+	if !t.active {
			248	+	t.clock.mu.Unlock()
			249	+	return false, false
			250	+	}
			251	+	if now.Before(t.deadline) {
			252	+	t.clock.mu.Unlock()
			253	+	return false, true
235		}	254		}
236		t.active = false	255		t.active = false
			256	+	t.clock.active--
			257	+	t.clock.mu.Unlock()
237		select {	258		select {
238		case t.c <- now:	259		case t.c <- now:
239		default:	260		default:
240		}	261		}
241	-	return true	262	+	return true, false
242		}	263		}
243			264
244		func waitForManualTimers(t testing.T, clk manualClock, n int) {	265		func waitForManualTimers(t testing.T, clk manualClock, n int) {
245		t.Helper()	266		t.Helper()
246		deadline := time.Now().Add(time.Second)	267		deadline := time.Now().Add(time.Second)
247		for time.Now().Before(deadline) {	268		for time.Now().Before(deadline) {
248	-	if len(clk.timers) >= n {	269	+	clk.mu.Lock()
			270	+	active := clk.active
			271	+	clk.mu.Unlock()
			272	+	if active >= n {
249		return	273		return
250		}	274		}
251		time.Sleep(time.Millisecond)	275		time.Sleep(time.Millisecond)
252		}	276		}
253	-	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)	277	+	clk.mu.Lock()
			278	+	active := clk.active
			279	+	clk.mu.Unlock()
			280	+	t.Fatalf("manual timers=%d, want at least %d", active, n)
254		}	281		}
255			282
256		type deadlineCapturePC struct {	283		type deadlineCapturePC struct {

@@ -315,21 +342,64 @@ func TestOutputFirstSendsArePacedAfterInitialDatagram(t *testing.T) {
315			342
316		func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {	343		func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
317		tr, _ := newResendTestTransport(t, 10)	344		tr, _ := newResendTestTransport(t, 10)
318	-	tr.srtt = time.Second	345	+	tests := []struct {
319	-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMaxDelay {	346	+	name string
320	-	t.Fatalf("high-rtt output pace delay=%v, want max clamp %v", got, defaultOutputPaceMaxDelay)	347	+	srtt time.Duration
			348	+	want time.Duration
			349	+	}{
			350	+	{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
			351	+	{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
			352	+	{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
			353	+	{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
			354	+	}
			355	+	for _, tt := range tests {
			356	+	t.Run(tt.name, func(t *testing.T) {
			357	+	tr.mu.Lock()
			358	+	tr.srtt = tt.srtt
			359	+	got := tr.outputPaceDelayLocked()
			360	+	tr.mu.Unlock()
			361	+	if got != tt.want {
			362	+	t.Fatalf("output pace delay=%v, want %v", got, tt.want)
			363	+	}
			364	+	})
			365	+	}
			366	+	}
			367	+
			368	+	func TestOutputDoesNotOvertakeQueuedOutputWhenPaceWindowElapsed(t *testing.T) {
			369	+	aPC, bPC := newPair()
			370	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			371	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			372	+	if err != nil {
			373	+	t.Fatal(err)
			374	+	}
			375	+	defer func() { _ = a.Close() }()
			376	+
			377	+	for _, payload := range []string{"first", "second"} {
			378	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
			379	+	t.Fatal(err)
			380	+	}
			381	+	}
			382	+	waitForManualTimers(t, clk, 3)
			383	+	a.mu.Lock()
			384	+	a.outputNext = clk.Now().Add(-time.Millisecond)
			385	+	// Leave the pacer waiting on its captured wake channel so Send's ordering
			386	+	// decision can be inspected before the queue is drained.
			387	+	a.outputWake = make(chan struct{})
			388	+	a.mu.Unlock()
			389	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")}); err != nil {
			390	+	t.Fatal(err)
321		}	391		}
322	-	tr.srtt = 30 * time.Millisecond	392	+
323	-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMinCadence {	393	+	if got := len(bPC.in); got != 1 {
324	-	t.Fatalf("low-rtt output pace delay=%v, want min cadence %v", got, defaultOutputPaceMinCadence)	394	+	t.Fatalf("immediate datagrams=%d, want only first output sent", got)
325		}	395		}
326	-	tr.srtt = 80 * time.Millisecond	396	+	a.mu.Lock()
327	-	if got := tr.outputPaceDelayLocked(); got != 40*time.Millisecond {	397	+	defer a.mu.Unlock()
328	-	t.Fatalf("mid-rtt output pace delay=%v, want srtt/2", got)	398	+	if got := string(a.outputQueue[0].frame.Payload); got != "second" {
			399	+	t.Fatalf("first queued output=%q, want second", got)
329		}	400		}
330	-	tr.srtt = 0	401	+	if got := string(a.outputQueue[1].frame.Payload); got != "third" {
331	-	if got := tr.outputPaceDelayLocked(); got != defaultOutputPaceMinDelay {	402	+	t.Fatalf("second queued output=%q, want third", got)
332	-	t.Fatalf("initial output pace delay=%v, want min delay %v", got, defaultOutputPaceMinDelay)
333		}	403		}
334		}	404		}
335			405

@@ -370,6 +440,36 @@ func TestInputFlushFailureRemovesUnsentQueuedPendingFrames(t *testing.T) {
370		}	440		}
371		}	441		}
372			442
			443	+	func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
			444	+	aPC, bPC := newPair()
			445	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			446	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
			447	+	if err != nil {
			448	+	t.Fatal(err)
			449	+	}
			450	+	defer func() { _ = a.Close() }()
			451	+
			452	+	for i := range 2 {
			453	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
			454	+	t.Fatal(err)
			455	+	}
			456	+	}
			457	+	waitForManualTimers(t, clk, 3)
			458	+	if err := bPC.Close(); err != nil {
			459	+	t.Fatal(err)
			460	+	}
			461	+	clk.advance(defaultOutputPaceMinDelay)
			462	+
			463	+	select {
			464	+	case <-a.done:
			465	+	case <-time.After(time.Second):
			466	+	t.Fatal("transport remained open after paced output write failed")
			467	+	}
			468	+	if err := a.Send(ports.Frame{Type: ports.MsgInput}); err == nil \|\| err.Error() != "peer closed" {
			469	+	t.Fatalf("send after paced write failure err=%v, want peer closed", err)
			470	+	}
			471	+	}
			472	+
373		func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {	473		func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
374		aPC, bPC := newPair()	474		aPC, bPC := newPair()
375		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	475		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))

@@ -394,11 +494,17 @@ func TestInputAndControlSendsBypassOutputPacing(t *testing.T) {
394		if got := len(bPC.in); got != 3 {	494		if got := len(bPC.in); got != 3 {
395		t.Fatalf("datagrams after input=%d, want queued output flushed plus input immediately", got)	495		t.Fatalf("datagrams after input=%d, want queued output flushed plus input immediately", got)
396		}	496		}
			497	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
			498	+	t.Fatal(err)
			499	+	}
			500	+	if got := len(bPC.in); got != 3 {
			501	+	t.Fatalf("datagrams before control=%d, want queued output still paced", got)
			502	+	}
397		if err := a.Send(ports.Frame{Type: ports.MsgPing}); err != nil {	503		if err := a.Send(ports.Frame{Type: ports.MsgPing}); err != nil {
398		t.Fatal(err)	504		t.Fatal(err)
399		}	505		}
400	-	if got := len(bPC.in); got != 4 {	506	+	if got := len(bPC.in); got != 5 {
401	-	t.Fatalf("datagrams after control=%d, want control sent immediately", got)	507	+	t.Fatalf("datagrams after control=%d, want queued output flushed plus control sent immediately", got)
402		}	508		}
403		}	509		}
404			510