attachment_suspension_expiry_test.go RawHistoryFolder
1package daemon
2
3import (
4 "errors"
5 "io"
6 "log/slog"
7 "slices"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/protocol"
17)
18
19const testSuspendedExpiry = 7 * time.Second
20
21// expiryAckGateTransport blocks the first suspension ACK until released, so a
22// test can observe the exact window between commit and record publication.
23type expiryAckGateTransport struct {
24 closeTrackingTransport
25 entered chan struct{}
26 release chan struct{}
27 block atomic.Bool
28 once sync.Once
29}
30
31func (t *expiryAckGateTransport) SendServer(m protocol.ServerMessage) error {
32 if t.block.Load() {
33 t.once.Do(func() { close(t.entered) })
34 <-t.release
35 }
36 return t.closeTrackingTransport.SendServer(m)
37}
38
39type expiryAckFailureTransport struct{ closeTrackingTransport }
40
41func (*expiryAckFailureTransport) SendServer(protocol.ServerMessage) error {
42 return errors.New("suspension ack failed")
43}
44
45// expiryCloseSignalTransport reports exactly when the daemon physically closes
46// the retained transport, which happens after eviction cleanup runs.
47type expiryCloseSignalTransport struct {
48 closeTrackingTransport
49 closed chan struct{}
50 once sync.Once
51}
52
53func newExpiryCloseSignalTransport() *expiryCloseSignalTransport {
54 return &expiryCloseSignalTransport{closed: make(chan struct{})}
55}
56
57func (t *expiryCloseSignalTransport) Close() error {
58 t.once.Do(func() { close(t.closed) })
59 return t.closeTrackingTransport.Close()
60}
61
62// newSuspendedExpiryFixture creates one manual session, suspends its sole
63// attachment, and returns the armed fake-clock safety timer plus a transport
64// whose Close call can be observed.
65func newSuspendedExpiryFixture(t *testing.T) (*Daemon, *session, *attachedClient, *signalTimer, *expiryCloseSignalTransport) {
66 t.Helper()
67 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
68 WithSuspendedSafetyExpiry(testSuspendedExpiry)(d)
69 d.attachCoordinator(sess, nil, ac, true)
70 tr := newExpiryCloseSignalTransport()
71 ac.replaceTransport(tr)
72 ac.resumeCapable, ac.resumeToken = true, 123
73 token := sess.captureAttachmentCapability(ac, tr)
74 ac.installTestAttachmentCapability(token)
75 clock := &signalClock{timers: make(chan *signalTimer, 64)}
76 d.clock = clock
77 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
78 timer := awaitTimerWithDuration(t, clock.timers, testSuspendedExpiry)
79 return d, sess, ac, timer, tr
80}
81
82// awaitTimerWithDuration drains unrelated watchdog timers (the suspension ACK
83// has its own bounded-send deadline) until the requested one is armed.
84func awaitTimerWithDuration(t *testing.T, timers <-chan *signalTimer, want time.Duration) *signalTimer {
85 t.Helper()
86 for range 8 {
87 timer := awaitTestValue(t, timers, "expected timer was never armed")
88 if timer.duration == want {
89 return timer
90 }
91 }
92 t.Fatalf("no timer with duration %s was armed", want)
93 return nil
94}
95
96func waitForAttachmentCleanup(t *testing.T, d *Daemon) {
97 t.Helper()
98 done := make(chan struct{})
99 go func() {
100 d.attachmentCleanupWg.Wait()
101 close(done)
102 }()
103 awaitTestCompletion(t, done, "suspension watcher was never joined")
104}
105
106func TestSuspendedAttachmentSafetyExpiryEvictsAfterTimer(t *testing.T) {
107 d, sess, ac, timer, tr := newSuspendedExpiryFixture(t)
108
109 evicted := make(chan struct{})
110 var once sync.Once
111 d.afterClientGoneDetach = func() { once.Do(func() { close(evicted) }) }
112
113 // Suspended-only presence is not observable, but membership and transport
114 // ownership are retained until expiry.
115 require.False(t, sess.snapshotView(viewOptions{}).attached, "suspended-only session must not be observably attached")
116 require.True(t, slices.Contains(sess.snapshotAttachments(), ac), "suspension retains session membership")
117 d.mu.Lock()
118 require.NotNil(t, d.suspended[ac], "suspension arms the daemon-owned safety expiry")
119 d.mu.Unlock()
120
121 timer.ch <- time.Now()
122 awaitTestCompletion(t, evicted, "safety expiry never evicted the suspended attachment")
123 awaitTestCompletion(t, tr.closed, "expiry never closed the retained transport")
124
125 require.Nil(t, ac.currentAttachmentSession())
126 sess.mu.Lock()
127 _, registered := sess.attachments[ac]
128 sess.mu.Unlock()
129 require.False(t, registered, "expiry releases session membership")
130 require.True(t, tr.Closed(), "expiry closes the retained transport")
131 d.mu.Lock()
132 _, retained := d.suspended[ac]
133 parked := d.resume.parked[123]
134 d.mu.Unlock()
135 require.False(t, retained)
136 require.Nil(t, parked, "expiry must not park a resume credential")
137 require.False(t, ac.parked)
138 require.False(t, sess.snapshotView(viewOptions{}).attached)
139 require.Same(t, sess, firstSession(d), "expiry leaves the session to headless lifetime policy")
140 waitForAttachmentCleanup(t, d)
141}
142
143func TestSuspendedAttachmentExpiryLeavesEphemeralSessionRegistered(t *testing.T) {
144 d, sess, ac, timer, _ := newSuspendedExpiryFixture(t)
145 d.ephemeralConfig.Store(&domain.EphemeralConfig{CloseOnExit: false})
146 sess.mu.Lock()
147 sess.ephemeral = true
148 sess.mu.Unlock()
149 evicted := make(chan struct{})
150 var once sync.Once
151 d.afterClientGoneDetach = func() { once.Do(func() { close(evicted) }) }
152 timer.ch <- time.Now()
153 awaitTestCompletion(t, evicted, "safety expiry never evicted the ephemeral attachment")
154 require.Same(t, sess, firstSession(d), "with close-on-exit off, ephemeral sessions follow headless lifetime policy")
155 require.Nil(t, ac.currentAttachmentSession())
156 waitForAttachmentCleanup(t, d)
157}
158
159func TestSuspendedAttachmentExpiryCancelledByActivation(t *testing.T) {
160 d, sess, ac, _, tr := newSuspendedExpiryFixture(t)
161 require.NoError(t, d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{
162 RequestID: 2,
163 Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name},
164 Size: ac.sizeSnapshot(),
165 }))
166 d.mu.Lock()
167 _, retained := d.suspended[ac]
168 d.mu.Unlock()
169 require.False(t, retained, "activation cancels the safety expiry")
170 require.Equal(t, attachmentActive, ac.attachmentActivity())
171 require.True(t, sess.snapshotView(viewOptions{}).attached)
172 require.False(t, tr.Closed())
173 waitForAttachmentCleanup(t, d)
174}
175
176func TestSuspendedAttachmentExpiryIsRenameSafe(t *testing.T) {
177 d, sess, ac, timer, _ := newSuspendedExpiryFixture(t)
178 // A rename keeps the daemon incarnation; expiry must not require the current
179 // name to equal the retained suspension target.
180 sess.mu.Lock()
181 sess.name = "renamed"
182 sess.mu.Unlock()
183 evicted := make(chan struct{})
184 var once sync.Once
185 d.afterClientGoneDetach = func() { once.Do(func() { close(evicted) }) }
186 timer.ch <- time.Now()
187 awaitTestCompletion(t, evicted, "rename must not postpone terminal eviction")
188 require.Nil(t, ac.currentAttachmentSession())
189 waitForAttachmentCleanup(t, d)
190}
191
192func TestSuspendedAttachmentExpiryLosesToDisconnect(t *testing.T) {
193 d, sess, ac, timer, _ := newSuspendedExpiryFixture(t)
194 var detaches atomic.Int64
195 d.afterClientGoneDetach = func() { detaches.Add(1) }
196 // Run the competing disconnect while the expiry watcher is paused before it
197 // freezes the gate.
198 d.beforeSuspendedExpiryFreeze = func(a *attachedClient) {
199 d.clientGone(sess, a, a.transport(), false)
200 }
201 timer.ch <- time.Now()
202 waitForAttachmentCleanup(t, d)
203 require.Nil(t, ac.currentAttachmentSession())
204 require.Equal(t, int64(1), detaches.Load(), "the losing expiry must not evict twice")
205 d.mu.Lock()
206 _, retained := d.suspended[ac]
207 d.mu.Unlock()
208 require.False(t, retained)
209}
210
211func TestSuspendedAttachmentExpiryLosesToActivation(t *testing.T) {
212 d, sess, ac, timer, _ := newSuspendedExpiryFixture(t)
213 var detaches atomic.Int64
214 d.afterClientGoneDetach = func() { detaches.Add(1) }
215 activationErr := make(chan error, 1)
216 d.beforeSuspendedExpiryFreeze = func(a *attachedClient) {
217 activationErr <- d.activateAttachment(a, a.transportSnapshot(), protocol.ActivateAttachment{
218 RequestID: 2,
219 Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name},
220 Size: a.sizeSnapshot(),
221 })
222 }
223 timer.ch <- time.Now()
224 require.NoError(t, awaitTestValue(t, activationErr, "activation never ran"))
225 waitForAttachmentCleanup(t, d)
226 require.Equal(t, attachmentActive, ac.attachmentActivity())
227 require.True(t, sess.attachmentRegistered(ac))
228 require.Equal(t, int64(0), detaches.Load(), "the losing expiry must not evict an activated attachment")
229}
230
231func TestSuspendedAttachmentExpiryRequiresExactRecord(t *testing.T) {
232 tests := []struct {
233 name string
234 prepare func(t *testing.T, d *Daemon, ac *attachedClient, record *suspendedAttachmentRetention) *suspendedAttachmentRetention
235 }{
236 {
237 name: "stale generation",
238 prepare: func(t *testing.T, d *Daemon, ac *attachedClient, record *suspendedAttachmentRetention) *suspendedAttachmentRetention {
239 stale := &suspendedAttachmentRetention{
240 ac: record.ac, sess: record.sess, transport: record.transport,
241 generation: record.generation + 1, timer: record.timer, done: record.done,
242 }
243 d.mu.Lock()
244 d.suspended[ac] = stale
245 d.mu.Unlock()
246 return stale
247 },
248 },
249 {
250 name: "stale transport incarnation",
251 prepare: func(t *testing.T, d *Daemon, ac *attachedClient, record *suspendedAttachmentRetention) *suspendedAttachmentRetention {
252 stale := &suspendedAttachmentRetention{
253 ac: record.ac, sess: record.sess, generation: record.generation,
254 transport: transportSnapshot{transport: record.transport.transport, incarnation: record.transport.incarnation + 1},
255 timer: record.timer, done: record.done,
256 }
257 d.mu.Lock()
258 d.suspended[ac] = stale
259 d.mu.Unlock()
260 return stale
261 },
262 },
263 {
264 name: "foreign record identity",
265 prepare: func(t *testing.T, d *Daemon, ac *attachedClient, record *suspendedAttachmentRetention) *suspendedAttachmentRetention {
266 return &suspendedAttachmentRetention{
267 ac: record.ac, sess: record.sess, generation: record.generation,
268 transport: record.transport, timer: record.timer, done: record.done,
269 }
270 },
271 },
272 {
273 name: "activity no longer suspended",
274 prepare: func(t *testing.T, d *Daemon, ac *attachedClient, record *suspendedAttachmentRetention) *suspendedAttachmentRetention {
275 ac.lifecycle.mu.Lock()
276 ac.lifecycle.activity = attachmentActive
277 ac.lifecycle.mu.Unlock()
278 return record
279 },
280 },
281 }
282 for _, tt := range tests {
283 t.Run(tt.name, func(t *testing.T) {
284 d, sess, ac, _, _ := newSuspendedExpiryFixture(t)
285 d.mu.Lock()
286 record := d.suspended[ac]
287 d.mu.Unlock()
288 require.NotNil(t, record)
289 expiring := tt.prepare(t, d, ac, record)
290 d.expireSuspendedAttachment(expiring)
291 require.Same(t, sess, ac.currentAttachmentSession(), "a stale record must not detach the live suspension")
292 require.True(t, sess.attachmentRegistered(ac))
293 d.mu.Lock()
294 live := d.suspended[ac]
295 d.mu.Unlock()
296 require.NotNil(t, live, "the live safety expiry must survive a stale expiry")
297 // Release the live record's watcher without triggering eviction.
298 d.clearSuspendedExpiry(ac)
299 waitForAttachmentCleanup(t, d)
300 })
301 }
302}
303
304func TestSuspendedAttachmentExpiryArmsOnlyAfterAck(t *testing.T) {
305 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
306 WithSuspendedSafetyExpiry(testSuspendedExpiry)(d)
307 d.attachCoordinator(sess, nil, ac, true)
308 tr := &expiryAckGateTransport{entered: make(chan struct{}), release: make(chan struct{})}
309 tr.block.Store(true)
310 ac.replaceTransport(tr)
311 token := sess.captureAttachmentCapability(ac, tr)
312 ac.installTestAttachmentCapability(token)
313
314 done := make(chan error, 1)
315 go func() { done <- d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}) }()
316 awaitTestCompletion(t, tr.entered, "suspension never attempted the ACK")
317 d.mu.Lock()
318 armed := d.suspended[ac] != nil
319 d.mu.Unlock()
320 require.False(t, armed, "the safety expiry must not be armed before the ACK is delivered")
321
322 close(tr.release)
323 require.NoError(t, awaitTestValue(t, done, "suspension never completed"))
324 d.mu.Lock()
325 armed = d.suspended[ac] != nil
326 d.mu.Unlock()
327 require.True(t, armed, "the safety expiry must be armed once the ACK is delivered")
328 d.clearSuspendedExpiry(ac)
329}
330
331func TestSuspendedAttachmentExpiryAckFailureArmsNothing(t *testing.T) {
332 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
333 WithSuspendedSafetyExpiry(testSuspendedExpiry)(d)
334 d.attachCoordinator(sess, nil, ac, true)
335 tr := &expiryAckFailureTransport{}
336 ac.replaceTransport(tr)
337 token := sess.captureAttachmentCapability(ac, tr)
338 ac.installTestAttachmentCapability(token)
339
340 require.Error(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
341 d.mu.Lock()
342 _, armed := d.suspended[ac]
343 d.mu.Unlock()
344 require.False(t, armed, "a failed ACK must not arm a safety expiry")
345}
346
347func TestSuspendedAttachmentExpiryRearmsOnResuspension(t *testing.T) {
348 d, sess, ac, _, _ := newSuspendedExpiryFixture(t)
349 d.mu.Lock()
350 first := d.suspended[ac]
351 d.mu.Unlock()
352 require.NotNil(t, first)
353 require.NoError(t, d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{
354 RequestID: 2,
355 Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name},
356 Size: ac.sizeSnapshot(),
357 }))
358 token := sess.captureAttachmentCapability(ac, ac.transport())
359 ac.installTestAttachmentCapability(token)
360 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 3}))
361 d.mu.Lock()
362 second := d.suspended[ac]
363 d.mu.Unlock()
364 require.NotNil(t, second)
365 require.NotSame(t, first, second, "re-suspension replaces the retained safety expiry")
366 select {
367 case <-first.done:
368 default:
369 t.Fatal("the superseded safety expiry was not released")
370 }
371 d.clearSuspendedExpiry(ac)
372 waitForAttachmentCleanup(t, d)
373}
374
375func TestSuspendedAttachmentShutdownJoinsWatcher(t *testing.T) {
376 d, sess, ac, _, _ := newSuspendedExpiryFixture(t)
377 require.False(t, d.shutdownAll(protocol.ReasonServerShutdown))
378 waitForAttachmentCleanup(t, d)
379 d.mu.Lock()
380 require.Empty(t, d.suspended)
381 d.mu.Unlock()
382 require.Nil(t, ac.currentAttachmentSession())
383 sess.mu.Lock()
384 _, registered := sess.attachments[ac]
385 sess.mu.Unlock()
386 require.False(t, registered)
387}
388
389func TestWithSuspendedSafetyExpiryOption(t *testing.T) {
390 log := slog.New(slog.NewTextHandler(io.Discard, nil))
391 configured := New(nil, stubClock{}, log, WithSuspendedSafetyExpiry(37*time.Second))
392 require.Equal(t, 37*time.Second, configured.suspendedSafetyExpiry)
393 unchanged := New(nil, stubClock{}, log, WithSuspendedSafetyExpiry(0))
394 require.Equal(t, defaultSuspendedSafetyExpiry, unchanged.suspendedSafetyExpiry)
395}
396
397func TestSessionViewAttachedRequiresInteractivePresence(t *testing.T) {
398 active := &attachedClient{}
399 activating := &attachedClient{}
400 activating.lifecycle.activity = attachmentActivating
401 suspended := &attachedClient{}
402 suspended.lifecycle.activity = attachmentSuspended
403
404 tests := []struct {
405 name string
406 attachments []*attachedClient
407 want bool
408 }{
409 {name: "active only", attachments: []*attachedClient{active}, want: true},
410 {name: "activating only", attachments: []*attachedClient{activating}, want: true},
411 {name: "suspended only", attachments: []*attachedClient{suspended}, want: false},
412 {name: "mixed active and suspended", attachments: []*attachedClient{suspended, active}, want: true},
413 {name: "mixed activating and suspended", attachments: []*attachedClient{suspended, activating}, want: true},
414 {name: "no attachments", attachments: nil, want: false},
415 }
416 for _, tt := range tests {
417 t.Run(tt.name, func(t *testing.T) {
418 s := &session{sessionCore: sessionCore{attachments: map[*attachedClient]struct{}{}}}
419 for _, ac := range tt.attachments {
420 s.attachments[ac] = struct{}{}
421 }
422 require.Equal(t, tt.want, s.snapshotView(viewOptions{}).attached)
423 s.mu.Lock()
424 require.Equal(t, tt.want, sessionInteractivelyAttachedLocked(s))
425 s.mu.Unlock()
426 // Membership is retained for every registered participant, suspended
427 // entries included, and remains activatable.
428 require.Len(t, s.attachments, len(tt.attachments))
429 })
430 }
431}
432
433func TestHandleListAttachedExcludesSuspendedOnly(t *testing.T) {
434 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
435 d.attachCoordinator(sess, nil, ac, true)
436 require.True(t, listedAttached(t, d, sess.name), "an active attachment is observably attached")
437
438 token := sess.captureAttachmentCapability(ac, ac.transport())
439 ac.installTestAttachmentCapability(token)
440 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
441 require.False(t, listedAttached(t, d, sess.name), "a suspended-only session is not attached in the listing")
442 require.True(t, slices.Contains(sess.snapshotAttachments(), ac), "suspension retains membership for activation")
443
444 require.NoError(t, d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{
445 RequestID: 2,
446 Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name},
447 Size: ac.sizeSnapshot(),
448 }))
449 require.True(t, listedAttached(t, d, sess.name), "activation restores observably attached presence")
450}
451
452func listedAttached(t *testing.T, d *Daemon, name string) bool {
453 t.Helper()
454 for _, info := range listSessions(t, d).Sessions {
455 if info.Name == name {
456 require.Equal(t, protocol.SessionUp, info.State)
457 return info.Attached
458 }
459 }
460 t.Fatalf("session %q missing from the listing", name)
461 return false
462}