1package daemon
2
3import (
4 "io"
5 "sync"
6 "sync/atomic"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/stretchr/testify/require"
13)
14
15type blockedAttachmentTransport struct {
16 entered chan struct{}
17 release chan struct{}
18 sends chan ports.Frame
19 blocked atomic.Bool
20 once sync.Once
21}
22
23func (t *blockedAttachmentTransport) Send(frame ports.Frame) error {
24 if frame.Type == ports.MsgOutput && t.blocked.Load() {
25 t.once.Do(func() { close(t.entered) })
26 <-t.release
27 }
28 t.sends <- frame
29 return nil
30}
31func (*blockedAttachmentTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
32func (*blockedAttachmentTransport) Close() error { return nil }
33
34func TestStaleCapacityReadinessRequeuesWithoutAnotherInvalidation(t *testing.T) {
35 d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
36 rc := d.attachCoordinator(sess, nil, ac, true)
37 require.True(t, sess.repairAttachmentView(ac))
38 rc.opts.clock = nil // drive fire explicitly; the interleaving is the test.
39
40 firstObserved := make(chan struct{})
41 retryObserved := make(chan struct{})
42 releaseFirst := make(chan struct{})
43 releaseRetry := make(chan struct{})
44 var probes atomic.Int32
45 rc.opts.ackReadyFor = func(*attachedClient) bool {
46 switch probes.Add(1) {
47 case 1:
48 close(firstObserved)
49 <-releaseFirst
50 case 2:
51 close(retryObserved)
52 <-releaseRetry
53 }
54 return true
55 }
56
57 pane := sess.tabs[0].focusedPane()
58 pane.mu.Lock()
59 pane.screen.Write([]byte("capacity race"))
60 pane.mu.Unlock()
61 require.True(t, rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "capacity-race-test"}))
62 fireDone := make(chan struct{})
63 go func() {
64 rc.fireCurrent(false)
65 close(fireDone)
66 }()
67 awaitTestCompletion(t, firstObserved, "readiness probe did not start")
68
69 // Readiness was observed while the window had room. Fill it before the
70 // wake reaches paint; paint must report the failed final capacity check.
71 ac.sendMu.Lock()
72 ac.output.maxOutstanding = 1
73 ac.output.next = 1
74 ac.output.acked = 0
75 ac.output.syncCapacityLocked()
76 ac.sendMu.Unlock()
77 close(releaseFirst)
78 awaitTestCompletion(t, retryObserved, "capacity failure did not schedule an internal retry")
79
80 // The retry is already in flight. Restore capacity without publishing any
81 // new invalidation or ACK notification, then let that retry paint.
82 ac.sendMu.Lock()
83 ac.output.maxOutstanding = 2
84 ac.output.syncCapacityLocked()
85 ac.sendMu.Unlock()
86 close(releaseRetry)
87 frame := awaitFrame(t, sends, ports.MsgOutput)
88 output, err := ports.UnmarshalOutput(frame.Payload)
89 require.NoError(t, err)
90 require.Contains(t, string(output.Data), "capacity race")
91 awaitTestCompletion(t, fireDone, "initial fire did not finish")
92
93 rc.mu.Lock()
94 requeuedPending := rc.pending
95 rc.mu.Unlock()
96 require.False(t, requeuedPending, "successful internal retry must consume the mutation")
97 require.Equal(t, int32(2), probes.Load(), "retry must not require another external wake")
98}
99
100func TestAttachmentResizeKeepsPeerWindowAndExpandsSharedContent(t *testing.T) {
101 pty, releasePTY := newBlockingPTY(t)
102 defer releasePTY()
103 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
104 firstTransport, _ := newCapturingTransport(t)
105 sess, first, err := d.route(ports.Hello{
106 Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "work",
107 Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{1},
108 }, firstTransport)
109 require.NoError(t, err)
110 secondTransport, _ := newCapturingTransport(t)
111 _, second, err := d.route(ports.Hello{
112 Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work",
113 Size: domain.Size{Cols: 100, Rows: 40}, ClientID: [16]byte{2},
114 }, secondTransport)
115 require.NoError(t, err)
116 require.True(t, sess.repairAttachmentView(second))
117
118 tb := sess.tabs[0]
119 firstSize := first.size
120 secondRevision := second.viewSnapshot().revision
121 secondEpoch := second.output.currentEpoch()
122
123 rc := sess.renderCoordinator()
124 lease := rc.attachmentLease(second)
125 require.True(t, rc.markAttachmentReady(lease))
126 token := sess.attachmentToken(second, secondTransport)
127 token.lease = lease
128 ticket, admitted := second.beginAttachmentEffect(token)
129 require.True(t, admitted)
130 token.effect = ticket
131 require.True(t, d.resizeAttachmentForLease(token, domain.Size{Cols: 120, Rows: 50}))
132 ticket.End()
133
134 tb.mu.Lock()
135 require.Equal(t, domain.Size{Cols: 120, Rows: 48}, tb.size)
136 tb.mu.Unlock()
137 require.Equal(t, firstSize, first.size)
138 require.Equal(t, domain.Size{Cols: 120, Rows: 50}, second.size)
139 require.Equal(t, secondRevision+1, second.viewSnapshot().revision)
140 require.Greater(t, second.output.currentEpoch(), secondEpoch)
141}
142
143func TestAttachmentFirstPaintDoesNotWaitForBlockedPeer(t *testing.T) {
144 pty, releasePTY := newBlockingPTY(t)
145 defer releasePTY()
146 clock := &signalClock{timers: make(chan *signalTimer, 16)}
147 d := newTestDaemon(t, newFactory(t, pty), clock)
148 oldTransport := &blockedAttachmentTransport{entered: make(chan struct{}), release: make(chan struct{}), sends: make(chan ports.Frame, 8)}
149 sess, old, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{1}}, oldTransport)
150 require.NoError(t, err)
151 newTransport := &blockedAttachmentTransport{entered: make(chan struct{}), release: make(chan struct{}), sends: make(chan ports.Frame, 8)}
152 _, fresh, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{2}}, newTransport)
153 require.NoError(t, err)
154
155 rc := sess.renderCoordinator()
156 oldLease := rc.attachmentLease(old)
157 freshLease := rc.attachmentLease(fresh)
158 require.True(t, rc.markAttachmentReady(oldLease))
159 oldTransport.blocked.Store(true)
160 pane := sess.tabs[0].focusedPane()
161 pane.mu.Lock()
162 pane.screen.Write([]byte("slow peer"))
163 pane.mu.Unlock()
164 require.True(t, rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "test"}))
165 rc.fireCurrent(false)
166 awaitTestCompletion(t, oldTransport.entered, "slow attachment did not begin its blocked output")
167
168 require.True(t, rc.markAttachmentReady(freshLease))
169 token := sess.attachmentToken(fresh, newTransport)
170 token.lease = freshLease
171 painted := make(chan bool, 1)
172 go func() { painted <- d.firstPaintForTransition(token) }()
173 frame := awaitTestValue(t, newTransport.sends, "healthy attachment first paint was gated by slow peer")
174 require.Equal(t, ports.MsgOutput, frame.Type)
175 require.True(t, <-painted)
176 close(oldTransport.release)
177}
178
179func TestMultiAttachmentHandshakeFirstPaintNotGatedByBlockedPeer(t *testing.T) {
180 pty, releasePTY := newBlockingPTY(t)
181 defer releasePTY()
182 clock := &signalClock{timers: make(chan *signalTimer, 16)}
183 d := newTestDaemon(t, newFactory(t, pty), clock)
184 oldTransport := &blockedAttachmentTransport{entered: make(chan struct{}), release: make(chan struct{}), sends: make(chan ports.Frame, 8)}
185 sess, old, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{1}}, oldTransport)
186 require.NoError(t, err)
187 rc := sess.renderCoordinator()
188 require.True(t, rc.markAttachmentReady(rc.attachmentLease(old)))
189
190 oldTransport.blocked.Store(true)
191 pane := sess.tabs[0].focusedPane()
192 pane.mu.Lock()
193 pane.screen.Write([]byte("slow peer"))
194 pane.mu.Unlock()
195 require.True(t, rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "test"}))
196 fireDone := make(chan struct{})
197 go func() {
198 rc.fireCurrent(false)
199 close(fireDone)
200 }()
201 awaitTestCompletion(t, oldTransport.entered, "slow attachment did not begin its blocked output")
202
203 hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{2}}
204 tr, sends, releaseConn := newConn(t, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)})
205 defer releaseConn()
206 handshakeDone := make(chan struct{})
207 go func() {
208 d.handleConn(tr)
209 close(handshakeDone)
210 }()
211 awaitFrame(t, sends, ports.MsgWelcome)
212 firstPaint := awaitFrame(t, sends, ports.MsgOutput)
213 require.Equal(t, ports.MsgOutput, firstPaint.Type)
214 releaseConn()
215 awaitTestCompletion(t, handshakeDone, "multi-attachment handshake did not complete after first paint")
216 close(oldTransport.release)
217 awaitTestCompletion(t, fireDone, "slow peer output did not finish after release")
218}
219
220func TestAttachmentPaintFanoutDoesNotWaitForBlockedPeer(t *testing.T) {
221 pty, releasePTY := newBlockingPTY(t)
222 defer releasePTY()
223 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
224 firstTransport := &blockedAttachmentTransport{entered: make(chan struct{}), release: make(chan struct{}), sends: make(chan ports.Frame, 8)}
225 sess, _, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{1}}, firstTransport)
226 require.NoError(t, err)
227 secondTransport := &blockedAttachmentTransport{entered: make(chan struct{}), release: make(chan struct{}), sends: make(chan ports.Frame, 8)}
228 _, _, err = d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{2}}, secondTransport)
229 require.NoError(t, err)
230 rc := sess.renderCoordinator()
231 for _, ac := range sess.snapshotAttachments() {
232 require.True(t, rc.markAttachmentReady(rc.attachmentLease(ac)))
233 }
234 firstTransport.blocked.Store(true)
235
236 sess.tabs[0].focusedPane().screen.Write([]byte("shared"))
237 rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "test"})
238 rc.fireCurrent(false)
239 select {
240 case <-firstTransport.entered:
241 case <-time.After(time.Second):
242 t.Fatal("first attachment did not reach blocked send")
243 }
244 select {
245 case frame := <-secondTransport.sends:
246 require.Equal(t, ports.MsgOutput, frame.Type)
247 out, err := ports.UnmarshalOutput(frame.Payload)
248 require.NoError(t, err)
249 require.True(t, out.Full, "fan-out peers must send a fresh first frame even after another attachment acknowledges shared damage")
250 case <-time.After(time.Second):
251 t.Fatal("second attachment waited for first attachment transport")
252 }
253 close(firstTransport.release)
254}
255
256func TestAttachmentResizeUsesLatestClaimedSessionGeometry(t *testing.T) {
257 pty, releasePTY := newBlockingPTY(t)
258 defer releasePTY()
259 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
260 firstTransport, _ := newCapturingTransport(t)
261 sess, first, err := d.route(ports.Hello{
262 Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "work",
263 Size: domain.Size{Cols: 80, Rows: 24}, ClientID: [16]byte{1},
264 }, firstTransport)
265 require.NoError(t, err)
266 secondTransport, _ := newCapturingTransport(t)
267 _, second, err := d.route(ports.Hello{
268 Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work",
269 Size: domain.Size{Cols: 100, Rows: 40}, ClientID: [16]byte{2},
270 }, secondTransport)
271 require.NoError(t, err)
272 d.firstPaint(sess, second)
273
274 tb := sess.tabs[0]
275 tb.mu.Lock()
276 require.Equal(t, domain.Size{Cols: 100, Rows: 38}, tb.size, "the newest attachment claim must win")
277 tb.mu.Unlock()
278
279 rc := sess.renderCoordinator()
280 firstLease := rc.attachmentLease(first)
281 secondLease := rc.attachmentLease(second)
282 require.True(t, rc.markAttachmentReady(firstLease))
283 require.True(t, rc.markAttachmentReady(secondLease))
284 resize := func(ac *attachedClient, tr ports.Transport, size domain.Size) {
285 token := sess.attachmentToken(ac, tr)
286 d.handleAttachmentClientFrame(token, ports.Frame{
287 Type: ports.MsgResize,
288 Payload: mustMarshalResize(ports.Resize{Size: size}),
289 })
290 }
291 resize(second, secondTransport, domain.Size{Cols: 120, Rows: 50})
292
293 tb.mu.Lock()
294 require.Equal(t, domain.Size{Cols: 120, Rows: 48}, tb.size)
295 tb.mu.Unlock()
296 require.Equal(t, domain.Size{Cols: 80, Rows: 24}, first.size)
297 require.Equal(t, domain.Size{Cols: 120, Rows: 50}, second.size)
298
299 // A same-size resize is still a claim, so the smaller peer can take
300 // authority without changing its local window state first.
301 resize(second, secondTransport, domain.Size{Cols: 90, Rows: 30})
302 resize(first, firstTransport, domain.Size{Cols: 80, Rows: 24})
303 tb.mu.Lock()
304 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, tb.size)
305 tb.mu.Unlock()
306
307 // The latest resize wins even when it is smaller than a live peer.
308 resize(first, firstTransport, domain.Size{Cols: 70, Rows: 20})
309 tb.mu.Lock()
310 require.Equal(t, domain.Size{Cols: 70, Rows: 18}, tb.size)
311 tb.mu.Unlock()
312 require.Equal(t, domain.Size{Cols: 70, Rows: 20}, first.size)
313 require.Equal(t, domain.Size{Cols: 90, Rows: 30}, second.size)
314
315 // Detaching the latest claimant falls back to the most recent remaining
316 // attachment claim rather than reverting to the historical maximum.
317 d.clientGone(sess, first, first.transport(), true)
318 tb.mu.Lock()
319 require.Equal(t, domain.Size{Cols: 90, Rows: 28}, tb.size)
320 tb.mu.Unlock()
321}
322
323func TestStaleAttachmentGeometryClaimCannotCommit(t *testing.T) {
324 d, sess, first, _ := newManualSessionWithPTYs(t, newQuietPTY())
325 second := &attachedClient{output: newOutputStateStream()}
326 second.setSize(domain.Size{Cols: 90, Rows: 30})
327 second.setSession(sess)
328 require.True(t, sess.registerAttachment(second))
329
330 first.setSize(domain.Size{Cols: 120, Rows: 40})
331 var superseded bool
332 d.beforeSessionResizePublication = func() {
333 if superseded {
334 return
335 }
336 superseded = true
337 sess.claimGeometryOwner(second)
338 }
339 require.False(t, d.recalculateSessionGeometry(sess, first), "a newer attachment claim must reject the stale layout commit")
340
341 tb := sess.tabs[0]
342 tb.mu.Lock()
343 require.Equal(t, domain.Size{Cols: 80, Rows: 23}, tb.size)
344 tb.mu.Unlock()
345
346 d.beforeSessionResizePublication = nil
347 require.True(t, d.recalculateSessionGeometry(sess, nil))
348 tb.mu.Lock()
349 require.Equal(t, domain.Size{Cols: 90, Rows: 28}, tb.size)
350 tb.mu.Unlock()
351}
352
353func TestAttachmentMoveReconcilesSourceGeometryAfterOwnerRemoval(t *testing.T) {
354 d, source, moved, _ := newManualSessionWithPTYs(t, newQuietPTY())
355 peer := &attachedClient{output: newOutputStateStream()}
356 peer.setSize(domain.Size{Cols: 80, Rows: 24})
357 peer.setSession(source)
358 require.True(t, source.registerAttachment(peer))
359
360 moved.setSize(domain.Size{Cols: 120, Rows: 40})
361 _, claimed := source.claimGeometryOwner(moved)
362 require.True(t, claimed)
363 require.True(t, d.recalculateSessionGeometry(source, nil))
364
365 target := &session{
366 sessionCore: sessionCore{id: "target-geometry", name: "target-geometry"},
367 ctx: source.ctx,
368 cancel: func() {},
369 tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})},
370 }
371 d.mu.Lock()
372 d.sessions[target.id] = target
373 d.mu.Unlock()
374
375 result, err := d.transitionAttachment(attachmentTransitionRequest{
376 source: source,
377 target: target,
378 next: moved,
379 expectedTransport: moved.transportSnapshot(),
380 ready: true,
381 })
382 require.NoError(t, err)
383 d.deferAttachmentTransitionCleanups(result)
384 d.firstPaint(target, moved)
385
386 target.mu.Lock()
387 targetTab := target.tabs[0]
388 target.mu.Unlock()
389 targetTab.mu.Lock()
390 require.Equal(t, domain.Size{Cols: 120, Rows: 38}, targetTab.size, "target geometry must use the moved attachment claim")
391 targetTab.mu.Unlock()
392
393 tb := source.tabs[0]
394 tb.mu.Lock()
395 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, tb.size, "source geometry must fall back after a moved owner leaves")
396 tb.mu.Unlock()
397 require.Same(t, source, peer.currentSession())
398 require.Same(t, target, moved.currentSession())
399}