7e909e8 Merge pull request #47 from bnema/fix/udp-remote-congestion-collapse · vev

M internal/adapters/dgram/proxy.go +28 -95


@@ -38,17 +38,9 @@ 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	41		type proxyCopier struct {
49	42		ctx context.Context
50	43		errCh chan<- proxyCopyErr
51		-	retryEvents chan<- proxyRetryEvent
52	44		retryBackoff time.Duration
53	45		clk ports.Clock
54	46		}

@@ -58,7 +50,7 @@ type proxyCopyDirection struct {
58	50		dst ports.Transport
59	51		recvKind proxyCopyErrKind
60	52		sendKind proxyCopyErrKind
61		-	retryPendingFull bool
	53	+	retryRecoverable bool
62	54		}
63	55
64	56		func (p ProxyRuntime) Run(ctx context.Context) error {

@@ -81,18 +73,11 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
81	73		}
82	74
83	75		errCh := make(chan proxyCopyErr, 2)
84		-	retryEvents := make(chan proxyRetryEvent, 2)
85		-	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
	76	+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}
86	77		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}
88		-	startClientCopy := func() {
89		-	go copier.copyFrames(clientToDaemon)
90		-	}
91		-	startDaemonCopy := func() {
92		-	go copier.copyFrames(daemonToClient)
93		-	}
94		-	startClientCopy()
95		-	startDaemonCopy()
	78	+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}
	79	+	go copier.copyFrames(clientToDaemon)
	80	+	go copier.copyFrames(daemonToClient)
96	81
97	82		var linkEvents <-chan ports.LinkEvent
98	83		if reporter, ok := p.Client.(ports.LinkStateReporter); ok {

@@ -119,16 +104,6 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
119	104		}
120	105		idle = nil
121	106		}
122		-	waitRetry := func() error {
123		-	t := clk.NewTimer(retryBackoff)
124		-	defer t.Stop()
125		-	select {
126		-	case <-t.C():
127		-	return nil
128		-	case <-ctx.Done():
129		-	return ctx.Err()
130		-	}
131		-	}
132	107		for {
133	108		select {
134	109		case <-ctx.Done():

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

@@ -162,32 +130,12 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
162	130		return nil
163	131		}
164	132		return ce.err
165		-	case proxyDaemonSend:
	133	+	default:
	134	+	// Recoverable client-side errors are retried inside the copier
	135	+	// without dropping the frame; anything that surfaces here (a
	136	+	// daemon send failure, or a non-recoverable client error such as
	137	+	// a dead link) is fatal.
166	138		return ce.err
167		-	case proxyClientSend:
168		-	if !recoverableClientLink(p.Client, ce.err) {
169		-	return ce.err
170		-	}
171		-	if p.Log != nil {
172		-	p.Log.Info("udp proxy restarting recoverable client send path", "err", ce.err)
173		-	}
174		-	startIdle()
175		-	if err := waitRetry(); err != nil {
176		-	return err
177		-	}
178		-	startDaemonCopy()
179		-	case proxyClientRecv:
180		-	if !recoverableClientLink(p.Client, ce.err) {
181		-	return ce.err
182		-	}
183		-	if p.Log != nil {
184		-	p.Log.Info("udp proxy restarting recoverable client receive path", "err", ce.err)
185		-	}
186		-	startIdle()
187		-	if err := waitRetry(); err != nil {
188		-	return err
189		-	}
190		-	startClientCopy()
191	139		}
192	140		}
193	141		}

@@ -200,46 +148,31 @@ func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
200	148		c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
201	149		return
202	150		}
203		-	retrying := false
	151	+	// Retry the same frame in place while the destination link error stays
	152	+	// recoverable (congestion, transient path loss). The frame is never
	153	+	// dropped — on a reliable stream a gap desynchronizes the client screen.
204	154		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
	155	+	err := dir.dst.Send(f)
	156	+	if err == nil {
	157	+	break
225	158		}
226		-	if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
227		-	return
	159	+	if dir.retryRecoverable && recoverableClientLink(dir.dst, err) {
	160	+	t := c.clk.NewTimer(c.retryBackoff)
	161	+	select {
	162	+	case <-t.C():
	163	+	t.Stop()
	164	+	continue
	165	+	case <-c.ctx.Done():
	166	+	t.Stop()
	167	+	return
	168	+	}
228	169		}
229		-	break
	170	+	c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
	171	+	return
230	172		}
231	173		}
232	174		}
233	175
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		-
243	176		func recoverableClientLink(t ports.Transport, err error) bool {
244	177		if errors.Is(err, ErrLinkDead) {
245	178		return false

 	err  error
 }
-type proxyRetryEvent uint8
-const (
-	proxyRetryStart proxyRetryEvent = iota + 1
-	proxyRetryDone
-)
 type proxyCopier struct {
 	ctx          context.Context
 	errCh        chan<- proxyCopyErr
-	retryEvents  chan<- proxyRetryEvent
 	retryBackoff time.Duration
 	clk          ports.Clock
 }
 	dst              ports.Transport
 	recvKind         proxyCopyErrKind
 	sendKind         proxyCopyErrKind
-	retryPendingFull bool
+	retryRecoverable bool
 }
 func (p ProxyRuntime) Run(ctx context.Context) error {
 	}
 	errCh := make(chan proxyCopyErr, 2)
-	retryEvents := make(chan proxyRetryEvent, 2)
+	copier := proxyCopier{ctx: ctx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}
-	copier := proxyCopier{ctx: ctx, errCh: errCh, retryEvents: retryEvents, retryBackoff: retryBackoff, clk: clk}
 	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
-	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryPendingFull: true}
+	daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}
-	startClientCopy := func() {
+	go copier.copyFrames(clientToDaemon)
-		go copier.copyFrames(clientToDaemon)
+	go copier.copyFrames(daemonToClient)
-	}
-	startDaemonCopy := func() {
-		go copier.copyFrames(daemonToClient)
-	}
-	startClientCopy()
-	startDaemonCopy()
 	var linkEvents <-chan ports.LinkEvent
 	if reporter, ok := p.Client.(ports.LinkStateReporter); ok {
 		}
 		idle = nil
 	}
-	waitRetry := func() error {
-		t := clk.NewTimer(retryBackoff)
-		defer t.Stop()
-		select {
-		case <-t.C():
-			return nil
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
 	for {
 		select {
 		case <-ctx.Done():
 			}
 		case <-idle:
 			return ErrLinkDead
-		case ev := <-retryEvents:
-			switch ev {
-			case proxyRetryStart:
-				startIdle()
-			case proxyRetryDone:
-				stopIdle()
-			}
 		case ce := <-errCh:
 			switch ce.kind {
 			case proxyDaemonRecv:
 					return nil
 				}
 				return ce.err
-			case proxyDaemonSend:
+			default:
+				// Recoverable client-side errors are retried inside the copier
+				// without dropping the frame; anything that surfaces here (a
+				// daemon send failure, or a non-recoverable client error such as
+				// a dead link) is fatal.
 				return ce.err
-			case proxyClientSend:
-				if !recoverableClientLink(p.Client, ce.err) {
-					return ce.err
-				}
-				if p.Log != nil {
-					p.Log.Info("udp proxy restarting recoverable client send path", "err", ce.err)
-				}
-				startIdle()
-				if err := waitRetry(); err != nil {
-					return err
-				}
-				startDaemonCopy()
-			case proxyClientRecv:
-				if !recoverableClientLink(p.Client, ce.err) {
-					return ce.err
-				}
-				if p.Log != nil {
-					p.Log.Info("udp proxy restarting recoverable client receive path", "err", ce.err)
-				}
-				startIdle()
-				if err := waitRetry(); err != nil {
-					return err
-				}
-				startClientCopy()
 			}
 		}
 	}
 			c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
 			return
 		}
-		retrying := false
+		// Retry the same frame in place while the destination link error stays
+		// recoverable (congestion, transient path loss). The frame is never
+		// dropped — on a reliable stream a gap desynchronizes the client screen.
 		for {
-			if err := dir.dst.Send(f); err != nil {
+			err := dir.dst.Send(f)
-				if dir.retryPendingFull && errors.Is(err, ErrPendingFull) && recoverableClientLink(dir.dst, err) {
+			if err == nil {
-					if !retrying {
+				break
-						retrying = true
-						if !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryStart) {
-							return
-						}
-					}
-					t := c.clk.NewTimer(c.retryBackoff)
-					select {
-					case <-t.C():
-						t.Stop()
-						continue
-					case <-c.ctx.Done():
-						t.Stop()
-						return
-					}
-				}
-				c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
-				return
 			}
-			if retrying && !sendProxyRetryEvent(c.ctx, c.retryEvents, proxyRetryDone) {
+			if dir.retryRecoverable && recoverableClientLink(dir.dst, err) {
-				return
+				t := c.clk.NewTimer(c.retryBackoff)
+				select {
+				case <-t.C():
+					t.Stop()
+					continue
+				case <-c.ctx.Done():
+					t.Stop()
+					return
+				}
 			}
-			break
+			c.errCh <- proxyCopyErr{kind: dir.sendKind, err: err}
+			return
 		}
 	}
 }
-func sendProxyRetryEvent(ctx context.Context, retryEvents chan<- proxyRetryEvent, ev proxyRetryEvent) bool {
-	select {
-	case retryEvents <- ev:
-		return true
-	case <-ctx.Done():
-		return false
-	}
-}
 func recoverableClientLink(t ports.Transport, err error) bool {
 	if errors.Is(err, ErrLinkDead) {
 		return false

M internal/adapters/dgram/proxy_test.go +45 -119


@@ -11,15 +11,13 @@ import (
11	11		)
12	12
13	13		type fakeTransport struct {
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
	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
23	21		}
24	22
25	23		type recvResult struct {

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

@@ -85,58 +77,27 @@ func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) {
85	77		}
86	78		}
87	79
88		-	func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T) {
89		-	client := newFakeTransport()
90		-	daemon := newFakeTransport()
91		-	client.linkState = ports.LinkStateOffline
92		-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
93		-	errCh := make(chan error, 1)
94		-	go func() {
95		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
96		-	}()
97		-	client.recv <- recvResult{err: errors.New("temporary receive failure")}
98		-	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgInput, Payload: []byte("after recovery")}}
99		-	select {
100		-	case got := <-daemon.sent:
101		-	if got.Type != ports.MsgInput \|\| string(got.Payload) != "after recovery" {
102		-	t.Fatalf("daemon got %+v, want recovered input", got)
103		-	}
104		-	case <-time.After(time.Second):
105		-	t.Fatal("timeout waiting for recovered frame")
106		-	}
107		-	daemon.recv <- recvResult{err: io.EOF}
108		-	select {
109		-	case err := <-errCh:
110		-	if err != nil {
111		-	t.Fatalf("Run err=%v, want nil", err)
112		-	}
113		-	case <-time.After(time.Second):
114		-	t.Fatal("timeout waiting for daemon EOF")
115		-	}
116		-	}
117		-
118	80		func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
119	81		tests := []struct {
120		-	name string
121		-	state ports.LinkState
122		-	linkEvent bool
	82	+	name string
	83	+	state ports.LinkState
	84	+	err error
123	85		}{
124		-	{name: "connected pending full", state: ports.LinkStateConnected},
125		-	{name: "offline pending full", state: ports.LinkStateOffline, linkEvent: true},
	86	+	{name: "connected pending full", state: ports.LinkStateConnected, err: ErrPendingFull},
	87	+	{name: "degraded transient error", state: ports.LinkStateDegraded, err: errors.New("degraded write drop")},
126	88		}
127	89		for _, tt := range tests {
128	90		t.Run(tt.name, func(t *testing.T) {
129	91		client := newFakeTransport()
130	92		daemon := newFakeTransport()
	93	+	// No link event is emitted, so the idle timer is never armed; the
	94	+	// retry alone must recover. IdleTTL is large to make that explicit.
131	95		client.linkState = tt.state
132		-	if tt.linkEvent {
133		-	client.linkEvents <- ports.LinkEvent{State: tt.state}
134		-	}
135	96		errCh := make(chan error, 1)
136	97		go func() {
137		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
	98	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
138	99		}()
139		-	client.sendErr <- ErrPendingFull
	100	+	client.sendErr <- tt.err
140	101		daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
141	102		select {
142	103		case got := <-client.sent:

@@ -159,28 +120,35 @@ func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing
159	120		}
160	121		}
161	122
162		-	func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
	123	+	// TestProxyRuntimeSurvivesPendingFullAndPreservesOrder is the congestion
	124	+	// regression: a persistently pending-full client window must not tear the proxy
	125	+	// down (the old idle TTL killed it after 30s) and every frame must arrive in
	126	+	// order. IdleTTL is a nanosecond: if a retry armed the idle timer the proxy
	127	+	// would die at once, so surviving proves retries no longer arm it.
	128	+	func TestProxyRuntimeSurvivesPendingFullAndPreservesOrder(t *testing.T) {
163	129		client := newFakeTransport()
164	130		daemon := newFakeTransport()
165		-	client.linkState = ports.LinkStateOffline
166		-	client.sideEffectOnErr = true
167		-	client.sendErr <- errors.New("write timed out after queuing")
	131	+	client.linkState = ports.LinkStateConnected
168	132		errCh := make(chan error, 1)
169	133		go func() {
170		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
	134	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Nanosecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
171	135		}()
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)
	136	+	const frames = 4
	137	+	for i := range frames {
	138	+	client.sendErr <- ErrPendingFull
	139	+	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}}
	140	+	select {
	141	+	case got := <-client.sent:
	142	+	if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
	143	+	t.Fatalf("frame %d delivered out of order or dropped: %+v", i, got)
	144	+	}
	145	+	case <-time.After(time.Second):
	146	+	t.Fatalf("frame %d dropped or proxy killed under pending-full", i)
177	147		}
178		-	case <-time.After(time.Second):
179		-	t.Fatal("timeout waiting for side-effected output")
180	148		}
181	149		select {
182		-	case got := <-client.sent:
183		-	t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
	150	+	case err := <-errCh:
	151	+	t.Fatalf("proxy returned under sustained pending-full: %v", err)
184	152		case <-time.After(20 * time.Millisecond):
185	153		}
186	154		daemon.recv <- recvResult{err: io.EOF}

@@ -194,82 +162,40 @@ func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
194	162		}
195	163		}
196	164
197		-	func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
	165	+	func TestProxyRuntimeOfflineLinkEventExpiresViaIdleTTL(t *testing.T) {
198	166		client := newFakeTransport()
199	167		daemon := newFakeTransport()
200	168		client.linkState = ports.LinkStateOffline
201		-	client.sendErrDefault = ErrPendingFull
202	169		errCh := make(chan error, 1)
203	170		go func() {
204		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
	171	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond}.Run(t.Context())
205	172		}()
206		-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
	173	+	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
207	174		select {
208	175		case err := <-errCh:
209	176		if !errors.Is(err, ErrLinkDead) {
210	177		t.Fatalf("Run err=%v, want ErrLinkDead", err)
211	178		}
212	179		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")}}
227		-	select {
228		-	case got := <-client.sent:
229		-	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "recovers" {
230		-	t.Fatalf("client got %+v, want retried output", got)
231		-	}
232		-	case <-time.After(time.Second):
233		-	t.Fatal("timeout waiting for retried output")
234		-	}
235		-	select {
236		-	case err := <-errCh:
237		-	t.Fatalf("Run returned after successful retry before daemon EOF: %v", err)
238		-	case <-time.After(50 * time.Millisecond):
239		-	}
240		-	daemon.recv <- recvResult{err: io.EOF}
241		-	select {
242		-	case err := <-errCh:
243		-	if err != nil {
244		-	t.Fatalf("Run err=%v, want nil", err)
245		-	}
246		-	case <-time.After(time.Second):
247		-	t.Fatal("timeout waiting for daemon EOF")
	180	+	t.Fatal("timeout waiting for idle expiry after offline link event")
248	181		}
249	182		}
250	183
251		-	func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
	184	+	func TestProxyRuntimeDeadLinkEventTearsDown(t *testing.T) {
252	185		client := newFakeTransport()
253	186		daemon := newFakeTransport()
254		-	client.linkState = ports.LinkStateOffline
255		-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
256	187		errCh := make(chan error, 1)
257	188		go func() {
258		-	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond}.Run(t.Context())
	189	+	errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
259	190		}()
260		-	client.recv <- recvResult{err: errors.New("temporary receive failure")}
261		-	select {
262		-	case err := <-errCh:
263		-	t.Fatalf("Run returned before idle TTL: %v", err)
264		-	case <-time.After(5 * time.Millisecond):
265		-	}
	191	+	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}
266	192		select {
267	193		case err := <-errCh:
268	194		if !errors.Is(err, ErrLinkDead) {
269	195		t.Fatalf("Run err=%v, want ErrLinkDead", err)
270	196		}
271	197		case <-time.After(time.Second):
272		-	t.Fatal("timeout waiting for idle expiry")
	198	+	t.Fatal("dead link event did not tear down proxy")
273	199		}
274	200		}
275	201

 )
 type fakeTransport struct {
-	recv            chan recvResult
+	recv       chan recvResult
-	sent            chan ports.Frame
+	sent       chan ports.Frame
-	sendErr         chan error
+	sendErr    chan error
-	sendErrDefault  error
+	linkState  ports.LinkState
-	sideEffectOnErr bool
+	linkEvents chan ports.LinkEvent
-	linkState       ports.LinkState
+	closed     chan struct{}
-	linkEvents      chan ports.LinkEvent
+	closeOnce  sync.Once
-	closed          chan struct{}
-	closeOnce       sync.Once
 }
 type recvResult struct {
 func (f *fakeTransport) Send(fr ports.Frame) error {
 	select {
 	case err := <-f.sendErr:
-		if f.sideEffectOnErr {
-			f.sent <- fr
-		}
 		return err
 	default:
-		if f.sendErrDefault != nil {
-			return f.sendErrDefault
-		}
 	}
 	f.sent <- fr
 	return nil
 	}
 }
-func TestProxyRuntimeRestartsRecoverableClientReceiveUntilIdleTTL(t *testing.T) {
-	client := newFakeTransport()
-	daemon := newFakeTransport()
-	client.linkState = ports.LinkStateOffline
-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
-	errCh := make(chan error, 1)
-	go func() {
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
-	}()
-	client.recv <- recvResult{err: errors.New("temporary receive failure")}
-	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgInput, Payload: []byte("after recovery")}}
-	select {
-	case got := <-daemon.sent:
-		if got.Type != ports.MsgInput || string(got.Payload) != "after recovery" {
-			t.Fatalf("daemon got %+v, want recovered input", got)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for recovered frame")
-	}
-	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")
-	}
-}
 func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) {
 	tests := []struct {
-		name      string
+		name  string
-		state     ports.LinkState
+		state ports.LinkState
-		linkEvent bool
+		err   error
 	}{
-		{name: "connected pending full", state: ports.LinkStateConnected},
+		{name: "connected pending full", state: ports.LinkStateConnected, err: ErrPendingFull},
-		{name: "offline pending full", state: ports.LinkStateOffline, linkEvent: true},
+		{name: "degraded transient error", state: ports.LinkStateDegraded, err: errors.New("degraded write drop")},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			client := newFakeTransport()
 			daemon := newFakeTransport()
+			// No link event is emitted, so the idle timer is never armed; the
+			// retry alone must recover. IdleTTL is large to make that explicit.
 			client.linkState = tt.state
-			if tt.linkEvent {
-				client.linkEvents <- ports.LinkEvent{State: tt.state}
-			}
 			errCh := make(chan error, 1)
 			go func() {
-				errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 50 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
+				errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
 			}()
-			client.sendErr <- ErrPendingFull
+			client.sendErr <- tt.err
 			daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("preserved during outage")}}
 			select {
 			case got := <-client.sent:
 	}
 }
-func TestProxyRuntimeDoesNotRetrySideEffectingClientSendError(t *testing.T) {
+// TestProxyRuntimeSurvivesPendingFullAndPreservesOrder is the congestion
+// regression: a persistently pending-full client window must not tear the proxy
+// down (the old idle TTL killed it after 30s) and every frame must arrive in
+// order. IdleTTL is a nanosecond: if a retry armed the idle timer the proxy
+// would die at once, so surviving proves retries no longer arm it.
+func TestProxyRuntimeSurvivesPendingFullAndPreservesOrder(t *testing.T) {
 	client := newFakeTransport()
 	daemon := newFakeTransport()
-	client.linkState = ports.LinkStateOffline
+	client.linkState = ports.LinkStateConnected
-	client.sideEffectOnErr = true
-	client.sendErr <- errors.New("write timed out after queuing")
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour, RetryBackoff: time.Nanosecond}.Run(t.Context())
+		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Nanosecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
 	}()
-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("maybe queued")}}
+	const frames = 4
-	select {
+	for i := range frames {
-	case got := <-client.sent:
+		client.sendErr <- ErrPendingFull
-		if got.Type != ports.MsgOutput || string(got.Payload) != "maybe queued" {
+		daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}}
-			t.Fatalf("client got %+v, want side-effected output", got)
+		select {
+		case got := <-client.sent:
+			if got.Type != ports.MsgOutput || len(got.Payload) != 1 || got.Payload[0] != byte(i) {
+				t.Fatalf("frame %d delivered out of order or dropped: %+v", i, got)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("frame %d dropped or proxy killed under pending-full", i)
 		}
-	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for side-effected output")
 	}
 	select {
-	case got := <-client.sent:
+	case err := <-errCh:
-		t.Fatalf("side-effecting send error was retried and duplicated output: %+v", got)
+		t.Fatalf("proxy returned under sustained pending-full: %v", err)
 	case <-time.After(20 * time.Millisecond):
 	}
 	daemon.recv <- recvResult{err: io.EOF}
 	}
 }
-func TestProxyRuntimeErrPendingFullRetryStartsIdleTTL(t *testing.T) {
+func TestProxyRuntimeOfflineLinkEventExpiresViaIdleTTL(t *testing.T) {
 	client := newFakeTransport()
 	daemon := newFakeTransport()
 	client.linkState = ports.LinkStateOffline
-	client.sendErrDefault = ErrPendingFull
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Millisecond}.Run(t.Context())
+		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond}.Run(t.Context())
 	}()
-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stalled")}}
+	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
 	select {
 	case err := <-errCh:
 		if !errors.Is(err, ErrLinkDead) {
 			t.Fatalf("Run err=%v, want ErrLinkDead", err)
 		}
 	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for idle expiry from ErrPendingFull retry")
+		t.Fatal("timeout waiting for idle expiry after offline link event")
-	}
-}
-func TestProxyRuntimeErrPendingFullRetrySuccessStopsIdleTTL(t *testing.T) {
-	client := newFakeTransport()
-	daemon := newFakeTransport()
-	client.linkState = ports.LinkStateConnected
-	errCh := make(chan error, 1)
-	go func() {
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond, RetryBackoff: time.Nanosecond}.Run(t.Context())
-	}()
-	client.sendErr <- ErrPendingFull
-	daemon.recv <- recvResult{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("recovers")}}
-	select {
-	case got := <-client.sent:
-		if got.Type != ports.MsgOutput || string(got.Payload) != "recovers" {
-			t.Fatalf("client got %+v, want retried output", got)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for retried output")
-	}
-	select {
-	case err := <-errCh:
-		t.Fatalf("Run returned after successful retry before daemon EOF: %v", err)
-	case <-time.After(50 * time.Millisecond):
-	}
-	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")
 	}
 }
-func TestProxyRuntimeKeepsRecoverableLinkUntilIdleTTL(t *testing.T) {
+func TestProxyRuntimeDeadLinkEventTearsDown(t *testing.T) {
 	client := newFakeTransport()
 	daemon := newFakeTransport()
-	client.linkState = ports.LinkStateOffline
-	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateOffline}
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: 20 * time.Millisecond}.Run(t.Context())
+		errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context())
 	}()
-	client.recv <- recvResult{err: errors.New("temporary receive failure")}
+	client.linkEvents <- ports.LinkEvent{State: ports.LinkStateDead}
-	select {
-	case err := <-errCh:
-		t.Fatalf("Run returned before idle TTL: %v", err)
-	case <-time.After(5 * time.Millisecond):
-	}
 	select {
 	case err := <-errCh:
 		if !errors.Is(err, ErrLinkDead) {
 			t.Fatalf("Run err=%v, want ErrLinkDead", err)
 		}
 	case <-time.After(time.Second):
-		t.Fatal("timeout waiting for idle expiry")
+		t.Fatal("dead link event did not tear down proxy")
 	}
 }

M internal/adapters/dgram/ssh_dialer.go +8 -1


@@ -172,7 +172,14 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
172	172		if err != nil {
173	173		return nil, udpUnavailable("listen UDP", err, &stderr)
174	174		}
175		-	t, err := NewTransport(pc, peer, ready.key, 1, 2)
	175	+	// RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
	176	+	// fresh local UDP socket once the link is silent long enough to probe. The
	177	+	// bind reuses listenUDPPacket so tests can stub it.
	178	+	t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
	179	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	180	+	return listenUDPPacket(ctx)
	181	+	},
	182	+	})
176	183		if err != nil {
177	184		_ = pc.Close()
178	185		return nil, udpUnavailable("create UDP transport", err, &stderr)

 	if err != nil {
 		return nil, udpUnavailable("listen UDP", err, &stderr)
 	}
-	t, err := NewTransport(pc, peer, ready.key, 1, 2)
+	// RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
+	// fresh local UDP socket once the link is silent long enough to probe. The
+	// bind reuses listenUDPPacket so tests can stub it.
+	t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
+		RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
+			return listenUDPPacket(ctx)
+		},
+	})
 	if err != nil {
 		_ = pc.Close()
 		return nil, udpUnavailable("create UDP transport", err, &stderr)

M internal/adapters/dgram/transport.go +42 -2


@@ -177,6 +177,7 @@ type Transport struct {
177	177
178	178		type pending struct {
179	179		frame ports.Frame
	180	+	first time.Time
180	181		last time.Time
181	182		attempts int
182	183		}

@@ -275,7 +276,8 @@ func (t *Transport) Send(f ports.Frame) error {
275	276		t.seq++
276	277		seq := t.seq
277	278		if reliable {
278		-	t.pending[seq] = &pending{frame: f, last: t.clock.Now()}
	279	+	now := t.clock.Now()
	280	+	t.pending[seq] = &pending{frame: f, first: now, last: now}
279	281		}
280	282		t.mu.Unlock()
281	283		if err := t.sendData(seq, reliable, f); err != nil {

@@ -624,10 +626,12 @@ func (t *Transport) resendLoop() {
624	626		case <-resend.C():
625	627		t.resendPending()
626	628		t.checkSilence()
	629	+	t.checkSendStall()
627	630		resend.Reset(t.resendAfter)
628	631		case <-heartbeat.C():
629	632		_ = t.sendProbe(recProbe, 0)
630	633		t.checkSilence()
	634	+	t.checkSendStall()
631	635		heartbeat.Reset(t.heartbeat)
632	636		case <-t.done:
633	637		return

@@ -732,14 +736,50 @@ func (t *Transport) checkSilence() {
732	736		t.closeWithError(ErrLinkDead)
733	737		case offlineAfter > 0 && silentFor >= offlineAfter:
734	738		t.setLinkState(ports.LinkStateOffline, nil)
735		-	t.hopPacketConnOnce()
736	739		case probeAfter > 0 && silentFor >= probeAfter:
737	740		t.setLinkState(ports.LinkStateProbing, nil)
	741	+	t.hopPacketConnOnce()
738	742		case degradedAfter > 0 && silentFor >= degradedAfter:
739	743		t.setLinkState(ports.LinkStateDegraded, nil)
740	744		}
741	745		}
742	746
	747	+	// checkSendStall declares the link dead when the oldest unacked reliable frame
	748	+	// has gone unacknowledged for deadAfter despite at least one retransmit. The
	749	+	// receive-silence check in checkSilence misses one-way-dead links whose receive
	750	+	// side stays fresh from peer heartbeats; this catches them at the layer that
	751	+	// owns death. first is set once when a frame is enqueued and never refreshed,
	752	+	// so acks (which delete pending entries) keep the oldest age bounded on a
	753	+	// progressing link — only a frame that is never acked crosses the threshold.
	754	+	func (t *Transport) checkSendStall() {
	755	+	t.mu.Lock()
	756	+	if t.closed \|\| t.deadAfter <= 0 {
	757	+	t.mu.Unlock()
	758	+	return
	759	+	}
	760	+	deadAfter := t.deadAfter
	761	+	now := t.clock.Now()
	762	+	var oldest time.Time
	763	+	var attempts int
	764	+	found := false
	765	+	for _, p := range t.pending {
	766	+	if p == nil \|\| p.first.IsZero() {
	767	+	continue
	768	+	}
	769	+	if !found \|\| p.first.Before(oldest) {
	770	+	oldest = p.first
	771	+	attempts = p.attempts
	772	+	found = true
	773	+	}
	774	+	}
	775	+	t.mu.Unlock()
	776	+	if !found \|\| attempts < 1 \|\| now.Sub(oldest) < deadAfter {
	777	+	return
	778	+	}
	779	+	t.setLinkState(ports.LinkStateDead, ErrLinkDead)
	780	+	t.closeWithError(ErrLinkDead)
	781	+	}
	782	+
743	783		func (t *Transport) setLinkState(state ports.LinkState, err error) {
744	784		now := t.clock.Now()
745	785		t.mu.Lock()


@@ -177,6 +177,7 @@ type Transport struct {
177			177
178		type pending struct {	178		type pending struct {
179		frame ports.Frame	179		frame ports.Frame
			180	+	first time.Time
180		last time.Time	181		last time.Time
181		attempts int	182		attempts int
182		}	183		}

@@ -275,7 +276,8 @@ func (t *Transport) Send(f ports.Frame) error {
275		t.seq++	276		t.seq++
276		seq := t.seq	277		seq := t.seq
277		if reliable {	278		if reliable {
278	-	t.pending[seq] = &pending{frame: f, last: t.clock.Now()}	279	+	now := t.clock.Now()
			280	+	t.pending[seq] = &pending{frame: f, first: now, last: now}
279		}	281		}
280		t.mu.Unlock()	282		t.mu.Unlock()
281		if err := t.sendData(seq, reliable, f); err != nil {	283		if err := t.sendData(seq, reliable, f); err != nil {

@@ -624,10 +626,12 @@ func (t *Transport) resendLoop() {
624		case <-resend.C():	626		case <-resend.C():
625		t.resendPending()	627		t.resendPending()
626		t.checkSilence()	628		t.checkSilence()
			629	+	t.checkSendStall()
627		resend.Reset(t.resendAfter)	630		resend.Reset(t.resendAfter)
628		case <-heartbeat.C():	631		case <-heartbeat.C():
629		_ = t.sendProbe(recProbe, 0)	632		_ = t.sendProbe(recProbe, 0)
630		t.checkSilence()	633		t.checkSilence()
			634	+	t.checkSendStall()
631		heartbeat.Reset(t.heartbeat)	635		heartbeat.Reset(t.heartbeat)
632		case <-t.done:	636		case <-t.done:
633		return	637		return

@@ -732,14 +736,50 @@ func (t *Transport) checkSilence() {
732		t.closeWithError(ErrLinkDead)	736		t.closeWithError(ErrLinkDead)
733		case offlineAfter > 0 && silentFor >= offlineAfter:	737		case offlineAfter > 0 && silentFor >= offlineAfter:
734		t.setLinkState(ports.LinkStateOffline, nil)	738		t.setLinkState(ports.LinkStateOffline, nil)
735	-	t.hopPacketConnOnce()
736		case probeAfter > 0 && silentFor >= probeAfter:	739		case probeAfter > 0 && silentFor >= probeAfter:
737		t.setLinkState(ports.LinkStateProbing, nil)	740		t.setLinkState(ports.LinkStateProbing, nil)
			741	+	t.hopPacketConnOnce()
738		case degradedAfter > 0 && silentFor >= degradedAfter:	742		case degradedAfter > 0 && silentFor >= degradedAfter:
739		t.setLinkState(ports.LinkStateDegraded, nil)	743		t.setLinkState(ports.LinkStateDegraded, nil)
740		}	744		}
741		}	745		}
742			746
			747	+	// checkSendStall declares the link dead when the oldest unacked reliable frame
			748	+	// has gone unacknowledged for deadAfter despite at least one retransmit. The
			749	+	// receive-silence check in checkSilence misses one-way-dead links whose receive
			750	+	// side stays fresh from peer heartbeats; this catches them at the layer that
			751	+	// owns death. first is set once when a frame is enqueued and never refreshed,
			752	+	// so acks (which delete pending entries) keep the oldest age bounded on a
			753	+	// progressing link — only a frame that is never acked crosses the threshold.
			754	+	func (t *Transport) checkSendStall() {
			755	+	t.mu.Lock()
			756	+	if t.closed \|\| t.deadAfter <= 0 {
			757	+	t.mu.Unlock()
			758	+	return
			759	+	}
			760	+	deadAfter := t.deadAfter
			761	+	now := t.clock.Now()
			762	+	var oldest time.Time
			763	+	var attempts int
			764	+	found := false
			765	+	for _, p := range t.pending {
			766	+	if p == nil \|\| p.first.IsZero() {
			767	+	continue
			768	+	}
			769	+	if !found \|\| p.first.Before(oldest) {
			770	+	oldest = p.first
			771	+	attempts = p.attempts
			772	+	found = true
			773	+	}
			774	+	}
			775	+	t.mu.Unlock()
			776	+	if !found \|\| attempts < 1 \|\| now.Sub(oldest) < deadAfter {
			777	+	return
			778	+	}
			779	+	t.setLinkState(ports.LinkStateDead, ErrLinkDead)
			780	+	t.closeWithError(ErrLinkDead)
			781	+	}
			782	+
743		func (t *Transport) setLinkState(state ports.LinkState, err error) {	783		func (t *Transport) setLinkState(state ports.LinkState, err error) {
744		now := t.clock.Now()	784		now := t.clock.Now()
745		t.mu.Lock()	785		t.mu.Lock()

M internal/adapters/dgram/transport_test.go +95 -8


@@ -408,9 +408,9 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
408	408		var hopped atomic.Bool
409	409		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
410	410		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
411		-	ResendAfter: 20 * time.Millisecond,
412		-	OfflineAfter: time.Millisecond,
413		-	DeadAfter: time.Hour,
	411	+	ResendAfter: 20 * time.Millisecond,
	412	+	ProbeAfter: time.Millisecond,
	413	+	DeadAfter: time.Hour,
414	414		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
415	415		newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
416	416		bPC.peers["a2"] = newPC

@@ -448,10 +448,10 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
448	448		aPC, bPC := newPair()
449	449		var attempts atomic.Int32
450	450		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
451		-	ResendAfter: time.Hour,
452		-	Heartbeat: time.Hour,
453		-	OfflineAfter: time.Millisecond,
454		-	DeadAfter: time.Hour,
	451	+	ResendAfter: time.Hour,
	452	+	Heartbeat: time.Hour,
	453	+	ProbeAfter: time.Millisecond,
	454	+	DeadAfter: time.Hour,
455	455		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
456	456		if attempts.Add(1) == 1 {
457	457		return nil, errors.New("temporary rebind failure")

@@ -487,7 +487,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
487	487
488	488		func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
489	489		aPC, bPC := newPair()
490		-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{OfflineAfter: time.Millisecond, DeadAfter: time.Hour})
	490	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ProbeAfter: time.Millisecond, DeadAfter: time.Hour})
491	491		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
492	492		defer func() { _ = a.Close() }()
493	493		defer func() { _ = b.Close() }()

@@ -1005,6 +1005,93 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1005	1005		}
1006	1006		}
1007	1007
	1008	+	func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) {
	1009	+	tests := []struct {
	1010	+	name string
	1011	+	age time.Duration
	1012	+	attempts int
	1013	+	wantDead bool
	1014	+	}{
	1015	+	{name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false},
	1016	+	{name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false},
	1017	+	{name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true},
	1018	+	}
	1019	+	for _, tt := range tests {
	1020	+	t.Run(tt.name, func(t *testing.T) {
	1021	+	aPC, bPC := newPair()
	1022	+	// Drop everything a sends so the frame never gets acked, and keep the
	1023	+	// receive side fresh so only the send-stall check can declare death.
	1024	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	1025	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	1026	+	ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second,
	1027	+	})
	1028	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	1029	+	defer func() { _ = a.Close() }()
	1030	+	defer func() { _ = b.Close() }()
	1031	+
	1032	+	now := time.Now()
	1033	+	a.mu.Lock()
	1034	+	a.lastHeard = now
	1035	+	a.pending[1] = &pending{
	1036	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")},
	1037	+	first: now.Add(-tt.age),
	1038	+	last: now,
	1039	+	attempts: tt.attempts,
	1040	+	}
	1041	+	a.mu.Unlock()
	1042	+
	1043	+	a.checkSendStall()
	1044	+	if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead {
	1045	+	t.Fatalf("dead=%v, want %v", dead, tt.wantDead)
	1046	+	}
	1047	+	select {
	1048	+	case <-a.done:
	1049	+	if !tt.wantDead {
	1050	+	t.Fatal("transport closed without a send stall")
	1051	+	}
	1052	+	default:
	1053	+	if tt.wantDead {
	1054	+	t.Fatal("transport not closed after send stall")
	1055	+	}
	1056	+	}
	1057	+	})
	1058	+	}
	1059	+	}
	1060	+
	1061	+	func TestHopFiresAtProbingNotDegraded(t *testing.T) {
	1062	+	aPC, bPC := newPair()
	1063	+	var hopped atomic.Bool
	1064	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
	1065	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	1066	+	DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: time.Hour,
	1067	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	1068	+	newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
	1069	+	bPC.peers["a2"] = newPC
	1070	+	hopped.Store(true)
	1071	+	return newPC, nil
	1072	+	},
	1073	+	})
	1074	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	1075	+	defer func() { _ = a.Close() }()
	1076	+	defer func() { _ = b.Close() }()
	1077	+
	1078	+	a.mu.Lock()
	1079	+	a.lastHeard = time.Now().Add(-time.Second)
	1080	+	a.mu.Unlock()
	1081	+	a.checkSilence()
	1082	+	if hopped.Load() {
	1083	+	t.Fatal("socket hopped at degraded silence")
	1084	+	}
	1085	+
	1086	+	a.mu.Lock()
	1087	+	a.lastHeard = time.Now().Add(-2 * time.Second)
	1088	+	a.mu.Unlock()
	1089	+	a.checkSilence()
	1090	+	if !hopped.Load() {
	1091	+	t.Fatal("socket did not hop at probing silence")
	1092	+	}
	1093	+	}
	1094	+
1008	1095		func TestShortSilenceDoesNotKillTransport(t *testing.T) {
1009	1096		aPC, bPC := newPair()
1010	1097		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour})


@@ -408,9 +408,9 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
408		var hopped atomic.Bool	408		var hopped atomic.Bool
409		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	409		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
410		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{	410		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
411	-	ResendAfter: 20 * time.Millisecond,	411	+	ResendAfter: 20 * time.Millisecond,
412	-	OfflineAfter: time.Millisecond,	412	+	ProbeAfter: time.Millisecond,
413	-	DeadAfter: time.Hour,	413	+	DeadAfter: time.Hour,
414		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {	414		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
415		newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}	415		newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
416		bPC.peers["a2"] = newPC	416		bPC.peers["a2"] = newPC

@@ -448,10 +448,10 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
448		aPC, bPC := newPair()	448		aPC, bPC := newPair()
449		var attempts atomic.Int32	449		var attempts atomic.Int32
450		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{	450		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
451	-	ResendAfter: time.Hour,	451	+	ResendAfter: time.Hour,
452	-	Heartbeat: time.Hour,	452	+	Heartbeat: time.Hour,
453	-	OfflineAfter: time.Millisecond,	453	+	ProbeAfter: time.Millisecond,
454	-	DeadAfter: time.Hour,	454	+	DeadAfter: time.Hour,
455		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {	455		RebindPacketConn: func(old net.PacketConn) (net.PacketConn, error) {
456		if attempts.Add(1) == 1 {	456		if attempts.Add(1) == 1 {
457		return nil, errors.New("temporary rebind failure")	457		return nil, errors.New("temporary rebind failure")

@@ -487,7 +487,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
487			487
488		func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {	488		func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
489		aPC, bPC := newPair()	489		aPC, bPC := newPair()
490	-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{OfflineAfter: time.Millisecond, DeadAfter: time.Hour})	490	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{ProbeAfter: time.Millisecond, DeadAfter: time.Hour})
491		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)	491		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
492		defer func() { _ = a.Close() }()	492		defer func() { _ = a.Close() }()
493		defer func() { _ = b.Close() }()	493		defer func() { _ = b.Close() }()

@@ -1005,6 +1005,93 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1005		}	1005		}
1006		}	1006		}
1007			1007
			1008	+	func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) {
			1009	+	tests := []struct {
			1010	+	name string
			1011	+	age time.Duration
			1012	+	attempts int
			1013	+	wantDead bool
			1014	+	}{
			1015	+	{name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false},
			1016	+	{name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false},
			1017	+	{name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true},
			1018	+	}
			1019	+	for _, tt := range tests {
			1020	+	t.Run(tt.name, func(t *testing.T) {
			1021	+	aPC, bPC := newPair()
			1022	+	// Drop everything a sends so the frame never gets acked, and keep the
			1023	+	// receive side fresh so only the send-stall check can declare death.
			1024	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
			1025	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
			1026	+	ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second,
			1027	+	})
			1028	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
			1029	+	defer func() { _ = a.Close() }()
			1030	+	defer func() { _ = b.Close() }()
			1031	+
			1032	+	now := time.Now()
			1033	+	a.mu.Lock()
			1034	+	a.lastHeard = now
			1035	+	a.pending[1] = &pending{
			1036	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")},
			1037	+	first: now.Add(-tt.age),
			1038	+	last: now,
			1039	+	attempts: tt.attempts,
			1040	+	}
			1041	+	a.mu.Unlock()
			1042	+
			1043	+	a.checkSendStall()
			1044	+	if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead {
			1045	+	t.Fatalf("dead=%v, want %v", dead, tt.wantDead)
			1046	+	}
			1047	+	select {
			1048	+	case <-a.done:
			1049	+	if !tt.wantDead {
			1050	+	t.Fatal("transport closed without a send stall")
			1051	+	}
			1052	+	default:
			1053	+	if tt.wantDead {
			1054	+	t.Fatal("transport not closed after send stall")
			1055	+	}
			1056	+	}
			1057	+	})
			1058	+	}
			1059	+	}
			1060	+
			1061	+	func TestHopFiresAtProbingNotDegraded(t *testing.T) {
			1062	+	aPC, bPC := newPair()
			1063	+	var hopped atomic.Bool
			1064	+	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
			1065	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
			1066	+	DegradedAfter: time.Second, ProbeAfter: 2 * time.Second, OfflineAfter: 3 * time.Second, DeadAfter: time.Hour,
			1067	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
			1068	+	newPC := &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}
			1069	+	bPC.peers["a2"] = newPC
			1070	+	hopped.Store(true)
			1071	+	return newPC, nil
			1072	+	},
			1073	+	})
			1074	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
			1075	+	defer func() { _ = a.Close() }()
			1076	+	defer func() { _ = b.Close() }()
			1077	+
			1078	+	a.mu.Lock()
			1079	+	a.lastHeard = time.Now().Add(-time.Second)
			1080	+	a.mu.Unlock()
			1081	+	a.checkSilence()
			1082	+	if hopped.Load() {
			1083	+	t.Fatal("socket hopped at degraded silence")
			1084	+	}
			1085	+
			1086	+	a.mu.Lock()
			1087	+	a.lastHeard = time.Now().Add(-2 * time.Second)
			1088	+	a.mu.Unlock()
			1089	+	a.checkSilence()
			1090	+	if !hopped.Load() {
			1091	+	t.Fatal("socket did not hop at probing silence")
			1092	+	}
			1093	+	}
			1094	+
1008		func TestShortSilenceDoesNotKillTransport(t *testing.T) {	1095		func TestShortSilenceDoesNotKillTransport(t *testing.T) {
1009		aPC, bPC := newPair()	1096		aPC, bPC := newPair()
1010		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour})	1097		a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{DegradedAfter: 50 * time.Millisecond, DeadAfter: time.Hour})

M internal/usecase/client/client.go +6 -0


@@ -458,6 +458,12 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
458	458		continue
459	459		}
460	460		drawStatusStage(stageForLinkState(ev.State))
	461	+	if ev.State == ports.LinkStateOffline {
	462	+	// Stop waiting for the transport to reach Dead at 60s: exit with
	463	+	// a retryable error so Run re-dials over ssh with the resume
	464	+	// token. Run closes this transport on the way out.
	465	+	return welcomedResult(errLinkOffline)
	466	+	}
461	467		case <-colorQueryCh:
462	468		if err := term.QueryColors(); err != nil {
463	469		log.Warn("querying terminal colors", "err", err)


@@ -458,6 +458,12 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
458	continue	458		continue
459	}	459		}
460	drawStatusStage(stageForLinkState(ev.State))	460		drawStatusStage(stageForLinkState(ev.State))
		461	+	if ev.State == ports.LinkStateOffline {
		462	+	// Stop waiting for the transport to reach Dead at 60s: exit with
		463	+	// a retryable error so Run re-dials over ssh with the resume
		464	+	// token. Run closes this transport on the way out.
		465	+	return welcomedResult(errLinkOffline)
		466	+	}
461	case <-colorQueryCh:	467		case <-colorQueryCh:
462	if err := term.QueryColors(); err != nil {	468		if err := term.QueryColors(); err != nil {
463	log.Warn("querying terminal colors", "err", err)	469		log.Warn("querying terminal colors", "err", err)

M internal/usecase/client/reconnect.go +5 -0


@@ -14,6 +14,11 @@ type reconnectBackoff struct {
14	14
15	15		var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second}
16	16
	17	+	// errLinkOffline is returned by attachOnce when the transport link reaches the
	18	+	// Offline state. It is retryable so Run re-dials over ssh with the resume token
	19	+	// at ~30s of silence instead of waiting the full Dead timeout at 60s.
	20	+	var errLinkOffline = errors.New("vev: link offline; reconnecting")
	21	+
17	22		type reconnectStage int
18	23
19	24		const (


@@ -14,6 +14,11 @@ type reconnectBackoff struct {
14		14
15	var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second}	15		var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second}
16		16
		17	+	// errLinkOffline is returned by attachOnce when the transport link reaches the
		18	+	// Offline state. It is retryable so Run re-dials over ssh with the resume token
		19	+	// at ~30s of silence instead of waiting the full Dead timeout at 60s.
		20	+	var errLinkOffline = errors.New("vev: link offline; reconnecting")
		21	+
17	type reconnectStage int	22		type reconnectStage int
18		23
19	const (	24		const (

M internal/usecase/client/reconnect_toast_test.go +26 -0


@@ -187,6 +187,32 @@ func TestAttachOnceLinkStateEventsDriveReconnectUIStage(t *testing.T) {
187	187		require.True(t, result.welcomed)
188	188		}
189	189
	190	+	func TestAttachOnceOfflineLinkEventReturnsReconnectableError(t *testing.T) {
	191	+	term := newReconnectToastTerminalHarness(t)
	192	+	defer term.closeInput()
	193	+	tr := newReconnectToastLinkTransport()
	194	+	tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)}
	195	+
	196	+	stages := make(chan reconnectStage, 4)
	197	+	resultCh := make(chan attachResult, 1)
	198	+	ms := milestones{}
	199	+	go func() {
	200	+	resultCh <- attachOnce(context.Background(), tr, term.term, portsmocks.NewMockClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, func() error {
	201	+	_, err := term.term.EnterRaw()
	202	+	return err
	203	+	}, func() {}, func(stage reconnectStage) {
	204	+	stages <- stage
	205	+	}, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler))
	206	+	}()
	207	+
	208	+	tr.events <- ports.LinkEvent{State: ports.LinkStateOffline}
	209	+	require.Equal(t, reconnectStageOfflineRetrying, <-stages)
	210	+	result := <-resultCh
	211	+	require.ErrorIs(t, result.err, errLinkOffline)
	212	+	require.True(t, result.welcomed)
	213	+	require.True(t, shouldReconnect(result.err), "offline exit must be reconnectable")
	214	+	}
	215	+
190	216		func TestReconnectToastDrawAndClearHelpers(t *testing.T) {
191	217		var out bytes.Buffer
192	218		size := domain.Size{Cols: 80, Rows: 24}


@@ -187,6 +187,32 @@ func TestAttachOnceLinkStateEventsDriveReconnectUIStage(t *testing.T) {
187	require.True(t, result.welcomed)	187		require.True(t, result.welcomed)
188	}	188		}
189		189
		190	+	func TestAttachOnceOfflineLinkEventReturnsReconnectableError(t *testing.T) {
		191	+	term := newReconnectToastTerminalHarness(t)
		192	+	defer term.closeInput()
		193	+	tr := newReconnectToastLinkTransport()
		194	+	tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)}
		195	+
		196	+	stages := make(chan reconnectStage, 4)
		197	+	resultCh := make(chan attachResult, 1)
		198	+	ms := milestones{}
		199	+	go func() {
		200	+	resultCh <- attachOnce(context.Background(), tr, term.term, portsmocks.NewMockClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, func() error {
		201	+	_, err := term.term.EnterRaw()
		202	+	return err
		203	+	}, func() {}, func(stage reconnectStage) {
		204	+	stages <- stage
		205	+	}, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler))
		206	+	}()
		207	+
		208	+	tr.events <- ports.LinkEvent{State: ports.LinkStateOffline}
		209	+	require.Equal(t, reconnectStageOfflineRetrying, <-stages)
		210	+	result := <-resultCh
		211	+	require.ErrorIs(t, result.err, errLinkOffline)
		212	+	require.True(t, result.welcomed)
		213	+	require.True(t, shouldReconnect(result.err), "offline exit must be reconnectable")
		214	+	}
		215	+
190	func TestReconnectToastDrawAndClearHelpers(t *testing.T) {	216		func TestReconnectToastDrawAndClearHelpers(t *testing.T) {
191	var out bytes.Buffer	217		var out bytes.Buffer
192	size := domain.Size{Cols: 80, Rows: 24}	218		size := domain.Size{Cols: 80, Rows: 24}

M internal/usecase/daemon/client.go +32 -0


@@ -49,6 +49,7 @@ type attachedClient struct {
49	49		nextStateNum uint64
50	50		echoAck atomic.Uint64
51	51		outputAck atomic.Uint64
	52	+	paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag
52	53		bars barCache // only touched while sendMu is held
53	54		size domain.Size
54	55		keys *keys.Router

@@ -225,6 +226,34 @@ func (ac *attachedClient) advanceOutputAck(state uint64) {
225	226		}
226	227		}
227	228
	229	+	// outputLagAtCapLocked reports whether the client has at least
	230	+	// maxUnackedOutputStates output states in flight (sent but unacked). Caller
	231	+	// holds sendMu.
	232	+	func (ac *attachedClient) outputLagAtCapLocked() bool {
	233	+	return ac.nextStateNum >= ac.outputAck.Load()+maxUnackedOutputStates
	234	+	}
	235	+
	236	+	// takeDeferredPaint reports whether a deferred diff paint is now flushable
	237	+	// because acks have caught up, clearing the flag when it is. The caller must
	238	+	// then paint outside sendMu (paint re-acquires it).
	239	+	func (ac *attachedClient) takeDeferredPaint() bool {
	240	+	ac.sendMu.Lock()
	241	+	defer ac.sendMu.Unlock()
	242	+	if ac.paintDeferred && !ac.outputLagAtCapLocked() {
	243	+	ac.paintDeferred = false
	244	+	return true
	245	+	}
	246	+	return false
	247	+	}
	248	+
	249	+	// clearPaintDeferred drops any pending deferred-paint flag. Used on park so a
	250	+	// resumed client does not inherit a stale deferral.
	251	+	func (ac *attachedClient) clearPaintDeferred() {
	252	+	ac.sendMu.Lock()
	253	+	ac.paintDeferred = false
	254	+	ac.sendMu.Unlock()
	255	+	}
	256	+
228	257		// send serialises a frame onto the client's transport.
229	258		func (ac *attachedClient) send(f ports.Frame) error {
230	259		_, err := ac.sendTransport(f)

@@ -492,6 +521,9 @@ func (d Daemon) runConnLoop(ac attachedClient) {
492	521		case ports.MsgAck:
493	522		if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {
494	523		ac.advanceOutputAck(ack.AckedStateNum)
	524	+	if ac.takeDeferredPaint() {
	525	+	d.paint(sess, ac, false)
	526	+	}
495	527		}
496	528		case ports.MsgPing:
497	529		_ = ac.send(framePong())


@@ -49,6 +49,7 @@ type attachedClient struct {
49	nextStateNum uint64	49		nextStateNum uint64
50	echoAck atomic.Uint64	50		echoAck atomic.Uint64
51	outputAck atomic.Uint64	51		outputAck atomic.Uint64
		52	+	paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag
52	bars barCache // only touched while sendMu is held	53		bars barCache // only touched while sendMu is held
53	size domain.Size	54		size domain.Size
54	keys *keys.Router	55		keys *keys.Router

@@ -225,6 +226,34 @@ func (ac *attachedClient) advanceOutputAck(state uint64) {
225	}	226		}
226	}	227		}
227		228
		229	+	// outputLagAtCapLocked reports whether the client has at least
		230	+	// maxUnackedOutputStates output states in flight (sent but unacked). Caller
		231	+	// holds sendMu.
		232	+	func (ac *attachedClient) outputLagAtCapLocked() bool {
		233	+	return ac.nextStateNum >= ac.outputAck.Load()+maxUnackedOutputStates
		234	+	}
		235	+
		236	+	// takeDeferredPaint reports whether a deferred diff paint is now flushable
		237	+	// because acks have caught up, clearing the flag when it is. The caller must
		238	+	// then paint outside sendMu (paint re-acquires it).
		239	+	func (ac *attachedClient) takeDeferredPaint() bool {
		240	+	ac.sendMu.Lock()
		241	+	defer ac.sendMu.Unlock()
		242	+	if ac.paintDeferred && !ac.outputLagAtCapLocked() {
		243	+	ac.paintDeferred = false
		244	+	return true
		245	+	}
		246	+	return false
		247	+	}
		248	+
		249	+	// clearPaintDeferred drops any pending deferred-paint flag. Used on park so a
		250	+	// resumed client does not inherit a stale deferral.
		251	+	func (ac *attachedClient) clearPaintDeferred() {
		252	+	ac.sendMu.Lock()
		253	+	ac.paintDeferred = false
		254	+	ac.sendMu.Unlock()
		255	+	}
		256	+
228	// send serialises a frame onto the client's transport.	257		// send serialises a frame onto the client's transport.
229	func (ac *attachedClient) send(f ports.Frame) error {	258		func (ac *attachedClient) send(f ports.Frame) error {
230	_, err := ac.sendTransport(f)	259		_, err := ac.sendTransport(f)

@@ -492,6 +521,9 @@ func (d Daemon) runConnLoop(ac attachedClient) {
492	case ports.MsgAck:	521		case ports.MsgAck:
493	if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {	522		if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {
494	ac.advanceOutputAck(ack.AckedStateNum)	523		ac.advanceOutputAck(ack.AckedStateNum)
		524	+	if ac.takeDeferredPaint() {
		525	+	d.paint(sess, ac, false)
		526	+	}
495	}	527		}
496	case ports.MsgPing:	528		case ports.MsgPing:
497	_ = ac.send(framePong())	529		_ = ac.send(framePong())

M internal/usecase/daemon/daemon.go +8 -0


@@ -65,6 +65,14 @@ const defaultScrollbackRows = 10_000
65	65		// the in-flight send. Teardown is never gated on a client draining its socket.
66	66		const detachNotifyTimeout = time.Second
67	67
	68	+	// maxUnackedOutputStates caps how many output states may be in flight (sent but
	69	+	// not yet acked by the client) before paint defers rather than composing
	70	+	// another diff. It bounds the daemon's paint rate to the client's ack rate, so
	71	+	// heavy output degrades to lower fps on a slow link instead of overflowing the
	72	+	// transport. It must stay well under the UDP proxy's 32-frame client window so
	73	+	// the proxy's reliable queue never fills from painting alone.
	74	+	const maxUnackedOutputStates = 8
	75	+
68	76		const defaultResumeParkGrace = 15 * time.Minute
69	77
70	78		// defaultSize is used when a client's Hello carries no valid dimensions.


@@ -65,6 +65,14 @@ const defaultScrollbackRows = 10_000
65	// the in-flight send. Teardown is never gated on a client draining its socket.	65		// the in-flight send. Teardown is never gated on a client draining its socket.
66	const detachNotifyTimeout = time.Second	66		const detachNotifyTimeout = time.Second
67		67
		68	+	// maxUnackedOutputStates caps how many output states may be in flight (sent but
		69	+	// not yet acked by the client) before paint defers rather than composing
		70	+	// another diff. It bounds the daemon's paint rate to the client's ack rate, so
		71	+	// heavy output degrades to lower fps on a slow link instead of overflowing the
		72	+	// transport. It must stay well under the UDP proxy's 32-frame client window so
		73	+	// the proxy's reliable queue never fills from painting alone.
		74	+	const maxUnackedOutputStates = 8
		75	+
68	const defaultResumeParkGrace = 15 * time.Minute	76		const defaultResumeParkGrace = 15 * time.Minute
69		77
70	// defaultSize is used when a client's Hello carries no valid dimensions.	78		// defaultSize is used when a client's Hello carries no valid dimensions.

A internal/usecase/daemon/paint_gate_test.go +98 -0

@@ -0,0 +1,98 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"io"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	10	+	"github.com/stretchr/testify/require"
	11	+	)
	12	+
	13	+	func requireNoOutputFrame(t *testing.T, sends chan ports.Frame) {
	14	+	t.Helper()
	15	+	deadline := time.After(50 * time.Millisecond)
	16	+	for {
	17	+	select {
	18	+	case f := <-sends:
	19	+	if f.Type == ports.MsgOutput {
	20	+	t.Fatalf("unexpected output frame: %+v", f)
	21	+	}
	22	+	case <-deadline:
	23	+	return
	24	+	}
	25	+	}
	26	+	}
	27	+
	28	+	func TestPaintDefersWhenOutputAckLagsAtCap(t *testing.T) {
	29	+	p, _ := newBlockingPTY(t)
	30	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	31	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	32	+
	33	+	// Client is maxUnackedOutputStates behind (outputAck stays 0).
	34	+	ac.nextStateNum = maxUnackedOutputStates
	35	+	d.paint(sess, ac, false)
	36	+
	37	+	requireNoOutputFrame(t, sends)
	38	+	ac.sendMu.Lock()
	39	+	deferred := ac.paintDeferred
	40	+	ac.sendMu.Unlock()
	41	+	require.True(t, deferred, "diff paint at ack cap must defer")
	42	+	}
	43	+
	44	+	func TestDeferredPaintFlushesOnceWhenAcksCatchUp(t *testing.T) {
	45	+	p, _ := newBlockingPTY(t)
	46	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	47	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	48	+
	49	+	ac.nextStateNum = maxUnackedOutputStates
	50	+	d.paint(sess, ac, false)
	51	+	requireNoOutputFrame(t, sends)
	52	+
	53	+	// Drive a real MsgAck through the connection loop so the flush is exercised
	54	+	// end-to-end: UnmarshalAck → advanceOutputAck → takeDeferredPaint → paint.
	55	+	// One ack frame, then Recv blocks until the test ends.
	56	+	recvCh := make(chan ports.Frame, 1)
	57	+	done := make(chan struct{})
	58	+	t.Cleanup(func() { close(done) })
	59	+	tr, ok := ac.transport().(*portsmocks.MockTransport)
	60	+	require.True(t, ok, "manual session transport must be the capturing mock")
	61	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	62	+	select {
	63	+	case f := <-recvCh:
	64	+	return f, nil
	65	+	case <-done:
	66	+	return ports.Frame{}, io.EOF
	67	+	}
	68	+	}).Maybe()
	69	+	recvCh <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: maxUnackedOutputStates})}
	70	+	go d.runConnLoop(ac)
	71	+
	72	+	// The cumulative flush paint must be triggered by the MsgAck handler itself
	73	+	// (the renderer shadow only advanced on send, so the skip coalesces here).
	74	+	awaitFrame(t, sends, ports.MsgOutput)
	75	+	requireNoOutputFrame(t, sends)
	76	+	ac.sendMu.Lock()
	77	+	deferred := ac.paintDeferred
	78	+	ac.sendMu.Unlock()
	79	+	require.False(t, deferred, "flag must clear after the flush paint")
	80	+	}
	81	+
	82	+	func TestResetPaintBypassesAckGate(t *testing.T) {
	83	+	p, _ := newBlockingPTY(t)
	84	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	85	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	86	+
	87	+	// At the cap and already deferred: a reset paint is a full invalidation that
	88	+	// must paint anyway and clear the deferral.
	89	+	ac.nextStateNum = maxUnackedOutputStates
	90	+	ac.paintDeferred = true
	91	+	d.paint(sess, ac, true)
	92	+
	93	+	awaitFrame(t, sends, ports.MsgOutput)
	94	+	ac.sendMu.Lock()
	95	+	deferred := ac.paintDeferred
	96	+	ac.sendMu.Unlock()
	97	+	require.False(t, deferred, "reset paint must clear the deferral")
	98	+	}

@@ -0,0 +1,98 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"io"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	10	+	"github.com/stretchr/testify/require"
	11	+	)
	12	+
	13	+	func requireNoOutputFrame(t *testing.T, sends chan ports.Frame) {
	14	+	t.Helper()
	15	+	deadline := time.After(50 * time.Millisecond)
	16	+	for {
	17	+	select {
	18	+	case f := <-sends:
	19	+	if f.Type == ports.MsgOutput {
	20	+	t.Fatalf("unexpected output frame: %+v", f)
	21	+	}
	22	+	case <-deadline:
	23	+	return
	24	+	}
	25	+	}
	26	+	}
	27	+
	28	+	func TestPaintDefersWhenOutputAckLagsAtCap(t *testing.T) {
	29	+	p, _ := newBlockingPTY(t)
	30	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	31	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	32	+
	33	+	// Client is maxUnackedOutputStates behind (outputAck stays 0).
	34	+	ac.nextStateNum = maxUnackedOutputStates
	35	+	d.paint(sess, ac, false)
	36	+
	37	+	requireNoOutputFrame(t, sends)
	38	+	ac.sendMu.Lock()
	39	+	deferred := ac.paintDeferred
	40	+	ac.sendMu.Unlock()
	41	+	require.True(t, deferred, "diff paint at ack cap must defer")
	42	+	}
	43	+
	44	+	func TestDeferredPaintFlushesOnceWhenAcksCatchUp(t *testing.T) {
	45	+	p, _ := newBlockingPTY(t)
	46	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	47	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	48	+
	49	+	ac.nextStateNum = maxUnackedOutputStates
	50	+	d.paint(sess, ac, false)
	51	+	requireNoOutputFrame(t, sends)
	52	+
	53	+	// Drive a real MsgAck through the connection loop so the flush is exercised
	54	+	// end-to-end: UnmarshalAck → advanceOutputAck → takeDeferredPaint → paint.
	55	+	// One ack frame, then Recv blocks until the test ends.
	56	+	recvCh := make(chan ports.Frame, 1)
	57	+	done := make(chan struct{})
	58	+	t.Cleanup(func() { close(done) })
	59	+	tr, ok := ac.transport().(*portsmocks.MockTransport)
	60	+	require.True(t, ok, "manual session transport must be the capturing mock")
	61	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	62	+	select {
	63	+	case f := <-recvCh:
	64	+	return f, nil
	65	+	case <-done:
	66	+	return ports.Frame{}, io.EOF
	67	+	}
	68	+	}).Maybe()
	69	+	recvCh <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: maxUnackedOutputStates})}
	70	+	go d.runConnLoop(ac)
	71	+
	72	+	// The cumulative flush paint must be triggered by the MsgAck handler itself
	73	+	// (the renderer shadow only advanced on send, so the skip coalesces here).
	74	+	awaitFrame(t, sends, ports.MsgOutput)
	75	+	requireNoOutputFrame(t, sends)
	76	+	ac.sendMu.Lock()
	77	+	deferred := ac.paintDeferred
	78	+	ac.sendMu.Unlock()
	79	+	require.False(t, deferred, "flag must clear after the flush paint")
	80	+	}
	81	+
	82	+	func TestResetPaintBypassesAckGate(t *testing.T) {
	83	+	p, _ := newBlockingPTY(t)
	84	+	d, sess, ac, sends := newManualSessionWithPTYs(t, p)
	85	+	sess.tabs[0].focusedPane().screen.Write([]byte("A"))
	86	+
	87	+	// At the cap and already deferred: a reset paint is a full invalidation that
	88	+	// must paint anyway and clear the deferral.
	89	+	ac.nextStateNum = maxUnackedOutputStates
	90	+	ac.paintDeferred = true
	91	+	d.paint(sess, ac, true)
	92	+
	93	+	awaitFrame(t, sends, ports.MsgOutput)
	94	+	ac.sendMu.Lock()
	95	+	deferred := ac.paintDeferred
	96	+	ac.sendMu.Unlock()
	97	+	require.False(t, deferred, "reset paint must clear the deferral")
	98	+	}

M internal/usecase/daemon/render.go +12 -0


@@ -280,6 +280,18 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool) {
280	280
281	281		ac.initOverlays()
282	282		ac.sendMu.Lock()
	283	+	// Ack-gated coalescing: while the client is at least maxUnackedOutputStates
	284	+	// behind, skip composing another diff and mark the paint deferred. The
	285	+	// MsgOutputAck handler flushes one cumulative paint once acks catch up.
	286	+	// A reset paint is an explicit full invalidation that supersedes any queued
	287	+	// diffs, so it bypasses the gate and clears the flag.
	288	+	if reset {
	289	+	ac.paintDeferred = false
	290	+	} else if ac.outputLagAtCapLocked() {
	291	+	ac.paintDeferred = true
	292	+	ac.sendMu.Unlock()
	293	+	return
	294	+	}
283	295		overlays := ac.overlays.SnapshotForRender()
284	296		repaintAttachedClients := false
285	297		defer func() {


@@ -280,6 +280,18 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool) {
280		280
281	ac.initOverlays()	281		ac.initOverlays()
282	ac.sendMu.Lock()	282		ac.sendMu.Lock()
		283	+	// Ack-gated coalescing: while the client is at least maxUnackedOutputStates
		284	+	// behind, skip composing another diff and mark the paint deferred. The
		285	+	// MsgOutputAck handler flushes one cumulative paint once acks catch up.
		286	+	// A reset paint is an explicit full invalidation that supersedes any queued
		287	+	// diffs, so it bypasses the gate and clears the flag.
		288	+	if reset {
		289	+	ac.paintDeferred = false
		290	+	} else if ac.outputLagAtCapLocked() {
		291	+	ac.paintDeferred = true
		292	+	ac.sendMu.Unlock()
		293	+	return
		294	+	}
283	overlays := ac.overlays.SnapshotForRender()	295		overlays := ac.overlays.SnapshotForRender()
284	repaintAttachedClients := false	296		repaintAttachedClients := false
285	defer func() {	297		defer func() {

M internal/usecase/daemon/resume.go +3 -0


@@ -48,6 +48,9 @@ func (d Daemon) parkAttachment(sess session, ac *attachedClient) bool {
48	48		if !d.prepareParkAttachment(sess, ac) {
49	49		return false
50	50		}
	51	+	// Drop any deferred diff so a resumed client starts clean; resume forces a
	52	+	// full reset repaint anyway.
	53	+	ac.clearPaintDeferred()
51	54		d.mu.Lock()
52	55		if d.closing {
53	56		d.mu.Unlock()


@@ -48,6 +48,9 @@ func (d Daemon) parkAttachment(sess session, ac *attachedClient) bool {
48	if !d.prepareParkAttachment(sess, ac) {	48		if !d.prepareParkAttachment(sess, ac) {
49	return false	49		return false
50	}	50		}
		51	+	// Drop any deferred diff so a resumed client starts clean; resume forces a
		52	+	// full reset repaint anyway.
		53	+	ac.clearPaintDeferred()
51	d.mu.Lock()	54		d.mu.Lock()
52	if d.closing {	55		if d.closing {
53	d.mu.Unlock()	56		d.mu.Unlock()