fix(dgram): pace UDP remote retransmits

commit: f08136d0f08fecf016e5717f0af553f9bcafc362
author: brice · 3mo ago · 2026-07-08 07:30 UTC
committer: brice · 3mo ago · 2026-07-08 07:48 UTC
parent: e2fc930
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +487 -99

M internal/adapters/dgram/proxy.go +31 -6


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

@@ -121,6 +122,8 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
121	122		}
122	123		case <-idle:
123	124		return ErrLinkDead
	125	+	case <-retryCh:
	126	+	startIdle()
124	127		case ce := <-errCh:
125	128		switch ce.kind {
126	129		case proxyDaemonRecv:

@@ -159,16 +162,38 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
159	162		}
160	163		}
161	164
162		-	func copyProxyFrames(dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr) {
	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) {
163	166		for {
164	167		f, err := src.Recv()
165	168		if err != nil {
166	169		errCh <- proxyCopyErr{kind: recvKind, err: err}
167	170		return
168	171		}
169		-	if err := dst.Send(f); err != nil {
170		-	errCh <- proxyCopyErr{kind: sendKind, err: err}
171		-	return
	172	+	retrying := false
	173	+	for {
	174	+	if err := dst.Send(f); err != nil {
	175	+	if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
	176	+	if !retrying {
	177	+	retrying = true
	178	+	select {
	179	+	case retryCh <- struct{}{}:
	180	+	default:
	181	+	}
	182	+	}
	183	+	t := clk.NewTimer(retryBackoff)
	184	+	select {
	185	+	case <-t.C():
	186	+	t.Stop()
	187	+	continue
	188	+	case <-ctx.Done():
	189	+	t.Stop()
	190	+	return
	191	+	}
	192	+	}
	193	+	errCh <- proxyCopyErr{kind: sendKind, err: err}
	194	+	return
	195	+	}
	196	+	break
172	197		}
173	198		}
174	199		}

 	}
 	errCh := make(chan proxyCopyErr, 2)
+	retryCh := make(chan struct{}, 1)
 	startClientCopy := func() {
-		go copyProxyFrames(p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh)
+		go copyProxyFrames(ctx, p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh, retryCh, retryBackoff, clk, false)
 	}
 	startDaemonCopy := func() {
-		go copyProxyFrames(p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh)
+		go copyProxyFrames(ctx, p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh, retryCh, retryBackoff, clk, true)
 	}
 	startClientCopy()
 	startDaemonCopy()
 			}
 		case <-idle:
 			return ErrLinkDead
+		case <-retryCh:
+			startIdle()
 		case ce := <-errCh:
 			switch ce.kind {
 			case proxyDaemonRecv:
 	}
 }
-func copyProxyFrames(dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr) {
+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) {
 	for {
 		f, err := src.Recv()
 		if err != nil {
 			errCh <- proxyCopyErr{kind: recvKind, err: err}
 			return
 		}
-		if err := dst.Send(f); err != nil {
+		retrying := false
-			errCh <- proxyCopyErr{kind: sendKind, err: err}
+		for {
-			return
+			if err := dst.Send(f); err != nil {
+				if retryRecoverable && errors.Is(err, ErrPendingFull) && recoverableClientLink(dst, err) {
+					if !retrying {
+						retrying = true
+						select {
+						case retryCh <- struct{}{}:
+						default:
+						}
+					}
+					t := clk.NewTimer(retryBackoff)
+					select {
+					case <-t.C():
+						t.Stop()
+						continue
+					case <-ctx.Done():
+						t.Stop()
+						return
+					}
+				}
+				errCh <- proxyCopyErr{kind: sendKind, err: err}
+				return
+			}
+			break
 		}
 	}
 }

M internal/adapters/dgram/proxy_test.go +75 -13


@@ -11,13 +11,15 @@ import (
11	11		)
12	12
13	13		type fakeTransport struct {
14		-	recv chan recvResult
15		-	sent chan ports.Frame
16		-	sendErr chan error
17		-	linkState ports.LinkState
18		-	linkEvents chan ports.LinkEvent
19		-	closed chan struct{}
20		-	closeOnce sync.Once
	14	+	recv chan recvResult
	15	+	sent chan ports.Frame
	16	+	sendErr chan error
	17	+	sendErrDefault error
	18	+	sideEffectOnErr bool
	19	+	linkState ports.LinkState
	20	+	linkEvents chan ports.LinkEvent
	21	+	closed chan struct{}
	22	+	closeOnce sync.Once
21	23		}
22	24
23	25		type recvResult struct {

@@ -31,8 +33,14 @@ func newFakeTransport() *fakeTransport {
31	33		func (f *fakeTransport) Send(fr ports.Frame) error {
32	34		select {
33	35		case err := <-f.sendErr:
	36	+	if f.sideEffectOnErr {
	37	+	f.sent <- fr
	38	+	}
34	39		return err
35	40		default:
	41	+	if f.sendErrDefault != nil {
	42	+	return f.sendErrDefault
	43	+	}
36	44		}
37	45		f.sent <- fr
38	46		return nil

@@ -107,7 +115,7 @@ func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T)
107	115		}
108	116		}
109	117
110		-	func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
	118	+	func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
111	119		client := newFakeTransport()
112	120		daemon := newFakeTransport()
113	121		client.linkState = ports.LinkStateOffline

@@ -116,13 +124,12 @@ func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
116	124		go func() {
117	125		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
118	126		}()
119		-	client.sendErr <- errors.New("temporary send failure")
120		-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("lost during outage")}}
121		-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("after recovery")}}
	127	+	client.sendErr <- ErrPendingFull
	128	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
122	129		select {
123	130		case got := <-client.sent:
124		-	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "after recovery" {
125		-	t.Fatalf("client got %+v, want recovered output", got)
	131	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "preserved during outage" {
	132	+	t.Fatalf("client got %+v, want retried output", got)
126	133		}
127	134		case <-time.After(time.Second):
128	135		t.Fatal("timeout waiting for recovered daemon output")

@@ -138,6 +145,61 @@ func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
138	145		}
139	146		}
140	147
	148	+	func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
	149	+	client := newFakeTransport()
	150	+	daemon := newFakeTransport()
	151	+	client.linkState = ports.LinkStateOffline
	152	+	client.sideEffectOnErr = true
	153	+	client.sendErr <- errors.New("write timed out after queuing")
	154	+	errCh := make(chan error, 1)
	155	+	go func() {
	156	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
	157	+	}()
	158	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("maybe queued")}}
	159	+	select {
	160	+	case got := <-client.sent:
	161	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "maybe queued" {
	162	+	t.Fatalf("client got %+v, want side-effected output", got)
	163	+	}
	164	+	case <-time.After(time.Second):
	165	+	t.Fatal("timeout waiting for side-effected output")
	166	+	}
	167	+	select {
	168	+	case got := <-client.sent:
	169	+	t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
	170	+	case <-time.After(20 * time.Millisecond):
	171	+	}
	172	+	daemon.recv <- recvResult{err: io.EOF}
	173	+	select {
	174	+	case err := <-errCh:
	175	+	if err != nil {
	176	+	t.Fatalf("Run err=%v, want nil", err)
	177	+	}
	178	+	case <-time.After(time.Second):
	179	+	t.Fatal("timeout waiting for daemon EOF")
	180	+	}
	181	+	}
	182	+
	183	+	func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
	184	+	client := newFakeTransport()
	185	+	daemon := newFakeTransport()
	186	+	client.linkState = ports.LinkStateOffline
	187	+	client.sendErrDefault = ErrPendingFull
	188	+	errCh := make(chan error, 1)
	189	+	go func() {
	190	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
	191	+	}()
	192	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
	193	+	select {
	194	+	case err := <-errCh:
	195	+	if !errors.Is(err, ErrLinkDead) {
	196	+	t.Fatalf("Run err=%v, want ErrLinkDead", err)
	197	+	}
	198	+	case <-time.After(time.Second):
	199	+	t.Fatal("timeout waiting for idle expiry from ErrPendingFull retry")
	200	+	}
	201	+	}
	202	+
141	203		func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
142	204		client := newFakeTransport()
143	205		daemon := newFakeTransport()


@@ -11,13 +11,15 @@ import (
11		)	11		)
12			12
13		type fakeTransport struct {	13		type fakeTransport struct {
14	-	recv chan recvResult	14	+	recv chan recvResult
15	-	sent chan ports.Frame	15	+	sent chan ports.Frame
16	-	sendErr chan error	16	+	sendErr chan error
17	-	linkState ports.LinkState	17	+	sendErrDefault error
18	-	linkEvents chan ports.LinkEvent	18	+	sideEffectOnErr bool
19	-	closed chan struct{}	19	+	linkState ports.LinkState
20	-	closeOnce sync.Once	20	+	linkEvents chan ports.LinkEvent
			21	+	closed chan struct{}
			22	+	closeOnce sync.Once
21		}	23		}
22			24
23		type recvResult struct {	25		type recvResult struct {

@@ -31,8 +33,14 @@ func newFakeTransport() *fakeTransport {
31		func (f *fakeTransport) Send(fr ports.Frame) error {	33		func (f *fakeTransport) Send(fr ports.Frame) error {
32		select {	34		select {
33		case err := <-f.sendErr:	35		case err := <-f.sendErr:
			36	+	if f.sideEffectOnErr {
			37	+	f.sent <- fr
			38	+	}
34		return err	39		return err
35		default:	40		default:
			41	+	if f.sendErrDefault != nil {
			42	+	return f.sendErrDefault
			43	+	}
36		}	44		}
37		f.sent <- fr	45		f.sent <- fr
38		return nil	46		return nil

@@ -107,7 +115,7 @@ func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T)
107		}	115		}
108		}	116		}
109			117
110	-	func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {	118	+	func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
111		client := newFakeTransport()	119		client := newFakeTransport()
112		daemon := newFakeTransport()	120		daemon := newFakeTransport()
113		client.linkState = ports.LinkStateOffline	121		client.linkState = ports.LinkStateOffline

@@ -116,13 +124,12 @@ func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
116		go func() {	124		go func() {
117		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())	125		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
118		}()	126		}()
119	-	client.sendErr <- errors.New("temporary send failure")	127	+	client.sendErr <- ErrPendingFull
120	-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("lost during outage")}}	128	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
121	-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("after recovery")}}
122		select {	129		select {
123		case got := <-client.sent:	130		case got := <-client.sent:
124	-	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "after recovery" {	131	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "preserved during outage" {
125	-	t.Fatalf("client got %+v, want recovered output", got)	132	+	t.Fatalf("client got %+v, want retried output", got)
126		}	133		}
127		case <-time.After(time.Second):	134		case <-time.After(time.Second):
128		t.Fatal("timeout waiting for recovered daemon output")	135		t.Fatal("timeout waiting for recovered daemon output")

@@ -138,6 +145,61 @@ func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
138		}	145		}
139		}	146		}
140			147
			148	+	func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
			149	+	client := newFakeTransport()
			150	+	daemon := newFakeTransport()
			151	+	client.linkState = ports.LinkStateOffline
			152	+	client.sideEffectOnErr = true
			153	+	client.sendErr <- errors.New("write timed out after queuing")
			154	+	errCh := make(chan error, 1)
			155	+	go func() {
			156	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
			157	+	}()
			158	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("maybe queued")}}
			159	+	select {
			160	+	case got := <-client.sent:
			161	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "maybe queued" {
			162	+	t.Fatalf("client got %+v, want side-effected output", got)
			163	+	}
			164	+	case <-time.After(time.Second):
			165	+	t.Fatal("timeout waiting for side-effected output")
			166	+	}
			167	+	select {
			168	+	case got := <-client.sent:
			169	+	t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
			170	+	case <-time.After(20 * time.Millisecond):
			171	+	}
			172	+	daemon.recv <- recvResult{err: io.EOF}
			173	+	select {
			174	+	case err := <-errCh:
			175	+	if err != nil {
			176	+	t.Fatalf("Run err=%v, want nil", err)
			177	+	}
			178	+	case <-time.After(time.Second):
			179	+	t.Fatal("timeout waiting for daemon EOF")
			180	+	}
			181	+	}
			182	+
			183	+	func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
			184	+	client := newFakeTransport()
			185	+	daemon := newFakeTransport()
			186	+	client.linkState = ports.LinkStateOffline
			187	+	client.sendErrDefault = ErrPendingFull
			188	+	errCh := make(chan error, 1)
			189	+	go func() {
			190	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
			191	+	}()
			192	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
			193	+	select {
			194	+	case err := <-errCh:
			195	+	if !errors.Is(err, ErrLinkDead) {
			196	+	t.Fatalf("Run err=%v, want ErrLinkDead", err)
			197	+	}
			198	+	case <-time.After(time.Second):
			199	+	t.Fatal("timeout waiting for idle expiry from ErrPendingFull retry")
			200	+	}
			201	+	}
			202	+
141		func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {	203		func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
142		client := newFakeTransport()	204		client := newFakeTransport()
143		daemon := newFakeTransport()	205		daemon := newFakeTransport()

M internal/adapters/dgram/transport.go +146 -65


@@ -6,6 +6,7 @@ import (
6	6		"encoding/binary"
7	7		"errors"
8	8		"net"
	9	+	"sort"
9	10		"sync"
10	11		"time"
11	12

@@ -19,18 +20,20 @@ const (
19	20		recProbe byte = 3
20	21		recPong byte = 4
21	22
22		-	defaultMTU = pdgram.DefaultMTU
23		-	defaultResend = 40 * time.Millisecond
24		-	defaultHeartbeat = 3 * time.Second
25		-	defaultDegraded = 10 * time.Second
26		-	defaultProbe = 20 * time.Second
27		-	defaultOffline = 30 * time.Second
28		-	defaultDead = 60 * time.Second
29		-	defaultMaxPending = 1024
30		-	defaultMaxRecvBuf = 1024
31		-	maxPendingReliable = defaultMaxPending
32		-	maxRecvBuffer = defaultMaxRecvBuf
33		-	linkEventBufferSize = 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	+	defaultMaxRecvBuf = 1024
	35	+	maxRecvBuffer = defaultMaxRecvBuf
	36	+	linkEventBufferSize = 16
34	37		)
35	38
36	39		var (

@@ -39,16 +42,19 @@ var (
39	42		)
40	43
41	44		type Options struct {
42		-	MTU int
43		-	ResendAfter time.Duration
44		-	Heartbeat time.Duration
45		-	DegradedAfter time.Duration
46		-	ProbeAfter time.Duration
47		-	OfflineAfter time.Duration
48		-	DeadAfter time.Duration
49		-	MaxPending int
50		-	MaxRecvBuffer int
51		-	Clock ports.Clock
	45	+	MTU int
	46	+	ResendAfter time.Duration
	47	+	MaxResendAfter time.Duration
	48	+	MaxResendPerTick int
	49	+	WriteTimeout time.Duration
	50	+	Heartbeat time.Duration
	51	+	DegradedAfter time.Duration
	52	+	ProbeAfter time.Duration
	53	+	OfflineAfter time.Duration
	54	+	DeadAfter time.Duration
	55	+	MaxPending int
	56	+	MaxRecvBuffer int
	57	+	Clock ports.Clock
52	58		// RebindPacketConn is an optional client-side hook used to hop the local UDP
53	59		// socket when the peer is offline. Server/proxy transports should leave it nil.
54	60		RebindPacketConn func(net.PacketConn) (net.PacketConn, error)

@@ -74,6 +80,18 @@ func normalizeOptions(opts Options) Options {
74	80		if opts.ResendAfter <= 0 {
75	81		opts.ResendAfter = defaultResend
76	82		}
	83	+	if opts.MaxResendAfter <= 0 {
	84	+	opts.MaxResendAfter = defaultMaxResendAfter
	85	+	}
	86	+	if opts.MaxResendAfter < opts.ResendAfter {
	87	+	opts.MaxResendAfter = opts.ResendAfter
	88	+	}
	89	+	if opts.MaxResendPerTick <= 0 {
	90	+	opts.MaxResendPerTick = defaultMaxResendPerTick
	91	+	}
	92	+	if opts.WriteTimeout <= 0 {
	93	+	opts.WriteTimeout = defaultWriteTimeout
	94	+	}
77	95		if opts.Heartbeat <= 0 {
78	96		opts.Heartbeat = defaultHeartbeat
79	97		}

@@ -107,29 +125,33 @@ type Transport struct {
107	125		sendDir, recvDir uint32
108	126		mtu int
109	127
110		-	mu sync.Mutex
111		-	peer net.Addr
112		-	ctr uint64
113		-	seq uint64
114		-	probeSeq uint64
115		-	pending map[uint64]*pending
116		-	closed bool
117		-	closeErr error
118		-	lastHeard time.Time
119		-	heartbeat time.Duration
120		-	resendAfter time.Duration
121		-	degradedAfter time.Duration
122		-	probeAfter time.Duration
123		-	offlineAfter time.Duration
124		-	deadAfter time.Duration
125		-	maxPending int
126		-	maxRecvBuffer int
127		-	clock ports.Clock
128		-	linkState ports.LinkState
129		-	linkEvents chan ports.LinkEvent
130		-	probeWait map[uint64]chan struct{}
131		-	rebind func(net.PacketConn) (net.PacketConn, error)
132		-	hoppedOffline bool
	128	+	mu sync.Mutex
	129	+	peer net.Addr
	130	+	ctr uint64
	131	+	seq uint64
	132	+	probeSeq uint64
	133	+	pending map[uint64]*pending
	134	+	closed bool
	135	+	closeErr error
	136	+	lastHeard time.Time
	137	+	heartbeat time.Duration
	138	+	resendAfter time.Duration
	139	+	maxResendAfter time.Duration
	140	+	maxResendPerTick int
	141	+	writeTimeout time.Duration
	142	+	degradedAfter time.Duration
	143	+	probeAfter time.Duration
	144	+	offlineAfter time.Duration
	145	+	deadAfter time.Duration
	146	+	maxPending int
	147	+	maxRecvBuffer int
	148	+	clock ports.Clock
	149	+	linkState ports.LinkState
	150	+	linkEvents chan ports.LinkEvent
	151	+	probeWait map[uint64]chan struct{}
	152	+	sendCond *sync.Cond
	153	+	rebind func(net.PacketConn) (net.PacketConn, error)
	154	+	hoppedOffline bool
133	155
134	156		recvMu sync.Mutex
135	157		replay *pdgram.ReplayWindow

@@ -137,6 +159,8 @@ type Transport struct {
137	159		in chan ports.Frame
138	160		done chan struct{}
139	161
	162	+	writeMu sync.Mutex
	163	+
140	164		deliverMu sync.Mutex
141	165		deliverCond *sync.Cond
142	166		nextRecvSeq uint64

@@ -144,8 +168,9 @@ type Transport struct {
144	168		}
145	169
146	170		type pending struct {
147		-	frame ports.Frame
148		-	last time.Time
	171	+	frame ports.Frame
	172	+	last time.Time
	173	+	attempts int
149	174		}
150	175
151	176		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {

@@ -158,8 +183,9 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
158	183		return nil, err
159	184		}
160	185		opts = normalizeOptions(opts)
161		-	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, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, 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)}
	186	+	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, 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)}
162	187		t.deliverCond = sync.NewCond(&t.deliverMu)
	188	+	t.sendCond = sync.NewCond(&t.mu)
163	189		go t.readLoop(pc)
164	190		go t.resendLoop()
165	191		go t.deliveryLoop()

@@ -171,6 +197,13 @@ func (t *Transport) Send(f ports.Frame) error {
171	197		// dropping MsgOutput can permanently desynchronize the client screen.
172	198		reliable := true
173	199		t.mu.Lock()
	200	+	for reliable && len(t.pending) >= t.maxPending && !t.closed {
	201	+	if t.linkState != ports.LinkStateConnected {
	202	+	t.mu.Unlock()
	203	+	return ErrPendingFull
	204	+	}
	205	+	t.sendCond.Wait()
	206	+	}
174	207		if t.closed {
175	208		err := t.closeErr
176	209		t.mu.Unlock()

@@ -179,14 +212,10 @@ func (t *Transport) Send(f ports.Frame) error {
179	212		}
180	213		return errors.New("dgram: closed")
181	214		}
182		-	if reliable && len(t.pending) >= t.maxPending {
183		-	t.mu.Unlock()
184		-	return ErrPendingFull
185		-	}
186	215		t.seq++
187	216		seq := t.seq
188	217		if reliable {
189		-	t.pending[seq] = &pending{frame: f}
	218	+	t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
190	219		}
191	220		t.mu.Unlock()
192	221		return t.sendData(seq, reliable, f)

@@ -314,15 +343,32 @@ func (t *Transport) sendPayload(p []byte) error {
314	343		if peer == nil {
315	344		return errors.New("dgram: no peer")
316	345		}
317		-	t.mu.Lock()
318		-	pc := t.pc
319		-	t.mu.Unlock()
320		-	if _, err := pc.WriteTo(pkt, peer); err != nil {
	346	+	if err := t.writePacket(pkt, peer); err != nil {
321	347		return err
322	348		}
323	349		}
324	350		return nil
325	351		}
	352	+
	353	+	func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
	354	+	t.writeMu.Lock()
	355	+	defer t.writeMu.Unlock()
	356	+
	357	+	t.mu.Lock()
	358	+	pc := t.pc
	359	+	writeTimeout := t.writeTimeout
	360	+	t.mu.Unlock()
	361	+
	362	+	if writeTimeout > 0 {
	363	+	if err := pc.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
	364	+	return err
	365	+	}
	366	+	defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()
	367	+	}
	368	+	_, err := pc.WriteTo(pkt, peer)
	369	+	return err
	370	+	}
	371	+
326	372		func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
327	373
328	374		func (t *Transport) readLoop(pc net.PacketConn) {

@@ -380,7 +426,12 @@ func (t *Transport) handleRecord(p []byte) {
380	426		}
381	427		seq := binary.BigEndian.Uint64(p[1:])
382	428		t.mu.Lock()
383		-	delete(t.pending, seq)
	429	+	if _, ok := t.pending[seq]; ok {
	430	+	delete(t.pending, seq)
	431	+	if t.sendCond != nil {
	432	+	t.sendCond.Signal()
	433	+	}
	434	+	}
384	435		t.mu.Unlock()
385	436		case recProbe:
386	437		if len(p) != 9 {

@@ -500,14 +551,27 @@ func (t *Transport) resendPending() {
500	551		f ports.Frame
501	552		}
502	553		t.mu.Lock()
503		-	for seq, p := range t.pending {
504		-	if p.last.IsZero() \|\| now.Sub(p.last) >= t.resendAfter {
505		-	p.last = now
506		-	resend = append(resend, struct {
507		-	seq uint64
508		-	f ports.Frame
509		-	}{seq, p.frame})
	554	+	seqs := make([]uint64, 0, len(t.pending))
	555	+	for seq := range t.pending {
	556	+	seqs = append(seqs, seq)
	557	+	}
	558	+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
	559	+	limit := t.maxResendPerTick
	560	+	for _, seq := range seqs {
	561	+	if limit <= 0 {
	562	+	break
510	563		}
	564	+	p := t.pending[seq]
	565	+	if p == nil \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
	566	+	continue
	567	+	}
	568	+	p.last = now
	569	+	p.attempts++
	570	+	resend = append(resend, struct {
	571	+	seq uint64
	572	+	f ports.Frame
	573	+	}{seq, p.frame})
	574	+	limit--
511	575		}
512	576		t.mu.Unlock()
513	577		for _, r := range resend {

@@ -515,6 +579,17 @@ func (t *Transport) resendPending() {
515	579		}
516	580		}
517	581
	582	+	func (t Transport) resendDelayLocked(p pending) time.Duration {
	583	+	d := t.resendAfter
	584	+	for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
	585	+	d *= 2
	586	+	if d > t.maxResendAfter {
	587	+	return t.maxResendAfter
	588	+	}
	589	+	}
	590	+	return d
	591	+	}
	592	+
518	593		func (t *Transport) hopPacketConnOnce() {
519	594		t.mu.Lock()
520	595		if t.rebind == nil \|\| t.hoppedOffline \|\| t.closed {

@@ -582,6 +657,9 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
582	657		return
583	658		}
584	659		t.linkState = state
	660	+	if t.sendCond != nil {
	661	+	t.sendCond.Broadcast()
	662	+	}
585	663		t.mu.Unlock()
586	664		event := ports.LinkEvent{State: state, At: now, Err: err}
587	665		select {

@@ -600,6 +678,9 @@ func (t *Transport) closeWithError(err error) {
600	678		t.closeErr = err
601	679		t.probeWait = make(map[uint64]chan struct{})
602	680		close(t.done)
	681	+	if t.sendCond != nil {
	682	+	t.sendCond.Broadcast()
	683	+	}
603	684		t.mu.Unlock()
604	685		t.deliverMu.Lock()
605	686		t.deliverCond.Broadcast()

 	"encoding/binary"
 	"errors"
 	"net"
+	"sort"
 	"sync"
 	"time"
 	recProbe byte = 3
 	recPong  byte = 4
-	defaultMTU          = pdgram.DefaultMTU
+	defaultMTU              = pdgram.DefaultMTU
-	defaultResend       = 40 * time.Millisecond
+	defaultResend           = 250 * time.Millisecond
-	defaultHeartbeat    = 3 * time.Second
+	defaultMaxResendAfter   = 2 * time.Second
-	defaultDegraded     = 10 * time.Second
+	defaultMaxResendPerTick = 64
-	defaultProbe        = 20 * time.Second
+	defaultWriteTimeout     = 250 * time.Millisecond
-	defaultOffline      = 30 * time.Second
+	defaultHeartbeat        = 3 * time.Second
-	defaultDead         = 60 * time.Second
+	defaultDegraded         = 10 * time.Second
-	defaultMaxPending   = 1024
+	defaultProbe            = 20 * time.Second
-	defaultMaxRecvBuf   = 1024
+	defaultOffline          = 30 * time.Second
-	maxPendingReliable  = defaultMaxPending
+	defaultDead             = 60 * time.Second
-	maxRecvBuffer       = defaultMaxRecvBuf
+	defaultMaxPending       = 1024
-	linkEventBufferSize = 16
+	defaultMaxRecvBuf       = 1024
+	maxRecvBuffer           = defaultMaxRecvBuf
+	linkEventBufferSize     = 16
 )
 var (
 )
 type Options struct {
-	MTU           int
+	MTU              int
-	ResendAfter   time.Duration
+	ResendAfter      time.Duration
-	Heartbeat     time.Duration
+	MaxResendAfter   time.Duration
-	DegradedAfter time.Duration
+	MaxResendPerTick int
-	ProbeAfter    time.Duration
+	WriteTimeout     time.Duration
-	OfflineAfter  time.Duration
+	Heartbeat        time.Duration
-	DeadAfter     time.Duration
+	DegradedAfter    time.Duration
-	MaxPending    int
+	ProbeAfter       time.Duration
-	MaxRecvBuffer int
+	OfflineAfter     time.Duration
-	Clock         ports.Clock
+	DeadAfter        time.Duration
+	MaxPending       int
+	MaxRecvBuffer    int
+	Clock            ports.Clock
 	// RebindPacketConn is an optional client-side hook used to hop the local UDP
 	// socket when the peer is offline. Server/proxy transports should leave it nil.
 	RebindPacketConn func(net.PacketConn) (net.PacketConn, error)
 	if opts.ResendAfter <= 0 {
 		opts.ResendAfter = defaultResend
 	}
+	if opts.MaxResendAfter <= 0 {
+		opts.MaxResendAfter = defaultMaxResendAfter
+	}
+	if opts.MaxResendAfter < opts.ResendAfter {
+		opts.MaxResendAfter = opts.ResendAfter
+	}
+	if opts.MaxResendPerTick <= 0 {
+		opts.MaxResendPerTick = defaultMaxResendPerTick
+	}
+	if opts.WriteTimeout <= 0 {
+		opts.WriteTimeout = defaultWriteTimeout
+	}
 	if opts.Heartbeat <= 0 {
 		opts.Heartbeat = defaultHeartbeat
 	}
 	sendDir, recvDir uint32
 	mtu              int
-	mu            sync.Mutex
+	mu               sync.Mutex
-	peer          net.Addr
+	peer             net.Addr
-	ctr           uint64
+	ctr              uint64
-	seq           uint64
+	seq              uint64
-	probeSeq      uint64
+	probeSeq         uint64
-	pending       map[uint64]*pending
+	pending          map[uint64]*pending
-	closed        bool
+	closed           bool
-	closeErr      error
+	closeErr         error
-	lastHeard     time.Time
+	lastHeard        time.Time
-	heartbeat     time.Duration
+	heartbeat        time.Duration
-	resendAfter   time.Duration
+	resendAfter      time.Duration
-	degradedAfter time.Duration
+	maxResendAfter   time.Duration
-	probeAfter    time.Duration
+	maxResendPerTick int
-	offlineAfter  time.Duration
+	writeTimeout     time.Duration
-	deadAfter     time.Duration
+	degradedAfter    time.Duration
-	maxPending    int
+	probeAfter       time.Duration
-	maxRecvBuffer int
+	offlineAfter     time.Duration
-	clock         ports.Clock
+	deadAfter        time.Duration
-	linkState     ports.LinkState
+	maxPending       int
-	linkEvents    chan ports.LinkEvent
+	maxRecvBuffer    int
-	probeWait     map[uint64]chan struct{}
+	clock            ports.Clock
-	rebind        func(net.PacketConn) (net.PacketConn, error)
+	linkState        ports.LinkState
-	hoppedOffline bool
+	linkEvents       chan ports.LinkEvent
+	probeWait        map[uint64]chan struct{}
+	sendCond         *sync.Cond
+	rebind           func(net.PacketConn) (net.PacketConn, error)
+	hoppedOffline    bool
 	recvMu sync.Mutex
 	replay *pdgram.ReplayWindow
 	in     chan ports.Frame
 	done   chan struct{}
+	writeMu sync.Mutex
 	deliverMu   sync.Mutex
 	deliverCond *sync.Cond
 	nextRecvSeq uint64
 }
 type pending struct {
-	frame ports.Frame
+	frame    ports.Frame
-	last  time.Time
+	last     time.Time
+	attempts int
 }
 func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
 		return nil, err
 	}
 	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, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, 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 := &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, 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.deliverCond = sync.NewCond(&t.deliverMu)
+	t.sendCond = sync.NewCond(&t.mu)
 	go t.readLoop(pc)
 	go t.resendLoop()
 	go t.deliveryLoop()
 	// dropping MsgOutput can permanently desynchronize the client screen.
 	reliable := true
 	t.mu.Lock()
+	for reliable && len(t.pending) >= t.maxPending && !t.closed {
+		if t.linkState != ports.LinkStateConnected {
+			t.mu.Unlock()
+			return ErrPendingFull
+		}
+		t.sendCond.Wait()
+	}
 	if t.closed {
 		err := t.closeErr
 		t.mu.Unlock()
 		}
 		return errors.New("dgram: closed")
 	}
-	if reliable && len(t.pending) >= t.maxPending {
-		t.mu.Unlock()
-		return ErrPendingFull
-	}
 	t.seq++
 	seq := t.seq
 	if reliable {
-		t.pending[seq] = &pending{frame: f}
+		t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
 	}
 	t.mu.Unlock()
 	return t.sendData(seq, reliable, f)
 		if peer == nil {
 			return errors.New("dgram: no peer")
 		}
-		t.mu.Lock()
+		if err := t.writePacket(pkt, peer); err != nil {
-		pc := t.pc
-		t.mu.Unlock()
-		if _, err := pc.WriteTo(pkt, peer); err != nil {
 			return err
 		}
 	}
 	return nil
 }
+func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
+	t.writeMu.Lock()
+	defer t.writeMu.Unlock()
+	t.mu.Lock()
+	pc := t.pc
+	writeTimeout := t.writeTimeout
+	t.mu.Unlock()
+	if writeTimeout > 0 {
+		if err := pc.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
+			return err
+		}
+		defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()
+	}
+	_, err := pc.WriteTo(pkt, peer)
+	return err
+}
 func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
 func (t *Transport) readLoop(pc net.PacketConn) {
 		}
 		seq := binary.BigEndian.Uint64(p[1:])
 		t.mu.Lock()
-		delete(t.pending, seq)
+		if _, ok := t.pending[seq]; ok {
+			delete(t.pending, seq)
+			if t.sendCond != nil {
+				t.sendCond.Signal()
+			}
+		}
 		t.mu.Unlock()
 	case recProbe:
 		if len(p) != 9 {
 		f   ports.Frame
 	}
 	t.mu.Lock()
-	for seq, p := range t.pending {
+	seqs := make([]uint64, 0, len(t.pending))
-		if p.last.IsZero() || now.Sub(p.last) >= t.resendAfter {
+	for seq := range t.pending {
-			p.last = now
+		seqs = append(seqs, seq)
-			resend = append(resend, struct {
+	}
-				seq uint64
+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
-				f   ports.Frame
+	limit := t.maxResendPerTick
-			}{seq, p.frame})
+	for _, seq := range seqs {
+		if limit <= 0 {
+			break
 		}
+		p := t.pending[seq]
+		if p == nil || now.Sub(p.last) < t.resendDelayLocked(p) {
+			continue
+		}
+		p.last = now
+		p.attempts++
+		resend = append(resend, struct {
+			seq uint64
+			f   ports.Frame
+		}{seq, p.frame})
+		limit--
 	}
 	t.mu.Unlock()
 	for _, r := range resend {
 	}
 }
+func (t *Transport) resendDelayLocked(p *pending) time.Duration {
+	d := t.resendAfter
+	for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
+		d *= 2
+		if d > t.maxResendAfter {
+			return t.maxResendAfter
+		}
+	}
+	return d
+}
 func (t *Transport) hopPacketConnOnce() {
 	t.mu.Lock()
 	if t.rebind == nil || t.hoppedOffline || t.closed {
 		return
 	}
 	t.linkState = state
+	if t.sendCond != nil {
+		t.sendCond.Broadcast()
+	}
 	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)
+	if t.sendCond != nil {
+		t.sendCond.Broadcast()
+	}
 	t.mu.Unlock()
 	t.deliverMu.Lock()
 	t.deliverCond.Broadcast()

M internal/adapters/dgram/transport_test.go +235 -15


@@ -3,6 +3,7 @@ package dgram
3	3		import (
4	4		"bytes"
5	5		"context"
	6	+	"encoding/binary"
6	7		"errors"
7	8		"net"
8	9		"sync"

@@ -82,6 +83,112 @@ func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
82	83
83	84		func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
84	85
	86	+	type timeoutErr struct{}
	87	+
	88	+	func (timeoutErr) Error() string { return "i/o timeout" }
	89	+	func (timeoutErr) Timeout() bool { return true }
	90	+
	91	+	type blockingWritePC struct {
	92	+	addr net.Addr
	93	+
	94	+	mu sync.Mutex
	95	+	closed bool
	96	+	deadline time.Time
	97	+	done chan struct{}
	98	+	}
	99	+
	100	+	func newBlockingWritePC() *blockingWritePC {
	101	+	return &blockingWritePC{addr: testAddr("blocked"), done: make(chan struct{})}
	102	+	}
	103	+
	104	+	func (p *blockingWritePC) ReadFrom([]byte) (int, net.Addr, error) {
	105	+	<-p.done
	106	+	return 0, nil, errors.New("closed")
	107	+	}
	108	+
	109	+	func (p *blockingWritePC) WriteTo([]byte, net.Addr) (int, error) {
	110	+	for {
	111	+	p.mu.Lock()
	112	+	closed := p.closed
	113	+	deadline := p.deadline
	114	+	p.mu.Unlock()
	115	+	if closed {
	116	+	return 0, errors.New("closed")
	117	+	}
	118	+	if !deadline.IsZero() {
	119	+	wait := time.Until(deadline)
	120	+	if wait > 0 {
	121	+	timer := time.NewTimer(wait)
	122	+	select {
	123	+	case <-timer.C:
	124	+	case <-p.done:
	125	+	timer.Stop()
	126	+	return 0, errors.New("closed")
	127	+	}
	128	+	}
	129	+	return 0, timeoutErr{}
	130	+	}
	131	+	select {
	132	+	case <-time.After(time.Millisecond):
	133	+	case <-p.done:
	134	+	return 0, errors.New("closed")
	135	+	}
	136	+	}
	137	+	}
	138	+
	139	+	func (p *blockingWritePC) Close() error {
	140	+	p.mu.Lock()
	141	+	defer p.mu.Unlock()
	142	+	if !p.closed {
	143	+	p.closed = true
	144	+	close(p.done)
	145	+	}
	146	+	return nil
	147	+	}
	148	+
	149	+	func (p *blockingWritePC) LocalAddr() net.Addr { return p.addr }
	150	+	func (p *blockingWritePC) SetDeadline(t time.Time) error {
	151	+	p.mu.Lock()
	152	+	p.deadline = t
	153	+	p.mu.Unlock()
	154	+	return nil
	155	+	}
	156	+	func (p *blockingWritePC) SetReadDeadline(time.Time) error { return nil }
	157	+	func (p *blockingWritePC) SetWriteDeadline(t time.Time) error {
	158	+	p.mu.Lock()
	159	+	p.deadline = t
	160	+	p.mu.Unlock()
	161	+	return nil
	162	+	}
	163	+
	164	+	func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
	165	+	pc := newBlockingWritePC()
	166	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
	167	+	WriteTimeout: 10 * time.Millisecond,
	168	+	ResendAfter: time.Hour,
	169	+	Heartbeat: time.Hour,
	170	+	})
	171	+	if err != nil {
	172	+	t.Fatal(err)
	173	+	}
	174	+	defer func() { _ = tr.Close() }()
	175	+
	176	+	done := make(chan error, 1)
	177	+	go func() { done <- tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("blocked")}) }()
	178	+	select {
	179	+	case err := <-done:
	180	+	if err == nil {
	181	+	t.Fatal("Send returned nil, want write timeout")
	182	+	}
	183	+	var timeout interface{ Timeout() bool }
	184	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
	185	+	t.Fatalf("Send err=%v, want timeout error", err)
	186	+	}
	187	+	case <-time.After(250 * time.Millisecond):
	188	+	t.Fatal("Send blocked without honoring WriteTimeout")
	189	+	}
	190	+	}
	191	+
85	192		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {
86	193		aPC, bPC := newPair()
87	194		var droppedAck atomic.Bool

@@ -342,12 +449,12 @@ func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
342	449		defer func() { _ = a.Close() }()
343	450		defer func() { _ = b.Close() }()
344	451
345		-	for i := 0; i < 100; i++ {
	452	+	for i := range 100 {
346	453		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
347	454		t.Fatal(err)
348	455		}
349	456		}
350		-	for i := 0; i < 100; i++ {
	457	+	for i := range 100 {
351	458		got := recvWithin(t, b, time.Second)
352	459		if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
353	460		t.Fatalf("frame %d delivered out of order: %+v", i, got)

@@ -362,7 +469,7 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
362	469		defer func() { _ = a.Close() }()
363	470		defer func() { _ = b.Close() }()
364	471
365		-	for i := 0; i < cap(b.in)+20; i++ {
	472	+	for i := range cap(b.in) + 20 {
366	473		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
367	474		t.Fatal(err)
368	475		}

@@ -383,6 +490,69 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
383	490		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)
384	491		}
385	492
	493	+	func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *atomic.Int32) {
	494	+	t.Helper()
	495	+	aPC, _ := newPair()
	496	+	codec, err := pdgram.NewCodec(key())
	497	+	if err != nil {
	498	+	t.Fatal(err)
	499	+	}
	500	+	var writes atomic.Int32
	501	+	aPC.drop = func(_ []byte, addr net.Addr) bool {
	502	+	if addr.String() == "b" {
	503	+	writes.Add(1)
	504	+	return true
	505	+	}
	506	+	return false
	507	+	}
	508	+	return &Transport{
	509	+	pc: aPC,
	510	+	peer: testAddr("b"),
	511	+	codec: codec,
	512	+	sendDir: 1,
	513	+	mtu: pdgram.DefaultMTU,
	514	+	pending: make(map[uint64]*pending),
	515	+	resendAfter: 100 * time.Millisecond,
	516	+	maxResendAfter: 500 * time.Millisecond,
	517	+	maxResendPerTick: maxResendPerTick,
	518	+	clock: realClock{},
	519	+	}, &writes
	520	+	}
	521	+
	522	+	func TestResendPendingCapsBurst(t *testing.T) {
	523	+	tr, writes := newResendTestTransport(t, 2)
	524	+	for i := range 5 {
	525	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: time.Now().Add(-time.Second)}
	526	+	}
	527	+
	528	+	tr.resendPending()
	529	+	if got := writes.Load(); got != 2 {
	530	+	t.Fatalf("resend writes=%d, want capped burst of 2", got)
	531	+	}
	532	+	}
	533	+
	534	+	func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
	535	+	tr, writes := newResendTestTransport(t, 10)
	536	+	now := time.Now()
	537	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}
	538	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}
	539	+	tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}
	540	+
	541	+	tr.resendPending()
	542	+	if got := writes.Load(); got != 2 {
	543	+	t.Fatalf("resend writes=%d, want ready and capped frames only", got)
	544	+	}
	545	+	if attempts := tr.pending[1].attempts; attempts != 1 {
	546	+	t.Fatalf("too-soon attempts=%d, want unchanged", attempts)
	547	+	}
	548	+	if attempts := tr.pending[2].attempts; attempts != 2 {
	549	+	t.Fatalf("ready attempts=%d, want incremented", attempts)
	550	+	}
	551	+	if attempts := tr.pending[3].attempts; attempts != 11 {
	552	+	t.Fatalf("capped attempts=%d, want incremented", attempts)
	553	+	}
	554	+	}
	555	+
386	556		func TestProbeTimeoutAndSuccess(t *testing.T) {
387	557		aPC, bPC := newPair()
388	558		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)

@@ -421,34 +591,84 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
421	591		}
422	592		}
423	593
424		-	func TestPendingReliableQueueBounded(t *testing.T) {
	594	+	func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
425	595		aPC, bPC := newPair()
426	596		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
427		-	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
	597	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
428	598		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
429	599		defer func() { _ = a.Close() }()
430	600		defer func() { _ = b.Close() }()
431	601
432		-	for i := 0; i < maxPendingReliable; i++ {
433		-	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte{byte(i)}}); err != nil {
434		-	t.Fatalf("send %d: %v", i, err)
435		-	}
	602	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	603	+	t.Fatal(err)
	604	+	}
	605	+	done := make(chan error, 1)
	606	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	607	+
	608	+	select {
	609	+	case err := <-done:
	610	+	t.Fatalf("send returned before an ack released pending capacity: %v", err)
	611	+	case <-time.After(25 * time.Millisecond):
436	612		}
437		-	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("full")}); !errors.Is(err, ErrPendingFull) {
438		-	t.Fatalf("send past bound err=%v, want ErrPendingFull", err)
	613	+
	614	+	a.handleRecord(ackRecord(1))
	615	+	select {
	616	+	case err := <-done:
	617	+	if err != nil {
	618	+	t.Fatalf("send after ack: %v", err)
	619	+	}
	620	+	case <-time.After(time.Second):
	621	+	t.Fatal("send did not resume after ack released pending capacity")
439	622		}
440	623		a.mu.Lock()
441	624		pending := len(a.pending)
442	625		a.mu.Unlock()
443		-	if pending > maxPendingReliable {
444		-	t.Fatalf("pending=%d, want <= %d", pending, maxPendingReliable)
	626	+	if pending > 1 {
	627	+	t.Fatalf("pending=%d, want <= 1", pending)
445	628		}
446	629		}
447	630
	631	+	func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
	632	+	aPC, bPC := newPair()
	633	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	634	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
	635	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	636	+	defer func() { _ = a.Close() }()
	637	+	defer func() { _ = b.Close() }()
	638	+
	639	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	640	+	t.Fatal(err)
	641	+	}
	642	+	done := make(chan error, 1)
	643	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	644	+
	645	+	select {
	646	+	case err := <-done:
	647	+	t.Fatalf("send returned before link degraded: %v", err)
	648	+	case <-time.After(25 * time.Millisecond):
	649	+	}
	650	+	a.setLinkState(ports.LinkStateDegraded, nil)
	651	+	select {
	652	+	case err := <-done:
	653	+	if !errors.Is(err, ErrPendingFull) {
	654	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
	655	+	}
	656	+	case <-time.After(time.Second):
	657	+	t.Fatal("send did not unblock after link degraded")
	658	+	}
	659	+	}
	660	+
	661	+	func ackRecord(seq uint64) []byte {
	662	+	var b [9]byte
	663	+	b[0] = recAck
	664	+	binary.BigEndian.PutUint64(b[1:], seq)
	665	+	return b[:]
	666	+	}
	667	+
448	668		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
449	669		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
450	670		tr.deliverCond = sync.NewCond(&tr.deliverMu)
451		-	for i := 0; i < maxRecvBuffer+100; i++ {
	671	+	for i := range maxRecvBuffer + 100 {
452	672		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
453	673		}
454	674		if got := len(tr.recvBuf); got != maxRecvBuffer {

@@ -474,7 +694,7 @@ func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
474	694		defer func() { _ = aPC.Close() }()
475	695		defer func() { _ = bPC.Close() }()
476	696
477		-	for i := 0; i < maxRecvBuffer; i++ {
	697	+	for i := range maxRecvBuffer {
478	698		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
479	699		}
480	700		tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))


@@ -3,6 +3,7 @@ package dgram
3		import (	3		import (
4		"bytes"	4		"bytes"
5		"context"	5		"context"
			6	+	"encoding/binary"
6		"errors"	7		"errors"
7		"net"	8		"net"
8		"sync"	9		"sync"

@@ -82,6 +83,112 @@ func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
82			83
83		func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }	84		func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
84			85
			86	+	type timeoutErr struct{}
			87	+
			88	+	func (timeoutErr) Error() string { return "i/o timeout" }
			89	+	func (timeoutErr) Timeout() bool { return true }
			90	+
			91	+	type blockingWritePC struct {
			92	+	addr net.Addr
			93	+
			94	+	mu sync.Mutex
			95	+	closed bool
			96	+	deadline time.Time
			97	+	done chan struct{}
			98	+	}
			99	+
			100	+	func newBlockingWritePC() *blockingWritePC {
			101	+	return &blockingWritePC{addr: testAddr("blocked"), done: make(chan struct{})}
			102	+	}
			103	+
			104	+	func (p *blockingWritePC) ReadFrom([]byte) (int, net.Addr, error) {
			105	+	<-p.done
			106	+	return 0, nil, errors.New("closed")
			107	+	}
			108	+
			109	+	func (p *blockingWritePC) WriteTo([]byte, net.Addr) (int, error) {
			110	+	for {
			111	+	p.mu.Lock()
			112	+	closed := p.closed
			113	+	deadline := p.deadline
			114	+	p.mu.Unlock()
			115	+	if closed {
			116	+	return 0, errors.New("closed")
			117	+	}
			118	+	if !deadline.IsZero() {
			119	+	wait := time.Until(deadline)
			120	+	if wait > 0 {
			121	+	timer := time.NewTimer(wait)
			122	+	select {
			123	+	case <-timer.C:
			124	+	case <-p.done:
			125	+	timer.Stop()
			126	+	return 0, errors.New("closed")
			127	+	}
			128	+	}
			129	+	return 0, timeoutErr{}
			130	+	}
			131	+	select {
			132	+	case <-time.After(time.Millisecond):
			133	+	case <-p.done:
			134	+	return 0, errors.New("closed")
			135	+	}
			136	+	}
			137	+	}
			138	+
			139	+	func (p *blockingWritePC) Close() error {
			140	+	p.mu.Lock()
			141	+	defer p.mu.Unlock()
			142	+	if !p.closed {
			143	+	p.closed = true
			144	+	close(p.done)
			145	+	}
			146	+	return nil
			147	+	}
			148	+
			149	+	func (p *blockingWritePC) LocalAddr() net.Addr { return p.addr }
			150	+	func (p *blockingWritePC) SetDeadline(t time.Time) error {
			151	+	p.mu.Lock()
			152	+	p.deadline = t
			153	+	p.mu.Unlock()
			154	+	return nil
			155	+	}
			156	+	func (p *blockingWritePC) SetReadDeadline(time.Time) error { return nil }
			157	+	func (p *blockingWritePC) SetWriteDeadline(t time.Time) error {
			158	+	p.mu.Lock()
			159	+	p.deadline = t
			160	+	p.mu.Unlock()
			161	+	return nil
			162	+	}
			163	+
			164	+	func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
			165	+	pc := newBlockingWritePC()
			166	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
			167	+	WriteTimeout: 10 * time.Millisecond,
			168	+	ResendAfter: time.Hour,
			169	+	Heartbeat: time.Hour,
			170	+	})
			171	+	if err != nil {
			172	+	t.Fatal(err)
			173	+	}
			174	+	defer func() { _ = tr.Close() }()
			175	+
			176	+	done := make(chan error, 1)
			177	+	go func() { done <- tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("blocked")}) }()
			178	+	select {
			179	+	case err := <-done:
			180	+	if err == nil {
			181	+	t.Fatal("Send returned nil, want write timeout")
			182	+	}
			183	+	var timeout interface{ Timeout() bool }
			184	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
			185	+	t.Fatalf("Send err=%v, want timeout error", err)
			186	+	}
			187	+	case <-time.After(250 * time.Millisecond):
			188	+	t.Fatal("Send blocked without honoring WriteTimeout")
			189	+	}
			190	+	}
			191	+
85		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {	192		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {
86		aPC, bPC := newPair()	193		aPC, bPC := newPair()
87		var droppedAck atomic.Bool	194		var droppedAck atomic.Bool

@@ -342,12 +449,12 @@ func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
342		defer func() { _ = a.Close() }()	449		defer func() { _ = a.Close() }()
343		defer func() { _ = b.Close() }()	450		defer func() { _ = b.Close() }()
344			451
345	-	for i := 0; i < 100; i++ {	452	+	for i := range 100 {
346		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {	453		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
347		t.Fatal(err)	454		t.Fatal(err)
348		}	455		}
349		}	456		}
350	-	for i := 0; i < 100; i++ {	457	+	for i := range 100 {
351		got := recvWithin(t, b, time.Second)	458		got := recvWithin(t, b, time.Second)
352		if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {	459		if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
353		t.Fatalf("frame %d delivered out of order: %+v", i, got)	460		t.Fatalf("frame %d delivered out of order: %+v", i, got)

@@ -362,7 +469,7 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
362		defer func() { _ = a.Close() }()	469		defer func() { _ = a.Close() }()
363		defer func() { _ = b.Close() }()	470		defer func() { _ = b.Close() }()
364			471
365	-	for i := 0; i < cap(b.in)+20; i++ {	472	+	for i := range cap(b.in) + 20 {
366		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {	473		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
367		t.Fatal(err)	474		t.Fatal(err)
368		}	475		}

@@ -383,6 +490,69 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
383		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)	490		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)
384		}	491		}
385			492
			493	+	func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *atomic.Int32) {
			494	+	t.Helper()
			495	+	aPC, _ := newPair()
			496	+	codec, err := pdgram.NewCodec(key())
			497	+	if err != nil {
			498	+	t.Fatal(err)
			499	+	}
			500	+	var writes atomic.Int32
			501	+	aPC.drop = func(_ []byte, addr net.Addr) bool {
			502	+	if addr.String() == "b" {
			503	+	writes.Add(1)
			504	+	return true
			505	+	}
			506	+	return false
			507	+	}
			508	+	return &Transport{
			509	+	pc: aPC,
			510	+	peer: testAddr("b"),
			511	+	codec: codec,
			512	+	sendDir: 1,
			513	+	mtu: pdgram.DefaultMTU,
			514	+	pending: make(map[uint64]*pending),
			515	+	resendAfter: 100 * time.Millisecond,
			516	+	maxResendAfter: 500 * time.Millisecond,
			517	+	maxResendPerTick: maxResendPerTick,
			518	+	clock: realClock{},
			519	+	}, &writes
			520	+	}
			521	+
			522	+	func TestResendPendingCapsBurst(t *testing.T) {
			523	+	tr, writes := newResendTestTransport(t, 2)
			524	+	for i := range 5 {
			525	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: time.Now().Add(-time.Second)}
			526	+	}
			527	+
			528	+	tr.resendPending()
			529	+	if got := writes.Load(); got != 2 {
			530	+	t.Fatalf("resend writes=%d, want capped burst of 2", got)
			531	+	}
			532	+	}
			533	+
			534	+	func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
			535	+	tr, writes := newResendTestTransport(t, 10)
			536	+	now := time.Now()
			537	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * time.Millisecond), attempts: 1}
			538	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("ready")}, last: now.Add(-250 * time.Millisecond), attempts: 1}
			539	+	tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}
			540	+
			541	+	tr.resendPending()
			542	+	if got := writes.Load(); got != 2 {
			543	+	t.Fatalf("resend writes=%d, want ready and capped frames only", got)
			544	+	}
			545	+	if attempts := tr.pending[1].attempts; attempts != 1 {
			546	+	t.Fatalf("too-soon attempts=%d, want unchanged", attempts)
			547	+	}
			548	+	if attempts := tr.pending[2].attempts; attempts != 2 {
			549	+	t.Fatalf("ready attempts=%d, want incremented", attempts)
			550	+	}
			551	+	if attempts := tr.pending[3].attempts; attempts != 11 {
			552	+	t.Fatalf("capped attempts=%d, want incremented", attempts)
			553	+	}
			554	+	}
			555	+
386		func TestProbeTimeoutAndSuccess(t *testing.T) {	556		func TestProbeTimeoutAndSuccess(t *testing.T) {
387		aPC, bPC := newPair()	557		aPC, bPC := newPair()
388		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)	558		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)

@@ -421,34 +591,84 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
421		}	591		}
422		}	592		}
423			593
424	-	func TestPendingReliableQueueBounded(t *testing.T) {	594	+	func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
425		aPC, bPC := newPair()	595		aPC, bPC := newPair()
426		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	596		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
427	-	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)	597	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
428		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)	598		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
429		defer func() { _ = a.Close() }()	599		defer func() { _ = a.Close() }()
430		defer func() { _ = b.Close() }()	600		defer func() { _ = b.Close() }()
431			601
432	-	for i := 0; i < maxPendingReliable; i++ {	602	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
433	-	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte{byte(i)}}); err != nil {	603	+	t.Fatal(err)
434	-	t.Fatalf("send %d: %v", i, err)	604	+	}
435	-	}	605	+	done := make(chan error, 1)
			606	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
			607	+
			608	+	select {
			609	+	case err := <-done:
			610	+	t.Fatalf("send returned before an ack released pending capacity: %v", err)
			611	+	case <-time.After(25 * time.Millisecond):
436		}	612		}
437	-	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("full")}); !errors.Is(err, ErrPendingFull) {	613	+
438	-	t.Fatalf("send past bound err=%v, want ErrPendingFull", err)	614	+	a.handleRecord(ackRecord(1))
			615	+	select {
			616	+	case err := <-done:
			617	+	if err != nil {
			618	+	t.Fatalf("send after ack: %v", err)
			619	+	}
			620	+	case <-time.After(time.Second):
			621	+	t.Fatal("send did not resume after ack released pending capacity")
439		}	622		}
440		a.mu.Lock()	623		a.mu.Lock()
441		pending := len(a.pending)	624		pending := len(a.pending)
442		a.mu.Unlock()	625		a.mu.Unlock()
443	-	if pending > maxPendingReliable {	626	+	if pending > 1 {
444	-	t.Fatalf("pending=%d, want <= %d", pending, maxPendingReliable)	627	+	t.Fatalf("pending=%d, want <= 1", pending)
445		}	628		}
446		}	629		}
447			630
			631	+	func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
			632	+	aPC, bPC := newPair()
			633	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
			634	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
			635	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
			636	+	defer func() { _ = a.Close() }()
			637	+	defer func() { _ = b.Close() }()
			638	+
			639	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
			640	+	t.Fatal(err)
			641	+	}
			642	+	done := make(chan error, 1)
			643	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
			644	+
			645	+	select {
			646	+	case err := <-done:
			647	+	t.Fatalf("send returned before link degraded: %v", err)
			648	+	case <-time.After(25 * time.Millisecond):
			649	+	}
			650	+	a.setLinkState(ports.LinkStateDegraded, nil)
			651	+	select {
			652	+	case err := <-done:
			653	+	if !errors.Is(err, ErrPendingFull) {
			654	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
			655	+	}
			656	+	case <-time.After(time.Second):
			657	+	t.Fatal("send did not unblock after link degraded")
			658	+	}
			659	+	}
			660	+
			661	+	func ackRecord(seq uint64) []byte {
			662	+	var b [9]byte
			663	+	b[0] = recAck
			664	+	binary.BigEndian.PutUint64(b[1:], seq)
			665	+	return b[:]
			666	+	}
			667	+
448		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {	668		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
449		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}	669		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
450		tr.deliverCond = sync.NewCond(&tr.deliverMu)	670		tr.deliverCond = sync.NewCond(&tr.deliverMu)
451	-	for i := 0; i < maxRecvBuffer+100; i++ {	671	+	for i := range maxRecvBuffer + 100 {
452		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})	672		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
453		}	673		}
454		if got := len(tr.recvBuf); got != maxRecvBuffer {	674		if got := len(tr.recvBuf); got != maxRecvBuffer {

@@ -474,7 +694,7 @@ func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
474		defer func() { _ = aPC.Close() }()	694		defer func() { _ = aPC.Close() }()
475		defer func() { _ = bPC.Close() }()	695		defer func() { _ = bPC.Close() }()
476			696
477	-	for i := 0; i < maxRecvBuffer; i++ {	697	+	for i := range maxRecvBuffer {
478		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}	698		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
479		}	699		}
480		tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))	700		tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))