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