detached 98aa679

fix(remote): address proxy link, output, and render review findings

- distinguish a failed reply from a malformed frame with an errProxyLinkSend
  sentinel, so a broken write retires only that transport instead of stopping
  the whole proxy link
- ignore retransmitted base-zero output whose state is at or below the applied
  state, unless this proxy actually asked for the reset; a stale full paint
  could otherwise clobber a newer screen and regress appliedState
- split proxySession.resize into changed/sent, and repaint local chrome
  whenever the local content rectangle moved even if the remote resize never
  went out; a no-op resize no longer repaints
- repaint after a failed render transaction on attachments with no local
  session, keeping the error notice itself session-scoped
- clean up proxy attachments on transport failure via detachProxyOnSendError
  and detachProxyIfCurrentTransport
- drive proxy tests off an injected ports.Clock instead of wall-clock resume
  backoff, and assert against a registry snapshot taken under d.mu rather than
  reading d.sessions while registration may write it
commit
98aa679396f48975aa96e5238e4d2500ea7a38a5
author
brice · · 2026-07-31 12:42 UTC
parent
6a6edad
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
8 files · +338 -36
Unified
UnifiedSplit
M internal/usecase/daemon/client_transport_errors.go +28 -0
@@ -107,6 +107,34 @@ func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient, failed por
107107 d.clearParkingInFlightIfAbandoned(sess, ac, parkingToken)
108108 }
109109
110+// detachProxyOnSendError is detachOnSendError for an attachment that has no
111+// local session behind it. Proxy attachments own no resumable parking
112+// lifecycle, so the remote link is retained as a warm headless session instead
113+// of being parked for the same client to resume.
114+func (d *Daemon) detachProxyOnSendError(p *proxySession, ac *attachedClient, failed ports.Transport) {
115+ if d == nil || p == nil || ac == nil {
116+ return
117+ }
118+ expected := transportSnapshot{}
119+ if failed != nil {
120+ expected = ac.transportSnapshot()
121+ if expected.transport != failed {
122+ return
123+ }
124+ }
125+ if !d.detachProxyIfCurrentTransport(p, ac, expected) {
126+ return
127+ }
128+ if rc := p.coordinator.Load(); rc != nil {
129+ rc.noteDetach(ac)
130+ }
131+ d.unregisterPreview(ac)
132+ ac.clearPreviousSession()
133+ _ = ac.closeCapturedTransport(failed)
134+ d.armProxyWarm(p)
135+ d.log.Warn("detached proxy client after send error", "host", p.key.Host, "session", p.key.Name)
136+}
137+
110138 func (d *Daemon) detachOnRoleSendError(token attachmentRoleToken, failed ports.Transport) {
111139 d.detachOnRoleSendErrorUntil(token, failed, nil)
112140 }
@@ -107,6 +107,34 @@ func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient, failed por
107 d.clearParkingInFlightIfAbandoned(sess, ac, parkingToken)107 d.clearParkingInFlightIfAbandoned(sess, ac, parkingToken)
108 }108 }
109 109
110+// detachProxyOnSendError is detachOnSendError for an attachment that has no
111+// local session behind it. Proxy attachments own no resumable parking
112+// lifecycle, so the remote link is retained as a warm headless session instead
113+// of being parked for the same client to resume.
114+func (d *Daemon) detachProxyOnSendError(p *proxySession, ac *attachedClient, failed ports.Transport) {
115+ if d == nil || p == nil || ac == nil {
116+ return
117+ }
118+ expected := transportSnapshot{}
119+ if failed != nil {
120+ expected = ac.transportSnapshot()
121+ if expected.transport != failed {
122+ return
123+ }
124+ }
125+ if !d.detachProxyIfCurrentTransport(p, ac, expected) {
126+ return
127+ }
128+ if rc := p.coordinator.Load(); rc != nil {
129+ rc.noteDetach(ac)
130+ }
131+ d.unregisterPreview(ac)
132+ ac.clearPreviousSession()
133+ _ = ac.closeCapturedTransport(failed)
134+ d.armProxyWarm(p)
135+ d.log.Warn("detached proxy client after send error", "host", p.key.Host, "session", p.key.Name)
136+}
137+
110 func (d *Daemon) detachOnRoleSendError(token attachmentRoleToken, failed ports.Transport) {138 func (d *Daemon) detachOnRoleSendError(token attachmentRoleToken, failed ports.Transport) {
111 d.detachOnRoleSendErrorUntil(token, failed, nil)139 d.detachOnRoleSendErrorUntil(token, failed, nil)
112 }140 }
M internal/usecase/daemon/proxy_link.go +11 -1
@@ -22,6 +22,10 @@ var (
2222 errProxyCommandTimeout = errors.New("proxy session: attached command timed out")
2323 errProxyCommandUnavailable = errors.New("proxy session: attached command link is unavailable")
2424 errProxyCommandGeneration = errors.New("proxy session: attached command link was replaced")
25+ // errProxyLinkSend marks a reply that could not be written to the remote.
26+ // The received frame was well formed, so the link is resumable rather than
27+ // terminally broken.
28+ errProxyLinkSend = errors.New("proxy session: link reply send failed")
2529 )
2630
2731 type proxyCommandOutcome struct {
@@ -520,6 +524,12 @@ func (d *Daemon) runProxyTransport(ctx context.Context, p *proxySession, generat
520524 }
521525 result, err := d.handleLinkFrame(p, generation, received.frame)
522526 if err != nil {
527+ // A failed reply says nothing about the remote's framing, so it
528+ // retires this transport instead of the whole proxy link.
529+ if errors.Is(err, errProxyLinkSend) {
530+ d.log.Warn("proxy link reply failed", "host", p.key.Host, "session", p.key.Name, "err", err)
531+ return proxyLinkResume
532+ }
523533 d.log.Warn("invalid proxy link frame", "host", p.key.Host, "session", p.key.Name, "err", err)
524534 return proxyLinkStop
525535 }
@@ -567,7 +577,7 @@ func (d *Daemon) handleLinkFrame(p *proxySession, generation uint64, frame ports
567577 return proxyLinkStop, err
568578 }
569579 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}); err != nil {
570- return proxyLinkStop, err
580+ return proxyLinkStop, fmt.Errorf("proxy pong reply: %w: %w", errProxyLinkSend, err)
571581 }
572582 return proxyLinkResume, nil
573583 case ports.MsgDetached:
@@ -22,6 +22,10 @@ var (
22 errProxyCommandTimeout = errors.New("proxy session: attached command timed out")22 errProxyCommandTimeout = errors.New("proxy session: attached command timed out")
23 errProxyCommandUnavailable = errors.New("proxy session: attached command link is unavailable")23 errProxyCommandUnavailable = errors.New("proxy session: attached command link is unavailable")
24 errProxyCommandGeneration = errors.New("proxy session: attached command link was replaced")24 errProxyCommandGeneration = errors.New("proxy session: attached command link was replaced")
25+ // errProxyLinkSend marks a reply that could not be written to the remote.
26+ // The received frame was well formed, so the link is resumable rather than
27+ // terminally broken.
28+ errProxyLinkSend = errors.New("proxy session: link reply send failed")
25 )29 )
26 30
27 type proxyCommandOutcome struct {31 type proxyCommandOutcome struct {
@@ -520,6 +524,12 @@ func (d *Daemon) runProxyTransport(ctx context.Context, p *proxySession, generat
520 }524 }
521 result, err := d.handleLinkFrame(p, generation, received.frame)525 result, err := d.handleLinkFrame(p, generation, received.frame)
522 if err != nil {526 if err != nil {
527+ // A failed reply says nothing about the remote's framing, so it
528+ // retires this transport instead of the whole proxy link.
529+ if errors.Is(err, errProxyLinkSend) {
530+ d.log.Warn("proxy link reply failed", "host", p.key.Host, "session", p.key.Name, "err", err)
531+ return proxyLinkResume
532+ }
523 d.log.Warn("invalid proxy link frame", "host", p.key.Host, "session", p.key.Name, "err", err)533 d.log.Warn("invalid proxy link frame", "host", p.key.Host, "session", p.key.Name, "err", err)
524 return proxyLinkStop534 return proxyLinkStop
525 }535 }
@@ -567,7 +577,7 @@ func (d *Daemon) handleLinkFrame(p *proxySession, generation uint64, frame ports
567 return proxyLinkStop, err577 return proxyLinkStop, err
568 }578 }
569 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}); err != nil {579 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}); err != nil {
570- return proxyLinkStop, err580+ return proxyLinkStop, fmt.Errorf("proxy pong reply: %w: %w", errProxyLinkSend, err)
571 }581 }
572 return proxyLinkResume, nil582 return proxyLinkResume, nil
573 case ports.MsgDetached:583 case ports.MsgDetached:
M internal/usecase/daemon/proxy_output.go +8 -2
@@ -39,6 +39,12 @@ func (p *proxySession) applyOutputForGeneration(generation uint64, out ports.Out
3939 return 0, false, len(data) != 0
4040 }
4141 if out.BaseStateNum == 0 {
42+ // A retransmitted full paint must not clobber a newer screen or regress
43+ // the applied state. Only a reset this proxy actually asked for may
44+ // replay a state it has already applied.
45+ if !p.awaitingReset && out.NewStateNum <= p.appliedState {
46+ return 0, false, false
47+ }
4248 p.screen = vt.NewScreen(p.screen.Frame.Width, p.screen.Frame.Height)
4349 p.screen.Write(data)
4450 p.appliedState = out.NewStateNum
@@ -69,7 +75,7 @@ func (d *Daemon) handleProxyOutput(p *proxySession, generation uint64, out ports
6975 Type: ports.MsgAck,
7076 Payload: ports.MarshalAck(ports.Ack{AckedStateNum: ack}),
7177 }); err != nil {
72- return fmt.Errorf("proxy output ACK: %w", err)
78+ return fmt.Errorf("proxy output ACK: %w: %w", errProxyLinkSend, err)
7379 }
7480 }
7581 if requestReset {
@@ -77,7 +83,7 @@ func (d *Daemon) handleProxyOutput(p *proxySession, generation uint64, out ports
7783 Type: ports.MsgOutputResetRequest,
7884 Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{}),
7985 }); err != nil {
80- return fmt.Errorf("proxy output reset request: %w", err)
86+ return fmt.Errorf("proxy output reset request: %w: %w", errProxyLinkSend, err)
8187 }
8288 }
8389 if !changed || !p.currentLinkGeneration(generation) {
@@ -39,6 +39,12 @@ func (p *proxySession) applyOutputForGeneration(generation uint64, out ports.Out
39 return 0, false, len(data) != 039 return 0, false, len(data) != 0
40 }40 }
41 if out.BaseStateNum == 0 {41 if out.BaseStateNum == 0 {
42+ // A retransmitted full paint must not clobber a newer screen or regress
43+ // the applied state. Only a reset this proxy actually asked for may
44+ // replay a state it has already applied.
45+ if !p.awaitingReset && out.NewStateNum <= p.appliedState {
46+ return 0, false, false
47+ }
42 p.screen = vt.NewScreen(p.screen.Frame.Width, p.screen.Frame.Height)48 p.screen = vt.NewScreen(p.screen.Frame.Width, p.screen.Frame.Height)
43 p.screen.Write(data)49 p.screen.Write(data)
44 p.appliedState = out.NewStateNum50 p.appliedState = out.NewStateNum
@@ -69,7 +75,7 @@ func (d *Daemon) handleProxyOutput(p *proxySession, generation uint64, out ports
69 Type: ports.MsgAck,75 Type: ports.MsgAck,
70 Payload: ports.MarshalAck(ports.Ack{AckedStateNum: ack}),76 Payload: ports.MarshalAck(ports.Ack{AckedStateNum: ack}),
71 }); err != nil {77 }); err != nil {
72- return fmt.Errorf("proxy output ACK: %w", err)78+ return fmt.Errorf("proxy output ACK: %w: %w", errProxyLinkSend, err)
73 }79 }
74 }80 }
75 if requestReset {81 if requestReset {
@@ -77,7 +83,7 @@ func (d *Daemon) handleProxyOutput(p *proxySession, generation uint64, out ports
77 Type: ports.MsgOutputResetRequest,83 Type: ports.MsgOutputResetRequest,
78 Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{}),84 Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{}),
79 }); err != nil {85 }); err != nil {
80- return fmt.Errorf("proxy output reset request: %w", err)86+ return fmt.Errorf("proxy output reset request: %w: %w", errProxyLinkSend, err)
81 }87 }
82 }88 }
83 if !changed || !p.currentLinkGeneration(generation) {89 if !changed || !p.currentLinkGeneration(generation) {
M internal/usecase/daemon/proxy_output_test.go +131 -23
@@ -33,6 +33,10 @@ type proxyTestTransport struct {
3333 sending atomic.Int32
3434 concurrent atomic.Bool
3535 receiving atomic.Int32
36+ // sendFails rejects every write while leaving Recv able to deliver queued
37+ // frames, so a reply failure can be observed without also ending the receive
38+ // pump.
39+ sendFails atomic.Bool
3640 }
3741
3842 func newProxyTestTransport() *proxyTestTransport {
@@ -56,6 +60,9 @@ func (t *proxyTestTransport) Send(frame ports.Frame) error {
5660 t.concurrent.Store(true)
5761 }
5862 defer t.sending.Add(-1)
63+ if t.sendFails.Load() {
64+ return io.ErrClosedPipe
65+ }
5966 if t.sendGate != nil && t.ungatedSends.Add(-1) < 0 {
6067 if t.sendEntered != nil {
6168 select {
@@ -113,12 +120,30 @@ func proxyMeta(name string) ports.Frame {
113120 return ports.Frame{Type: ports.MsgSessionMeta, Payload: payload}
114121 }
115122
116-func newProxyTestDaemon(t *testing.T, factory ports.RemoteDialerFactory) *Daemon {
123+// newProxyTestDaemon always installs a controllable clock so no proxy test can
124+// depend on wall-clock deadlines. Pass a signalClock to drive resume backoff.
125+func newProxyTestDaemon(t *testing.T, factory ports.RemoteDialerFactory, clk ports.Clock) *Daemon {
117126 t.Helper()
118- return New(nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)),
127+ if clk == nil {
128+ clk = stubClock{}
129+ }
130+ return New(nil, clk, slog.New(slog.NewTextHandler(io.Discard, nil)),
119131 WithRemoteDiscovery(nil, nil, nil, factory, ports.RemoteTransportUDP))
120132 }
121133
134+// proxySessionSnapshot clones the registry under d.mu so assertions never read
135+// d.sessions while a registration writes it. Assertions run on the clone, after
136+// the lock has been released.
137+func proxySessionSnapshot(d *Daemon) map[domain.SessionID]attachmentSession {
138+ d.mu.Lock()
139+ defer d.mu.Unlock()
140+ clone := make(map[domain.SessionID]attachmentSession, len(d.sessions))
141+ for id, entry := range d.sessions {
142+ clone[id] = entry
143+ }
144+ return clone
145+}
146+
122147 func proxyFactoryFor(t *testing.T, key domain.RemoteSessionKey, transports ...ports.Transport) ports.RemoteDialerFactory {
123148 t.Helper()
124149 factory := portsmocks.NewMockRemoteDialerFactory(t)
@@ -157,7 +182,7 @@ func stopProxy(t *testing.T, p *proxySession) {
157182 func TestProxyHandshakePublishesOnlyAfterWelcomeAndMetadata(t *testing.T) {
158183 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
159184 tr := newProxyTestTransport()
160- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))
185+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
161186
162187 result := make(chan struct {
163188 proxy *proxySession
@@ -178,15 +203,15 @@ func TestProxyHandshakePublishesOnlyAfterWelcomeAndMetadata(t *testing.T) {
178203 require.Equal(t, key.Name, hello.Name)
179204 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, hello.Size)
180205 require.Equal(t, uint8(maxUnackedOutputStates), hello.MaxOutputInFlight)
181- require.Empty(t, d.sessions, "Hello alone must not publish a proxy")
206+ require.Empty(t, proxySessionSnapshot(d), "Hello alone must not publish a proxy")
182207
183208 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 44, ports.CapabilityResume|ports.CapabilityProxied)}
184- require.Empty(t, d.sessions, "Welcome alone must not publish a proxy")
209+ require.Empty(t, proxySessionSnapshot(d), "Welcome alone must not publish a proxy")
185210 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}
186211
187212 created := awaitTestValue(t, result, "proxy handshake did not complete")
188213 require.NoError(t, created.err)
189- require.Same(t, created.proxy, d.sessions[key.ID()])
214+ require.Same(t, created.proxy, proxySessionSnapshot(d)[key.ID()])
190215 require.Equal(t, key, created.proxy.key)
191216 require.Equal(t, uint64(44), created.proxy.resumeToken)
192217 stopProxy(t, created.proxy)
@@ -206,12 +231,12 @@ func TestProxyHandshakeRejectsInvalidWelcomeWithoutPublication(t *testing.T) {
206231 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
207232 tr := newProxyTestTransport()
208233 tr.recv <- proxyRecv{frame: tt.first}
209- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))
234+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
210235
211236 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
212237 require.Error(t, err)
213238 require.Nil(t, proxy)
214- require.Empty(t, d.sessions)
239+ require.Empty(t, proxySessionSnapshot(d))
215240 select {
216241 case <-tr.closed:
217242 default:
@@ -236,7 +261,7 @@ func TestProxyRegistryDedupesByExactStructuredKey(t *testing.T) {
236261 factory.EXPECT().DialerForRemote(key.Host, key.Name, ports.RemoteTransportUDP, mock.Anything).Return(dialer, nil).Once()
237262 dialer.EXPECT().Dial(mock.Anything).Return(transport, nil).Once()
238263 }
239- d := newProxyTestDaemon(t, factory)
264+ d := newProxyTestDaemon(t, factory, stubClock{})
240265
241266 first, err := d.openProxySession(context.Background(), firstKey, domain.Size{Cols: 80, Rows: 24})
242267 require.NoError(t, err)
@@ -246,7 +271,7 @@ func TestProxyRegistryDedupesByExactStructuredKey(t *testing.T) {
246271 second, err := d.openProxySession(context.Background(), secondKey, domain.Size{Cols: 80, Rows: 24})
247272 require.NoError(t, err)
248273 require.NotSame(t, first, second)
249- require.Len(t, d.sessions, 2)
274+ require.Len(t, proxySessionSnapshot(d), 2)
250275
251276 stopProxy(t, first)
252277 stopProxy(t, second)
@@ -264,7 +289,7 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
264289 <-ctx.Done()
265290 return nil, ctx.Err()
266291 }).Once()
267- d := newProxyTestDaemon(t, factory)
292+ d := newProxyTestDaemon(t, factory, stubClock{})
268293 ctx, cancel := context.WithCancel(context.Background())
269294 result := make(chan struct {
270295 proxy *proxySession
@@ -284,13 +309,13 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
284309 proxy, err := opened.proxy, opened.err
285310 require.ErrorIs(t, err, context.Canceled)
286311 require.Nil(t, proxy)
287- require.Empty(t, d.sessions)
312+ require.Empty(t, proxySessionSnapshot(d))
288313 })
289314
290315 t.Run("handshake", func(t *testing.T) {
291316 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
292317 tr := newProxyTestTransport()
293- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))
318+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
294319 ctx, cancel := context.WithCancel(context.Background())
295320 result := make(chan error, 1)
296321 go func() {
@@ -300,7 +325,7 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
300325 requireProxyHello(t, tr)
301326 cancel()
302327 require.ErrorIs(t, awaitTestValue(t, result, "canceled proxy handshake did not finish"), context.Canceled)
303- require.Empty(t, d.sessions)
328+ require.Empty(t, proxySessionSnapshot(d))
304329 select {
305330 case <-tr.closed:
306331 default:
@@ -316,7 +341,10 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
316341 first.recv <- proxyRecv{frame: proxyMeta(key.Name)}
317342 second := newProxyTestTransport()
318343 factory := proxyFactoryFor(t, key, first, second)
319- d := newProxyTestDaemon(t, factory)
344+ // The resume backoff is a ports.Clock deadline, so it is fired rather than
345+ // waited out. The buffer also absorbs the warm timer armed by replacement.
346+ clock := &signalClock{timers: make(chan *signalTimer, 8)}
347+ d := newProxyTestDaemon(t, factory, clock)
320348
321349 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
322350 require.NoError(t, err)
@@ -324,6 +352,9 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
324352 require.Equal(t, ports.IntentAttach, firstHello.Intent)
325353
326354 first.recv <- proxyRecv{err: io.EOF}
355+ backoff := awaitTestValue(t, clock.timers, "proxy resume backoff timer was not armed")
356+ require.Equal(t, proxyResumeInitialBackoff, backoff.duration)
357+ backoff.ch <- time.Time{}
327358 resumeHello := requireProxyHello(t, second)
328359 require.Equal(t, ports.IntentResume, resumeHello.Intent)
329360 require.Equal(t, uint64(55), resumeHello.ResumeToken)
@@ -341,7 +372,7 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
341372 require.True(t, proxy.expired)
342373 require.Equal(t, uint64(66), proxy.resumeToken)
343374 proxy.mu.Unlock()
344- require.Same(t, proxy, d.sessions[key.ID()], "replacement leaves the expired proxy available for local presentation")
375+ require.Same(t, proxy, proxySessionSnapshot(d)[key.ID()], "replacement leaves the expired proxy available for local presentation")
345376 }
346377
347378 func TestRunProxyTransportStopsReceivePumpAfterInvalidFrame(t *testing.T) {
@@ -368,7 +399,7 @@ func TestProxySenderSerializesAllTransportWrites(t *testing.T) {
368399 tr := newProxyTestTransportWithSendGate(gate, entered, 1)
369400 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
370401 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}
371- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))
402+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
372403 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
373404 require.NoError(t, err)
374405 _ = requireProxyHello(t, tr)
@@ -398,24 +429,24 @@ func TestProxyHandshakeRejectsInvalidInitialMetadata(t *testing.T) {
398429 tr := newProxyTestTransport()
399430 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
400431 tr.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: []byte{0xff}}}
401- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))
432+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
402433
403434 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
404435 require.Error(t, err)
405436 require.Nil(t, proxy)
406- require.Empty(t, d.sessions)
437+ require.Empty(t, proxySessionSnapshot(d))
407438 }
408439
409440 func TestProxyOpenValidatesDependenciesAndKey(t *testing.T) {
410- d := newProxyTestDaemon(t, nil)
441+ d := newProxyTestDaemon(t, nil, stubClock{})
411442 _, err := d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "work"}, domain.Size{Cols: 80, Rows: 24})
412443 require.Error(t, err)
413444
414445 factory := portsmocks.NewMockRemoteDialerFactory(t)
415- d = newProxyTestDaemon(t, factory)
446+ d = newProxyTestDaemon(t, factory, stubClock{})
416447 _, err = d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "bad:name"}, domain.Size{Cols: 80, Rows: 24})
417448 require.Error(t, err)
418- require.Empty(t, d.sessions)
449+ require.Empty(t, proxySessionSnapshot(d))
419450 }
420451
421452 func newProxyOutputSession(t *testing.T) (*Daemon, *proxySession, *proxyTestTransport, uint64) {
@@ -424,7 +455,7 @@ func newProxyOutputSession(t *testing.T) (*Daemon, *proxySession, *proxyTestTran
424455 require.NoError(t, err)
425456 transport := newProxyTestTransport()
426457 generation, _ := proxy.installTransport(transport)
427- return newProxyTestDaemon(t, nil), proxy, transport, generation
458+ return newProxyTestDaemon(t, nil, stubClock{}), proxy, transport, generation
428459 }
429460
430461 func proxyScreenText(p *proxySession) string {
@@ -492,6 +523,20 @@ func TestProxyOutputStateMachine(t *testing.T) {
492523 wantChanged: true,
493524 wantScreen: "! ",
494525 },
526+ {
527+ name: "stale base zero is ignored",
528+ initial: &ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")},
529+ output: ports.Output{BaseStateNum: 0, NewStateNum: 3, Data: []byte("B")},
530+ wantState: 5,
531+ wantScreen: "A ",
532+ },
533+ {
534+ name: "duplicate base zero is ignored",
535+ initial: &ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")},
536+ output: ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("B")},
537+ wantState: 5,
538+ wantScreen: "A ",
539+ },
495540 }
496541 for _, tt := range tests {
497542 t.Run(tt.name, func(t *testing.T) {
@@ -512,6 +557,69 @@ func TestProxyOutputStateMachine(t *testing.T) {
512557 }
513558 }
514559
560+// TestProxyOutputAwaitedResetAcceptsReplayedState fences the stale base-zero
561+// guard: a reset this proxy asked for must still be applied even when the
562+// remote numbers it at or below the state already applied.
563+func TestProxyOutputAwaitedResetAcceptsReplayedState(t *testing.T) {
564+ _, proxy, _, generation := newProxyOutputSession(t)
565+ _, _, _ = proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")})
566+ _, requestReset, _ := proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 9, NewStateNum: 10, Data: []byte("x")})
567+ require.True(t, requestReset)
568+
569+ ack, reset, changed := proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 0, NewStateNum: 3, Data: []byte("B")})
570+ require.Equal(t, uint64(3), ack)
571+ require.False(t, reset)
572+ require.True(t, changed)
573+ state, awaiting := proxyOutputState(proxy)
574+ require.Equal(t, uint64(3), state)
575+ require.False(t, awaiting)
576+ require.Equal(t, "B ", proxyScreenText(proxy))
577+}
578+
579+// TestProxyLinkReplySendFailureResumesInsteadOfStopping separates a broken
580+// write from a broken remote: the frame was well formed, so the proxy link is
581+// resumable and only this transport incarnation is retired.
582+func TestProxyLinkReplySendFailureResumesInsteadOfStopping(t *testing.T) {
583+ tests := []struct {
584+ name string
585+ frame ports.Frame
586+ }{
587+ {
588+ name: "pong reply",
589+ frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})},
590+ },
591+ {
592+ name: "output ack",
593+ frame: ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("A")})},
594+ },
595+ {
596+ name: "output reset request",
597+ frame: ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 7, NewStateNum: 8, Data: []byte("A")})},
598+ },
599+ }
600+ for _, tt := range tests {
601+ t.Run(tt.name, func(t *testing.T) {
602+ d, proxy, transport, generation := newProxyOutputSession(t)
603+ t.Cleanup(func() { _ = transport.Close() })
604+ transport.sendFails.Store(true)
605+ transport.recv <- proxyRecv{frame: tt.frame}
606+
607+ result := d.runProxyTransport(context.Background(), proxy, generation, transport)
608+ require.Equal(t, proxyLinkResume, result, "a failed reply must retire the transport, not the proxy link")
609+ })
610+ }
611+}
612+
613+// TestProxyLinkProtocolErrorStillStops fences the sentinel so it cannot widen
614+// into "every error is resumable".
615+func TestProxyLinkProtocolErrorStillStops(t *testing.T) {
616+ d, proxy, transport, generation := newProxyOutputSession(t)
617+ t.Cleanup(func() { _ = transport.Close() })
618+ transport.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{1}}}
619+
620+ require.Equal(t, proxyLinkStop, d.runProxyTransport(context.Background(), proxy, generation, transport))
621+}
622+
515623 func TestProxyOutputMismatchSendsOneResetUntilBaseZero(t *testing.T) {
516624 d, proxy, transport, generation := newProxyOutputSession(t)
517625 require.NoError(t, d.handleProxyOutput(proxy, generation, ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("A")}))
@@ -33,6 +33,10 @@ type proxyTestTransport struct {
33 sending atomic.Int3233 sending atomic.Int32
34 concurrent atomic.Bool34 concurrent atomic.Bool
35 receiving atomic.Int3235 receiving atomic.Int32
36+ // sendFails rejects every write while leaving Recv able to deliver queued
37+ // frames, so a reply failure can be observed without also ending the receive
38+ // pump.
39+ sendFails atomic.Bool
36 }40 }
37 41
38 func newProxyTestTransport() *proxyTestTransport {42 func newProxyTestTransport() *proxyTestTransport {
@@ -56,6 +60,9 @@ func (t *proxyTestTransport) Send(frame ports.Frame) error {
56 t.concurrent.Store(true)60 t.concurrent.Store(true)
57 }61 }
58 defer t.sending.Add(-1)62 defer t.sending.Add(-1)
63+ if t.sendFails.Load() {
64+ return io.ErrClosedPipe
65+ }
59 if t.sendGate != nil && t.ungatedSends.Add(-1) < 0 {66 if t.sendGate != nil && t.ungatedSends.Add(-1) < 0 {
60 if t.sendEntered != nil {67 if t.sendEntered != nil {
61 select {68 select {
@@ -113,12 +120,30 @@ func proxyMeta(name string) ports.Frame {
113 return ports.Frame{Type: ports.MsgSessionMeta, Payload: payload}120 return ports.Frame{Type: ports.MsgSessionMeta, Payload: payload}
114 }121 }
115 122
116-func newProxyTestDaemon(t *testing.T, factory ports.RemoteDialerFactory) *Daemon {123+// newProxyTestDaemon always installs a controllable clock so no proxy test can
124+// depend on wall-clock deadlines. Pass a signalClock to drive resume backoff.
125+func newProxyTestDaemon(t *testing.T, factory ports.RemoteDialerFactory, clk ports.Clock) *Daemon {
117 t.Helper()126 t.Helper()
118- return New(nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)),127+ if clk == nil {
128+ clk = stubClock{}
129+ }
130+ return New(nil, clk, slog.New(slog.NewTextHandler(io.Discard, nil)),
119 WithRemoteDiscovery(nil, nil, nil, factory, ports.RemoteTransportUDP))131 WithRemoteDiscovery(nil, nil, nil, factory, ports.RemoteTransportUDP))
120 }132 }
121 133
134+// proxySessionSnapshot clones the registry under d.mu so assertions never read
135+// d.sessions while a registration writes it. Assertions run on the clone, after
136+// the lock has been released.
137+func proxySessionSnapshot(d *Daemon) map[domain.SessionID]attachmentSession {
138+ d.mu.Lock()
139+ defer d.mu.Unlock()
140+ clone := make(map[domain.SessionID]attachmentSession, len(d.sessions))
141+ for id, entry := range d.sessions {
142+ clone[id] = entry
143+ }
144+ return clone
145+}
146+
122 func proxyFactoryFor(t *testing.T, key domain.RemoteSessionKey, transports ...ports.Transport) ports.RemoteDialerFactory {147 func proxyFactoryFor(t *testing.T, key domain.RemoteSessionKey, transports ...ports.Transport) ports.RemoteDialerFactory {
123 t.Helper()148 t.Helper()
124 factory := portsmocks.NewMockRemoteDialerFactory(t)149 factory := portsmocks.NewMockRemoteDialerFactory(t)
@@ -157,7 +182,7 @@ func stopProxy(t *testing.T, p *proxySession) {
157 func TestProxyHandshakePublishesOnlyAfterWelcomeAndMetadata(t *testing.T) {182 func TestProxyHandshakePublishesOnlyAfterWelcomeAndMetadata(t *testing.T) {
158 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}183 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
159 tr := newProxyTestTransport()184 tr := newProxyTestTransport()
160- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))185+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
161 186
162 result := make(chan struct {187 result := make(chan struct {
163 proxy *proxySession188 proxy *proxySession
@@ -178,15 +203,15 @@ func TestProxyHandshakePublishesOnlyAfterWelcomeAndMetadata(t *testing.T) {
178 require.Equal(t, key.Name, hello.Name)203 require.Equal(t, key.Name, hello.Name)
179 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, hello.Size)204 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, hello.Size)
180 require.Equal(t, uint8(maxUnackedOutputStates), hello.MaxOutputInFlight)205 require.Equal(t, uint8(maxUnackedOutputStates), hello.MaxOutputInFlight)
181- require.Empty(t, d.sessions, "Hello alone must not publish a proxy")206+ require.Empty(t, proxySessionSnapshot(d), "Hello alone must not publish a proxy")
182 207
183 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 44, ports.CapabilityResume|ports.CapabilityProxied)}208 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 44, ports.CapabilityResume|ports.CapabilityProxied)}
184- require.Empty(t, d.sessions, "Welcome alone must not publish a proxy")209+ require.Empty(t, proxySessionSnapshot(d), "Welcome alone must not publish a proxy")
185 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}210 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}
186 211
187 created := awaitTestValue(t, result, "proxy handshake did not complete")212 created := awaitTestValue(t, result, "proxy handshake did not complete")
188 require.NoError(t, created.err)213 require.NoError(t, created.err)
189- require.Same(t, created.proxy, d.sessions[key.ID()])214+ require.Same(t, created.proxy, proxySessionSnapshot(d)[key.ID()])
190 require.Equal(t, key, created.proxy.key)215 require.Equal(t, key, created.proxy.key)
191 require.Equal(t, uint64(44), created.proxy.resumeToken)216 require.Equal(t, uint64(44), created.proxy.resumeToken)
192 stopProxy(t, created.proxy)217 stopProxy(t, created.proxy)
@@ -206,12 +231,12 @@ func TestProxyHandshakeRejectsInvalidWelcomeWithoutPublication(t *testing.T) {
206 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}231 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
207 tr := newProxyTestTransport()232 tr := newProxyTestTransport()
208 tr.recv <- proxyRecv{frame: tt.first}233 tr.recv <- proxyRecv{frame: tt.first}
209- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))234+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
210 235
211 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})236 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
212 require.Error(t, err)237 require.Error(t, err)
213 require.Nil(t, proxy)238 require.Nil(t, proxy)
214- require.Empty(t, d.sessions)239+ require.Empty(t, proxySessionSnapshot(d))
215 select {240 select {
216 case <-tr.closed:241 case <-tr.closed:
217 default:242 default:
@@ -236,7 +261,7 @@ func TestProxyRegistryDedupesByExactStructuredKey(t *testing.T) {
236 factory.EXPECT().DialerForRemote(key.Host, key.Name, ports.RemoteTransportUDP, mock.Anything).Return(dialer, nil).Once()261 factory.EXPECT().DialerForRemote(key.Host, key.Name, ports.RemoteTransportUDP, mock.Anything).Return(dialer, nil).Once()
237 dialer.EXPECT().Dial(mock.Anything).Return(transport, nil).Once()262 dialer.EXPECT().Dial(mock.Anything).Return(transport, nil).Once()
238 }263 }
239- d := newProxyTestDaemon(t, factory)264+ d := newProxyTestDaemon(t, factory, stubClock{})
240 265
241 first, err := d.openProxySession(context.Background(), firstKey, domain.Size{Cols: 80, Rows: 24})266 first, err := d.openProxySession(context.Background(), firstKey, domain.Size{Cols: 80, Rows: 24})
242 require.NoError(t, err)267 require.NoError(t, err)
@@ -246,7 +271,7 @@ func TestProxyRegistryDedupesByExactStructuredKey(t *testing.T) {
246 second, err := d.openProxySession(context.Background(), secondKey, domain.Size{Cols: 80, Rows: 24})271 second, err := d.openProxySession(context.Background(), secondKey, domain.Size{Cols: 80, Rows: 24})
247 require.NoError(t, err)272 require.NoError(t, err)
248 require.NotSame(t, first, second)273 require.NotSame(t, first, second)
249- require.Len(t, d.sessions, 2)274+ require.Len(t, proxySessionSnapshot(d), 2)
250 275
251 stopProxy(t, first)276 stopProxy(t, first)
252 stopProxy(t, second)277 stopProxy(t, second)
@@ -264,7 +289,7 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
264 <-ctx.Done()289 <-ctx.Done()
265 return nil, ctx.Err()290 return nil, ctx.Err()
266 }).Once()291 }).Once()
267- d := newProxyTestDaemon(t, factory)292+ d := newProxyTestDaemon(t, factory, stubClock{})
268 ctx, cancel := context.WithCancel(context.Background())293 ctx, cancel := context.WithCancel(context.Background())
269 result := make(chan struct {294 result := make(chan struct {
270 proxy *proxySession295 proxy *proxySession
@@ -284,13 +309,13 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
284 proxy, err := opened.proxy, opened.err309 proxy, err := opened.proxy, opened.err
285 require.ErrorIs(t, err, context.Canceled)310 require.ErrorIs(t, err, context.Canceled)
286 require.Nil(t, proxy)311 require.Nil(t, proxy)
287- require.Empty(t, d.sessions)312+ require.Empty(t, proxySessionSnapshot(d))
288 })313 })
289 314
290 t.Run("handshake", func(t *testing.T) {315 t.Run("handshake", func(t *testing.T) {
291 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}316 key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
292 tr := newProxyTestTransport()317 tr := newProxyTestTransport()
293- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))318+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
294 ctx, cancel := context.WithCancel(context.Background())319 ctx, cancel := context.WithCancel(context.Background())
295 result := make(chan error, 1)320 result := make(chan error, 1)
296 go func() {321 go func() {
@@ -300,7 +325,7 @@ func TestProxyCancelDuringDialOrHandshakeNeverPublishes(t *testing.T) {
300 requireProxyHello(t, tr)325 requireProxyHello(t, tr)
301 cancel()326 cancel()
302 require.ErrorIs(t, awaitTestValue(t, result, "canceled proxy handshake did not finish"), context.Canceled)327 require.ErrorIs(t, awaitTestValue(t, result, "canceled proxy handshake did not finish"), context.Canceled)
303- require.Empty(t, d.sessions)328+ require.Empty(t, proxySessionSnapshot(d))
304 select {329 select {
305 case <-tr.closed:330 case <-tr.closed:
306 default:331 default:
@@ -316,7 +341,10 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
316 first.recv <- proxyRecv{frame: proxyMeta(key.Name)}341 first.recv <- proxyRecv{frame: proxyMeta(key.Name)}
317 second := newProxyTestTransport()342 second := newProxyTestTransport()
318 factory := proxyFactoryFor(t, key, first, second)343 factory := proxyFactoryFor(t, key, first, second)
319- d := newProxyTestDaemon(t, factory)344+ // The resume backoff is a ports.Clock deadline, so it is fired rather than
345+ // waited out. The buffer also absorbs the warm timer armed by replacement.
346+ clock := &signalClock{timers: make(chan *signalTimer, 8)}
347+ d := newProxyTestDaemon(t, factory, clock)
320 348
321 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})349 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
322 require.NoError(t, err)350 require.NoError(t, err)
@@ -324,6 +352,9 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
324 require.Equal(t, ports.IntentAttach, firstHello.Intent)352 require.Equal(t, ports.IntentAttach, firstHello.Intent)
325 353
326 first.recv <- proxyRecv{err: io.EOF}354 first.recv <- proxyRecv{err: io.EOF}
355+ backoff := awaitTestValue(t, clock.timers, "proxy resume backoff timer was not armed")
356+ require.Equal(t, proxyResumeInitialBackoff, backoff.duration)
357+ backoff.ch <- time.Time{}
327 resumeHello := requireProxyHello(t, second)358 resumeHello := requireProxyHello(t, second)
328 require.Equal(t, ports.IntentResume, resumeHello.Intent)359 require.Equal(t, ports.IntentResume, resumeHello.Intent)
329 require.Equal(t, uint64(55), resumeHello.ResumeToken)360 require.Equal(t, uint64(55), resumeHello.ResumeToken)
@@ -341,7 +372,7 @@ func TestProxyLinkResumesAndReplacementIsTerminal(t *testing.T) {
341 require.True(t, proxy.expired)372 require.True(t, proxy.expired)
342 require.Equal(t, uint64(66), proxy.resumeToken)373 require.Equal(t, uint64(66), proxy.resumeToken)
343 proxy.mu.Unlock()374 proxy.mu.Unlock()
344- require.Same(t, proxy, d.sessions[key.ID()], "replacement leaves the expired proxy available for local presentation")375+ require.Same(t, proxy, proxySessionSnapshot(d)[key.ID()], "replacement leaves the expired proxy available for local presentation")
345 }376 }
346 377
347 func TestRunProxyTransportStopsReceivePumpAfterInvalidFrame(t *testing.T) {378 func TestRunProxyTransportStopsReceivePumpAfterInvalidFrame(t *testing.T) {
@@ -368,7 +399,7 @@ func TestProxySenderSerializesAllTransportWrites(t *testing.T) {
368 tr := newProxyTestTransportWithSendGate(gate, entered, 1)399 tr := newProxyTestTransportWithSendGate(gate, entered, 1)
369 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}400 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
370 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}401 tr.recv <- proxyRecv{frame: proxyMeta(key.Name)}
371- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))402+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
372 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})403 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
373 require.NoError(t, err)404 require.NoError(t, err)
374 _ = requireProxyHello(t, tr)405 _ = requireProxyHello(t, tr)
@@ -398,24 +429,24 @@ func TestProxyHandshakeRejectsInvalidInitialMetadata(t *testing.T) {
398 tr := newProxyTestTransport()429 tr := newProxyTestTransport()
399 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}430 tr.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
400 tr.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: []byte{0xff}}}431 tr.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: []byte{0xff}}}
401- d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr))432+ d := newProxyTestDaemon(t, proxyFactoryFor(t, key, tr), stubClock{})
402 433
403 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})434 proxy, err := d.openProxySession(context.Background(), key, domain.Size{Cols: 80, Rows: 24})
404 require.Error(t, err)435 require.Error(t, err)
405 require.Nil(t, proxy)436 require.Nil(t, proxy)
406- require.Empty(t, d.sessions)437+ require.Empty(t, proxySessionSnapshot(d))
407 }438 }
408 439
409 func TestProxyOpenValidatesDependenciesAndKey(t *testing.T) {440 func TestProxyOpenValidatesDependenciesAndKey(t *testing.T) {
410- d := newProxyTestDaemon(t, nil)441+ d := newProxyTestDaemon(t, nil, stubClock{})
411 _, err := d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "work"}, domain.Size{Cols: 80, Rows: 24})442 _, err := d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "work"}, domain.Size{Cols: 80, Rows: 24})
412 require.Error(t, err)443 require.Error(t, err)
413 444
414 factory := portsmocks.NewMockRemoteDialerFactory(t)445 factory := portsmocks.NewMockRemoteDialerFactory(t)
415- d = newProxyTestDaemon(t, factory)446+ d = newProxyTestDaemon(t, factory, stubClock{})
416 _, err = d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "bad:name"}, domain.Size{Cols: 80, Rows: 24})447 _, err = d.openProxySession(context.Background(), domain.RemoteSessionKey{Host: "arch", Name: "bad:name"}, domain.Size{Cols: 80, Rows: 24})
417 require.Error(t, err)448 require.Error(t, err)
418- require.Empty(t, d.sessions)449+ require.Empty(t, proxySessionSnapshot(d))
419 }450 }
420 451
421 func newProxyOutputSession(t *testing.T) (*Daemon, *proxySession, *proxyTestTransport, uint64) {452 func newProxyOutputSession(t *testing.T) (*Daemon, *proxySession, *proxyTestTransport, uint64) {
@@ -424,7 +455,7 @@ func newProxyOutputSession(t *testing.T) (*Daemon, *proxySession, *proxyTestTran
424 require.NoError(t, err)455 require.NoError(t, err)
425 transport := newProxyTestTransport()456 transport := newProxyTestTransport()
426 generation, _ := proxy.installTransport(transport)457 generation, _ := proxy.installTransport(transport)
427- return newProxyTestDaemon(t, nil), proxy, transport, generation458+ return newProxyTestDaemon(t, nil, stubClock{}), proxy, transport, generation
428 }459 }
429 460
430 func proxyScreenText(p *proxySession) string {461 func proxyScreenText(p *proxySession) string {
@@ -492,6 +523,20 @@ func TestProxyOutputStateMachine(t *testing.T) {
492 wantChanged: true,523 wantChanged: true,
493 wantScreen: "! ",524 wantScreen: "! ",
494 },525 },
526+ {
527+ name: "stale base zero is ignored",
528+ initial: &ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")},
529+ output: ports.Output{BaseStateNum: 0, NewStateNum: 3, Data: []byte("B")},
530+ wantState: 5,
531+ wantScreen: "A ",
532+ },
533+ {
534+ name: "duplicate base zero is ignored",
535+ initial: &ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")},
536+ output: ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("B")},
537+ wantState: 5,
538+ wantScreen: "A ",
539+ },
495 }540 }
496 for _, tt := range tests {541 for _, tt := range tests {
497 t.Run(tt.name, func(t *testing.T) {542 t.Run(tt.name, func(t *testing.T) {
@@ -512,6 +557,69 @@ func TestProxyOutputStateMachine(t *testing.T) {
512 }557 }
513 }558 }
514 559
560+// TestProxyOutputAwaitedResetAcceptsReplayedState fences the stale base-zero
561+// guard: a reset this proxy asked for must still be applied even when the
562+// remote numbers it at or below the state already applied.
563+func TestProxyOutputAwaitedResetAcceptsReplayedState(t *testing.T) {
564+ _, proxy, _, generation := newProxyOutputSession(t)
565+ _, _, _ = proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 0, NewStateNum: 5, Data: []byte("A")})
566+ _, requestReset, _ := proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 9, NewStateNum: 10, Data: []byte("x")})
567+ require.True(t, requestReset)
568+
569+ ack, reset, changed := proxy.applyOutputForGeneration(generation, ports.Output{BaseStateNum: 0, NewStateNum: 3, Data: []byte("B")})
570+ require.Equal(t, uint64(3), ack)
571+ require.False(t, reset)
572+ require.True(t, changed)
573+ state, awaiting := proxyOutputState(proxy)
574+ require.Equal(t, uint64(3), state)
575+ require.False(t, awaiting)
576+ require.Equal(t, "B ", proxyScreenText(proxy))
577+}
578+
579+// TestProxyLinkReplySendFailureResumesInsteadOfStopping separates a broken
580+// write from a broken remote: the frame was well formed, so the proxy link is
581+// resumable and only this transport incarnation is retired.
582+func TestProxyLinkReplySendFailureResumesInsteadOfStopping(t *testing.T) {
583+ tests := []struct {
584+ name string
585+ frame ports.Frame
586+ }{
587+ {
588+ name: "pong reply",
589+ frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})},
590+ },
591+ {
592+ name: "output ack",
593+ frame: ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("A")})},
594+ },
595+ {
596+ name: "output reset request",
597+ frame: ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: 7, NewStateNum: 8, Data: []byte("A")})},
598+ },
599+ }
600+ for _, tt := range tests {
601+ t.Run(tt.name, func(t *testing.T) {
602+ d, proxy, transport, generation := newProxyOutputSession(t)
603+ t.Cleanup(func() { _ = transport.Close() })
604+ transport.sendFails.Store(true)
605+ transport.recv <- proxyRecv{frame: tt.frame}
606+
607+ result := d.runProxyTransport(context.Background(), proxy, generation, transport)
608+ require.Equal(t, proxyLinkResume, result, "a failed reply must retire the transport, not the proxy link")
609+ })
610+ }
611+}
612+
613+// TestProxyLinkProtocolErrorStillStops fences the sentinel so it cannot widen
614+// into "every error is resumable".
615+func TestProxyLinkProtocolErrorStillStops(t *testing.T) {
616+ d, proxy, transport, generation := newProxyOutputSession(t)
617+ t.Cleanup(func() { _ = transport.Close() })
618+ transport.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{1}}}
619+
620+ require.Equal(t, proxyLinkStop, d.runProxyTransport(context.Background(), proxy, generation, transport))
621+}
622+
515 func TestProxyOutputMismatchSendsOneResetUntilBaseZero(t *testing.T) {623 func TestProxyOutputMismatchSendsOneResetUntilBaseZero(t *testing.T) {
516 d, proxy, transport, generation := newProxyOutputSession(t)624 d, proxy, transport, generation := newProxyOutputSession(t)
517 require.NoError(t, d.handleProxyOutput(proxy, generation, ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("A")}))625 require.NoError(t, d.handleProxyOutput(proxy, generation, ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("A")}))
M internal/usecase/daemon/proxy_render.go +15 -8
@@ -73,7 +73,7 @@ func (d *Daemon) firstProxyPaintWithLease(p *proxySession, ac *attachedClient, l
7373 if p == nil || ac == nil {
7474 return
7575 }
76- p.resize(ac.size)
76+ _, _ = p.resize(ac.size)
7777 if rc := attachmentRenderCoordinator(p); rc != nil {
7878 if rc.invalidateForLease(ac, lease, renderInvalidation{class: invalidateUrgent, reset: true, producer: "proxy_render.go"}) {
7979 rc.fireCurrent(false)
@@ -84,8 +84,10 @@ func (d *Daemon) firstProxyPaintWithLease(p *proxySession, ac *attachedClient, l
8484 }
8585
8686 // resizeProxyForLease accepts a local client resize only for its exact active
87-// attachment lease. It repaints local chrome after the remote content resize
88-// has been sent by the proxy's serialized sender.
87+// attachment lease. Local chrome is repainted whenever the local content
88+// rectangle moved, including when the remote never learned about it: the local
89+// VT has already been resized either way, so skipping the repaint would leave
90+// the client showing chrome for the old geometry.
8991 func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease *attachmentLease, size domain.Size) {
9092 if p == nil || ac == nil || !size.Valid() {
9193 return
@@ -93,7 +95,7 @@ func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease
9395 if rc := attachmentRenderCoordinator(p); rc != nil && !rc.leaseCurrent(lease, false) {
9496 return
9597 }
96- if !p.resize(size) {
98+ if changed, _ := p.resize(size); !changed {
9799 return
98100 }
99101 if rc := attachmentRenderCoordinator(p); rc != nil {
@@ -106,16 +108,21 @@ func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease
106108 // resize updates the local VT to the local content rectangle and sends that
107109 // reduced geometry once to the remote daemon. The sender performs transport
108110 // I/O only after proxy.mu has been released.
109-func (p *proxySession) resize(size domain.Size) bool {
111+//
112+// changed reports whether the local content rectangle actually moved, which is
113+// what local chrome repaints depend on. sent reports whether the remote was
114+// told. The two are deliberately separate: a failed send leaves the local VT
115+// already resized, so the attached client still owes a repaint.
116+func (p *proxySession) resize(size domain.Size) (changed, sent bool) {
110117 if p == nil || !size.Valid() {
111- return false
118+ return false, false
112119 }
113120 content := contentSize(size, false)
114121 p.mu.Lock()
115122 if p.screen == nil || p.contentSize == content {
116123 available := p.screen != nil
117124 p.mu.Unlock()
118- return available
125+ return false, available
119126 }
120127 p.screen.Resize(content.Cols, content.Rows)
121128 p.contentSize = content
@@ -123,7 +130,7 @@ func (p *proxySession) resize(size domain.Size) bool {
123130 generation := p.linkGeneration
124131 p.mu.Unlock()
125132
126- return p.sendGeneration(generation, ports.Frame{
133+ return true, p.sendGeneration(generation, ports.Frame{
127134 Type: ports.MsgResize,
128135 Payload: ports.MarshalResize(ports.Resize{Size: content}),
129136 }) == nil
@@ -73,7 +73,7 @@ func (d *Daemon) firstProxyPaintWithLease(p *proxySession, ac *attachedClient, l
73 if p == nil || ac == nil {73 if p == nil || ac == nil {
74 return74 return
75 }75 }
76- p.resize(ac.size)76+ _, _ = p.resize(ac.size)
77 if rc := attachmentRenderCoordinator(p); rc != nil {77 if rc := attachmentRenderCoordinator(p); rc != nil {
78 if rc.invalidateForLease(ac, lease, renderInvalidation{class: invalidateUrgent, reset: true, producer: "proxy_render.go"}) {78 if rc.invalidateForLease(ac, lease, renderInvalidation{class: invalidateUrgent, reset: true, producer: "proxy_render.go"}) {
79 rc.fireCurrent(false)79 rc.fireCurrent(false)
@@ -84,8 +84,10 @@ func (d *Daemon) firstProxyPaintWithLease(p *proxySession, ac *attachedClient, l
84 }84 }
85 85
86 // resizeProxyForLease accepts a local client resize only for its exact active86 // resizeProxyForLease accepts a local client resize only for its exact active
87-// attachment lease. It repaints local chrome after the remote content resize87+// attachment lease. Local chrome is repainted whenever the local content
88-// has been sent by the proxy's serialized sender.88+// rectangle moved, including when the remote never learned about it: the local
89+// VT has already been resized either way, so skipping the repaint would leave
90+// the client showing chrome for the old geometry.
89 func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease *attachmentLease, size domain.Size) {91 func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease *attachmentLease, size domain.Size) {
90 if p == nil || ac == nil || !size.Valid() {92 if p == nil || ac == nil || !size.Valid() {
91 return93 return
@@ -93,7 +95,7 @@ func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease
93 if rc := attachmentRenderCoordinator(p); rc != nil && !rc.leaseCurrent(lease, false) {95 if rc := attachmentRenderCoordinator(p); rc != nil && !rc.leaseCurrent(lease, false) {
94 return96 return
95 }97 }
96- if !p.resize(size) {98+ if changed, _ := p.resize(size); !changed {
97 return99 return
98 }100 }
99 if rc := attachmentRenderCoordinator(p); rc != nil {101 if rc := attachmentRenderCoordinator(p); rc != nil {
@@ -106,16 +108,21 @@ func (d *Daemon) resizeProxyForLease(p *proxySession, ac *attachedClient, lease
106 // resize updates the local VT to the local content rectangle and sends that108 // resize updates the local VT to the local content rectangle and sends that
107 // reduced geometry once to the remote daemon. The sender performs transport109 // reduced geometry once to the remote daemon. The sender performs transport
108 // I/O only after proxy.mu has been released.110 // I/O only after proxy.mu has been released.
109-func (p *proxySession) resize(size domain.Size) bool {111+//
112+// changed reports whether the local content rectangle actually moved, which is
113+// what local chrome repaints depend on. sent reports whether the remote was
114+// told. The two are deliberately separate: a failed send leaves the local VT
115+// already resized, so the attached client still owes a repaint.
116+func (p *proxySession) resize(size domain.Size) (changed, sent bool) {
110 if p == nil || !size.Valid() {117 if p == nil || !size.Valid() {
111- return false118+ return false, false
112 }119 }
113 content := contentSize(size, false)120 content := contentSize(size, false)
114 p.mu.Lock()121 p.mu.Lock()
115 if p.screen == nil || p.contentSize == content {122 if p.screen == nil || p.contentSize == content {
116 available := p.screen != nil123 available := p.screen != nil
117 p.mu.Unlock()124 p.mu.Unlock()
118- return available125+ return false, available
119 }126 }
120 p.screen.Resize(content.Cols, content.Rows)127 p.screen.Resize(content.Cols, content.Rows)
121 p.contentSize = content128 p.contentSize = content
@@ -123,7 +130,7 @@ func (p *proxySession) resize(size domain.Size) bool {
123 generation := p.linkGeneration130 generation := p.linkGeneration
124 p.mu.Unlock()131 p.mu.Unlock()
125 132
126- return p.sendGeneration(generation, ports.Frame{133+ return true, p.sendGeneration(generation, ports.Frame{
127 Type: ports.MsgResize,134 Type: ports.MsgResize,
128 Payload: ports.MarshalResize(ports.Resize{Size: content}),135 Payload: ports.MarshalResize(ports.Resize{Size: content}),
129 }) == nil136 }) == nil
M internal/usecase/daemon/proxy_render_test.go +109 -1
@@ -1,11 +1,14 @@
11 package daemon
22
33 import (
4+ "io"
45 "testing"
56
67 "github.com/bnema/vev/internal/domain"
78 "github.com/bnema/vev/internal/ports"
9+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
810 "github.com/bnema/vev/pkg/renderer"
11+ "github.com/stretchr/testify/mock"
912 "github.com/stretchr/testify/require"
1013 )
1114
@@ -106,7 +109,9 @@ func TestProxyResizeUsesReducedRemoteGeometryOnce(t *testing.T) {
106109 proxy.linkGeneration = 1
107110 proxy.mu.Unlock()
108111
109- require.True(t, proxy.resize(domain.Size{Cols: 20, Rows: 10}))
112+ changed, sent := proxy.resize(domain.Size{Cols: 20, Rows: 10})
113+ require.True(t, changed)
114+ require.True(t, sent)
110115 frame := awaitTestValue(t, transport.sent, "proxy resize did not reach remote transport")
111116 require.Equal(t, ports.MsgResize, frame.Type)
112117 resize, err := ports.UnmarshalResize(frame.Payload)
@@ -125,6 +130,109 @@ func TestProxyResizeUsesReducedRemoteGeometryOnce(t *testing.T) {
125130 require.Equal(t, 8, state.layout.area.Height)
126131 }
127132
133+// newAttachedProxyFixture publishes one proxy with a local thin client bound to
134+// clientTr and its remote link bound to remoteTr, using no render coordinator so
135+// invalidation paints synchronously.
136+func newAttachedProxyFixture(t *testing.T, d *Daemon, clientTr ports.Transport, remoteTr ports.Transport) (*proxySession, *attachedClient) {
137+ t.Helper()
138+ proxy, err := newProxySession(domain.RemoteSessionKey{Host: "arch", Name: "work"}, domain.Size{Cols: 16, Rows: 6})
139+ require.NoError(t, err)
140+ proxy.mu.Lock()
141+ proxy.meta = ports.SessionMeta{SessionName: "work", Tabs: []ports.SessionTabMeta{{Index: 0, Name: "shell"}}}
142+ proxy.transport = remoteTr
143+ proxy.linkGeneration = 1
144+ proxy.mu.Unlock()
145+ d.mu.Lock()
146+ require.True(t, d.registerSessionLocked(proxy))
147+ d.mu.Unlock()
148+
149+ ac := &attachedClient{tr: clientTr, output: newOutputStateStream(), size: domain.Size{Cols: 16, Rows: 6}}
150+ ac.setSession(proxy)
151+ proxy.sessionCore.mu.Lock()
152+ proxy.client = ac
153+ proxy.sessionCore.mu.Unlock()
154+ return proxy, ac
155+}
156+
157+func newFailingTransport(t *testing.T) *portsmocks.MockTransport {
158+ t.Helper()
159+ tr := portsmocks.NewMockTransport(t)
160+ tr.EXPECT().Send(mock.Anything).Return(io.ErrClosedPipe).Maybe()
161+ tr.EXPECT().Close().Return(nil).Maybe()
162+ return tr
163+}
164+
165+// TestProxyResizeRepaintsLocalChromeWhenRemoteSendFails covers the split
166+// between "the local content rectangle moved" and "the remote was told". The
167+// local VT has already been resized, so the client owes a repaint either way.
168+func TestProxyResizeRepaintsLocalChromeWhenRemoteSendFails(t *testing.T) {
169+ d := newTestDaemon(t, nil, stubClock{})
170+ clientTr, sent := newCapturingTransport(t)
171+ remote := newProxyTestTransport()
172+ remote.sendFails.Store(true)
173+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, remote)
174+
175+ d.resizeProxyForLease(proxy, ac, nil, domain.Size{Cols: 24, Rows: 10})
176+
177+ frame := awaitTestValue(t, sent, "failed remote resize did not repaint local chrome")
178+ require.Equal(t, ports.MsgOutput, frame.Type)
179+ proxy.mu.Lock()
180+ content := proxy.contentSize
181+ proxy.mu.Unlock()
182+ require.Equal(t, contentSize(domain.Size{Cols: 24, Rows: 10}, false), content, "local VT resized despite the failed send")
183+}
184+
185+// TestProxyResizeWithoutGeometryChangeDoesNotRepaint is the other half: an
186+// identical size is not a reason to repaint.
187+func TestProxyResizeWithoutGeometryChangeDoesNotRepaint(t *testing.T) {
188+ d := newTestDaemon(t, nil, stubClock{})
189+ clientTr, sent := newCapturingTransport(t)
190+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
191+
192+ d.resizeProxyForLease(proxy, ac, nil, domain.Size{Cols: 16, Rows: 6})
193+
194+ select {
195+ case frame := <-sent:
196+ t.Fatalf("no-op resize repainted: %v", frame.Type)
197+ default:
198+ }
199+}
200+
201+// TestProxyPrepareFailureRepaintsWithoutLocalSession covers a failed render
202+// transaction on an attachment that has no local session to report through.
203+func TestProxyPrepareFailureRepaintsWithoutLocalSession(t *testing.T) {
204+ d := newTestDaemon(t, nil, stubClock{})
205+ clientTr, sent := newCapturingTransport(t)
206+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
207+
208+ ac.sendMu.Lock() // emitFrame releases the transaction lock.
209+ state, ok := proxy.capturePrimary(ac, primaryCaptureRequest{bars: barState{status: proxy.statusSegments(false)}, reset: true})
210+ require.True(t, ok)
211+ composed := composeFrame(*state, composeCacheInput{})
212+ // A malformed frame is what makes outputStateStream.prepare fail.
213+ composed.frame = renderer.Frame{Width: 1}
214+
215+ require.True(t, d.emitFrame(proxy, ac, state, composed))
216+ frame := awaitTestValue(t, sent, "prepare failure on a proxy did not schedule a repaint")
217+ require.Equal(t, ports.MsgOutput, frame.Type)
218+}
219+
220+// TestProxySendErrorDetachesProxyAttachment covers the transport-failure
221+// cleanup path for an attachment with no local session lifecycle.
222+func TestProxySendErrorDetachesProxyAttachment(t *testing.T) {
223+ d := newTestDaemon(t, nil, stubClock{})
224+ clientTr := newFailingTransport(t)
225+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
226+
227+ d.paint(proxy, ac, true, nil)
228+
229+ proxy.sessionCore.mu.Lock()
230+ client := proxy.client
231+ proxy.sessionCore.mu.Unlock()
232+ require.Nil(t, client, "a failed send must release the proxy attachment")
233+ require.Nil(t, ac.currentAttachmentSession(), "the client must no longer point at the proxy")
234+}
235+
128236 func proxyRowText(row []renderer.Cell) string {
129237 runes := make([]rune, len(row))
130238 for i, cell := range row {
@@ -1,11 +1,14 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4+ "io"
4 "testing"5 "testing"
5 6
6 "github.com/bnema/vev/internal/domain"7 "github.com/bnema/vev/internal/domain"
7 "github.com/bnema/vev/internal/ports"8 "github.com/bnema/vev/internal/ports"
9+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
8 "github.com/bnema/vev/pkg/renderer"10 "github.com/bnema/vev/pkg/renderer"
11+ "github.com/stretchr/testify/mock"
9 "github.com/stretchr/testify/require"12 "github.com/stretchr/testify/require"
10 )13 )
11 14
@@ -106,7 +109,9 @@ func TestProxyResizeUsesReducedRemoteGeometryOnce(t *testing.T) {
106 proxy.linkGeneration = 1109 proxy.linkGeneration = 1
107 proxy.mu.Unlock()110 proxy.mu.Unlock()
108 111
109- require.True(t, proxy.resize(domain.Size{Cols: 20, Rows: 10}))112+ changed, sent := proxy.resize(domain.Size{Cols: 20, Rows: 10})
113+ require.True(t, changed)
114+ require.True(t, sent)
110 frame := awaitTestValue(t, transport.sent, "proxy resize did not reach remote transport")115 frame := awaitTestValue(t, transport.sent, "proxy resize did not reach remote transport")
111 require.Equal(t, ports.MsgResize, frame.Type)116 require.Equal(t, ports.MsgResize, frame.Type)
112 resize, err := ports.UnmarshalResize(frame.Payload)117 resize, err := ports.UnmarshalResize(frame.Payload)
@@ -125,6 +130,109 @@ func TestProxyResizeUsesReducedRemoteGeometryOnce(t *testing.T) {
125 require.Equal(t, 8, state.layout.area.Height)130 require.Equal(t, 8, state.layout.area.Height)
126 }131 }
127 132
133+// newAttachedProxyFixture publishes one proxy with a local thin client bound to
134+// clientTr and its remote link bound to remoteTr, using no render coordinator so
135+// invalidation paints synchronously.
136+func newAttachedProxyFixture(t *testing.T, d *Daemon, clientTr ports.Transport, remoteTr ports.Transport) (*proxySession, *attachedClient) {
137+ t.Helper()
138+ proxy, err := newProxySession(domain.RemoteSessionKey{Host: "arch", Name: "work"}, domain.Size{Cols: 16, Rows: 6})
139+ require.NoError(t, err)
140+ proxy.mu.Lock()
141+ proxy.meta = ports.SessionMeta{SessionName: "work", Tabs: []ports.SessionTabMeta{{Index: 0, Name: "shell"}}}
142+ proxy.transport = remoteTr
143+ proxy.linkGeneration = 1
144+ proxy.mu.Unlock()
145+ d.mu.Lock()
146+ require.True(t, d.registerSessionLocked(proxy))
147+ d.mu.Unlock()
148+
149+ ac := &attachedClient{tr: clientTr, output: newOutputStateStream(), size: domain.Size{Cols: 16, Rows: 6}}
150+ ac.setSession(proxy)
151+ proxy.sessionCore.mu.Lock()
152+ proxy.client = ac
153+ proxy.sessionCore.mu.Unlock()
154+ return proxy, ac
155+}
156+
157+func newFailingTransport(t *testing.T) *portsmocks.MockTransport {
158+ t.Helper()
159+ tr := portsmocks.NewMockTransport(t)
160+ tr.EXPECT().Send(mock.Anything).Return(io.ErrClosedPipe).Maybe()
161+ tr.EXPECT().Close().Return(nil).Maybe()
162+ return tr
163+}
164+
165+// TestProxyResizeRepaintsLocalChromeWhenRemoteSendFails covers the split
166+// between "the local content rectangle moved" and "the remote was told". The
167+// local VT has already been resized, so the client owes a repaint either way.
168+func TestProxyResizeRepaintsLocalChromeWhenRemoteSendFails(t *testing.T) {
169+ d := newTestDaemon(t, nil, stubClock{})
170+ clientTr, sent := newCapturingTransport(t)
171+ remote := newProxyTestTransport()
172+ remote.sendFails.Store(true)
173+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, remote)
174+
175+ d.resizeProxyForLease(proxy, ac, nil, domain.Size{Cols: 24, Rows: 10})
176+
177+ frame := awaitTestValue(t, sent, "failed remote resize did not repaint local chrome")
178+ require.Equal(t, ports.MsgOutput, frame.Type)
179+ proxy.mu.Lock()
180+ content := proxy.contentSize
181+ proxy.mu.Unlock()
182+ require.Equal(t, contentSize(domain.Size{Cols: 24, Rows: 10}, false), content, "local VT resized despite the failed send")
183+}
184+
185+// TestProxyResizeWithoutGeometryChangeDoesNotRepaint is the other half: an
186+// identical size is not a reason to repaint.
187+func TestProxyResizeWithoutGeometryChangeDoesNotRepaint(t *testing.T) {
188+ d := newTestDaemon(t, nil, stubClock{})
189+ clientTr, sent := newCapturingTransport(t)
190+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
191+
192+ d.resizeProxyForLease(proxy, ac, nil, domain.Size{Cols: 16, Rows: 6})
193+
194+ select {
195+ case frame := <-sent:
196+ t.Fatalf("no-op resize repainted: %v", frame.Type)
197+ default:
198+ }
199+}
200+
201+// TestProxyPrepareFailureRepaintsWithoutLocalSession covers a failed render
202+// transaction on an attachment that has no local session to report through.
203+func TestProxyPrepareFailureRepaintsWithoutLocalSession(t *testing.T) {
204+ d := newTestDaemon(t, nil, stubClock{})
205+ clientTr, sent := newCapturingTransport(t)
206+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
207+
208+ ac.sendMu.Lock() // emitFrame releases the transaction lock.
209+ state, ok := proxy.capturePrimary(ac, primaryCaptureRequest{bars: barState{status: proxy.statusSegments(false)}, reset: true})
210+ require.True(t, ok)
211+ composed := composeFrame(*state, composeCacheInput{})
212+ // A malformed frame is what makes outputStateStream.prepare fail.
213+ composed.frame = renderer.Frame{Width: 1}
214+
215+ require.True(t, d.emitFrame(proxy, ac, state, composed))
216+ frame := awaitTestValue(t, sent, "prepare failure on a proxy did not schedule a repaint")
217+ require.Equal(t, ports.MsgOutput, frame.Type)
218+}
219+
220+// TestProxySendErrorDetachesProxyAttachment covers the transport-failure
221+// cleanup path for an attachment with no local session lifecycle.
222+func TestProxySendErrorDetachesProxyAttachment(t *testing.T) {
223+ d := newTestDaemon(t, nil, stubClock{})
224+ clientTr := newFailingTransport(t)
225+ proxy, ac := newAttachedProxyFixture(t, d, clientTr, newProxyTestTransport())
226+
227+ d.paint(proxy, ac, true, nil)
228+
229+ proxy.sessionCore.mu.Lock()
230+ client := proxy.client
231+ proxy.sessionCore.mu.Unlock()
232+ require.Nil(t, client, "a failed send must release the proxy attachment")
233+ require.Nil(t, ac.currentAttachmentSession(), "the client must no longer point at the proxy")
234+}
235+
128 func proxyRowText(row []renderer.Cell) string {236 func proxyRowText(row []renderer.Cell) string {
129 runes := make([]rune, len(row))237 runes := make([]rune, len(row))
130 for i, cell := range row {238 for i, cell := range row {
M internal/usecase/daemon/render_pipeline.go +6 -1
@@ -486,8 +486,11 @@ func (d *Daemon) emitFrame(entry attachmentSession, ac *attachedClient, state *c
486486 return true
487487 }
488488 d.reportError(sess, domain.UserErr(domain.NoticeInternal, "display update failed", err))
489- d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
490489 }
490+ // Notices are session-scoped, but the repaint is not: a proxy attachment
491+ // has no local session to report through and still needs its chrome
492+ // redrawn after a failed transaction.
493+ d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
491494 return true
492495 }
493496 data := append([]byte(nil), prepared.data...)
@@ -551,6 +554,8 @@ func (d *Daemon) emitFrame(entry attachmentSession, ac *attachedClient, state *c
551554 if marks.roleEffect == nil {
552555 if sess, ok := localSession(entry); ok {
553556 d.detachOnSendError(sess, ac, sendTr)
557+ } else if proxy, ok := entry.(*proxySession); ok {
558+ d.detachProxyOnSendError(proxy, ac, sendTr)
554559 }
555560 } else {
556561 // Capture the exact admitted capability, including its coordinator
@@ -486,8 +486,11 @@ func (d *Daemon) emitFrame(entry attachmentSession, ac *attachedClient, state *c
486 return true486 return true
487 }487 }
488 d.reportError(sess, domain.UserErr(domain.NoticeInternal, "display update failed", err))488 d.reportError(sess, domain.UserErr(domain.NoticeInternal, "display update failed", err))
489- d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
490 }489 }
490+ // Notices are session-scoped, but the repaint is not: a proxy attachment
491+ // has no local session to report through and still needs its chrome
492+ // redrawn after a failed transaction.
493+ d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
491 return true494 return true
492 }495 }
493 data := append([]byte(nil), prepared.data...)496 data := append([]byte(nil), prepared.data...)
@@ -551,6 +554,8 @@ func (d *Daemon) emitFrame(entry attachmentSession, ac *attachedClient, state *c
551 if marks.roleEffect == nil {554 if marks.roleEffect == nil {
552 if sess, ok := localSession(entry); ok {555 if sess, ok := localSession(entry); ok {
553 d.detachOnSendError(sess, ac, sendTr)556 d.detachOnSendError(sess, ac, sendTr)
557+ } else if proxy, ok := entry.(*proxySession); ok {
558+ d.detachProxyOnSendError(proxy, ac, sendTr)
554 }559 }
555 } else {560 } else {
556 // Capture the exact admitted capability, including its coordinator561 // Capture the exact admitted capability, including its coordinator
M internal/usecase/daemon/session.go +30 -0
@@ -751,6 +751,36 @@ func (d *Daemon) detachIfCurrentTransport(sess *session, ac *attachedClient, exp
751751 return current
752752 }
753753
754+// detachProxyIfCurrentTransport is detachIfCurrentTransport for a proxy
755+// attachment. It fences on the same published identity — exact session pointer,
756+// exact attachment, exact transport incarnation — but reads the attachment
757+// through the shared sessionCore rather than a local session.
758+func (d *Daemon) detachProxyIfCurrentTransport(p *proxySession, ac *attachedClient, expected transportSnapshot) bool {
759+ if p == nil || ac == nil {
760+ return false
761+ }
762+ frozen := freezeRoleEffectGates(ac)
763+ defer frozen.unfreeze()
764+ if d.afterDetachRoleEffectsFrozen != nil {
765+ d.afterDetachRoleEffectsFrozen()
766+ }
767+ d.notices.routingMu.Lock()
768+ p.sessionCore.mu.Lock()
769+ current := p.client == ac && ac.currentAttachmentSession() == attachmentSession(p)
770+ if current && expected.transport != nil && !ac.transportSnapshotCurrent(expected) {
771+ current = false
772+ }
773+ if current {
774+ p.client = nil
775+ ac.roleGeneration.Add(1)
776+ ac.setSession(nil)
777+ ac.invalidateFrozenRoleCapability()
778+ }
779+ p.sessionCore.mu.Unlock()
780+ d.notices.routingMu.Unlock()
781+ return current
782+}
783+
754784 // detachIfRoleCurrent invalidates only the exact role generation and transport
755785 // whose admitted send failed. A queued stale error cannot detach a later role
756786 // publication or a rebound transport on the same attachment object.
@@ -751,6 +751,36 @@ func (d *Daemon) detachIfCurrentTransport(sess *session, ac *attachedClient, exp
751 return current751 return current
752 }752 }
753 753
754+// detachProxyIfCurrentTransport is detachIfCurrentTransport for a proxy
755+// attachment. It fences on the same published identity — exact session pointer,
756+// exact attachment, exact transport incarnation — but reads the attachment
757+// through the shared sessionCore rather than a local session.
758+func (d *Daemon) detachProxyIfCurrentTransport(p *proxySession, ac *attachedClient, expected transportSnapshot) bool {
759+ if p == nil || ac == nil {
760+ return false
761+ }
762+ frozen := freezeRoleEffectGates(ac)
763+ defer frozen.unfreeze()
764+ if d.afterDetachRoleEffectsFrozen != nil {
765+ d.afterDetachRoleEffectsFrozen()
766+ }
767+ d.notices.routingMu.Lock()
768+ p.sessionCore.mu.Lock()
769+ current := p.client == ac && ac.currentAttachmentSession() == attachmentSession(p)
770+ if current && expected.transport != nil && !ac.transportSnapshotCurrent(expected) {
771+ current = false
772+ }
773+ if current {
774+ p.client = nil
775+ ac.roleGeneration.Add(1)
776+ ac.setSession(nil)
777+ ac.invalidateFrozenRoleCapability()
778+ }
779+ p.sessionCore.mu.Unlock()
780+ d.notices.routingMu.Unlock()
781+ return current
782+}
783+
754 // detachIfRoleCurrent invalidates only the exact role generation and transport784 // detachIfRoleCurrent invalidates only the exact role generation and transport
755 // whose admitted send failed. A queued stale error cannot detach a later role785 // whose admitted send failed. A queued stale error cannot detach a later role
756 // publication or a rebound transport on the same attachment object.786 // publication or a rebound transport on the same attachment object.

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close