Merge pull request #40 from bnema/fix/udp-remote-cmatrix-freeze

fix(dgram): pace UDP remote retransmits

commit: 91b82aabf8180876c908e834c2b80c60c6abd248
author: bnema · 3mo ago · 2026-07-08 11:20 UTC
committer: GitHub · 3mo ago · 2026-07-08 11:20 UTC
merge of: e2fc930 + 8916627 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

8 files · +901 -129

M internal/adapters/dgram/proxy.go +77 -8


@@ -38,6 +38,29 @@ 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	+
	48	+	type proxyCopier struct {
	49	+	ctx context.Context
	50	+	errCh chan<- proxyCopyErr
	51	+	retryEvents chan<- proxyRetryEvent
	52	+	retryBackoff time.Duration
	53	+	clk ports.Clock
	54	+	}
	55	+
	56	+	type proxyCopyDirection struct {
	57	+	src ports.Transport
	58	+	dst ports.Transport
	59	+	recvKind proxyCopyErrKind
	60	+	sendKind proxyCopyErrKind
	61	+	retryPendingFull bool
	62	+	}
	63	+
41	64		func (p ProxyRuntime) Run(ctx context.Context) error {
42	65		defer func() {
43	66		_ = p.Client.Close()

@@ -58,11 +81,15 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
58	81		}
59	82
60	83		errCh := make(chan proxyCopyErr, 2)
	84	+	retryEvents := make(chan proxyRetryEvent, 2)
	85	+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
	86	+	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
	87	+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryPendingFull: true}
61	88		startClientCopy := func() {
62		-	go copyProxyFrames(p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh)
	89	+	go copier.copyFrames(clientToDaemon)
63	90		}
64	91		startDaemonCopy := func() {
65		-	go copyProxyFrames(p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh)
	92	+	go copier.copyFrames(daemonToClient)
66	93		}
67	94		startClientCopy()
68	95		startDaemonCopy()

@@ -121,6 +148,13 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
121	148		}
122	149		case <-idle:
123	150		return ErrLinkDead
	151	+	case ev := <-retryEvents:
	152	+	switch ev {
	153	+	case proxyRetryStart:
	154	+	startIdle()
	155	+	case proxyRetryDone:
	156	+	stopIdle()
	157	+	}
124	158		case ce := <-errCh:
125	159		switch ce.kind {
126	160		case proxyDaemonRecv:

@@ -159,20 +193,53 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
159	193		}
160	194		}
161	195
162		-	func copyProxyFrames(dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr) {
	196	+	func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
163	197		for {
164		-	f, err := src.Recv()
	198	+	f, err := dir.src.Recv()
165	199		if err != nil {
166		-	errCh <- proxyCopyErr{kind: recvKind, err: err}
	200	+	c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
167	201		return
168	202		}
169		-	if err := dst.Send(f); err != nil {
170		-	errCh <- proxyCopyErr{kind: sendKind, err: err}
171		-	return
	203	+	retrying := false
	204	+	for {
	205	+	if err := dir.dst.Send(f); err != nil {
	206	+	if dir.retryPendingFull && errors.Is(err, ErrPendingFull) && recoverableClientLink(dir.dst, err) {
	207	+	if !retrying {
	208	+	retrying = true
	209	+	if !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryStart) {
	210	+	return
	211	+	}
	212	+	}
	213	+	t := c.clk.NewTimer(c.retryBackoff)
	214	+	select {
	215	+	case <-t.C():
	216	+	t.Stop()
	217	+	continue
	218	+	case <-c.ctx.Done():
	219	+	t.Stop()
	220	+	return
	221	+	}
	222	+	}
	223	+	c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
	224	+	return
	225	+	}
	226	+	if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
	227	+	return
	228	+	}
	229	+	break
172	230		}
173	231		}
174	232		}
175	233
	234	+	func sendProxyRetryEvent(ctx context.Context, retryEvents chan<- proxyRetryEvent, ev proxyRetryEvent) bool {
	235	+	select {
	236	+	case retryEvents <- ev:
	237	+	return true
	238	+	case <-ctx.Done():
	239	+	return false
	240	+	}
	241	+	}
	242	+
176	243		func recoverableClientLink(t ports.Transport, err error) bool {
177	244		if errors.Is(err, ErrLinkDead) {
178	245		return false

@@ -182,6 +249,8 @@ func recoverableClientLink(t ports.Transport, err error) bool {
182	249		return false
183	250		}
184	251		switch reporter.LinkState() {
	252	+	case ports.LinkStateConnected:
	253	+	return errors.Is(err, ErrPendingFull)
185	254		case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
186	255		return true
187	256		default:


@@ -38,6 +38,29 @@ type proxyCopyErr struct {
38		err error	38		err error
39		}	39		}
40			40
			41	+	type proxyRetryEvent uint8
			42	+
			43	+	const (
			44	+	proxyRetryStart proxyRetryEvent = iota + 1
			45	+	proxyRetryDone
			46	+	)
			47	+
			48	+	type proxyCopier struct {
			49	+	ctx context.Context
			50	+	errCh chan<- proxyCopyErr
			51	+	retryEvents chan<- proxyRetryEvent
			52	+	retryBackoff time.Duration
			53	+	clk ports.Clock
			54	+	}
			55	+
			56	+	type proxyCopyDirection struct {
			57	+	src ports.Transport
			58	+	dst ports.Transport
			59	+	recvKind proxyCopyErrKind
			60	+	sendKind proxyCopyErrKind
			61	+	retryPendingFull bool
			62	+	}
			63	+
41		func (p ProxyRuntime) Run(ctx context.Context) error {	64		func (p ProxyRuntime) Run(ctx context.Context) error {
42		defer func() {	65		defer func() {
43		_ = p.Client.Close()	66		_ = p.Client.Close()

@@ -58,11 +81,15 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
58		}	81		}
59			82
60		errCh := make(chan proxyCopyErr, 2)	83		errCh := make(chan proxyCopyErr, 2)
			84	+	retryEvents := make(chan proxyRetryEvent, 2)
			85	+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
			86	+	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
			87	+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryPendingFull: true}
61		startClientCopy := func() {	88		startClientCopy := func() {
62	-	go copyProxyFrames(p.Daemon, p.Client, proxyClientRecv, proxyDaemonSend, errCh)	89	+	go copier.copyFrames(clientToDaemon)
63		}	90		}
64		startDaemonCopy := func() {	91		startDaemonCopy := func() {
65	-	go copyProxyFrames(p.Client, p.Daemon, proxyDaemonRecv, proxyClientSend, errCh)	92	+	go copier.copyFrames(daemonToClient)
66		}	93		}
67		startClientCopy()	94		startClientCopy()
68		startDaemonCopy()	95		startDaemonCopy()

@@ -121,6 +148,13 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
121		}	148		}
122		case <-idle:	149		case <-idle:
123		return ErrLinkDead	150		return ErrLinkDead
			151	+	case ev := <-retryEvents:
			152	+	switch ev {
			153	+	case proxyRetryStart:
			154	+	startIdle()
			155	+	case proxyRetryDone:
			156	+	stopIdle()
			157	+	}
124		case ce := <-errCh:	158		case ce := <-errCh:
125		switch ce.kind {	159		switch ce.kind {
126		case proxyDaemonRecv:	160		case proxyDaemonRecv:

@@ -159,20 +193,53 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
159		}	193		}
160		}	194		}
161			195
162	-	func copyProxyFrames(dst, src ports.Transport, recvKind, sendKind proxyCopyErrKind, errCh chan<- proxyCopyErr) {	196	+	func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
163		for {	197		for {
164	-	f, err := src.Recv()	198	+	f, err := dir.src.Recv()
165		if err != nil {	199		if err != nil {
166	-	errCh <- proxyCopyErr{kind: recvKind, err: err}	200	+	c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
167		return	201		return
168		}	202		}
169	-	if err := dst.Send(f); err != nil {	203	+	retrying := false
170	-	errCh <- proxyCopyErr{kind: sendKind, err: err}	204	+	for {
171	-	return	205	+	if err := dir.dst.Send(f); err != nil {
			206	+	if dir.retryPendingFull && errors.Is(err, ErrPendingFull) && recoverableClientLink(dir.dst, err) {
			207	+	if !retrying {
			208	+	retrying = true
			209	+	if !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryStart) {
			210	+	return
			211	+	}
			212	+	}
			213	+	t := c.clk.NewTimer(c.retryBackoff)
			214	+	select {
			215	+	case <-t.C():
			216	+	t.Stop()
			217	+	continue
			218	+	case <-c.ctx.Done():
			219	+	t.Stop()
			220	+	return
			221	+	}
			222	+	}
			223	+	c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
			224	+	return
			225	+	}
			226	+	if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
			227	+	return
			228	+	}
			229	+	break
172		}	230		}
173		}	231		}
174		}	232		}
175			233
			234	+	func sendProxyRetryEvent(ctx context.Context, retryEvents chan<- proxyRetryEvent, ev proxyRetryEvent) bool {
			235	+	select {
			236	+	case retryEvents <- ev:
			237	+	return true
			238	+	case <-ctx.Done():
			239	+	return false
			240	+	}
			241	+	}
			242	+
176		func recoverableClientLink(t ports.Transport, err error) bool {	243		func recoverableClientLink(t ports.Transport, err error) bool {
177		if errors.Is(err, ErrLinkDead) {	244		if errors.Is(err, ErrLinkDead) {
178		return false	245		return false

@@ -182,6 +249,8 @@ func recoverableClientLink(t ports.Transport, err error) bool {
182		return false	249		return false
183		}	250		}
184		switch reporter.LinkState() {	251		switch reporter.LinkState() {
			252	+	case ports.LinkStateConnected:
			253	+	return errors.Is(err, ErrPendingFull)
185		case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:	254		case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
186		return true	255		return true
187		default:	256		default:

M internal/adapters/dgram/proxy_test.go +126 -16


@@ -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,25 +115,127 @@ func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T)
107	115		}
108	116		}
109	117
110		-	func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {
	118	+	func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
	119	+	tests := []struct {
	120	+	name string
	121	+	state ports.LinkState
	122	+	linkEvent bool
	123	+	}{
	124	+	{name: "connected pending full", state: ports.LinkStateConnected},
	125	+	{name: "offline pending full", state: ports.LinkStateOffline, linkEvent: true},
	126	+	}
	127	+	for _, tt := range tests {
	128	+	t.Run(tt.name, func(t *testing.T) {
	129	+	client := newFakeTransport()
	130	+	daemon := newFakeTransport()
	131	+	client.linkState = tt.state
	132	+	if tt.linkEvent {
	133	+	client.linkEvents <- ports.LinkEvent{State: tt.state}
	134	+	}
	135	+	errCh := make(chan error, 1)
	136	+	go func() {
	137	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
	138	+	}()
	139	+	client.sendErr <- ErrPendingFull
	140	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
	141	+	select {
	142	+	case got := <-client.sent:
	143	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "preserved during outage" {
	144	+	t.Fatalf("client got %+v, want retried output", got)
	145	+	}
	146	+	case <-time.After(time.Second):
	147	+	t.Fatal("timeout waiting for recovered daemon output")
	148	+	}
	149	+	daemon.recv <- recvResult{err: io.EOF}
	150	+	select {
	151	+	case err := <-errCh:
	152	+	if err != nil {
	153	+	t.Fatalf("Run err=%v, want nil", err)
	154	+	}
	155	+	case <-time.After(time.Second):
	156	+	t.Fatal("timeout waiting for daemon EOF")
	157	+	}
	158	+	})
	159	+	}
	160	+	}
	161	+
	162	+	func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
111	163		client := newFakeTransport()
112	164		daemon := newFakeTransport()
113	165		client.linkState = ports.LinkStateOffline
114		-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
	166	+	client.sideEffectOnErr = true
	167	+	client.sendErr <- errors.New("write timed out after queuing")
115	168		errCh := make(chan error, 1)
116	169		go func() {
117		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
	170	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
	171	+	}()
	172	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("maybe queued")}}
	173	+	select {
	174	+	case got := <-client.sent:
	175	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "maybe queued" {
	176	+	t.Fatalf("client got %+v, want side-effected output", got)
	177	+	}
	178	+	case <-time.After(time.Second):
	179	+	t.Fatal("timeout waiting for side-effected output")
	180	+	}
	181	+	select {
	182	+	case got := <-client.sent:
	183	+	t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
	184	+	case <-time.After(20 * time.Millisecond):
	185	+	}
	186	+	daemon.recv <- recvResult{err: io.EOF}
	187	+	select {
	188	+	case err := <-errCh:
	189	+	if err != nil {
	190	+	t.Fatalf("Run err=%v, want nil", err)
	191	+	}
	192	+	case <-time.After(time.Second):
	193	+	t.Fatal("timeout waiting for daemon EOF")
	194	+	}
	195	+	}
	196	+
	197	+	func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
	198	+	client := newFakeTransport()
	199	+	daemon := newFakeTransport()
	200	+	client.linkState = ports.LinkStateOffline
	201	+	client.sendErrDefault = ErrPendingFull
	202	+	errCh := make(chan error, 1)
	203	+	go func() {
	204	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
118	205		}()
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")}}
	206	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
	207	+	select {
	208	+	case err := <-errCh:
	209	+	if !errors.Is(err, ErrLinkDead) {
	210	+	t.Fatalf("Run err=%v, want ErrLinkDead", err)
	211	+	}
	212	+	case <-time.After(time.Second):
	213	+	t.Fatal("timeout waiting for idle expiry from ErrPendingFull retry")
	214	+	}
	215	+	}
	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")}}
122	227		select {
123	228		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)
	229	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "recovers" {
	230	+	t.Fatalf("client got %+v, want retried output", got)
126	231		}
127	232		case <-time.After(time.Second):
128		-	t.Fatal("timeout waiting for recovered daemon output")
	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):
129	239		}
130	240		daemon.recv <- recvResult{err: io.EOF}
131	241		select {


@@ -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,25 +115,127 @@ func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T)
107		}	115		}
108		}	116		}
109			117
110	-	func TestProxyRuntimeRestartsRecoverableClientSendUntilIdleTTL(t *testing.T) {	118	+	func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
			119	+	tests := []struct {
			120	+	name string
			121	+	state ports.LinkState
			122	+	linkEvent bool
			123	+	}{
			124	+	{name: "connected pending full", state: ports.LinkStateConnected},
			125	+	{name: "offline pending full", state: ports.LinkStateOffline, linkEvent: true},
			126	+	}
			127	+	for _, tt := range tests {
			128	+	t.Run(tt.name, func(t *testing.T) {
			129	+	client := newFakeTransport()
			130	+	daemon := newFakeTransport()
			131	+	client.linkState = tt.state
			132	+	if tt.linkEvent {
			133	+	client.linkEvents <- ports.LinkEvent{State: tt.state}
			134	+	}
			135	+	errCh := make(chan error, 1)
			136	+	go func() {
			137	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
			138	+	}()
			139	+	client.sendErr <- ErrPendingFull
			140	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
			141	+	select {
			142	+	case got := <-client.sent:
			143	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "preserved during outage" {
			144	+	t.Fatalf("client got %+v, want retried output", got)
			145	+	}
			146	+	case <-time.After(time.Second):
			147	+	t.Fatal("timeout waiting for recovered daemon output")
			148	+	}
			149	+	daemon.recv <- recvResult{err: io.EOF}
			150	+	select {
			151	+	case err := <-errCh:
			152	+	if err != nil {
			153	+	t.Fatalf("Run err=%v, want nil", err)
			154	+	}
			155	+	case <-time.After(time.Second):
			156	+	t.Fatal("timeout waiting for daemon EOF")
			157	+	}
			158	+	})
			159	+	}
			160	+	}
			161	+
			162	+	func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
111		client := newFakeTransport()	163		client := newFakeTransport()
112		daemon := newFakeTransport()	164		daemon := newFakeTransport()
113		client.linkState = ports.LinkStateOffline	165		client.linkState = ports.LinkStateOffline
114	-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}	166	+	client.sideEffectOnErr = true
			167	+	client.sendErr <- errors.New("write timed out after queuing")
115		errCh := make(chan error, 1)	168		errCh := make(chan error, 1)
116		go func() {	169		go func() {
117	-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())	170	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
			171	+	}()
			172	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("maybe queued")}}
			173	+	select {
			174	+	case got := <-client.sent:
			175	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "maybe queued" {
			176	+	t.Fatalf("client got %+v, want side-effected output", got)
			177	+	}
			178	+	case <-time.After(time.Second):
			179	+	t.Fatal("timeout waiting for side-effected output")
			180	+	}
			181	+	select {
			182	+	case got := <-client.sent:
			183	+	t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
			184	+	case <-time.After(20 * time.Millisecond):
			185	+	}
			186	+	daemon.recv <- recvResult{err: io.EOF}
			187	+	select {
			188	+	case err := <-errCh:
			189	+	if err != nil {
			190	+	t.Fatalf("Run err=%v, want nil", err)
			191	+	}
			192	+	case <-time.After(time.Second):
			193	+	t.Fatal("timeout waiting for daemon EOF")
			194	+	}
			195	+	}
			196	+
			197	+	func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
			198	+	client := newFakeTransport()
			199	+	daemon := newFakeTransport()
			200	+	client.linkState = ports.LinkStateOffline
			201	+	client.sendErrDefault = ErrPendingFull
			202	+	errCh := make(chan error, 1)
			203	+	go func() {
			204	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
118		}()	205		}()
119	-	client.sendErr <- errors.New("temporary send failure")	206	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
120	-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("lost during outage")}}	207	+	select {
121	-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("after recovery")}}	208	+	case err := <-errCh:
			209	+	if !errors.Is(err, ErrLinkDead) {
			210	+	t.Fatalf("Run err=%v, want ErrLinkDead", err)
			211	+	}
			212	+	case <-time.After(time.Second):
			213	+	t.Fatal("timeout waiting for idle expiry from ErrPendingFull retry")
			214	+	}
			215	+	}
			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")}}
122		select {	227		select {
123		case got := <-client.sent:	228		case got := <-client.sent:
124	-	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "after recovery" {	229	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "recovers" {
125	-	t.Fatalf("client got %+v, want recovered output", got)	230	+	t.Fatalf("client got %+v, want retried output", got)
126		}	231		}
127		case <-time.After(time.Second):	232		case <-time.After(time.Second):
128	-	t.Fatal("timeout waiting for recovered daemon output")	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):
129		}	239		}
130		daemon.recv <- recvResult{err: io.EOF}	240		daemon.recv <- recvResult{err: io.EOF}
131		select {	241		select {

M internal/adapters/dgram/transport.go +242 -67


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

@@ -19,18 +20,22 @@ 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	+	defaultMaxPendingWait = 50 * time.Millisecond
	35	+	defaultMaxRecvBuf = 1024
	36	+	maxRecvBuffer = defaultMaxRecvBuf
	37	+	linkEventBufferSize = 16
	38	+	probeReplyBufferSize = 16
34	39		)
35	40
36	41		var (

@@ -39,16 +44,20 @@ var (
39	44		)
40	45
41	46		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
	47	+	MTU int
	48	+	ResendAfter time.Duration
	49	+	MaxResendAfter time.Duration
	50	+	MaxResendPerTick int
	51	+	WriteTimeout time.Duration
	52	+	Heartbeat time.Duration
	53	+	DegradedAfter time.Duration
	54	+	ProbeAfter time.Duration
	55	+	OfflineAfter time.Duration
	56	+	DeadAfter time.Duration
	57	+	MaxPending int
	58	+	MaxPendingWait time.Duration
	59	+	MaxRecvBuffer int
	60	+	Clock ports.Clock
52	61		// RebindPacketConn is an optional client-side hook used to hop the local UDP
53	62		// socket when the peer is offline. Server/proxy transports should leave it nil.
54	63		RebindPacketConn func(net.PacketConn) (net.PacketConn, error)

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

@@ -92,6 +113,9 @@ func normalizeOptions(opts Options) Options {
92	113		if opts.MaxPending <= 0 {
93	114		opts.MaxPending = defaultMaxPending
94	115		}
	116	+	if opts.MaxPendingWait <= 0 {
	117	+	opts.MaxPendingWait = defaultMaxPendingWait
	118	+	}
95	119		if opts.MaxRecvBuffer <= 0 {
96	120		opts.MaxRecvBuffer = defaultMaxRecvBuf
97	121		}

@@ -107,29 +131,35 @@ type Transport struct {
107	131		sendDir, recvDir uint32
108	132		mtu int
109	133
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
	134	+	mu sync.Mutex
	135	+	peer net.Addr
	136	+	ctr uint64
	137	+	seq uint64
	138	+	probeSeq uint64
	139	+	pending map[uint64]*pending
	140	+	closed bool
	141	+	closeErr error
	142	+	lastHeard time.Time
	143	+	heartbeat time.Duration
	144	+	resendAfter time.Duration
	145	+	maxResendAfter time.Duration
	146	+	maxResendPerTick int
	147	+	writeTimeout time.Duration
	148	+	degradedAfter time.Duration
	149	+	probeAfter time.Duration
	150	+	offlineAfter time.Duration
	151	+	deadAfter time.Duration
	152	+	maxPending int
	153	+	maxPendingWait time.Duration
	154	+	maxRecvBuffer int
	155	+	clock ports.Clock
	156	+	linkState ports.LinkState
	157	+	linkEvents chan ports.LinkEvent
	158	+	probeWait map[uint64]chan struct{}
	159	+	sendWake chan struct{}
	160	+	probeReply chan uint64
	161	+	rebind func(net.PacketConn) (net.PacketConn, error)
	162	+	hoppedOffline bool
133	163
134	164		recvMu sync.Mutex
135	165		replay *pdgram.ReplayWindow

@@ -137,6 +167,8 @@ type Transport struct {
137	167		in chan ports.Frame
138	168		done chan struct{}
139	169
	170	+	writeMu sync.Mutex
	171	+
140	172		deliverMu sync.Mutex
141	173		deliverCond *sync.Cond
142	174		nextRecvSeq uint64

@@ -144,8 +176,9 @@ type Transport struct {
144	176		}
145	177
146	178		type pending struct {
147		-	frame ports.Frame
148		-	last time.Time
	179	+	frame ports.Frame
	180	+	last time.Time
	181	+	attempts int
149	182		}
150	183
151	184		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {

@@ -158,11 +191,46 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
158	191		return nil, err
159	192		}
160	193		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)}
	194	+	t := &Transport{
	195	+	pc: pc,
	196	+	codec: c,
	197	+	sendDir: sendDir,
	198	+	recvDir: recvDir,
	199	+	mtu: opts.MTU,
	200	+	peer: peer,
	201	+	pending: make(map[uint64]*pending),
	202	+	replay: pdgram.NewReplayWindow(),
	203	+	reasm: pdgram.NewReassembler(),
	204	+	in: make(chan ports.Frame, 32),
	205	+	done: make(chan struct{}),
	206	+	lastHeard: opts.Clock.Now(),
	207	+	heartbeat: opts.Heartbeat,
	208	+	resendAfter: opts.ResendAfter,
	209	+	maxResendAfter: opts.MaxResendAfter,
	210	+	maxResendPerTick: opts.MaxResendPerTick,
	211	+	writeTimeout: opts.WriteTimeout,
	212	+	degradedAfter: opts.DegradedAfter,
	213	+	probeAfter: opts.ProbeAfter,
	214	+	offlineAfter: opts.OfflineAfter,
	215	+	deadAfter: opts.DeadAfter,
	216	+	maxPending: opts.MaxPending,
	217	+	maxPendingWait: opts.MaxPendingWait,
	218	+	maxRecvBuffer: opts.MaxRecvBuffer,
	219	+	clock: opts.Clock,
	220	+	linkState: ports.LinkStateConnected,
	221	+	linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
	222	+	probeWait: make(map[uint64]chan struct{}),
	223	+	rebind: opts.RebindPacketConn,
	224	+	nextRecvSeq: 1,
	225	+	recvBuf: make(map[uint64]ports.Frame),
	226	+	}
	227	+	t.sendWake = make(chan struct{})
	228	+	t.probeReply = make(chan uint64, probeReplyBufferSize)
162	229		t.deliverCond = sync.NewCond(&t.deliverMu)
163	230		go t.readLoop(pc)
164	231		go t.resendLoop()
165	232		go t.deliveryLoop()
	233	+	go t.probeReplyLoop()
166	234		return t, nil
167	235		}
168	236

@@ -171,6 +239,31 @@ func (t *Transport) Send(f ports.Frame) error {
171	239		// dropping MsgOutput can permanently desynchronize the client screen.
172	240		reliable := true
173	241		t.mu.Lock()
	242	+	var pendingWaitStarted time.Time
	243	+	for reliable && len(t.pending) >= t.maxPending && !t.closed {
	244	+	if t.linkState != ports.LinkStateConnected {
	245	+	t.mu.Unlock()
	246	+	return ErrPendingFull
	247	+	}
	248	+	if pendingWaitStarted.IsZero() {
	249	+	pendingWaitStarted = t.clock.Now()
	250	+	}
	251	+	remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
	252	+	if remaining <= 0 {
	253	+	t.mu.Unlock()
	254	+	return ErrPendingFull
	255	+	}
	256	+	wake := t.sendWake
	257	+	timer := t.clock.NewTimer(remaining)
	258	+	t.mu.Unlock()
	259	+	select {
	260	+	case <-timer.C():
	261	+	case <-wake:
	262	+	case <-t.done:
	263	+	}
	264	+	timer.Stop()
	265	+	t.mu.Lock()
	266	+	}
174	267		if t.closed {
175	268		err := t.closeErr
176	269		t.mu.Unlock()

@@ -179,17 +272,24 @@ func (t *Transport) Send(f ports.Frame) error {
179	272		}
180	273		return errors.New("dgram: closed")
181	274		}
182		-	if reliable && len(t.pending) >= t.maxPending {
183		-	t.mu.Unlock()
184		-	return ErrPendingFull
185		-	}
186	275		t.seq++
187	276		seq := t.seq
188	277		if reliable {
189		-	t.pending[seq] = &pending{frame: f}
	278	+	t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
190	279		}
191	280		t.mu.Unlock()
192		-	return t.sendData(seq, reliable, f)
	281	+	if err := t.sendData(seq, reliable, f); err != nil {
	282	+	if reliable {
	283	+	t.mu.Lock()
	284	+	if _, ok := t.pending[seq]; ok {
	285	+	delete(t.pending, seq)
	286	+	t.notifySendWaitersLocked()
	287	+	}
	288	+	t.mu.Unlock()
	289	+	}
	290	+	return err
	291	+	}
	292	+	return nil
193	293		}
194	294
195	295		func (t *Transport) Recv() (ports.Frame, error) {

@@ -314,15 +414,32 @@ func (t *Transport) sendPayload(p []byte) error {
314	414		if peer == nil {
315	415		return errors.New("dgram: no peer")
316	416		}
317		-	t.mu.Lock()
318		-	pc := t.pc
319		-	t.mu.Unlock()
320		-	if _, err := pc.WriteTo(pkt, peer); err != nil {
	417	+	if err := t.writePacket(pkt, peer); err != nil {
321	418		return err
322	419		}
323	420		}
324	421		return nil
325	422		}
	423	+
	424	+	func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
	425	+	t.writeMu.Lock()
	426	+	defer t.writeMu.Unlock()
	427	+
	428	+	t.mu.Lock()
	429	+	pc := t.pc
	430	+	writeTimeout := t.writeTimeout
	431	+	t.mu.Unlock()
	432	+
	433	+	if writeTimeout > 0 {
	434	+	if err := pc.SetWriteDeadline(t.clock.Now().Add(writeTimeout)); err != nil {
	435	+	return err
	436	+	}
	437	+	defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()
	438	+	}
	439	+	_, err := pc.WriteTo(pkt, peer)
	440	+	return err
	441	+	}
	442	+
326	443		func (t *Transport) nextCounter() uint64 { t.mu.Lock(); defer t.mu.Unlock(); t.ctr++; return t.ctr }
327	444
328	445		func (t *Transport) readLoop(pc net.PacketConn) {

@@ -380,13 +497,19 @@ func (t *Transport) handleRecord(p []byte) {
380	497		}
381	498		seq := binary.BigEndian.Uint64(p[1:])
382	499		t.mu.Lock()
383		-	delete(t.pending, seq)
	500	+	if _, ok := t.pending[seq]; ok {
	501	+	delete(t.pending, seq)
	502	+	t.notifySendWaitersLocked()
	503	+	}
384	504		t.mu.Unlock()
385	505		case recProbe:
386	506		if len(p) != 9 {
387	507		return
388	508		}
389		-	_ = t.sendProbe(recPong, binary.BigEndian.Uint64(p[1:]))
	509	+	select {
	510	+	case t.probeReply <- binary.BigEndian.Uint64(p[1:]):
	511	+	default:
	512	+	}
390	513		case recPong:
391	514		if len(p) != 9 {
392	515		return

@@ -472,6 +595,25 @@ func (t *Transport) deliver(f ports.Frame) {
472	595		}
473	596		}
474	597
	598	+	func (t *Transport) probeReplyLoop() {
	599	+	for {
	600	+	select {
	601	+	case id := <-t.probeReply:
	602	+	_ = t.sendProbe(recPong, id)
	603	+	case <-t.done:
	604	+	return
	605	+	}
	606	+	}
	607	+	}
	608	+
	609	+	func (t *Transport) notifySendWaitersLocked() {
	610	+	if t.sendWake == nil {
	611	+	return
	612	+	}
	613	+	close(t.sendWake)
	614	+	t.sendWake = make(chan struct{})
	615	+	}
	616	+
475	617		func (t *Transport) resendLoop() {
476	618		resend := t.clock.NewTimer(t.resendAfter)
477	619		defer resend.Stop()

@@ -500,14 +642,27 @@ func (t *Transport) resendPending() {
500	642		f ports.Frame
501	643		}
502	644		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})
	645	+	seqs := make([]uint64, 0, len(t.pending))
	646	+	for seq := range t.pending {
	647	+	seqs = append(seqs, seq)
	648	+	}
	649	+	slices.Sort(seqs)
	650	+	limit := t.maxResendPerTick
	651	+	for _, seq := range seqs {
	652	+	if limit <= 0 {
	653	+	break
	654	+	}
	655	+	p := t.pending[seq]
	656	+	if p == nil \|\| now.Sub(p.last) < t.resendDelayLocked(p) {
	657	+	continue
510	658		}
	659	+	p.last = now
	660	+	p.attempts++
	661	+	resend = append(resend, struct {
	662	+	seq uint64
	663	+	f ports.Frame
	664	+	}{seq, p.frame})
	665	+	limit--
511	666		}
512	667		t.mu.Unlock()
513	668		for _, r := range resend {

@@ -515,6 +670,17 @@ func (t *Transport) resendPending() {
515	670		}
516	671		}
517	672
	673	+	func (t Transport) resendDelayLocked(p pending) time.Duration {
	674	+	d := t.resendAfter
	675	+	for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
	676	+	d *= 2
	677	+	if d > t.maxResendAfter {
	678	+	return t.maxResendAfter
	679	+	}
	680	+	}
	681	+	return d
	682	+	}
	683	+
518	684		func (t *Transport) hopPacketConnOnce() {
519	685		t.mu.Lock()
520	686		if t.rebind == nil \|\| t.hoppedOffline \|\| t.closed {

@@ -582,6 +748,14 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
582	748		return
583	749		}
584	750		t.linkState = state
	751	+	if state == ports.LinkStateConnected {
	752	+	for _, p := range t.pending {
	753	+	if p != nil {
	754	+	p.attempts = 0
	755	+	}
	756	+	}
	757	+	}
	758	+	t.notifySendWaitersLocked()
585	759		t.mu.Unlock()
586	760		event := ports.LinkEvent{State: state, At: now, Err: err}
587	761		select {

@@ -600,6 +774,7 @@ func (t *Transport) closeWithError(err error) {
600	774		t.closeErr = err
601	775		t.probeWait = make(map[uint64]chan struct{})
602	776		close(t.done)
	777	+	t.notifySendWaitersLocked()
603	778		t.mu.Unlock()
604	779		t.deliverMu.Lock()
605	780		t.deliverCond.Broadcast()

 	"encoding/binary"
 	"errors"
 	"net"
+	"slices"
 	"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
+	defaultMaxPendingWait   = 50 * time.Millisecond
+	defaultMaxRecvBuf       = 1024
+	maxRecvBuffer           = defaultMaxRecvBuf
+	linkEventBufferSize     = 16
+	probeReplyBufferSize    = 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
+	MaxPendingWait   time.Duration
+	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
 	}
 	if opts.MaxPending <= 0 {
 		opts.MaxPending = defaultMaxPending
 	}
+	if opts.MaxPendingWait <= 0 {
+		opts.MaxPendingWait = defaultMaxPendingWait
+	}
 	if opts.MaxRecvBuffer <= 0 {
 		opts.MaxRecvBuffer = defaultMaxRecvBuf
 	}
 	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
+	maxPendingWait   time.Duration
-	probeWait     map[uint64]chan struct{}
+	maxRecvBuffer    int
-	rebind        func(net.PacketConn) (net.PacketConn, error)
+	clock            ports.Clock
-	hoppedOffline bool
+	linkState        ports.LinkState
+	linkEvents       chan ports.LinkEvent
+	probeWait        map[uint64]chan struct{}
+	sendWake         chan struct{}
+	probeReply       chan uint64
+	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,
+		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{})
+	t.probeReply = make(chan uint64, probeReplyBufferSize)
 	t.deliverCond = sync.NewCond(&t.deliverMu)
 	go t.readLoop(pc)
 	go t.resendLoop()
 	go t.deliveryLoop()
+	go t.probeReplyLoop()
 	return t, nil
 }
 	// dropping MsgOutput can permanently desynchronize the client screen.
 	reliable := true
 	t.mu.Lock()
+	var pendingWaitStarted time.Time
+	for reliable && len(t.pending) >= t.maxPending && !t.closed {
+		if t.linkState != ports.LinkStateConnected {
+			t.mu.Unlock()
+			return ErrPendingFull
+		}
+		if pendingWaitStarted.IsZero() {
+			pendingWaitStarted = t.clock.Now()
+		}
+		remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
+		if remaining <= 0 {
+			t.mu.Unlock()
+			return ErrPendingFull
+		}
+		wake := t.sendWake
+		timer := t.clock.NewTimer(remaining)
+		t.mu.Unlock()
+		select {
+		case <-timer.C():
+		case <-wake:
+		case <-t.done:
+		}
+		timer.Stop()
+		t.mu.Lock()
+	}
 	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 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) {
 		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(t.clock.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)
+			t.notifySendWaitersLocked()
+		}
 		t.mu.Unlock()
 	case recProbe:
 		if len(p) != 9 {
 			return
 		}
-		_ = t.sendProbe(recPong, binary.BigEndian.Uint64(p[1:]))
+		select {
+		case t.probeReply <- binary.BigEndian.Uint64(p[1:]):
+		default:
+		}
 	case recPong:
 		if len(p) != 9 {
 			return
 	}
 }
+func (t *Transport) probeReplyLoop() {
+	for {
+		select {
+		case id := <-t.probeReply:
+			_ = t.sendProbe(recPong, id)
+		case <-t.done:
+			return
+		}
+	}
+}
+func (t *Transport) notifySendWaitersLocked() {
+	if t.sendWake == nil {
+		return
+	}
+	close(t.sendWake)
+	t.sendWake = make(chan struct{})
+}
 func (t *Transport) resendLoop() {
 	resend := t.clock.NewTimer(t.resendAfter)
 	defer resend.Stop()
 		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
+	slices.Sort(seqs)
-				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 state == ports.LinkStateConnected {
+		for _, p := range t.pending {
+			if p != nil {
+				p.attempts = 0
+			}
+		}
+	}
+	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.notifySendWaitersLocked()
 	t.mu.Unlock()
 	t.deliverMu.Lock()
 	t.deliverCond.Broadcast()

M internal/adapters/dgram/transport_test.go +435 -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,212 @@ 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	+	type fixedClock struct{ now time.Time }
	165	+
	166	+	func (c fixedClock) Now() time.Time { return c.now }
	167	+	func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
	168	+	return realTimer{t: time.NewTimer(d)}
	169	+	}
	170	+
	171	+	type manualClock struct {
	172	+	mu sync.Mutex
	173	+	now time.Time
	174	+	timers chan *manualTimer
	175	+	}
	176	+
	177	+	func newManualClock(now time.Time) *manualClock {
	178	+	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}
	179	+	}
	180	+
	181	+	func (c *manualClock) Now() time.Time {
	182	+	c.mu.Lock()
	183	+	defer c.mu.Unlock()
	184	+	return c.now
	185	+	}
	186	+
	187	+	func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
	188	+	tm := &manualTimer{c: make(chan time.Time, 1), d: d}
	189	+	c.timers <- tm
	190	+	return tm
	191	+	}
	192	+
	193	+	func (c *manualClock) advance(d time.Duration) {
	194	+	c.mu.Lock()
	195	+	c.now = c.now.Add(d)
	196	+	now := c.now
	197	+	c.mu.Unlock()
	198	+	for {
	199	+	select {
	200	+	case tm := <-c.timers:
	201	+	tm.fire(now)
	202	+	default:
	203	+	return
	204	+	}
	205	+	}
	206	+	}
	207	+
	208	+	type manualTimer struct {
	209	+	c chan time.Time
	210	+	d time.Duration
	211	+	}
	212	+
	213	+	func (t *manualTimer) C() <-chan time.Time { return t.c }
	214	+	func (t *manualTimer) Reset(d time.Duration) bool {
	215	+	t.d = d
	216	+	return true
	217	+	}
	218	+	func (t *manualTimer) Stop() bool { return true }
	219	+	func (t *manualTimer) fire(now time.Time) {
	220	+	select {
	221	+	case t.c <- now:
	222	+	default:
	223	+	}
	224	+	}
	225	+
	226	+	func waitForManualTimers(t testing.T, clk manualClock, n int) {
	227	+	t.Helper()
	228	+	deadline := time.Now().Add(time.Second)
	229	+	for time.Now().Before(deadline) {
	230	+	if len(clk.timers) >= n {
	231	+	return
	232	+	}
	233	+	time.Sleep(time.Millisecond)
	234	+	}
	235	+	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
	236	+	}
	237	+
	238	+	type deadlineCapturePC struct {
	239	+	addr net.Addr
	240	+	deadline atomic.Value
	241	+	}
	242	+
	243	+	func (p *deadlineCapturePC) ReadFrom([]byte) (int, net.Addr, error) {
	244	+	return 0, nil, errors.New("closed")
	245	+	}
	246	+	func (p *deadlineCapturePC) WriteTo(b []byte, _ net.Addr) (int, error) { return len(b), nil }
	247	+	func (p *deadlineCapturePC) Close() error { return nil }
	248	+	func (p *deadlineCapturePC) LocalAddr() net.Addr { return p.addr }
	249	+	func (p *deadlineCapturePC) SetDeadline(time.Time) error { return nil }
	250	+	func (p *deadlineCapturePC) SetReadDeadline(time.Time) error { return nil }
	251	+	func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
	252	+	if !t.IsZero() {
	253	+	p.deadline.Store(t)
	254	+	}
	255	+	return nil
	256	+	}
	257	+
	258	+	func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
	259	+	pc := newBlockingWritePC()
	260	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
	261	+	WriteTimeout: 10 * time.Millisecond,
	262	+	ResendAfter: time.Hour,
	263	+	Heartbeat: time.Hour,
	264	+	})
	265	+	if err != nil {
	266	+	t.Fatal(err)
	267	+	}
	268	+	defer func() { _ = tr.Close() }()
	269	+
	270	+	done := make(chan error, 1)
	271	+	go func() { done <- tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("blocked")}) }()
	272	+	select {
	273	+	case err := <-done:
	274	+	if err == nil {
	275	+	t.Fatal("Send returned nil, want write timeout")
	276	+	}
	277	+	var timeout interface{ Timeout() bool }
	278	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
	279	+	t.Fatalf("Send err=%v, want timeout error", err)
	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	+	}
	287	+	case <-time.After(250 * time.Millisecond):
	288	+	t.Fatal("Send blocked without honoring WriteTimeout")
	289	+	}
	290	+	}
	291	+
85	292		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {
86	293		aPC, bPC := newPair()
87	294		var droppedAck atomic.Bool

@@ -342,12 +549,12 @@ func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
342	549		defer func() { _ = a.Close() }()
343	550		defer func() { _ = b.Close() }()
344	551
345		-	for i := 0; i < 100; i++ {
	552	+	for i := range 100 {
346	553		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
347	554		t.Fatal(err)
348	555		}
349	556		}
350		-	for i := 0; i < 100; i++ {
	557	+	for i := range 100 {
351	558		got := recvWithin(t, b, time.Second)
352	559		if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
353	560		t.Fatalf("frame %d delivered out of order: %+v", i, got)

@@ -362,7 +569,7 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
362	569		defer func() { _ = a.Close() }()
363	570		defer func() { _ = b.Close() }()
364	571
365		-	for i := 0; i < cap(b.in)+20; i++ {
	572	+	for i := range cap(b.in) + 20 {
366	573		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
367	574		t.Fatal(err)
368	575		}

@@ -383,6 +590,70 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
383	590		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)
384	591		}
385	592
	593	+	func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *atomic.Int32) {
	594	+	t.Helper()
	595	+	aPC, _ := newPair()
	596	+	codec, err := pdgram.NewCodec(key())
	597	+	if err != nil {
	598	+	t.Fatal(err)
	599	+	}
	600	+	var writes atomic.Int32
	601	+	aPC.drop = func(_ []byte, addr net.Addr) bool {
	602	+	if addr.String() == "b" {
	603	+	writes.Add(1)
	604	+	return true
	605	+	}
	606	+	return false
	607	+	}
	608	+	return &Transport{
	609	+	pc: aPC,
	610	+	peer: testAddr("b"),
	611	+	codec: codec,
	612	+	sendDir: 1,
	613	+	mtu: pdgram.DefaultMTU,
	614	+	pending: make(map[uint64]*pending),
	615	+	resendAfter: 100 * time.Millisecond,
	616	+	maxResendAfter: 500 * time.Millisecond,
	617	+	maxResendPerTick: maxResendPerTick,
	618	+	clock: realClock{},
	619	+	}, &writes
	620	+	}
	621	+
	622	+	func TestResendPendingCapsBurst(t *testing.T) {
	623	+	tr, writes := newResendTestTransport(t, 2)
	624	+	now := tr.clock.Now()
	625	+	for i := range 5 {
	626	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: now.Add(-time.Second)}
	627	+	}
	628	+
	629	+	tr.resendPending()
	630	+	if got := writes.Load(); got != 2 {
	631	+	t.Fatalf("resend writes=%d, want capped burst of 2", got)
	632	+	}
	633	+	}
	634	+
	635	+	func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
	636	+	tr, writes := newResendTestTransport(t, 10)
	637	+	now := tr.clock.Now()
	638	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * 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}
	640	+	tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}
	641	+
	642	+	tr.resendPending()
	643	+	if got := writes.Load(); got != 2 {
	644	+	t.Fatalf("resend writes=%d, want ready and capped frames only", got)
	645	+	}
	646	+	if attempts := tr.pending[1].attempts; attempts != 1 {
	647	+	t.Fatalf("too-soon attempts=%d, want unchanged", attempts)
	648	+	}
	649	+	if attempts := tr.pending[2].attempts; attempts != 2 {
	650	+	t.Fatalf("ready attempts=%d, want incremented", attempts)
	651	+	}
	652	+	if attempts := tr.pending[3].attempts; attempts != 11 {
	653	+	t.Fatalf("capped attempts=%d, want incremented", attempts)
	654	+	}
	655	+	}
	656	+
386	657		func TestProbeTimeoutAndSuccess(t *testing.T) {
387	658		aPC, bPC := newPair()
388	659		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)

@@ -421,34 +692,183 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
421	692		}
422	693		}
423	694
424		-	func TestPendingReliableQueueBounded(t *testing.T) {
	695	+	func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
425	696		aPC, bPC := newPair()
426	697		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
427		-	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
	698	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
428	699		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
429	700		defer func() { _ = a.Close() }()
430	701		defer func() { _ = b.Close() }()
431	702
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		-	}
	703	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	704	+	t.Fatal(err)
436	705		}
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)
	706	+	done := make(chan error, 1)
	707	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	708	+
	709	+	select {
	710	+	case err := <-done:
	711	+	t.Fatalf("send returned before an ack released pending capacity: %v", err)
	712	+	case <-time.After(25 * time.Millisecond):
	713	+	}
	714	+
	715	+	a.handleRecord(ackRecord(1))
	716	+	select {
	717	+	case err := <-done:
	718	+	if err != nil {
	719	+	t.Fatalf("send after ack: %v", err)
	720	+	}
	721	+	case <-time.After(time.Second):
	722	+	t.Fatal("send did not resume after ack released pending capacity")
439	723		}
440	724		a.mu.Lock()
441	725		pending := len(a.pending)
442	726		a.mu.Unlock()
443		-	if pending > maxPendingReliable {
444		-	t.Fatalf("pending=%d, want <= %d", pending, maxPendingReliable)
	727	+	if pending > 1 {
	728	+	t.Fatalf("pending=%d, want <= 1", pending)
	729	+	}
	730	+	}
	731	+
	732	+	func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
	733	+	aPC, bPC := newPair()
	734	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	735	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	736	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: 50 * time.Millisecond, Clock: clk})
	737	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	738	+	defer func() { _ = a.Close() }()
	739	+	defer func() { _ = b.Close() }()
	740	+
	741	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	742	+	t.Fatal(err)
	743	+	}
	744	+	done := make(chan error, 1)
	745	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	746	+	waitForManualTimers(t, clk, 3)
	747	+
	748	+	clk.advance(50 * time.Millisecond)
	749	+	select {
	750	+	case err := <-done:
	751	+	if !errors.Is(err, ErrPendingFull) {
	752	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
	753	+	}
	754	+	case <-time.After(time.Second):
	755	+	t.Fatal("send did not return after clock advanced past MaxPendingWait")
	756	+	}
	757	+	}
	758	+
	759	+	func TestPendingAttemptsResetWhenLinkRecovers(t *testing.T) {
	760	+	tr, _ := newResendTestTransport(t, 10)
	761	+	tr.linkState = ports.LinkStateDegraded
	762	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 4}
	763	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 7}
	764	+
	765	+	tr.setLinkState(ports.LinkStateConnected, nil)
	766	+
	767	+	for seq, p := range tr.pending {
	768	+	if p.attempts != 0 {
	769	+	t.Fatalf("pending[%d] attempts=%d, want reset on connected recovery", seq, p.attempts)
	770	+	}
	771	+	}
	772	+	}
	773	+
	774	+	func TestWritePacketDeadlineUsesTransportClock(t *testing.T) {
	775	+	wantNow := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
	776	+	pc := &deadlineCapturePC{addr: testAddr("a")}
	777	+	tr, err := NewTransportWithOptions(pc, testAddr("b"), key(), 1, 2, Options{
	778	+	Clock: fixedClock{now: wantNow},
	779	+	WriteTimeout: 250 * time.Millisecond,
	780	+	ResendAfter: time.Hour,
	781	+	Heartbeat: time.Hour,
	782	+	})
	783	+	if err != nil {
	784	+	t.Fatal(err)
	785	+	}
	786	+	defer func() { _ = tr.Close() }()
	787	+
	788	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("deadline")}); err != nil {
	789	+	t.Fatal(err)
	790	+	}
	791	+	got, _ := pc.deadline.Load().(time.Time)
	792	+	if want := wantNow.Add(250 * time.Millisecond); !got.Equal(want) {
	793	+	t.Fatalf("write deadline=%v, want transport clock deadline %v", got, want)
	794	+	}
	795	+	}
	796	+
	797	+	func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
	798	+	aPC, bPC := newPair()
	799	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
	800	+	if err != nil {
	801	+	t.Fatal(err)
445	802		}
	803	+	defer func() { _ = tr.Close() }()
	804	+	defer func() { _ = bPC.Close() }()
	805	+
	806	+	tr.writeMu.Lock()
	807	+	done := make(chan struct{})
	808	+	go func() {
	809	+	for id := uint64(1); id <= probeReplyBufferSize*4; id++ {
	810	+	tr.handleRecord(probeRecord(recProbe, id))
	811	+	}
	812	+	close(done)
	813	+	}()
	814	+
	815	+	select {
	816	+	case <-done:
	817	+	case <-time.After(time.Second):
	818	+	tr.writeMu.Unlock()
	819	+	<-done
	820	+	t.Fatal("probe reply burst blocked behind writeMu")
	821	+	}
	822	+	tr.writeMu.Unlock()
	823	+	}
	824	+
	825	+	func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
	826	+	aPC, bPC := newPair()
	827	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	828	+	clk := newManualClock(time.Now())
	829	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: time.Hour, Clock: clk})
	830	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	831	+	defer func() { _ = a.Close() }()
	832	+	defer func() { _ = b.Close() }()
	833	+
	834	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	835	+	t.Fatal(err)
	836	+	}
	837	+	done := make(chan error, 2)
	838	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	839	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("third")}) }()
	840	+	waitForManualTimers(t, clk, 2)
	841	+	a.setLinkState(ports.LinkStateDegraded, nil)
	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)
	850	+	}
	851	+	}
	852	+	}
	853	+
	854	+	func ackRecord(seq uint64) []byte {
	855	+	var b [9]byte
	856	+	b[0] = recAck
	857	+	binary.BigEndian.PutUint64(b[1:], seq)
	858	+	return b[:]
	859	+	}
	860	+
	861	+	func probeRecord(kind byte, id uint64) []byte {
	862	+	var b [9]byte
	863	+	b[0] = kind
	864	+	binary.BigEndian.PutUint64(b[1:], id)
	865	+	return b[:]
446	866		}
447	867
448	868		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
449	869		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
450	870		tr.deliverCond = sync.NewCond(&tr.deliverMu)
451		-	for i := 0; i < maxRecvBuffer+100; i++ {
	871	+	for i := range maxRecvBuffer + 100 {
452	872		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
453	873		}
454	874		if got := len(tr.recvBuf); got != maxRecvBuffer {

@@ -474,7 +894,7 @@ func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
474	894		defer func() { _ = aPC.Close() }()
475	895		defer func() { _ = bPC.Close() }()
476	896
477		-	for i := 0; i < maxRecvBuffer; i++ {
	897	+	for i := range maxRecvBuffer {
478	898		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
479	899		}
480	900		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,212 @@ 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	+	type fixedClock struct{ now time.Time }
			165	+
			166	+	func (c fixedClock) Now() time.Time { return c.now }
			167	+	func (c fixedClock) NewTimer(d time.Duration) ports.Timer {
			168	+	return realTimer{t: time.NewTimer(d)}
			169	+	}
			170	+
			171	+	type manualClock struct {
			172	+	mu sync.Mutex
			173	+	now time.Time
			174	+	timers chan *manualTimer
			175	+	}
			176	+
			177	+	func newManualClock(now time.Time) *manualClock {
			178	+	return &manualClock{now: now, timers: make(chan *manualTimer, 16)}
			179	+	}
			180	+
			181	+	func (c *manualClock) Now() time.Time {
			182	+	c.mu.Lock()
			183	+	defer c.mu.Unlock()
			184	+	return c.now
			185	+	}
			186	+
			187	+	func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
			188	+	tm := &manualTimer{c: make(chan time.Time, 1), d: d}
			189	+	c.timers <- tm
			190	+	return tm
			191	+	}
			192	+
			193	+	func (c *manualClock) advance(d time.Duration) {
			194	+	c.mu.Lock()
			195	+	c.now = c.now.Add(d)
			196	+	now := c.now
			197	+	c.mu.Unlock()
			198	+	for {
			199	+	select {
			200	+	case tm := <-c.timers:
			201	+	tm.fire(now)
			202	+	default:
			203	+	return
			204	+	}
			205	+	}
			206	+	}
			207	+
			208	+	type manualTimer struct {
			209	+	c chan time.Time
			210	+	d time.Duration
			211	+	}
			212	+
			213	+	func (t *manualTimer) C() <-chan time.Time { return t.c }
			214	+	func (t *manualTimer) Reset(d time.Duration) bool {
			215	+	t.d = d
			216	+	return true
			217	+	}
			218	+	func (t *manualTimer) Stop() bool { return true }
			219	+	func (t *manualTimer) fire(now time.Time) {
			220	+	select {
			221	+	case t.c <- now:
			222	+	default:
			223	+	}
			224	+	}
			225	+
			226	+	func waitForManualTimers(t testing.T, clk manualClock, n int) {
			227	+	t.Helper()
			228	+	deadline := time.Now().Add(time.Second)
			229	+	for time.Now().Before(deadline) {
			230	+	if len(clk.timers) >= n {
			231	+	return
			232	+	}
			233	+	time.Sleep(time.Millisecond)
			234	+	}
			235	+	t.Fatalf("manual timers=%d, want at least %d", len(clk.timers), n)
			236	+	}
			237	+
			238	+	type deadlineCapturePC struct {
			239	+	addr net.Addr
			240	+	deadline atomic.Value
			241	+	}
			242	+
			243	+	func (p *deadlineCapturePC) ReadFrom([]byte) (int, net.Addr, error) {
			244	+	return 0, nil, errors.New("closed")
			245	+	}
			246	+	func (p *deadlineCapturePC) WriteTo(b []byte, _ net.Addr) (int, error) { return len(b), nil }
			247	+	func (p *deadlineCapturePC) Close() error { return nil }
			248	+	func (p *deadlineCapturePC) LocalAddr() net.Addr { return p.addr }
			249	+	func (p *deadlineCapturePC) SetDeadline(time.Time) error { return nil }
			250	+	func (p *deadlineCapturePC) SetReadDeadline(time.Time) error { return nil }
			251	+	func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
			252	+	if !t.IsZero() {
			253	+	p.deadline.Store(t)
			254	+	}
			255	+	return nil
			256	+	}
			257	+
			258	+	func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
			259	+	pc := newBlockingWritePC()
			260	+	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
			261	+	WriteTimeout: 10 * time.Millisecond,
			262	+	ResendAfter: time.Hour,
			263	+	Heartbeat: time.Hour,
			264	+	})
			265	+	if err != nil {
			266	+	t.Fatal(err)
			267	+	}
			268	+	defer func() { _ = tr.Close() }()
			269	+
			270	+	done := make(chan error, 1)
			271	+	go func() { done <- tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("blocked")}) }()
			272	+	select {
			273	+	case err := <-done:
			274	+	if err == nil {
			275	+	t.Fatal("Send returned nil, want write timeout")
			276	+	}
			277	+	var timeout interface{ Timeout() bool }
			278	+	if !errors.As(err, &timeout) \|\| !timeout.Timeout() {
			279	+	t.Fatalf("Send err=%v, want timeout error", err)
			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	+	}
			287	+	case <-time.After(250 * time.Millisecond):
			288	+	t.Fatal("Send blocked without honoring WriteTimeout")
			289	+	}
			290	+	}
			291	+
85		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {	292		func TestReliableDuplicateAckedButNotDeliveredTwice(t *testing.T) {
86		aPC, bPC := newPair()	293		aPC, bPC := newPair()
87		var droppedAck atomic.Bool	294		var droppedAck atomic.Bool

@@ -342,12 +549,12 @@ func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
342		defer func() { _ = a.Close() }()	549		defer func() { _ = a.Close() }()
343		defer func() { _ = b.Close() }()	550		defer func() { _ = b.Close() }()
344			551
345	-	for i := 0; i < 100; i++ {	552	+	for i := range 100 {
346		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {	553		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
347		t.Fatal(err)	554		t.Fatal(err)
348		}	555		}
349		}	556		}
350	-	for i := 0; i < 100; i++ {	557	+	for i := range 100 {
351		got := recvWithin(t, b, time.Second)	558		got := recvWithin(t, b, time.Second)
352		if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {	559		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)	560		t.Fatalf("frame %d delivered out of order: %+v", i, got)

@@ -362,7 +569,7 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
362		defer func() { _ = a.Close() }()	569		defer func() { _ = a.Close() }()
363		defer func() { _ = b.Close() }()	570		defer func() { _ = b.Close() }()
364			571
365	-	for i := 0; i < cap(b.in)+20; i++ {	572	+	for i := range cap(b.in) + 20 {
366		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {	573		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
367		t.Fatal(err)	574		t.Fatal(err)
368		}	575		}

@@ -383,6 +590,70 @@ func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
383		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)	590		t.Fatalf("acks blocked behind undrained consumer; pending=%d", pending)
384		}	591		}
385			592
			593	+	func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *atomic.Int32) {
			594	+	t.Helper()
			595	+	aPC, _ := newPair()
			596	+	codec, err := pdgram.NewCodec(key())
			597	+	if err != nil {
			598	+	t.Fatal(err)
			599	+	}
			600	+	var writes atomic.Int32
			601	+	aPC.drop = func(_ []byte, addr net.Addr) bool {
			602	+	if addr.String() == "b" {
			603	+	writes.Add(1)
			604	+	return true
			605	+	}
			606	+	return false
			607	+	}
			608	+	return &Transport{
			609	+	pc: aPC,
			610	+	peer: testAddr("b"),
			611	+	codec: codec,
			612	+	sendDir: 1,
			613	+	mtu: pdgram.DefaultMTU,
			614	+	pending: make(map[uint64]*pending),
			615	+	resendAfter: 100 * time.Millisecond,
			616	+	maxResendAfter: 500 * time.Millisecond,
			617	+	maxResendPerTick: maxResendPerTick,
			618	+	clock: realClock{},
			619	+	}, &writes
			620	+	}
			621	+
			622	+	func TestResendPendingCapsBurst(t *testing.T) {
			623	+	tr, writes := newResendTestTransport(t, 2)
			624	+	now := tr.clock.Now()
			625	+	for i := range 5 {
			626	+	tr.pending[uint64(i+1)] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, last: now.Add(-time.Second)}
			627	+	}
			628	+
			629	+	tr.resendPending()
			630	+	if got := writes.Load(); got != 2 {
			631	+	t.Fatalf("resend writes=%d, want capped burst of 2", got)
			632	+	}
			633	+	}
			634	+
			635	+	func TestResendPendingUsesExponentialBackoffAndMaxDelay(t *testing.T) {
			636	+	tr, writes := newResendTestTransport(t, 10)
			637	+	now := tr.clock.Now()
			638	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("too-soon")}, last: now.Add(-150 * 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}
			640	+	tr.pending[3] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("capped")}, last: now.Add(-time.Second), attempts: 10}
			641	+
			642	+	tr.resendPending()
			643	+	if got := writes.Load(); got != 2 {
			644	+	t.Fatalf("resend writes=%d, want ready and capped frames only", got)
			645	+	}
			646	+	if attempts := tr.pending[1].attempts; attempts != 1 {
			647	+	t.Fatalf("too-soon attempts=%d, want unchanged", attempts)
			648	+	}
			649	+	if attempts := tr.pending[2].attempts; attempts != 2 {
			650	+	t.Fatalf("ready attempts=%d, want incremented", attempts)
			651	+	}
			652	+	if attempts := tr.pending[3].attempts; attempts != 11 {
			653	+	t.Fatalf("capped attempts=%d, want incremented", attempts)
			654	+	}
			655	+	}
			656	+
386		func TestProbeTimeoutAndSuccess(t *testing.T) {	657		func TestProbeTimeoutAndSuccess(t *testing.T) {
387		aPC, bPC := newPair()	658		aPC, bPC := newPair()
388		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)	659		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)

@@ -421,34 +692,183 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
421		}	692		}
422		}	693		}
423			694
424	-	func TestPendingReliableQueueBounded(t *testing.T) {	695	+	func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
425		aPC, bPC := newPair()	696		aPC, bPC := newPair()
426		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	697		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
427	-	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)	698	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
428		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)	699		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
429		defer func() { _ = a.Close() }()	700		defer func() { _ = a.Close() }()
430		defer func() { _ = b.Close() }()	701		defer func() { _ = b.Close() }()
431			702
432	-	for i := 0; i < maxPendingReliable; i++ {	703	+	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 {	704	+	t.Fatal(err)
434	-	t.Fatalf("send %d: %v", i, err)
435	-	}
436		}	705		}
437	-	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("full")}); !errors.Is(err, ErrPendingFull) {	706	+	done := make(chan error, 1)
438	-	t.Fatalf("send past bound err=%v, want ErrPendingFull", err)	707	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
			708	+
			709	+	select {
			710	+	case err := <-done:
			711	+	t.Fatalf("send returned before an ack released pending capacity: %v", err)
			712	+	case <-time.After(25 * time.Millisecond):
			713	+	}
			714	+
			715	+	a.handleRecord(ackRecord(1))
			716	+	select {
			717	+	case err := <-done:
			718	+	if err != nil {
			719	+	t.Fatalf("send after ack: %v", err)
			720	+	}
			721	+	case <-time.After(time.Second):
			722	+	t.Fatal("send did not resume after ack released pending capacity")
439		}	723		}
440		a.mu.Lock()	724		a.mu.Lock()
441		pending := len(a.pending)	725		pending := len(a.pending)
442		a.mu.Unlock()	726		a.mu.Unlock()
443	-	if pending > maxPendingReliable {	727	+	if pending > 1 {
444	-	t.Fatalf("pending=%d, want <= %d", pending, maxPendingReliable)	728	+	t.Fatalf("pending=%d, want <= 1", pending)
			729	+	}
			730	+	}
			731	+
			732	+	func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
			733	+	aPC, bPC := newPair()
			734	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
			735	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			736	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: 50 * time.Millisecond, Clock: clk})
			737	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
			738	+	defer func() { _ = a.Close() }()
			739	+	defer func() { _ = b.Close() }()
			740	+
			741	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
			742	+	t.Fatal(err)
			743	+	}
			744	+	done := make(chan error, 1)
			745	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
			746	+	waitForManualTimers(t, clk, 3)
			747	+
			748	+	clk.advance(50 * time.Millisecond)
			749	+	select {
			750	+	case err := <-done:
			751	+	if !errors.Is(err, ErrPendingFull) {
			752	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
			753	+	}
			754	+	case <-time.After(time.Second):
			755	+	t.Fatal("send did not return after clock advanced past MaxPendingWait")
			756	+	}
			757	+	}
			758	+
			759	+	func TestPendingAttemptsResetWhenLinkRecovers(t *testing.T) {
			760	+	tr, _ := newResendTestTransport(t, 10)
			761	+	tr.linkState = ports.LinkStateDegraded
			762	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 4}
			763	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 7}
			764	+
			765	+	tr.setLinkState(ports.LinkStateConnected, nil)
			766	+
			767	+	for seq, p := range tr.pending {
			768	+	if p.attempts != 0 {
			769	+	t.Fatalf("pending[%d] attempts=%d, want reset on connected recovery", seq, p.attempts)
			770	+	}
			771	+	}
			772	+	}
			773	+
			774	+	func TestWritePacketDeadlineUsesTransportClock(t *testing.T) {
			775	+	wantNow := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
			776	+	pc := &deadlineCapturePC{addr: testAddr("a")}
			777	+	tr, err := NewTransportWithOptions(pc, testAddr("b"), key(), 1, 2, Options{
			778	+	Clock: fixedClock{now: wantNow},
			779	+	WriteTimeout: 250 * time.Millisecond,
			780	+	ResendAfter: time.Hour,
			781	+	Heartbeat: time.Hour,
			782	+	})
			783	+	if err != nil {
			784	+	t.Fatal(err)
			785	+	}
			786	+	defer func() { _ = tr.Close() }()
			787	+
			788	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("deadline")}); err != nil {
			789	+	t.Fatal(err)
			790	+	}
			791	+	got, _ := pc.deadline.Load().(time.Time)
			792	+	if want := wantNow.Add(250 * time.Millisecond); !got.Equal(want) {
			793	+	t.Fatalf("write deadline=%v, want transport clock deadline %v", got, want)
			794	+	}
			795	+	}
			796	+
			797	+	func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
			798	+	aPC, bPC := newPair()
			799	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
			800	+	if err != nil {
			801	+	t.Fatal(err)
445		}	802		}
			803	+	defer func() { _ = tr.Close() }()
			804	+	defer func() { _ = bPC.Close() }()
			805	+
			806	+	tr.writeMu.Lock()
			807	+	done := make(chan struct{})
			808	+	go func() {
			809	+	for id := uint64(1); id <= probeReplyBufferSize*4; id++ {
			810	+	tr.handleRecord(probeRecord(recProbe, id))
			811	+	}
			812	+	close(done)
			813	+	}()
			814	+
			815	+	select {
			816	+	case <-done:
			817	+	case <-time.After(time.Second):
			818	+	tr.writeMu.Unlock()
			819	+	<-done
			820	+	t.Fatal("probe reply burst blocked behind writeMu")
			821	+	}
			822	+	tr.writeMu.Unlock()
			823	+	}
			824	+
			825	+	func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
			826	+	aPC, bPC := newPair()
			827	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
			828	+	clk := newManualClock(time.Now())
			829	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: time.Hour, Clock: clk})
			830	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
			831	+	defer func() { _ = a.Close() }()
			832	+	defer func() { _ = b.Close() }()
			833	+
			834	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
			835	+	t.Fatal(err)
			836	+	}
			837	+	done := make(chan error, 2)
			838	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
			839	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("third")}) }()
			840	+	waitForManualTimers(t, clk, 2)
			841	+	a.setLinkState(ports.LinkStateDegraded, nil)
			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)
			850	+	}
			851	+	}
			852	+	}
			853	+
			854	+	func ackRecord(seq uint64) []byte {
			855	+	var b [9]byte
			856	+	b[0] = recAck
			857	+	binary.BigEndian.PutUint64(b[1:], seq)
			858	+	return b[:]
			859	+	}
			860	+
			861	+	func probeRecord(kind byte, id uint64) []byte {
			862	+	var b [9]byte
			863	+	b[0] = kind
			864	+	binary.BigEndian.PutUint64(b[1:], id)
			865	+	return b[:]
446		}	866		}
447			867
448		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {	868		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
449		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}	869		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
450		tr.deliverCond = sync.NewCond(&tr.deliverMu)	870		tr.deliverCond = sync.NewCond(&tr.deliverMu)
451	-	for i := 0; i < maxRecvBuffer+100; i++ {	871	+	for i := range maxRecvBuffer + 100 {
452		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})	872		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
453		}	873		}
454		if got := len(tr.recvBuf); got != maxRecvBuffer {	874		if got := len(tr.recvBuf); got != maxRecvBuffer {

@@ -474,7 +894,7 @@ func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
474		defer func() { _ = aPC.Close() }()	894		defer func() { _ = aPC.Close() }()
475		defer func() { _ = bPC.Close() }()	895		defer func() { _ = bPC.Close() }()
476			896
477	-	for i := 0; i < maxRecvBuffer; i++ {	897	+	for i := range maxRecvBuffer {
478		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}	898		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
479		}	899		}
480		tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))	900		tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))

M internal/app/run.go +3 -1


@@ -730,7 +730,7 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
730	730		return err
731	731		}
732	732		defer func() { _ = daemonTr.Close() }()
733		-	dg, err := dgram.NewTransport(conn, nil, key, 2, 1)
	733	+	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpProxyClientTransportOptions)
734	734		if err != nil {
735	735		return err
736	736		}

@@ -758,6 +758,8 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
758	758		return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log}.Run(ctx)
759	759		}
760	760
	761	+	var udpProxyClientTransportOptions = dgram.Options{MaxPending: 32}
	762	+
761	763		const (
762	764		// envUDPPortRange configures the remote UDP proxy's listen port range so a
763	765		// host firewall can allow a known range instead of a random ephemeral port.

 		return err
 	}
 	defer func() { _ = daemonTr.Close() }()
-	dg, err := dgram.NewTransport(conn, nil, key, 2, 1)
+	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpProxyClientTransportOptions)
 	if err != nil {
 		return err
 	}
 	return dgram.ProxyRuntime{Client: dgramTr, Daemon: daemonTr, Log: log}.Run(ctx)
 }
+var udpProxyClientTransportOptions = dgram.Options{MaxPending: 32}
 const (
 	// envUDPPortRange configures the remote UDP proxy's listen port range so a
 	// host firewall can allow a known range instead of a random ephemeral port.

M internal/app/run_test.go +4 -0


@@ -25,6 +25,10 @@ import (
25	25		"github.com/stretchr/testify/require"
26	26		)
27	27
	28	+	func TestRunUDPProxyUsesBoundedClientMaxPending(t *testing.T) {
	29	+	require.Equal(t, 32, udpProxyClientTransportOptions.MaxPending)
	30	+	}
	31	+
28	32		func TestParseArgs(t *testing.T) {
29	33		tests := []struct {
30	34		name string


@@ -25,6 +25,10 @@ import (
25	"github.com/stretchr/testify/require"	25		"github.com/stretchr/testify/require"
26	)	26		)
27		27
		28	+	func TestRunUDPProxyUsesBoundedClientMaxPending(t *testing.T) {
		29	+	require.Equal(t, 32, udpProxyClientTransportOptions.MaxPending)
		30	+	}
		31	+
28	func TestParseArgs(t *testing.T) {	32		func TestParseArgs(t *testing.T) {
29	tests := []struct {	33		tests := []struct {
30	name string	34		name string

M internal/usecase/daemon/client.go +3 -11


@@ -202,18 +202,10 @@ func (ac *attachedClient) nextOutputFrameLocked(b []byte, reset bool) ports.Fram
202	202		return frameOutputState(b, baseStateNum, ac.nextStateNum, ac.echoAck.Load())
203	203		}
204	204
205		-	// shouldResetOutputState reads nextStateNum and must only be called while
206		-	// sendMu is held, matching nextOutputFrameLocked.
	205	+	// shouldResetOutputState honors explicit invalidations; reliable transport ack
	206	+	// lag alone must not force a full output-state repaint.
207	207		func (ac *attachedClient) shouldResetOutputState(reset bool) bool {
208		-	if reset {
209		-	return true
210		-	}
211		-	latest := ac.nextStateNum
212		-	if latest == 0 {
213		-	return false
214		-	}
215		-	acked := ac.outputAck.Load()
216		-	return acked+1 < latest
	208	+	return reset
217	209		}
218	210
219	211		func (ac *attachedClient) advanceOutputAck(state uint64) {

 	return frameOutputState(b, baseStateNum, ac.nextStateNum, ac.echoAck.Load())
 }
-// shouldResetOutputState reads nextStateNum and must only be called while
+// shouldResetOutputState honors explicit invalidations; reliable transport ack
-// sendMu is held, matching nextOutputFrameLocked.
+// lag alone must not force a full output-state repaint.
 func (ac *attachedClient) shouldResetOutputState(reset bool) bool {
-	if reset {
+	return reset
-		return true
-	}
-	latest := ac.nextStateNum
-	if latest == 0 {
-		return false
-	}
-	acked := ac.outputAck.Load()
-	return acked+1 < latest
 }
 func (ac *attachedClient) advanceOutputAck(state uint64) {

M internal/usecase/daemon/resume_test.go +11 -11


@@ -126,35 +126,35 @@ func TestResumeRebindsRotatesAndDoesNotOpenPTY(t *testing.T) {
126	126		require.Same(t, resumedAC, sess.client)
127	127		}
128	128
129		-	func TestOutputAckGapForcesFullStateRepaint(t *testing.T) {
	129	+	func TestOutputAckLagAloneDoesNotForceFullStateRepaint(t *testing.T) {
130	130		ac := &attachedClient{}
131	131		ac.nextStateNum = 3
132	132		ac.advanceOutputAck(3)
133	133		ac.advanceOutputAck(2)
134	134		ac.advanceOutputAck(4)
135	135		require.Equal(t, uint64(3), ac.outputAck.Load(), "stale or future ACKs must not move output state incorrectly")
	136	+
136	137		ac.sendMu.Lock()
137	138		ac.nextStateNum = 5
138	139		reset := ac.shouldResetOutputState(false)
139		-	require.True(t, reset, "client ack gap should force dependency-free full repaint")
	140	+	require.False(t, reset, "reliable output ack lag alone must not force dependency-free full repaint")
140	141
141		-	f := ac.nextOutputFrameLocked([]byte("full repaint"), reset)
	142	+	f := ac.nextOutputFrameLocked([]byte("incremental while reliable backlog drains"), reset)
142	143		ac.sendMu.Unlock()
143	144		out, err := ports.UnmarshalOutput(f.Payload)
144	145		require.NoError(t, err)
145		-	require.Equal(t, uint64(0), out.BaseStateNum, "full repaint must not depend on dropped prior output states")
	146	+	require.Equal(t, uint64(5), out.BaseStateNum, "output should remain incremental unless an explicit reset is requested")
146	147		require.Equal(t, uint64(6), out.NewStateNum)
147	148
148		-	ac.outputAck.Store(out.NewStateNum)
149	149		ac.sendMu.Lock()
150		-	reset = ac.shouldResetOutputState(false)
151		-	require.False(t, reset, "acked current state can resume incremental output")
152		-	incremental := ac.nextOutputFrameLocked([]byte("incremental"), reset)
	150	+	reset = ac.shouldResetOutputState(true)
	151	+	require.True(t, reset, "explicit reset should still force full repaint")
	152	+	full := ac.nextOutputFrameLocked([]byte("explicit full repaint"), reset)
153	153		ac.sendMu.Unlock()
154		-	incrementalOut, err := ports.UnmarshalOutput(incremental.Payload)
	154	+	fullOut, err := ports.UnmarshalOutput(full.Payload)
155	155		require.NoError(t, err)
156		-	require.Equal(t, uint64(6), incrementalOut.BaseStateNum)
157		-	require.Equal(t, uint64(7), incrementalOut.NewStateNum)
	156	+	require.Equal(t, uint64(0), fullOut.BaseStateNum)
	157	+	require.Equal(t, uint64(7), fullOut.NewStateNum)
158	158		}
159	159
160	160		func TestResumeClientIDMismatchDoesNotConsumeParkedToken(t *testing.T) {

 	require.Same(t, resumedAC, sess.client)
 }
-func TestOutputAckGapForcesFullStateRepaint(t *testing.T) {
+func TestOutputAckLagAloneDoesNotForceFullStateRepaint(t *testing.T) {
 	ac := &attachedClient{}
 	ac.nextStateNum = 3
 	ac.advanceOutputAck(3)
 	ac.advanceOutputAck(2)
 	ac.advanceOutputAck(4)
 	require.Equal(t, uint64(3), ac.outputAck.Load(), "stale or future ACKs must not move output state incorrectly")
 	ac.sendMu.Lock()
 	ac.nextStateNum = 5
 	reset := ac.shouldResetOutputState(false)
-	require.True(t, reset, "client ack gap should force dependency-free full repaint")
+	require.False(t, reset, "reliable output ack lag alone must not force dependency-free full repaint")
-	f := ac.nextOutputFrameLocked([]byte("full repaint"), reset)
+	f := ac.nextOutputFrameLocked([]byte("incremental while reliable backlog drains"), reset)
 	ac.sendMu.Unlock()
 	out, err := ports.UnmarshalOutput(f.Payload)
 	require.NoError(t, err)
-	require.Equal(t, uint64(0), out.BaseStateNum, "full repaint must not depend on dropped prior output states")
+	require.Equal(t, uint64(5), out.BaseStateNum, "output should remain incremental unless an explicit reset is requested")
 	require.Equal(t, uint64(6), out.NewStateNum)
-	ac.outputAck.Store(out.NewStateNum)
 	ac.sendMu.Lock()
-	reset = ac.shouldResetOutputState(false)
+	reset = ac.shouldResetOutputState(true)
-	require.False(t, reset, "acked current state can resume incremental output")
+	require.True(t, reset, "explicit reset should still force full repaint")
-	incremental := ac.nextOutputFrameLocked([]byte("incremental"), reset)
+	full := ac.nextOutputFrameLocked([]byte("explicit full repaint"), reset)
 	ac.sendMu.Unlock()
-	incrementalOut, err := ports.UnmarshalOutput(incremental.Payload)
+	fullOut, err := ports.UnmarshalOutput(full.Payload)
 	require.NoError(t, err)
-	require.Equal(t, uint64(6), incrementalOut.BaseStateNum)
+	require.Equal(t, uint64(0), fullOut.BaseStateNum)
-	require.Equal(t, uint64(7), incrementalOut.NewStateNum)
+	require.Equal(t, uint64(7), fullOut.NewStateNum)
 }
 func TestResumeClientIDMismatchDoesNotConsumeParkedToken(t *testing.T) {