1package daemon
2
3import (
4 "errors"
5 "io"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 renderer "github.com/bnema/vev-vt"
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/usecase/keys"
16)
17
18func releaseTestGate(t *testing.T, ch chan struct{}) func() {
19 t.Helper()
20 var once sync.Once
21 release := func() { once.Do(func() { close(ch) }) }
22 t.Cleanup(release)
23 return release
24}
25
26type blockedRenderReplacementTransport struct {
27 sendEntered chan struct{}
28 closed chan struct{}
29 sendOnce sync.Once
30 closeOnce sync.Once
31 locksFree chan bool
32 d *Daemon
33 sess *session
34}
35
36func newBlockedRenderReplacementTransport(d *Daemon, sess *session) *blockedRenderReplacementTransport {
37 return &blockedRenderReplacementTransport{
38 sendEntered: make(chan struct{}),
39 closed: make(chan struct{}),
40 locksFree: make(chan bool, 1),
41 d: d,
42 sess: sess,
43 }
44}
45
46func (t *blockedRenderReplacementTransport) Send(ports.Frame) error {
47 t.sendOnce.Do(func() { close(t.sendEntered) })
48 <-t.closed
49 return errors.New("transport closed")
50}
51
52func (*blockedRenderReplacementTransport) Recv() (ports.Frame, error) {
53 return ports.Frame{}, io.EOF
54}
55
56func (t *blockedRenderReplacementTransport) Close() error {
57 locksFree := t.d.mu.TryLock()
58 if locksFree {
59 t.d.mu.Unlock()
60 }
61 if locksFree {
62 locksFree = t.d.notices.routingMu.TryLock()
63 if locksFree {
64 t.d.notices.routingMu.Unlock()
65 }
66 }
67 if locksFree {
68 locksFree = t.sess.mu.TryLock()
69 if locksFree {
70 t.sess.mu.Unlock()
71 }
72 }
73 if locksFree {
74 if rc := t.sess.renderCoordinator(); rc != nil {
75 locksFree = rc.mu.TryLock()
76 if locksFree {
77 rc.mu.Unlock()
78 }
79 }
80 }
81 t.closeOnce.Do(func() {
82 t.locksFree <- locksFree
83 close(t.closed)
84 })
85 return nil
86}
87
88type teardownBlockingTransport struct {
89 sendEntered chan struct{}
90 closed chan struct{}
91 sendOnce sync.Once
92 closeOnce sync.Once
93}
94
95func newTeardownBlockingTransport() *teardownBlockingTransport {
96 return &teardownBlockingTransport{sendEntered: make(chan struct{}), closed: make(chan struct{})}
97}
98
99func (t *teardownBlockingTransport) Send(ports.Frame) error {
100 t.sendOnce.Do(func() { close(t.sendEntered) })
101 <-t.closed
102 return errors.New("transport closed")
103}
104
105func (*teardownBlockingTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
106
107func (t *teardownBlockingTransport) Close() error {
108 t.closeOnce.Do(func() { close(t.closed) })
109 return nil
110}
111
112func (t *teardownBlockingTransport) Closed() bool {
113 select {
114 case <-t.closed:
115 return true
116 default:
117 return false
118 }
119}
120
121func startBlockedAttachmentSend(t *testing.T, token attachmentConnectionToken) <-chan struct{} {
122 t.Helper()
123 ticket, admitted := token.ac.beginAttachmentEffect(token)
124 require.True(t, admitted)
125 done := make(chan struct{})
126 go func() {
127 defer close(done)
128 defer ticket.End()
129 _ = token.ac.sendExpectedTransportForAttachment(token.transport, framePong(), ticket)
130 }()
131 return done
132}
133
134func requireAttachmentEffectGateRetired(t *testing.T, ac *attachedClient) {
135 t.Helper()
136 ac.attachmentEffects.mu.Lock()
137 defer ac.attachmentEffects.mu.Unlock()
138 require.Equal(t, attachmentEffectsStable, ac.attachmentEffects.phase)
139 require.Zero(t, ac.attachmentEffects.inFlight)
140 require.Empty(t, ac.attachmentEffects.transportEffects)
141}
142
143func TestBeginCurrentAttachmentEffectFailsClosedOnStableCapabilityMismatch(t *testing.T) {
144 _, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
145 current := sess.attachmentToken(ac, ac.transport())
146 stale := current
147 stale.generation++
148 ac.publishAttachmentCapability(stale)
149
150 result := make(chan bool, 1)
151 go func() {
152 _, ticket, admitted := ac.beginCurrentAttachmentEffect(sess, ac.transport())
153 if ticket != nil {
154 ticket.End()
155 }
156 result <- admitted
157 }()
158 select {
159 case admitted := <-result:
160 require.False(t, admitted)
161 case <-time.After(time.Second):
162 t.Fatal("stable capability mismatch spun instead of failing closed")
163 }
164}
165
166func TestBeginAttachmentLeaseEffectRejectsForeignLease(t *testing.T) {
167 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
168 rc := d.attachCoordinator(sess, nil, ac, true)
169 foreign := &attachmentLease{attachment: &attachedClient{}}
170 _, admitted := beginAttachmentLeaseEffect(sess, ac, foreign)
171 require.False(t, admitted)
172 require.NotNil(t, rc.attachmentLease(ac))
173}
174
175func attachmentEffectGatePublicationSnapshot(ac *attachedClient) (attachmentEffectPhase, attachmentCapability) {
176 ac.attachmentEffects.mu.Lock()
177 defer ac.attachmentEffects.mu.Unlock()
178 return ac.attachmentEffects.phase, ac.attachmentEffects.capability
179}
180
181func TestKillSessionAcquisitionTimeoutDoesNotPublishPartialInvalidation(t *testing.T) {
182 d, sess, first, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
183 firstTransport := &closeTrackingTransport{}
184 first.replaceTransport(firstTransport)
185
186 blockedTransport := &closeTrackingTransport{}
187 blocked := &attachedClient{tr: blockedTransport, output: newOutputStateStream(), size: first.size}
188 blocked.initOverlays()
189 blocked.setSession(sess)
190 laterTransport := &closeTrackingTransport{}
191 later := &attachedClient{tr: laterTransport, output: newOutputStateStream(), size: first.size}
192 later.initOverlays()
193 later.setSession(sess)
194 sess.mu.Lock()
195 sess.registerAttachmentLocked(blocked)
196 sess.registerAttachmentLocked(later)
197 sess.mu.Unlock()
198
199 rc := d.attachCoordinator(sess, nil, first, true)
200 firstToken := sess.attachmentToken(first, firstTransport)
201 firstToken.lease = rc.attachmentLease(first)
202 first.publishAttachmentCapability(firstToken)
203 blockedToken := sess.attachmentToken(blocked, blockedTransport)
204 blocked.publishAttachmentCapability(blockedToken)
205 laterToken := sess.attachmentToken(later, laterTransport)
206 later.publishAttachmentCapability(laterToken)
207
208 // Fix the canonical order so teardown acquires first, waits behind the gate
209 // owned by this test, and never reaches the still-stable later participant.
210 first.attachmentEffects.immutableOrder()
211 blocked.attachmentEffects.immutableOrder()
212 later.attachmentEffects.immutableOrder()
213 require.Less(t, first.attachmentEffects.order.Load(), blocked.attachmentEffects.order.Load())
214 require.Less(t, blocked.attachmentEffects.order.Load(), later.attachmentEffects.order.Load())
215
216 blockedOwner := freezeAttachmentEffectGates(blocked)
217 blockedOwnerHeld := true
218 t.Cleanup(func() {
219 if blockedOwnerHeld {
220 blockedOwner.unfreeze()
221 }
222 })
223
224 firstPhase, firstCapability := attachmentEffectGatePublicationSnapshot(first)
225 blockedPhase, blockedCapability := attachmentEffectGatePublicationSnapshot(blocked)
226 laterPhase, laterCapability := attachmentEffectGatePublicationSnapshot(later)
227 require.Equal(t, attachmentEffectsStable, firstPhase)
228 require.Equal(t, attachmentEffectsFrozen, blockedPhase)
229 require.Equal(t, attachmentEffectsStable, laterPhase)
230 firstGeneration := first.connectionGeneration.Load()
231 blockedGeneration := blocked.connectionGeneration.Load()
232 laterGeneration := later.connectionGeneration.Load()
233 firstSnapshot := first.transportSnapshot()
234 blockedSnapshot := blocked.transportSnapshot()
235 laterSnapshot := later.transportSnapshot()
236
237 clock := &signalClock{timers: make(chan *signalTimer, 1)}
238 d.clock = clock
239 frozenByKill := make(chan *attachedClient, 3)
240 d.afterAttachmentEffectGateFrozen = func(_ string, ac *attachedClient) { frozenByKill <- ac }
241 killDone := make(chan error, 1)
242 go func() { killDone <- d.killSession(sess, ports.ReasonSessionKilled, false) }()
243
244 require.Same(t, first, awaitTestValue(t, frozenByKill, "teardown did not acquire the first attachment effect gate"))
245 deadline := awaitTestValue(t, clock.timers, "teardown did not arm the gate acquisition deadline")
246 require.Equal(t, detachNotifyTimeout, deadline.duration)
247 deadline.ch <- time.Now()
248 select {
249 case err := <-killDone:
250 require.NoError(t, err)
251 case <-time.After(time.Second):
252 t.Fatal("teardown did not abort after partial gate acquisition timed out")
253 }
254 require.Empty(t, frozenByKill, "teardown acquired a participant after its deadline expired")
255
256 d.mu.Lock()
257 require.Same(t, sess, d.sessions[sess.id], "partial gate acquisition removed the session registry owner")
258 d.mu.Unlock()
259 sess.mu.Lock()
260 require.Contains(t, sess.snapshotAttachmentsLocked(), first)
261 require.Contains(t, sess.snapshotAttachmentsLocked(), blocked)
262 require.Contains(t, sess.snapshotAttachmentsLocked(), later)
263 sess.mu.Unlock()
264 for _, ac := range []*attachedClient{first, blocked, later} {
265 require.Same(t, sess, ac.currentSession(), "partial gate acquisition cleared attachment ownership")
266 }
267 require.Equal(t, firstGeneration, first.connectionGeneration.Load())
268 require.Equal(t, blockedGeneration, blocked.connectionGeneration.Load())
269 require.Equal(t, laterGeneration, later.connectionGeneration.Load())
270 require.True(t, first.transportSnapshotCurrent(firstSnapshot))
271 require.True(t, blocked.transportSnapshotCurrent(blockedSnapshot))
272 require.True(t, later.transportSnapshotCurrent(laterSnapshot))
273 require.False(t, firstTransport.Closed())
274 require.False(t, blockedTransport.Closed())
275 require.False(t, laterTransport.Closed())
276
277 firstPhase, gotFirstCapability := attachmentEffectGatePublicationSnapshot(first)
278 blockedPhase, gotBlockedCapability := attachmentEffectGatePublicationSnapshot(blocked)
279 laterPhase, gotLaterCapability := attachmentEffectGatePublicationSnapshot(later)
280 require.Equal(t, attachmentEffectsStable, firstPhase, "partially acquired gate was not rolled back")
281 require.Equal(t, attachmentEffectsFrozen, blockedPhase, "teardown released a gate owned by another transition")
282 require.Equal(t, attachmentEffectsStable, laterPhase, "unacquired later gate was frozen or invalidated")
283 require.Equal(t, firstCapability, gotFirstCapability)
284 require.Equal(t, blockedCapability, gotBlockedCapability)
285 require.Equal(t, laterCapability, gotLaterCapability)
286
287 blockedOwner.unfreeze()
288 blockedOwnerHeld = false
289 complete := freezeAttachmentEffectGates(first, blocked, later)
290 require.True(t, complete.acquired, "rolled-back gate set could not be acquired again")
291 require.True(t, complete.drained)
292 complete.unfreeze()
293 for _, token := range []attachmentConnectionToken{firstToken, blockedToken, laterToken} {
294 ticket, admitted := token.ac.beginAttachmentEffect(token)
295 require.True(t, admitted, "pre-publication abort changed an attachment capability")
296 ticket.End()
297 requireAttachmentEffectGateRetired(t, token.ac)
298 }
299}
300
301func TestKillSessionInterruptsOnlyExactParticipantBlockedSends(t *testing.T) {
302 d, sess, snatched, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
303 snatchedTransport := newTeardownBlockingTransport()
304 snatched.replaceTransport(snatchedTransport)
305 d.attachCoordinator(sess, nil, snatched, true)
306 snatched.publishAttachmentCapability(sess.attachmentToken(snatched, snatchedTransport))
307
308 activeTransport := newTeardownBlockingTransport()
309 active := &attachedClient{tr: activeTransport, output: newOutputStateStream(), size: snatched.size}
310 active.initOverlays()
311 transition, err := d.transitionAttachment(attachmentTransitionRequest{
312 target: sess, next: active, expectedTransport: active.transportSnapshot(), ready: true,
313 })
314 require.NoError(t, err)
315 defer d.deferAttachmentTransitionCleanups(transition)
316
317 unrelatedTransport := &closeTrackingTransport{}
318 unrelatedClient := &attachedClient{tr: unrelatedTransport, output: newOutputStateStream(), size: active.size}
319 unrelatedClient.initOverlays()
320 unrelated := &session{sessionCore: sessionCore{id: "unrelated", name: "unrelated", attachments: map[*attachedClient]struct{}{unrelatedClient: {}}}, ctx: sess.ctx, cancel: func() {}}
321 unrelatedClient.setSession(unrelated)
322 d.mu.Lock()
323 d.sessions[unrelated.id] = unrelated
324 d.mu.Unlock()
325 d.attachCoordinator(unrelated, nil, unrelatedClient, true)
326
327 snatchedToken := sess.attachmentToken(snatched, snatchedTransport)
328 activeToken := sess.attachmentToken(active, activeTransport)
329 snatchedDone := startBlockedAttachmentSend(t, snatchedToken)
330 activeDone := startBlockedAttachmentSend(t, activeToken)
331 awaitTestCompletion(t, snatchedTransport.sendEntered, "snatched send did not block")
332 awaitTestCompletion(t, activeTransport.sendEntered, "active send did not block")
333
334 killDone := make(chan error, 1)
335 go func() { killDone <- d.killSession(sess, ports.ReasonSessionKilled, false) }()
336 select {
337 case err := <-killDone:
338 require.NoError(t, err)
339 case <-time.After(500 * time.Millisecond):
340 _ = snatchedTransport.Close()
341 _ = activeTransport.Close()
342 t.Fatal("session kill remained pinned behind admitted participant sends")
343 }
344 awaitTestCompletion(t, snatchedDone, "snatched attachment effect ticket did not retire")
345 awaitTestCompletion(t, activeDone, "active attachment effect ticket did not retire")
346 d.waitNotifies()
347
348 require.True(t, snatchedTransport.Closed())
349 require.True(t, activeTransport.Closed())
350 require.False(t, unrelatedTransport.Closed(), "unrelated session transport was interrupted")
351 require.Contains(t, unrelated.snapshotAttachments(), unrelatedClient)
352 require.Same(t, unrelated, unrelatedClient.currentSession())
353 requireAttachmentEffectGateRetired(t, snatched)
354 requireAttachmentEffectGateRetired(t, active)
355}
356
357func TestKillSessionInterruptsCapturedSendWithoutClosingNewerIncarnation(t *testing.T) {
358 d, sess, ac, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
359 stale := newTeardownBlockingTransport()
360 ac.replaceTransport(stale)
361 d.attachCoordinator(sess, nil, ac, true)
362 ac.publishAttachmentCapability(sess.attachmentToken(ac, stale))
363
364 token := sess.attachmentToken(ac, stale)
365 sendDone := startBlockedAttachmentSend(t, token)
366 awaitTestCompletion(t, stale.sendEntered, "captured incarnation send did not block")
367 fresh := &closeTrackingTransport{}
368 d.afterAttachmentEffectParticipantsSnapshotted = func(string, []*attachedClient) {
369 ac.replaceTransport(fresh)
370 }
371
372 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
373 awaitTestCompletion(t, sendDone, "captured incarnation attachment effect ticket did not retire")
374 require.True(t, stale.Closed(), "exact captured in-flight transport was not interrupted")
375 require.False(t, fresh.Closed(), "newer transport incarnation was interrupted")
376 require.Same(t, fresh, ac.transport())
377 d.mu.Lock()
378 require.Same(t, sess, d.sessions[sess.id], "transport incarnation change should abort stale teardown publication")
379 d.mu.Unlock()
380 sess.mu.Lock()
381 require.Contains(t, sess.snapshotAttachmentsLocked(), ac)
382 sess.mu.Unlock()
383 requireAttachmentEffectGateRetired(t, ac)
384}
385
386func TestShutdownSignalsServeWhenParticipantGateAcquisitionTimesOut(t *testing.T) {
387 d, sess, active, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
388 transport := &closeTrackingTransport{}
389 active.replaceTransport(transport)
390 d.attachCoordinator(sess, nil, active, true)
391 active.publishAttachmentCapability(sess.attachmentToken(active, transport))
392
393 owner := freezeAttachmentEffectGates(active)
394 require.True(t, owner.acquired)
395 defer owner.unfreeze()
396
397 clock := &signalClock{timers: make(chan *signalTimer, 1)}
398 d.clock = clock
399 shutdownDone := make(chan bool, 1)
400 go func() { shutdownDone <- d.shutdownAll(ports.ReasonServerShutdown) }()
401
402 deadline := awaitTestValue(t, clock.timers, "shutdown did not arm the gate acquisition deadline")
403 deadline.ch <- clock.Now()
404 require.False(t, awaitTestValue(t, shutdownDone, "shutdown did not return after its gate acquisition deadline"))
405
406 d.mu.Lock()
407 registered := d.sessions[sess.id]
408 d.mu.Unlock()
409 require.Same(t, sess, registered, "timed-out initial teardown should leave the session for Serve's shutdown pass")
410 select {
411 case <-d.done:
412 case <-time.After(time.Second):
413 t.Fatal("shutdown request did not signal Serve after the initial teardown pass timed out")
414 }
415}
416
417func TestShutdownInterruptsBlockedParticipantSend(t *testing.T) {
418 d, sess, active, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
419 blocked := newTeardownBlockingTransport()
420 active.replaceTransport(blocked)
421 d.attachCoordinator(sess, nil, active, true)
422 active.publishAttachmentCapability(sess.attachmentToken(active, blocked))
423
424 token := sess.attachmentToken(active, blocked)
425 sendDone := startBlockedAttachmentSend(t, token)
426 awaitTestCompletion(t, blocked.sendEntered, "active send did not block")
427
428 shutdownDone := make(chan bool, 1)
429 go func() { shutdownDone <- d.shutdownAll(ports.ReasonServerShutdown) }()
430 select {
431 case incomplete := <-shutdownDone:
432 require.False(t, incomplete)
433 case <-time.After(500 * time.Millisecond):
434 _ = blocked.Close()
435 t.Fatal("shutdown remained pinned behind admitted participant send")
436 }
437 awaitTestCompletion(t, sendDone, "shutdown-interrupted attachment effect ticket did not retire")
438 d.waitNotifies()
439 require.True(t, blocked.Closed())
440 requireAttachmentEffectGateRetired(t, active)
441}
442
443func TestRenderSendFailureCleanupIsDeadlineBoundedWhileAttachmentEffectGateIsBusy(t *testing.T) {
444 d, sess, ac, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
445 transport := &closeTrackingTransport{}
446 ac.replaceTransport(transport)
447 rc := d.attachCoordinator(sess, nil, ac, true)
448 token := sess.attachmentToken(ac, transport)
449 token.lease = rc.attachmentLease(ac)
450 ac.publishAttachmentCapability(token)
451
452 clock := &signalClock{timers: make(chan *signalTimer, 1)}
453 d.clock = clock
454 frozen := freezeAttachmentEffectGates(ac)
455 defer frozen.unfreeze()
456 cleanupDone := make(chan struct{})
457 d.afterAttachmentSendErrorCleanup = func() { close(cleanupDone) }
458
459 launchCleanup := d.reserveAttachmentSendErrorCleanup(token, transport)
460 launchCleanup()
461 deadline := awaitTestValue(t, clock.timers, "render cleanup did not arm its deadline")
462 require.Equal(t, detachNotifyTimeout, deadline.duration)
463 deadline.ch <- time.Now()
464 awaitTestCompletion(t, cleanupDone, "deadline did not release blocked render cleanup")
465 awaitTestCompletion(t, waitGroupDone(&d.attachmentCleanupWg), "bounded render cleanup leaked its tracked goroutine")
466
467 sess.mu.Lock()
468 require.Contains(t, sess.snapshotAttachmentsLocked(), ac)
469 sess.mu.Unlock()
470 require.False(t, transport.Closed())
471}
472
473func TestAttachmentEffectGateReplacementInterruptsBlockedOldRenderBeforePublication(t *testing.T) {
474 d, sess, old, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
475 oldTransport := newBlockedRenderReplacementTransport(d, sess)
476 old.replaceTransport(oldTransport)
477 rc := d.attachCoordinator(sess, nil, old, true)
478 lease := rc.attachmentLease(old)
479 token := sess.attachmentToken(old, oldTransport)
480 token.lease = lease
481 old.publishAttachmentCapability(token)
482
483 paintDone := make(chan struct{})
484 go func() {
485 d.paint(sess, old, true, lease)
486 close(paintDone)
487 }()
488 <-oldTransport.sendEntered
489
490 newTransport := &closeTrackingTransport{}
491 next := &attachedClient{tr: newTransport, output: newOutputStateStream(), size: old.size}
492 next.initOverlays()
493 transitionDone := make(chan attachmentTransitionResult, 1)
494 transitionErr := make(chan error, 1)
495 go func() {
496 result, err := d.transitionAttachment(attachmentTransitionRequest{
497 target: sess, next: next, expectedTransport: next.transportSnapshot(), ready: true,
498 })
499 transitionDone <- result
500 transitionErr <- err
501 }()
502
503 var result attachmentTransitionResult
504 select {
505 case result = <-transitionDone:
506 case <-time.After(time.Second):
507 t.Fatal("replacement publication remained blocked behind the old render send")
508 }
509 require.NoError(t, <-transitionErr)
510 require.Contains(t, sess.snapshotAttachments(), next)
511 require.Contains(t, sess.snapshotAttachments(), old)
512 require.False(t, newTransport.Closed(), "new attachment transport was affected by publication")
513
514 // Generic attachment publication does not interrupt or retire another
515 // registered connection. Release the test transport explicitly.
516 require.NoError(t, oldTransport.Close())
517 select {
518 case <-paintDone:
519 case <-time.After(time.Second):
520 t.Fatal("closing the test transport did not unblock its render")
521 }
522 d.deferAttachmentTransitionCleanups(result)
523 d.attachmentCleanupWg.Wait()
524 require.Same(t, oldTransport, old.transport(), "unrelated attachment link was retired")
525 require.False(t, newTransport.Closed())
526}
527
528func TestAttachmentEffectGateAdmittedEffectCompletesBeforeConflictingTransition(t *testing.T) {
529 d, sess, old, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
530 oldTransport := newDatagramTestTransport()
531 old.replaceTransport(oldTransport)
532 rc := d.attachCoordinator(sess, nil, old, true)
533 token := sess.attachmentToken(old, oldTransport)
534 token.lease = rc.attachmentLease(old)
535 old.publishAttachmentCapability(token)
536
537 ticket, ok := old.beginAttachmentEffect(token)
538 require.True(t, ok)
539
540 next := &attachedClient{tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: old.size}
541 next.initOverlays()
542 done := make(chan error, 1)
543 go func() {
544 _, err := d.transitionAttachment(attachmentTransitionRequest{
545 target: sess, next: next, expectedTransport: next.transportSnapshot(), ready: true,
546 })
547 done <- err
548 }()
549
550 // Adding an independent attachment does not wait on another connection's
551 // effect gate.
552 require.NoError(t, <-done)
553 ticket.End()
554}
555
556func TestAttachmentEffectGateAdmittedActiveEffectsFinishBeforeReplacement(t *testing.T) {
557 tests := []struct {
558 name string
559 setup func(*Daemon, *session, *attachedClient, *transactionalResizePTY)
560 frame ports.Frame
561 assert func(*testing.T, *attachedClient, *transactionalResizePTY)
562 }{
563 {
564 name: "PTY input",
565 frame: frameInput([]byte("reserved")),
566 assert: func(t *testing.T, _ *attachedClient, pty *transactionalResizePTY) {
567 require.Equal(t, [][]byte{[]byte("reserved")}, pty.writes())
568 },
569 },
570 {
571 name: "mouse input",
572 setup: func(_ *Daemon, sess *session, _ *attachedClient, _ *transactionalResizePTY) {
573 p := testAttachmentTab(sess).focusedPane()
574 p.mu.Lock()
575 p.screen.Write([]byte("\x1b[?1000h\x1b[?1006h"))
576 p.mu.Unlock()
577 },
578 frame: frameInput([]byte("\x1b[<0;1;2M")),
579 assert: func(t *testing.T, _ *attachedClient, pty *transactionalResizePTY) {
580 require.NotEmpty(t, pty.writes())
581 },
582 },
583 {
584 name: "key action",
585 frame: frameInput([]byte("\x1b ")),
586 assert: func(t *testing.T, ac *attachedClient, _ *transactionalResizePTY) {
587 require.True(t, ac.overlays.paletteActive())
588 },
589 },
590 {
591 name: "overlay mutation",
592 setup: func(d *Daemon, sess *session, ac *attachedClient, _ *transactionalResizePTY) {
593 d.enterPrompt(sess, ac, "Rename", "", func(string) error { return nil })
594 },
595 frame: frameInput([]byte("x")),
596 assert: func(t *testing.T, ac *attachedClient, _ *transactionalResizePTY) {
597 ac.overlays.promptMu.Lock()
598 defer ac.overlays.promptMu.Unlock()
599 require.Equal(t, "x", ac.overlays.prompt.Value())
600 },
601 },
602 {
603 name: "theme",
604 frame: ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(ports.Theme{
605 HasForeground: true, Foreground: renderer.RGB{R: 1, G: 2, B: 3},
606 HasBackground: true, Background: renderer.RGB{R: 4, G: 5, B: 6},
607 })},
608 assert: func(t *testing.T, ac *attachedClient, _ *transactionalResizePTY) {
609 require.True(t, ac.getClientTheme().HasFG)
610 },
611 },
612 {
613 name: "ACK",
614 setup: func(_ *Daemon, _ *session, ac *attachedClient, _ *transactionalResizePTY) {
615 ac.output.next = 3
616 },
617 frame: ports.Frame{Type: ports.MsgAck, Payload: mustMarshalAck(ports.Ack{Epoch: 1, State: 1})},
618 assert: func(t *testing.T, ac *attachedClient, _ *transactionalResizePTY) {
619 ac.sendMu.Lock()
620 defer ac.sendMu.Unlock()
621 require.Equal(t, uint64(1), ac.output.acked)
622 },
623 },
624 }
625
626 for _, tt := range tests {
627 t.Run(tt.name, func(t *testing.T) {
628 pty := &transactionalResizePTY{}
629 d, sess, old, _ := newManualSessionWithPTYs(t, pty)
630 oldTransport := newDatagramTestTransport()
631 old.replaceTransport(oldTransport)
632 old.keys = keys.NewRouter(d.clock, daemonKeyHandler{d: d, ac: old}, &d.bindings)
633 rc := d.attachCoordinator(sess, nil, old, true)
634 token := sess.attachmentToken(old, oldTransport)
635 token.lease = rc.attachmentLease(old)
636 if tt.setup != nil {
637 tt.setup(d, sess, old, pty)
638 }
639
640 admitted := make(chan struct{})
641 release := make(chan struct{})
642 d.afterAttachmentEffectAdmitted = func(attachmentConnectionToken) {
643 close(admitted)
644 <-release
645 }
646 effectDone := make(chan struct{})
647 go func() {
648 d.handleAttachmentClientFrame(token, tt.frame)
649 close(effectDone)
650 }()
651 <-admitted
652
653 next := &attachedClient{tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: old.size}
654 next.initOverlays()
655 transitionDone := make(chan error, 1)
656 go func() {
657 _, err := d.transitionAttachment(attachmentTransitionRequest{
658 target: sess, next: next, expectedTransport: next.transportSnapshot(), ready: true,
659 })
660 transitionDone <- err
661 }()
662 require.NoError(t, <-transitionDone)
663 close(release)
664 <-effectDone
665 tt.assert(t, old, pty)
666 })
667 }
668}
669
670func TestAttachedRouteSnapshotIsAcceptedAsAnAttachmentValue(t *testing.T) {
671 d, source, ac, _ := newManualSessionWithPTYs(t, nil)
672 transport := &closeTrackingTransport{}
673 ac.replaceTransport(transport)
674 rc := d.attachCoordinator(source, nil, ac, true)
675 token := source.attachmentToken(ac, transport)
676 token.lease = rc.attachmentLease(ac)
677 ac.publishAttachmentCapability(token)
678
679 snapshot := ports.RecentRouteSnapshot{
680 Generation: 4,
681 Active: ports.RouteRef{Key: 8, Generation: 4},
682 Home: ports.RouteRef{Key: 8, Generation: 4},
683 Entries: []ports.RecentRouteEntry{{
684 Key: 7, Generation: 3, Name: "previous", Kind: ports.RouteKindLocal,
685 }},
686 }
687 payload, err := ports.MarshalRecentRouteSnapshot(snapshot)
688 require.NoError(t, err)
689
690 require.False(t, d.handleAttachmentClientFrame(token, ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: payload}))
691 require.Equal(t, snapshot, ac.routeSnapshotCopy())
692}
693
694func TestAttachedNavigationCommandSendsResultAfterLocalTransition(t *testing.T) {
695 d, source, ac, _ := newManualSessionWithPTYs(t, nil)
696 target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
697 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
698 }}
699 d.mu.Lock()
700 d.sessions[target.id] = target
701 d.mu.Unlock()
702 ac.setRouteSnapshot(ports.RecentRouteSnapshot{
703 Generation: 2,
704 Active: ports.RouteRef{Key: 2, Generation: 2},
705 Previous: ports.RouteRef{Key: 1, Generation: 1},
706 Entries: []ports.RecentRouteEntry{{Key: 1, Generation: 1, Name: "target", Kind: ports.RouteKindLocal}},
707 })
708
709 transport := &closeTrackingTransport{}
710 ac.replaceTransport(transport)
711 rc := d.attachCoordinator(source, nil, ac, true)
712 token := source.attachmentToken(ac, transport)
713 token.lease = rc.attachmentLease(ac)
714 ac.publishAttachmentCapability(token)
715 payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
716 Version: ports.ProtocolVersion, RequestID: 1, Slug: "back-session", Attached: true,
717 })
718 require.NoError(t, err)
719
720 require.False(t, d.handleAttachmentClientFrame(token, ports.Frame{Type: ports.MsgCommand, Payload: payload}))
721 require.Same(t, source, ac.currentAttachmentSession())
722 frames := transport.Sends()
723 require.NotEmpty(t, frames)
724 var result ports.CommandResult
725 for _, frame := range frames {
726 if frame.Type != ports.MsgCommandResult {
727 continue
728 }
729 result, err = ports.UnmarshalCommandResult(frame.Payload)
730 require.NoError(t, err)
731 break
732 }
733 require.True(t, result.OK, result.Text)
734 var action ports.RouteNavigationAction
735 for _, frame := range frames {
736 if frame.Type != ports.MsgNavigateRecentRoute {
737 continue
738 }
739 action, err = ports.UnmarshalRouteNavigationAction(frame.Payload)
740 require.NoError(t, err)
741 }
742 require.Equal(t, ports.RouteNavigationAction{SnapshotGeneration: 2, Key: 1, Generation: 1}, action)
743 require.False(t, transport.Closed(), "client navigation must not tear down the attachment")
744}
745
746func TestJumpAttentionAdmittedHandoffCrossesSessions(t *testing.T) {
747 p1, release1 := newBlockingPTY(t)
748 p2, release2 := newBlockingPTY(t)
749 p3, release3 := newBlockingPTY(t)
750 defer release1()
751 defer release2()
752 defer release3()
753 d, source, ac, _ := newManualSessionWithPTYs(t, p1)
754 target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
755 newTab(p2, domain.Size{Cols: 80, Rows: 23}),
756 newTab(p3, domain.Size{Cols: 80, Rows: 23}),
757 }}
758 target.tabs[1].attention = true
759 target.tabs[1].attentionAt = time.Unix(1, 0)
760 d.sessions[target.id] = target
761
762 rc := d.attachCoordinator(source, nil, ac, true)
763 token := source.attachmentToken(ac, ac.transport())
764 token.lease = rc.attachmentLease(ac)
765 ac.publishAttachmentCapability(token)
766
767 d.handleAttachmentClientFrame(token, frameInput([]byte("\x1ba")))
768
769 require.Same(t, target, ac.currentSession())
770 require.Equal(t, 1, testAttachmentTabIndex(target))
771}
772
773func TestPickerDeleteDoesNotDeleteSourceAfterInitiatorReplacement(t *testing.T) {
774 p, releasePTY := newBlockingPTY(t)
775 defer releasePTY()
776 d, sess, old, _ := newManualSessionWithPTYs(t, p)
777 d.enterPicker(sess, old)
778
779 oldTransport := old.transport()
780 rc := d.attachCoordinator(sess, nil, old, true)
781 token := sess.attachmentToken(old, oldTransport)
782 token.lease = rc.attachmentLease(old)
783 old.publishAttachmentCapability(token)
784
785 admissionEnded := make(chan struct{})
786 releaseAction := make(chan struct{})
787 release := releaseTestGate(t, releaseAction)
788 var admissionEndedOnce sync.Once
789 d.afterActionAttachmentEffectEnded = func(action string) {
790 if action != "picker-delete" {
791 return
792 }
793 admissionEndedOnce.Do(func() { close(admissionEnded) })
794 <-releaseAction
795 }
796 actionDone := make(chan struct{})
797 go func() {
798 d.handleAttachmentClientFrame(token, frameInput([]byte("x")))
799 close(actionDone)
800 }()
801 <-admissionEnded
802
803 next := &attachedClient{tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: old.size}
804 next.initOverlays()
805 result, err := d.transitionAttachment(attachmentTransitionRequest{
806 target: sess, next: next, expectedTransport: next.transportSnapshot(), ready: true,
807 })
808 require.NoError(t, err)
809 release()
810 <-actionDone
811
812 d.mu.Lock()
813 registered := d.sessions[sess.id]
814 d.mu.Unlock()
815 require.Same(t, sess, registered, "the replaced initiator deleted its former source session")
816 d.deferAttachmentTransitionCleanups(result)
817}
818
819func TestPickerDeleteSourceForCurrentInitiatorDoesNotDeadlock(t *testing.T) {
820 p, release := newBlockingPTY(t)
821 defer release()
822 d, sess, ac, _ := newManualSessionWithPTYs(t, p)
823 d.enterPicker(sess, ac)
824 rc := d.attachCoordinator(sess, nil, ac, true)
825 token := sess.attachmentToken(ac, ac.transport())
826 token.lease = rc.attachmentLease(ac)
827 ac.publishAttachmentCapability(token)
828
829 d.handleAttachmentClientFrame(token, frameInput([]byte("x")))
830
831 d.mu.Lock()
832 _, registered := d.sessions[sess.id]
833 d.mu.Unlock()
834 require.False(t, registered)
835}
836
837func TestAttachmentEffectGateAdmittedFirstPaintFinishesBeforeReplacement(t *testing.T) {
838 d, sess, old, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
839 oldTransport := newDatagramTestTransport()
840 old.replaceTransport(oldTransport)
841 rc := d.attachCoordinator(sess, nil, old, true)
842 token := sess.attachmentToken(old, oldTransport)
843 token.lease = rc.attachmentLease(old)
844
845 admitted := make(chan struct{})
846 release := make(chan struct{})
847 d.afterAttachmentEffectAdmitted = func(attachmentConnectionToken) {
848 close(admitted)
849 <-release
850 }
851 paintDone := make(chan bool, 1)
852 go func() { paintDone <- d.firstPaintForTransition(token) }()
853 <-admitted
854
855 next := &attachedClient{tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: old.size}
856 next.initOverlays()
857 transitionDone := make(chan error, 1)
858 go func() {
859 _, err := d.transitionAttachment(attachmentTransitionRequest{
860 target: sess, next: next, expectedTransport: next.transportSnapshot(), ready: true,
861 })
862 transitionDone <- err
863 }()
864 // An independent attachment publication does not wait on this connection's
865 // admitted first paint.
866 require.NoError(t, <-transitionDone)
867 close(release)
868 require.True(t, <-paintDone)
869}
870
871func TestAttachmentEffectGateReversedConcurrentTransitionsDoNotDeadlock(t *testing.T) {
872 d, first, a, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
873 aTransport := newDatagramTestTransport()
874 a.replaceTransport(aTransport)
875 d.attachCoordinator(first, nil, a, true)
876 first.attachmentToken(a, aTransport)
877
878 bTransport := newDatagramTestTransport()
879 b := &attachedClient{tr: bTransport, output: newOutputStateStream(), size: a.size}
880 b.initOverlays()
881 second := &session{sessionCore: sessionCore{id: "second", name: "second", attachments: map[*attachedClient]struct{}{b: {}}}}
882 b.setSession(second)
883 d.mu.Lock()
884 d.sessions[second.id] = second
885 d.mu.Unlock()
886 d.attachCoordinator(second, nil, b, true)
887 second.attachmentToken(b, bTransport)
888
889 start := make(chan struct{})
890 results := make(chan error, 2)
891 go func() {
892 <-start
893 _, err := d.transitionAttachment(attachmentTransitionRequest{
894 source: first, target: second, next: a,
895
896 expectedTransport: a.transportSnapshot(), ready: true,
897 })
898 results <- err
899 }()
900 go func() {
901 <-start
902 _, err := d.transitionAttachment(attachmentTransitionRequest{
903 source: second, target: first, next: b,
904
905 expectedTransport: b.transportSnapshot(), ready: true,
906 })
907 results <- err
908 }()
909 close(start)
910
911 errs := make([]error, 0, 2)
912 for range 2 {
913 select {
914 case err := <-results:
915 errs = append(errs, err)
916 case <-time.After(time.Second):
917 t.Fatal("reversed transitions deadlocked")
918 }
919 }
920 successes := 0
921 for _, err := range errs {
922 if err == nil {
923 successes++
924 }
925 }
926 require.Equal(t, 2, successes)
927}
928
929func TestAttachmentEffectGateFrozenConnectionRejectsLateEffect(t *testing.T) {
930 d, sess, ac, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
931 transport := newDatagramTestTransport()
932 ac.replaceTransport(transport)
933 rc := d.attachCoordinator(sess, nil, ac, true)
934 token := sess.attachmentToken(ac, transport)
935 token.lease = rc.attachmentLease(ac)
936 ac.publishAttachmentCapability(token)
937
938 frozen := freezeAttachmentEffectGates(ac)
939 require.True(t, frozen.acquired)
940 require.True(t, frozen.drained)
941 _, ok := ac.beginAttachmentEffect(token)
942 require.False(t, ok, "a frozen capability must admit no new effect")
943 frozen.unfreeze()
944}