3b58b7b fix(remote): reduce UDP output backlog · vev-vt

M internal/adapters/dgram/proxy.go +2 -0


@@ -207,6 +207,8 @@ func recoverableClientLink(t ports.Transport, err error) bool {
207	207		return false
208	208		}
209	209		switch reporter.LinkState() {
	210	+	case ports.LinkStateConnected:
	211	+	return errors.Is(err, ErrPendingFull)
210	212		case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
211	213		return true
212	214		default:


@@ -207,6 +207,8 @@ func recoverableClientLink(t ports.Transport, err error) bool {
207	return false	207		return false
208	}	208		}
209	switch reporter.LinkState() {	209		switch reporter.LinkState() {
		210	+	case ports.LinkStateConnected:
		211	+	return errors.Is(err, ErrPendingFull)
210	case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:	212		case ports.LinkStateDegraded, ports.LinkStateProbing, ports.LinkStateOffline:
211	return true	213		return true
212	default:	214		default:

M internal/adapters/dgram/proxy_test.go +39 -25


@@ -116,32 +116,46 @@ func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T)
116	116		}
117	117
118	118		func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
119		-	client := newFakeTransport()
120		-	daemon := newFakeTransport()
121		-	client.linkState = ports.LinkStateOffline
122		-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
123		-	errCh := make(chan error, 1)
124		-	go func() {
125		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
126		-	}()
127		-	client.sendErr <- ErrPendingFull
128		-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
129		-	select {
130		-	case got := <-client.sent:
131		-	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "preserved during outage" {
132		-	t.Fatalf("client got %+v, want retried output", got)
133		-	}
134		-	case <-time.After(time.Second):
135		-	t.Fatal("timeout waiting for recovered daemon output")
	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},
136	126		}
137		-	daemon.recv <- recvResult{err: io.EOF}
138		-	select {
139		-	case err := <-errCh:
140		-	if err != nil {
141		-	t.Fatalf("Run err=%v, want nil", err)
142		-	}
143		-	case <-time.After(time.Second):
144		-	t.Fatal("timeout waiting for daemon EOF")
	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	+	})
145	159		}
146	160		}
147	161

 }
 func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
-	client := newFakeTransport()
+	tests := []struct {
-	daemon := newFakeTransport()
+		name      string
-	client.linkState = ports.LinkStateOffline
+		state     ports.LinkState
-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
+		linkEvent bool
-	errCh := make(chan error, 1)
+	}{
-	go func() {
+		{name: "connected pending full", state: ports.LinkStateConnected},
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
+		{name: "offline pending full", state: ports.LinkStateOffline, linkEvent: true},
-	}()
-	client.sendErr <- ErrPendingFull
-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
-	select {
-	case got := <-client.sent:
-		if got.Type != ports.MsgOutput || string(got.Payload) != "preserved during outage" {
-			t.Fatalf("client got %+v, want retried output", got)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for recovered daemon output")
 	}
-	daemon.recv <- recvResult{err: io.EOF}
+	for _, tt := range tests {
-	select {
+		t.Run(tt.name, func(t *testing.T) {
-	case err := <-errCh:
+			client := newFakeTransport()
-		if err != nil {
+			daemon := newFakeTransport()
-			t.Fatalf("Run err=%v, want nil", err)
+			client.linkState = tt.state
-		}
+			if tt.linkEvent {
-	case <-time.After(time.Second):
+				client.linkEvents <- ports.LinkEvent{State: tt.state}
-		t.Fatal("timeout waiting for daemon EOF")
+			}
+			errCh := make(chan error, 1)
+			go func() {
+				errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
+			}()
+			client.sendErr <- ErrPendingFull
+			daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
+			select {
+			case got := <-client.sent:
+				if got.Type != ports.MsgOutput || string(got.Payload) != "preserved during outage" {
+					t.Fatalf("client got %+v, want retried output", got)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("timeout waiting for recovered daemon output")
+			}
+			daemon.recv <- recvResult{err: io.EOF}
+			select {
+			case err := <-errCh:
+				if err != nil {
+					t.Fatalf("Run err=%v, want nil", err)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("timeout waiting for daemon EOF")
+			}
+		})
 	}
 }

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


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

@@ -31,9 +31,11 @@ const (
31	31		defaultOffline = 30 * time.Second
32	32		defaultDead = 60 * time.Second
33	33		defaultMaxPending = 1024
	34	+	defaultMaxPendingWait = 50 * time.Millisecond
34	35		defaultMaxRecvBuf = 1024
35	36		maxRecvBuffer = defaultMaxRecvBuf
36	37		linkEventBufferSize = 16
	38	+	probeReplyBufferSize = 16
37	39		)
38	40
39	41		var (

@@ -53,6 +55,7 @@ type Options struct {
53	55		OfflineAfter time.Duration
54	56		DeadAfter time.Duration
55	57		MaxPending int
	58	+	MaxPendingWait time.Duration
56	59		MaxRecvBuffer int
57	60		Clock ports.Clock
58	61		// RebindPacketConn is an optional client-side hook used to hop the local UDP

@@ -110,6 +113,9 @@ func normalizeOptions(opts Options) Options {
110	113		if opts.MaxPending <= 0 {
111	114		opts.MaxPending = defaultMaxPending
112	115		}
	116	+	if opts.MaxPendingWait <= 0 {
	117	+	opts.MaxPendingWait = defaultMaxPendingWait
	118	+	}
113	119		if opts.MaxRecvBuffer <= 0 {
114	120		opts.MaxRecvBuffer = defaultMaxRecvBuf
115	121		}

@@ -144,12 +150,14 @@ type Transport struct {
144	150		offlineAfter time.Duration
145	151		deadAfter time.Duration
146	152		maxPending int
	153	+	maxPendingWait time.Duration
147	154		maxRecvBuffer int
148	155		clock ports.Clock
149	156		linkState ports.LinkState
150	157		linkEvents chan ports.LinkEvent
151	158		probeWait map[uint64]chan struct{}
152		-	sendCond *sync.Cond
	159	+	sendWake chan struct{}
	160	+	probeReply chan uint64
153	161		rebind func(net.PacketConn) (net.PacketConn, error)
154	162		hoppedOffline bool
155	163

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

@@ -197,12 +207,29 @@ func (t *Transport) Send(f ports.Frame) error {
197	207		// dropping MsgOutput can permanently desynchronize the client screen.
198	208		reliable := true
199	209		t.mu.Lock()
	210	+	var pendingWaitStarted time.Time
200	211		for reliable && len(t.pending) >= t.maxPending && !t.closed {
201	212		if t.linkState != ports.LinkStateConnected {
202	213		t.mu.Unlock()
203	214		return ErrPendingFull
204	215		}
205		-	t.sendCond.Wait()
	216	+	if pendingWaitStarted.IsZero() {
	217	+	pendingWaitStarted = t.clock.Now()
	218	+	}
	219	+	remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
	220	+	if remaining <= 0 {
	221	+	t.mu.Unlock()
	222	+	return ErrPendingFull
	223	+	}
	224	+	timer := t.clock.NewTimer(remaining)
	225	+	t.mu.Unlock()
	226	+	select {
	227	+	case <-timer.C():
	228	+	case <-t.sendWake:
	229	+	case <-t.done:
	230	+	}
	231	+	timer.Stop()
	232	+	t.mu.Lock()
206	233		}
207	234		if t.closed {
208	235		err := t.closeErr

@@ -360,7 +387,7 @@ func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
360	387		t.mu.Unlock()
361	388
362	389		if writeTimeout > 0 {
363		-	if err := pc.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
	390	+	if err := pc.SetWriteDeadline(t.clock.Now().Add(writeTimeout)); err != nil {
364	391		return err
365	392		}
366	393		defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()

@@ -428,16 +455,17 @@ func (t *Transport) handleRecord(p []byte) {
428	455		t.mu.Lock()
429	456		if _, ok := t.pending[seq]; ok {
430	457		delete(t.pending, seq)
431		-	if t.sendCond != nil {
432		-	t.sendCond.Signal()
433		-	}
	458	+	t.notifySendWaiter()
434	459		}
435	460		t.mu.Unlock()
436	461		case recProbe:
437	462		if len(p) != 9 {
438	463		return
439	464		}
440		-	_ = t.sendProbe(recPong, binary.BigEndian.Uint64(p[1:]))
	465	+	select {
	466	+	case t.probeReply <- binary.BigEndian.Uint64(p[1:]):
	467	+	default:
	468	+	}
441	469		case recPong:
442	470		if len(p) != 9 {
443	471		return

@@ -523,6 +551,24 @@ func (t *Transport) deliver(f ports.Frame) {
523	551		}
524	552		}
525	553
	554	+	func (t *Transport) probeReplyLoop() {
	555	+	for {
	556	+	select {
	557	+	case id := <-t.probeReply:
	558	+	_ = t.sendProbe(recPong, id)
	559	+	case <-t.done:
	560	+	return
	561	+	}
	562	+	}
	563	+	}
	564	+
	565	+	func (t *Transport) notifySendWaiter() {
	566	+	select {
	567	+	case t.sendWake <- struct{}{}:
	568	+	default:
	569	+	}
	570	+	}
	571	+
526	572		func (t *Transport) resendLoop() {
527	573		resend := t.clock.NewTimer(t.resendAfter)
528	574		defer resend.Stop()

@@ -555,7 +601,7 @@ func (t *Transport) resendPending() {
555	601		for seq := range t.pending {
556	602		seqs = append(seqs, seq)
557	603		}
558		-	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
	604	+	slices.Sort(seqs)
559	605		limit := t.maxResendPerTick
560	606		for _, seq := range seqs {
561	607		if limit <= 0 {

@@ -657,9 +703,14 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
657	703		return
658	704		}
659	705		t.linkState = state
660		-	if t.sendCond != nil {
661		-	t.sendCond.Broadcast()
	706	+	if state == ports.LinkStateConnected {
	707	+	for _, p := range t.pending {
	708	+	if p != nil {
	709	+	p.attempts = 0
	710	+	}
	711	+	}
662	712		}
	713	+	t.notifySendWaiter()
663	714		t.mu.Unlock()
664	715		event := ports.LinkEvent{State: state, At: now, Err: err}
665	716		select {

@@ -678,9 +729,7 @@ func (t *Transport) closeWithError(err error) {
678	729		t.closeErr = err
679	730		t.probeWait = make(map[uint64]chan struct{})
680	731		close(t.done)
681		-	if t.sendCond != nil {
682		-	t.sendCond.Broadcast()
683		-	}
	732	+	t.notifySendWaiter()
684	733		t.mu.Unlock()
685	734		t.deliverMu.Lock()
686	735		t.deliverCond.Broadcast()

 	"encoding/binary"
 	"errors"
 	"net"
-	"sort"
+	"slices"
 	"sync"
 	"time"
 	defaultOffline          = 30 * time.Second
 	defaultDead             = 60 * time.Second
 	defaultMaxPending       = 1024
+	defaultMaxPendingWait   = 50 * time.Millisecond
 	defaultMaxRecvBuf       = 1024
 	maxRecvBuffer           = defaultMaxRecvBuf
 	linkEventBufferSize     = 16
+	probeReplyBufferSize    = 16
 )
 var (
 	OfflineAfter     time.Duration
 	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
 	if opts.MaxPending <= 0 {
 		opts.MaxPending = defaultMaxPending
 	}
+	if opts.MaxPendingWait <= 0 {
+		opts.MaxPendingWait = defaultMaxPendingWait
+	}
 	if opts.MaxRecvBuffer <= 0 {
 		opts.MaxRecvBuffer = defaultMaxRecvBuf
 	}
 	offlineAfter     time.Duration
 	deadAfter        time.Duration
 	maxPending       int
+	maxPendingWait   time.Duration
 	maxRecvBuffer    int
 	clock            ports.Clock
 	linkState        ports.LinkState
 	linkEvents       chan ports.LinkEvent
 	probeWait        map[uint64]chan struct{}
-	sendCond         *sync.Cond
+	sendWake         chan struct{}
+	probeReply       chan uint64
 	rebind           func(net.PacketConn) (net.PacketConn, error)
 	hoppedOffline    bool
 		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, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
+	t := &Transport{pc: pc, codec: c, sendDir: sendDir, recvDir: recvDir, mtu: opts.MTU, peer: peer, pending: make(map[uint64]*pending), replay: pdgram.NewReplayWindow(), reasm: pdgram.NewReassembler(), in: make(chan ports.Frame, 32), done: make(chan struct{}), lastHeard: opts.Clock.Now(), heartbeat: opts.Heartbeat, resendAfter: opts.ResendAfter, maxResendAfter: opts.MaxResendAfter, maxResendPerTick: opts.MaxResendPerTick, writeTimeout: opts.WriteTimeout, degradedAfter: opts.DegradedAfter, probeAfter: opts.ProbeAfter, offlineAfter: opts.OfflineAfter, deadAfter: opts.DeadAfter, maxPending: opts.MaxPending, maxPendingWait: opts.MaxPendingWait, maxRecvBuffer: opts.MaxRecvBuffer, clock: opts.Clock, linkState: ports.LinkStateConnected, linkEvents: make(chan ports.LinkEvent, linkEventBufferSize), probeWait: make(map[uint64]chan struct{}), rebind: opts.RebindPacketConn, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
+	t.sendWake = make(chan struct{}, 1)
+	t.probeReply = make(chan uint64, probeReplyBufferSize)
 	t.deliverCond = sync.NewCond(&t.deliverMu)
-	t.sendCond = sync.NewCond(&t.mu)
 	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
 		}
-		t.sendCond.Wait()
+		if pendingWaitStarted.IsZero() {
+			pendingWaitStarted = t.clock.Now()
+		}
+		remaining := t.maxPendingWait - t.clock.Now().Sub(pendingWaitStarted)
+		if remaining <= 0 {
+			t.mu.Unlock()
+			return ErrPendingFull
+		}
+		timer := t.clock.NewTimer(remaining)
+		t.mu.Unlock()
+		select {
+		case <-timer.C():
+		case <-t.sendWake:
+		case <-t.done:
+		}
+		timer.Stop()
+		t.mu.Lock()
 	}
 	if t.closed {
 		err := t.closeErr
 	t.mu.Unlock()
 	if writeTimeout > 0 {
-		if err := pc.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
+		if err := pc.SetWriteDeadline(t.clock.Now().Add(writeTimeout)); err != nil {
 			return err
 		}
 		defer func() { _ = pc.SetWriteDeadline(time.Time{}) }()
 		t.mu.Lock()
 		if _, ok := t.pending[seq]; ok {
 			delete(t.pending, seq)
-			if t.sendCond != nil {
+			t.notifySendWaiter()
-				t.sendCond.Signal()
-			}
 		}
 		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) notifySendWaiter() {
+	select {
+	case t.sendWake <- struct{}{}:
+	default:
+	}
+}
 func (t *Transport) resendLoop() {
 	resend := t.clock.NewTimer(t.resendAfter)
 	defer resend.Stop()
 	for seq := range t.pending {
 		seqs = append(seqs, seq)
 	}
-	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
+	slices.Sort(seqs)
 	limit := t.maxResendPerTick
 	for _, seq := range seqs {
 		if limit <= 0 {
 		return
 	}
 	t.linkState = state
-	if t.sendCond != nil {
+	if state == ports.LinkStateConnected {
-		t.sendCond.Broadcast()
+		for _, p := range t.pending {
+			if p != nil {
+				p.attempts = 0
+			}
+		}
 	}
+	t.notifySendWaiter()
 	t.mu.Unlock()
 	event := ports.LinkEvent{State: state, At: now, Err: err}
 	select {
 	t.closeErr = err
 	t.probeWait = make(map[uint64]chan struct{})
 	close(t.done)
-	if t.sendCond != nil {
+	t.notifySendWaiter()
-		t.sendCond.Broadcast()
-	}
 	t.mu.Unlock()
 	t.deliverMu.Lock()
 	t.deliverCond.Broadcast()

M internal/adapters/dgram/transport_test.go +194 -0


@@ -161,6 +161,100 @@ func (p *blockingWritePC) SetWriteDeadline(t time.Time) error {
161	161		return nil
162	162		}
163	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	+
164	258		func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
165	259		pc := newBlockingWritePC()
166	260		tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{

@@ -628,6 +722,99 @@ func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
628	722		}
629	723		}
630	724
	725	+	func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
	726	+	aPC, bPC := newPair()
	727	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	728	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	729	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: 50 * time.Millisecond, Clock: clk})
	730	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	731	+	defer func() { _ = a.Close() }()
	732	+	defer func() { _ = b.Close() }()
	733	+
	734	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
	735	+	t.Fatal(err)
	736	+	}
	737	+	done := make(chan error, 1)
	738	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
	739	+	waitForManualTimers(t, clk, 3)
	740	+
	741	+	clk.advance(50 * time.Millisecond)
	742	+	select {
	743	+	case err := <-done:
	744	+	if !errors.Is(err, ErrPendingFull) {
	745	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
	746	+	}
	747	+	case <-time.After(time.Second):
	748	+	t.Fatal("send did not return after clock advanced past MaxPendingWait")
	749	+	}
	750	+	}
	751	+
	752	+	func TestPendingAttemptsResetWhenLinkRecovers(t *testing.T) {
	753	+	tr, _ := newResendTestTransport(t, 10)
	754	+	tr.linkState = ports.LinkStateDegraded
	755	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 4}
	756	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 7}
	757	+
	758	+	tr.setLinkState(ports.LinkStateConnected, nil)
	759	+
	760	+	for seq, p := range tr.pending {
	761	+	if p.attempts != 0 {
	762	+	t.Fatalf("pending[%d] attempts=%d, want reset on connected recovery", seq, p.attempts)
	763	+	}
	764	+	}
	765	+	}
	766	+
	767	+	func TestWritePacketDeadlineUsesTransportClock(t *testing.T) {
	768	+	wantNow := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
	769	+	pc := &deadlineCapturePC{addr: testAddr("a")}
	770	+	tr, err := NewTransportWithOptions(pc, testAddr("b"), key(), 1, 2, Options{
	771	+	Clock: fixedClock{now: wantNow},
	772	+	WriteTimeout: 250 * time.Millisecond,
	773	+	ResendAfter: time.Hour,
	774	+	Heartbeat: time.Hour,
	775	+	})
	776	+	if err != nil {
	777	+	t.Fatal(err)
	778	+	}
	779	+	defer func() { _ = tr.Close() }()
	780	+
	781	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("deadline")}); err != nil {
	782	+	t.Fatal(err)
	783	+	}
	784	+	got, _ := pc.deadline.Load().(time.Time)
	785	+	if want := wantNow.Add(250 * time.Millisecond); !got.Equal(want) {
	786	+	t.Fatalf("write deadline=%v, want transport clock deadline %v", got, want)
	787	+	}
	788	+	}
	789	+
	790	+	func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
	791	+	aPC, bPC := newPair()
	792	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
	793	+	if err != nil {
	794	+	t.Fatal(err)
	795	+	}
	796	+	defer func() { _ = tr.Close() }()
	797	+	defer func() { _ = bPC.Close() }()
	798	+
	799	+	tr.writeMu.Lock()
	800	+	done := make(chan struct{})
	801	+	go func() {
	802	+	for id := uint64(1); id <= probeReplyBufferSize*4; id++ {
	803	+	tr.handleRecord(probeRecord(recProbe, id))
	804	+	}
	805	+	close(done)
	806	+	}()
	807	+
	808	+	select {
	809	+	case <-done:
	810	+	case <-time.After(time.Second):
	811	+	tr.writeMu.Unlock()
	812	+	<-done
	813	+	t.Fatal("probe reply burst blocked behind writeMu")
	814	+	}
	815	+	tr.writeMu.Unlock()
	816	+	}
	817	+
631	818		func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
632	819		aPC, bPC := newPair()
633	820		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }

@@ -665,6 +852,13 @@ func ackRecord(seq uint64) []byte {
665	852		return b[:]
666	853		}
667	854
	855	+	func probeRecord(kind byte, id uint64) []byte {
	856	+	var b [9]byte
	857	+	b[0] = kind
	858	+	binary.BigEndian.PutUint64(b[1:], id)
	859	+	return b[:]
	860	+	}
	861	+
668	862		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
669	863		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
670	864		tr.deliverCond = sync.NewCond(&tr.deliverMu)


@@ -161,6 +161,100 @@ func (p *blockingWritePC) SetWriteDeadline(t time.Time) error {
161	return nil	161		return nil
162	}	162		}
163		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	+
164	func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {	258		func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
165	pc := newBlockingWritePC()	259		pc := newBlockingWritePC()
166	tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{	260		tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{

@@ -628,6 +722,99 @@ func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
628	}	722		}
629	}	723		}
630		724
		725	+	func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
		726	+	aPC, bPC := newPair()
		727	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
		728	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
		729	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: 50 * time.Millisecond, Clock: clk})
		730	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
		731	+	defer func() { _ = a.Close() }()
		732	+	defer func() { _ = b.Close() }()
		733	+
		734	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
		735	+	t.Fatal(err)
		736	+	}
		737	+	done := make(chan error, 1)
		738	+	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
		739	+	waitForManualTimers(t, clk, 3)
		740	+
		741	+	clk.advance(50 * time.Millisecond)
		742	+	select {
		743	+	case err := <-done:
		744	+	if !errors.Is(err, ErrPendingFull) {
		745	+	t.Fatalf("send err=%v, want ErrPendingFull", err)
		746	+	}
		747	+	case <-time.After(time.Second):
		748	+	t.Fatal("send did not return after clock advanced past MaxPendingWait")
		749	+	}
		750	+	}
		751	+
		752	+	func TestPendingAttemptsResetWhenLinkRecovers(t *testing.T) {
		753	+	tr, _ := newResendTestTransport(t, 10)
		754	+	tr.linkState = ports.LinkStateDegraded
		755	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 4}
		756	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stale")}, attempts: 7}
		757	+
		758	+	tr.setLinkState(ports.LinkStateConnected, nil)
		759	+
		760	+	for seq, p := range tr.pending {
		761	+	if p.attempts != 0 {
		762	+	t.Fatalf("pending[%d] attempts=%d, want reset on connected recovery", seq, p.attempts)
		763	+	}
		764	+	}
		765	+	}
		766	+
		767	+	func TestWritePacketDeadlineUsesTransportClock(t *testing.T) {
		768	+	wantNow := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
		769	+	pc := &deadlineCapturePC{addr: testAddr("a")}
		770	+	tr, err := NewTransportWithOptions(pc, testAddr("b"), key(), 1, 2, Options{
		771	+	Clock: fixedClock{now: wantNow},
		772	+	WriteTimeout: 250 * time.Millisecond,
		773	+	ResendAfter: time.Hour,
		774	+	Heartbeat: time.Hour,
		775	+	})
		776	+	if err != nil {
		777	+	t.Fatal(err)
		778	+	}
		779	+	defer func() { _ = tr.Close() }()
		780	+
		781	+	if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("deadline")}); err != nil {
		782	+	t.Fatal(err)
		783	+	}
		784	+	got, _ := pc.deadline.Load().(time.Time)
		785	+	if want := wantNow.Add(250 * time.Millisecond); !got.Equal(want) {
		786	+	t.Fatalf("write deadline=%v, want transport clock deadline %v", got, want)
		787	+	}
		788	+	}
		789	+
		790	+	func TestProbeReplyDoesNotBlockOnWriteMu(t *testing.T) {
		791	+	aPC, bPC := newPair()
		792	+	tr, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ResendAfter: time.Hour, Heartbeat: time.Hour})
		793	+	if err != nil {
		794	+	t.Fatal(err)
		795	+	}
		796	+	defer func() { _ = tr.Close() }()
		797	+	defer func() { _ = bPC.Close() }()
		798	+
		799	+	tr.writeMu.Lock()
		800	+	done := make(chan struct{})
		801	+	go func() {
		802	+	for id := uint64(1); id <= probeReplyBufferSize*4; id++ {
		803	+	tr.handleRecord(probeRecord(recProbe, id))
		804	+	}
		805	+	close(done)
		806	+	}()
		807	+
		808	+	select {
		809	+	case <-done:
		810	+	case <-time.After(time.Second):
		811	+	tr.writeMu.Unlock()
		812	+	<-done
		813	+	t.Fatal("probe reply burst blocked behind writeMu")
		814	+	}
		815	+	tr.writeMu.Unlock()
		816	+	}
		817	+
631	func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {	818		func TestPendingReliableQueueReturnsWhenLinkDegrades(t *testing.T) {
632	aPC, bPC := newPair()	819		aPC, bPC := newPair()
633	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	820		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }

@@ -665,6 +852,13 @@ func ackRecord(seq uint64) []byte {
665	return b[:]	852		return b[:]
666	}	853		}
667		854
		855	+	func probeRecord(kind byte, id uint64) []byte {
		856	+	var b [9]byte
		857	+	b[0] = kind
		858	+	binary.BigEndian.PutUint64(b[1:], id)
		859	+	return b[:]
		860	+	}
		861	+
668	func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {	862		func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
669	tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}	863		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
670	tr.deliverCond = sync.NewCond(&tr.deliverMu)	864		tr.deliverCond = sync.NewCond(&tr.deliverMu)

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) {