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