f1350fe fix(dgram): resolve UDP retry review findings · vev

M internal/adapters/dgram/proxy.go +32 -9


@@ -38,6 +38,13 @@ type proxyCopyErr struct {
38	38		err error
39	39		}
40	40
	41	+	type proxyRetryEvent uint8
	42	+
	43	+	const (
	44	+	proxyRetryStart proxyRetryEvent = iota + 1
	45	+	proxyRetryDone
	46	+	)
	47	+
41	48		func (p ProxyRuntime) Run(ctx context.Context) error {
42	49		defer func() {
43	50		_ = p.Client.Close()

@@ -58,12 +65,12 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
58	65		}
59	66
60	67		errCh := make(chan proxyCopyErr, 2)
61		-	retryCh := make(chan struct{}, 1)
	68	+	retryEvents := make(chan proxyRetryEvent, 2)
62	69		startClientCopy := func() {
63		-	go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryCh, retryBackoff, clk, false)
	70	+	go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryEvents, retryBackoff, clk, false)
64	71		}
65	72		startDaemonCopy := func() {
66		-	go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryCh, retryBackoff, clk, true)
	73	+	go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryEvents, retryBackoff, clk, true)
67	74		}
68	75		startClientCopy()
69	76		startDaemonCopy()

@@ -122,8 +129,13 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
122	129		}
123	130		case <-idle:
124	131		return ErrLinkDead
125		-	case <-retryCh:
126		-	startIdle()
	132	+	case ev := <-retryEvents:
	133	+	switch ev {
	134	+	case proxyRetryStart:
	135	+	startIdle()
	136	+	case proxyRetryDone:
	137	+	stopIdle()
	138	+	}
127	139		case ce := <-errCh:
128	140		switch ce.kind {
129	141		case proxyDaemonRecv:

@@ -162,7 +174,7 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
162	174		}
163	175		}
164	176
165		-	func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryCh chan<- struct{}, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
	177	+	func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryEvents chan<- proxyRetryEvent, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
166	178		for {
167	179		f, err := src.Recv()
168	180		if err != nil {

@@ -175,9 +187,8 @@ func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, se
175	187		if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
176	188		if !retrying {
177	189		retrying = true
178		-	select {
179		-	case retryCh <- struct{}{}:
180		-	default:
	190	+	if !sendProxyRetryEvent(ctx, retryEvents, proxyRetryStart) {
	191	+	return
181	192		}
182	193		}
183	194		t := clk.NewTimer(retryBackoff)

@@ -193,11 +204,23 @@ func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, se
193	204		errCh <- proxyCopyErr{kind: sendKind, err: err}
194	205		return
195	206		}
	207	+	if retrying && !sendProxyRetryEvent(ctx, retryEvents, proxyRetryDone) {
	208	+	return
	209	+	}
196	210		break
197	211		}
198	212		}
199	213		}
200	214
	215	+	func sendProxyRetryEvent(ctx context.Context, retryEvents chan<- proxyRetryEvent, ev proxyRetryEvent) bool {
	216	+	select {
	217	+	case retryEvents <- ev:
	218	+	return true
	219	+	case <-ctx.Done():
	220	+	return false
	221	+	}
	222	+	}
	223	+
201	224		func recoverableClientLink(t ports.Transport, err error) bool {
202	225		if errors.Is(err, ErrLinkDead) {
203	226		return false

 	err  error
 }
+type proxyRetryEvent uint8
+const (
+	proxyRetryStart proxyRetryEvent = iota + 1
+	proxyRetryDone
+)
 func (p ProxyRuntime) Run(ctx context.Context) error {
 	defer func() {
 		_ = p.Client.Close()
 	}
 	errCh := make(chan proxyCopyErr, 2)
-	retryCh := make(chan struct{}, 1)
+	retryEvents := make(chan proxyRetryEvent, 2)
 	startClientCopy := func() {
-		go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryCh, retryBackoff, clk, false)
+		go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryEvents, retryBackoff, clk, false)
 	}
 	startDaemonCopy := func() {
-		go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryCh, retryBackoff, clk, true)
+		go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryEvents, retryBackoff, clk, true)
 	}
 	startClientCopy()
 	startDaemonCopy()
 			}
 		case <-idle:
 			return ErrLinkDead
-		case <-retryCh:
+		case ev := <-retryEvents:
-			startIdle()
+			switch ev {
+			case proxyRetryStart:
+				startIdle()
+			case proxyRetryDone:
+				stopIdle()
+			}
 		case ce := <-errCh:
 			switch ce.kind {
 			case proxyDaemonRecv:
 	}
 }
-func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryCh chan<- struct{}, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
+func copyProxyFrames(ctx context.Context, dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr, retryEvents chan<- proxyRetryEvent, retryBackoff time.Duration, clk ports.Clock, retryRecoverable bool) {
 	for {
 		f, err := src.Recv()
 		if err != nil {
 				if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
 					if !retrying {
 						retrying = true
-						select {
+						if !sendProxyRetryEvent(ctx, retryEvents, proxyRetryStart) {
-						case retryCh <- struct{}{}:
+							return
-						default:
 						}
 					}
 					t := clk.NewTimer(retryBackoff)
 				errCh <- proxyCopyErr{kind: sendKind, err: err}
 				return
 			}
+			if retrying && !sendProxyRetryEvent(ctx, retryEvents, proxyRetryDone) {
+				return
+			}
 			break
 		}
 	}
 }
+func sendProxyRetryEvent(ctx context.Context, retryEvents chan<- proxyRetryEvent, ev proxyRetryEvent) bool {
+	select {
+	case retryEvents <- ev:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
 func recoverableClientLink(t ports.Transport, err error) bool {
 	if errors.Is(err, ErrLinkDead) {
 		return false

M internal/adapters/dgram/proxy_test.go +34 -0


@@ -214,6 +214,40 @@ func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
214	214		}
215	215		}
216	216
	217	+	func TestProxyRuntimeErrPendingFullRetrySuccessStopsIdleTTL(t *testing.T) {
	218	+	client := newFakeTransport()
	219	+	daemon := newFakeTransport()
	220	+	client.linkState = ports.LinkStateConnected
	221	+	errCh := make(chan error, 1)
	222	+	go func() {
	223	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
	224	+	}()
	225	+	client.sendErr <- ErrPendingFull
	226	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("recovers")}}
	227	+	select {
	228	+	case got := <-client.sent:
	229	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "recovers" {
	230	+	t.Fatalf("client got %+v, want retried output", got)
	231	+	}
	232	+	case <-time.After(time.Second):
	233	+	t.Fatal("timeout waiting for retried output")
	234	+	}
	235	+	select {
	236	+	case err := <-errCh:
	237	+	t.Fatalf("Run returned after successful retry before daemon EOF: %v", err)
	238	+	case <-time.After(50 * time.Millisecond):
	239	+	}
	240	+	daemon.recv <- recvResult{err: io.EOF}
	241	+	select {
	242	+	case err := <-errCh:
	243	+	if err != nil {
	244	+	t.Fatalf("Run err=%v, want nil", err)
	245	+	}
	246	+	case <-time.After(time.Second):
	247	+	t.Fatal("timeout waiting for daemon EOF")
	248	+	}
	249	+	}
	250	+
217	251		func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
218	252		client := newFakeTransport()
219	253		daemon := newFakeTransport()


@@ -214,6 +214,40 @@ func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
214	}	214		}
215	}	215		}
216		216
		217	+	func TestProxyRuntimeErrPendingFullRetrySuccessStopsIdleTTL(t *testing.T) {
		218	+	client := newFakeTransport()
		219	+	daemon := newFakeTransport()
		220	+	client.linkState = ports.LinkStateConnected
		221	+	errCh := make(chan error, 1)
		222	+	go func() {
		223	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
		224	+	}()
		225	+	client.sendErr <- ErrPendingFull
		226	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("recovers")}}
		227	+	select {
		228	+	case got := <-client.sent:
		229	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "recovers" {
		230	+	t.Fatalf("client got %+v, want retried output", got)
		231	+	}
		232	+	case <-time.After(time.Second):
		233	+	t.Fatal("timeout waiting for retried output")
		234	+	}
		235	+	select {
		236	+	case err := <-errCh:
		237	+	t.Fatalf("Run returned after successful retry before daemon EOF: %v", err)
		238	+	case <-time.After(50 * time.Millisecond):
		239	+	}
		240	+	daemon.recv <- recvResult{err: io.EOF}
		241	+	select {
		242	+	case err := <-errCh:
		243	+	if err != nil {
		244	+	t.Fatalf("Run err=%v, want nil", err)
		245	+	}
		246	+	case <-time.After(time.Second):
		247	+	t.Fatal("timeout waiting for daemon EOF")
		248	+	}
		249	+	}
		250	+
217	func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {	251		func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
218	client := newFakeTransport()	252		client := newFakeTransport()
219	daemon := newFakeTransport()	253		daemon := newFakeTransport()

M internal/adapters/dgram/transport.go +23 -10


@@ -192,7 +192,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
192	192		}
193	193		opts = normalizeOptions(opts)
194	194		t := &Transport{pc: pc, codec: c, sendDir: sendDir, recvDir: recvDir, mtu: opts.MTU, peer: peer, pending: make(map[uint64]*pending), replay: pdgram.NewReplayWindow(), reasm: pdgram.NewReassembler(), in: make(chan ports.Frame, 32), done: make(chan struct{}), lastHeard: opts.Clock.Now(), heartbeat: opts.Heartbeat, resendAfter: opts.ResendAfter, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxPendingWait: opts.MaxPendingWait, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
195		-	t.sendWake = make(chan struct{}, 1)
	195	+	t.sendWake = make(chan struct{})
196	196		t.probeReply = make(chan uint64, probeReplyBufferSize)
197	197		t.deliverCond = sync.NewCond(&t.deliverMu)
198	198		go t.readLoop(pc)

@@ -221,11 +221,12 @@ func (t *Transport) Send(f ports.Frame) error {
221	221		t.mu.Unlock()
222	222		return ErrPendingFull
223	223		}
	224	+	wake := t.sendWake
224	225		timer := t.clock.NewTimer(remaining)
225	226		t.mu.Unlock()
226	227		select {
227	228		case <-timer.C():
228		-	case <-t.sendWake:
	229	+	case <-wake:
229	230		case <-t.done:
230	231		}
231	232		timer.Stop()

@@ -245,7 +246,18 @@ func (t *Transport) Send(f ports.Frame) error {
245	246		t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
246	247		}
247	248		t.mu.Unlock()
248		-	return t.sendData(seq, reliable, f)
	249	+	if err := t.sendData(seq, reliable, f); err != nil {
	250	+	if reliable {
	251	+	t.mu.Lock()
	252	+	if _, ok := t.pending[seq]; ok {
	253	+	delete(t.pending, seq)
	254	+	t.notifySendWaitersLocked()
	255	+	}
	256	+	t.mu.Unlock()
	257	+	}
	258	+	return err
	259	+	}
	260	+	return nil
249	261		}
250	262
251	263		func (t *Transport) Recv() (ports.Frame, error) {

@@ -455,7 +467,7 @@ func (t *Transport) handleRecord(p []byte) {
455	467		t.mu.Lock()
456	468		if _, ok := t.pending[seq]; ok {
457	469		delete(t.pending, seq)
458		-	t.notifySendWaiter()
	470	+	t.notifySendWaitersLocked()
459	471		}
460	472		t.mu.Unlock()
461	473		case recProbe:

@@ -562,11 +574,12 @@ func (t *Transport) probeReplyLoop() {
562	574		}
563	575		}
564	576
565		-	func (t *Transport) notifySendWaiter() {
566		-	select {
567		-	case t.sendWake <- struct{}{}:
568		-	default:
	577	+	func (t *Transport) notifySendWaitersLocked() {
	578	+	if t.sendWake == nil {
	579	+	return
569	580		}
	581	+	close(t.sendWake)
	582	+	t.sendWake = make(chan struct{})
570	583		}
571	584
572	585		func (t *Transport) resendLoop() {

@@ -710,7 +723,7 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
710	723		}
711	724		}
712	725		}
713		-	t.notifySendWaiter()
	726	+	t.notifySendWaitersLocked()
714	727		t.mu.Unlock()
715	728		event := ports.LinkEvent{State: state, At: now, Err: err}
716	729		select {

@@ -729,7 +742,7 @@ func (t *Transport) closeWithError(err error) {
729	742		t.closeErr = err
730	743		t.probeWait = make(map[uint64]chan struct{})
731	744		close(t.done)
732		-	t.notifySendWaiter()
	745	+	t.notifySendWaitersLocked()
733	746		t.mu.Unlock()
734	747		t.deliverMu.Lock()
735	748		t.deliverCond.Broadcast()

 	}
 	opts = normalizeOptions(opts)
 	t := &Transport{pc: pc, codec: c, sendDir: sendDir, recvDir: recvDir, mtu: opts.MTU, peer: peer, pending: make(map[uint64]*pending), replay: pdgram.NewReplayWindow(), reasm: pdgram.NewReassembler(), in: make(chan ports.Frame, 32), done: make(chan struct{}), lastHeard: opts.Clock.Now(), heartbeat: opts.Heartbeat, resendAfter: opts.ResendAfter, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxPendingWait: opts.MaxPendingWait, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
-	t.sendWake = make(chan struct{}, 1)
+	t.sendWake = make(chan struct{})
 	t.probeReply = make(chan uint64, probeReplyBufferSize)
 	t.deliverCond = sync.NewCond(&t.deliverMu)
 	go t.readLoop(pc)
 			t.mu.Unlock()
 			return ErrPendingFull
 		}
+		wake := t.sendWake
 		timer := t.clock.NewTimer(remaining)
 		t.mu.Unlock()
 		select {
 		case <-timer.C():
-		case <-t.sendWake:
+		case <-wake:
 		case <-t.done:
 		}
 		timer.Stop()
 		t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
 	}
 	t.mu.Unlock()
-	return t.sendData(seq, reliable, f)
+	if err := t.sendData(seq, reliable, f); err != nil {
+		if reliable {
+			t.mu.Lock()
+			if _, ok := t.pending[seq]; ok {
+				delete(t.pending, seq)
+				t.notifySendWaitersLocked()
+			}
+			t.mu.Unlock()
+		}
+		return err
+	}
+	return nil
 }
 func (t *Transport) Recv() (ports.Frame, error) {
 		t.mu.Lock()
 		if _, ok := t.pending[seq]; ok {
 			delete(t.pending, seq)
-			t.notifySendWaiter()
+			t.notifySendWaitersLocked()
 		}
 		t.mu.Unlock()
 	case recProbe:
 	}
 }
-func (t *Transport) notifySendWaiter() {
+func (t *Transport) notifySendWaitersLocked() {
-	select {
+	if t.sendWake == nil {
-	case t.sendWake <- struct{}{}:
+		return
-	default:
 	}
+	close(t.sendWake)
+	t.sendWake = make(chan struct{})
 }
 func (t *Transport) resendLoop() {
 			}
 		}
 	}
-	t.notifySendWaiter()
+	t.notifySendWaitersLocked()
 	t.mu.Unlock()
 	event := ports.LinkEvent{State: state, At: now, Err: err}
 	select {
 	t.closeErr = err
 	t.probeWait = make(map[uint64]chan struct{})
 	close(t.done)
-	t.notifySendWaiter()
+	t.notifySendWaitersLocked()
 	t.mu.Unlock()
 	t.deliverMu.Lock()
 	t.deliverCond.Broadcast()

M internal/adapters/dgram/transport_test.go +22 -16


@@ -278,6 +278,12 @@ func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
278	278		if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
279	279		t.Fatalf("Send err=%v, want timeout error", err)
280	280		}
	281	+	tr.mu.Lock()
	282	+	pending := len(tr.pending)
	283	+	tr.mu.Unlock()
	284	+	if pending != 0 {
	285	+	t.Fatalf("pending after failed initial send=%d, want 0", pending)
	286	+	}
281	287		case <-time.After(250 * time.Millisecond):
282	288		t.Fatal("Send blocked without honoring WriteTimeout")
283	289		}

@@ -615,8 +621,9 @@ func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *at
615	621
616	622		func TestResendPendingCapsBurst(t *testing.T) {
617	623		tr, writes := newResendTestTransport(t, 2)
	624	+	now := tr.clock.Now()
618	625		for i := range 5 {
619		-	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: time.Now().Add(-time.Second)}
	626	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: now.Add(-time.Second)}
620	627		}
621	628
622	629		tr.resendPending()

@@ -627,7 +634,7 @@ func TestResendPendingCapsBurst(t *testing.T) {
627	634
628	635		func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
629	636		tr, writes := newResendTestTransport(t, 10)
630		-	now := time.Now()
	637	+	now := tr.clock.Now()
631	638		tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}
632	639		tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}
633	640		tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}

@@ -818,7 +825,8 @@ func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
818	825		func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
819	826		aPC, bPC := newPair()
820	827		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
821		-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
	828	+	clk := newManualClock(time.Now())
	829	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: time.Hour, Clock: clk})
822	830		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
823	831		defer func() { _ = a.Close() }()
824	832		defer func() { _ = b.Close() }()

@@ -826,22 +834,20 @@ func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
826	834		if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
827	835		t.Fatal(err)
828	836		}
829		-	done := make(chan error, 1)
	837	+	done := make(chan error, 2)
830	838		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
831		-
832		-	select {
833		-	case err := <-done:
834		-	t.Fatalf("send returned before link degraded: %v", err)
835		-	case <-time.After(25 * time.Millisecond):
836		-	}
	839	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("third")}) }()
	840	+	waitForManualTimers(t, clk, 2)
837	841		a.setLinkState(ports.LinkStateDegraded, nil)
838		-	select {
839		-	case err := <-done:
840		-	if !errors.Is(err, ErrPendingFull) {
841		-	t.Fatalf("send err=%v, want ErrPendingFull", err)
	842	+	for i := range 2 {
	843	+	select {
	844	+	case err := <-done:
	845	+	if !errors.Is(err, ErrPendingFull) {
	846	+	t.Fatalf("send %d err=%v, want ErrPendingFull", i, err)
	847	+	}
	848	+	case <-time.After(time.Second):
	849	+	t.Fatalf("send %d did not unblock after link degraded", i)
842	850		}
843		-	case <-time.After(time.Second):
844		-	t.Fatal("send did not unblock after link degraded")
845	851		}
846	852		}
847	853


@@ -278,6 +278,12 @@ func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
278		if !errors.As(err, &timeout) \|\| !timeout.Timeout() {	278		if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
279		t.Fatalf("Send err=%v, want timeout error", err)	279		t.Fatalf("Send err=%v, want timeout error", err)
280		}	280		}
			281	+	tr.mu.Lock()
			282	+	pending := len(tr.pending)
			283	+	tr.mu.Unlock()
			284	+	if pending != 0 {
			285	+	t.Fatalf("pending after failed initial send=%d, want 0", pending)
			286	+	}
281		case <-time.After(250 * time.Millisecond):	287		case <-time.After(250 * time.Millisecond):
282		t.Fatal("Send blocked without honoring WriteTimeout")	288		t.Fatal("Send blocked without honoring WriteTimeout")
283		}	289		}

@@ -615,8 +621,9 @@ func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *at
615			621
616		func TestResendPendingCapsBurst(t *testing.T) {	622		func TestResendPendingCapsBurst(t *testing.T) {
617		tr, writes := newResendTestTransport(t, 2)	623		tr, writes := newResendTestTransport(t, 2)
			624	+	now := tr.clock.Now()
618		for i := range 5 {	625		for i := range 5 {
619	-	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: time.Now().Add(-time.Second)}	626	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: now.Add(-time.Second)}
620		}	627		}
621			628
622		tr.resendPending()	629		tr.resendPending()

@@ -627,7 +634,7 @@ func TestResendPendingCapsBurst(t *testing.T) {
627			634
628		func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {	635		func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
629		tr, writes := newResendTestTransport(t, 10)	636		tr, writes := newResendTestTransport(t, 10)
630	-	now := time.Now()	637	+	now := tr.clock.Now()
631		tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}	638		tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}
632		tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}	639		tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}
633		tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}	640		tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}

@@ -818,7 +825,8 @@ func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
818		func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {	825		func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
819		aPC, bPC := newPair()	826		aPC, bPC := newPair()
820		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	827		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
821	-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})	828	+	clk := newManualClock(time.Now())
			829	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: time.Hour, Clock: clk})
822		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)	830		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
823		defer func() { _ = a.Close() }()	831		defer func() { _ = a.Close() }()
824		defer func() { _ = b.Close() }()	832		defer func() { _ = b.Close() }()

@@ -826,22 +834,20 @@ func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
826		if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {	834		if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
827		t.Fatal(err)	835		t.Fatal(err)
828		}	836		}
829	-	done := make(chan error, 1)	837	+	done := make(chan error, 2)
830		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()	838		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
831	-		839	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("third")}) }()
832	-	select {	840	+	waitForManualTimers(t, clk, 2)
833	-	case err := <-done:
834	-	t.Fatalf("send returned before link degraded: %v", err)
835	-	case <-time.After(25 * time.Millisecond):
836	-	}
837		a.setLinkState(ports.LinkStateDegraded, nil)	841		a.setLinkState(ports.LinkStateDegraded, nil)
838	-	select {	842	+	for i := range 2 {
839	-	case err := <-done:	843	+	select {
840	-	if !errors.Is(err, ErrPendingFull) {	844	+	case err := <-done:
841	-	t.Fatalf("send err=%v, want ErrPendingFull", err)	845	+	if !errors.Is(err, ErrPendingFull) {
			846	+	t.Fatalf("send %d err=%v, want ErrPendingFull", i, err)
			847	+	}
			848	+	case <-time.After(time.Second):
			849	+	t.Fatalf("send %d did not unblock after link degraded", i)
842		}	850		}
843	-	case <-time.After(time.Second):
844	-	t.Fatal("send did not unblock after link degraded")
845		}	851		}
846		}	852		}
847			853