fix(dgram): keep the UDP proxy alive under congestion

Retry the same daemon->client frame in place while the link error stays
recoverable instead of restarting the copier and dropping the in-flight
frame, which desynchronized the client screen on a reliable stream.

Retries no longer arm the idle TTL: it is driven solely by link-state
events (armed on Degraded/Probing/Offline, disarmed on Connected, Dead is
immediate). A persistently pending-full but Connected link therefore
survives rather than being killed after 30s. Send-stall death now comes
from the transport layer.

commit: 5e4db397661f28a354dabb170621351d7024c4c1
author: brice · 3mo ago · 2026-07-08 16:52 UTC
parent: ce08837
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +73 -214

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")
 	}
 }