1package daemon
2
3import (
4 "context"
5 "io"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/adapters/sessionwire"
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/wire"
16)
17
18type handshakeBlockingTransport struct {
19 blockWelcome bool
20 welcome chan struct{}
21 closed chan struct{}
22 closeOnce sync.Once
23}
24
25func newHandshakeBlockingTransport(blockWelcome bool) *handshakeBlockingTransport {
26 return &handshakeBlockingTransport{
27 blockWelcome: blockWelcome,
28 welcome: make(chan struct{}),
29 closed: make(chan struct{}),
30 }
31}
32
33func (t *handshakeBlockingTransport) Send(frame wire.Envelope) error {
34 if t.blockWelcome && envelopeMessageName(nil, frame.Payload) == "Welcome" {
35 close(t.welcome)
36 <-t.closed
37 return io.ErrClosedPipe
38 }
39 return nil
40}
41
42func (t *handshakeBlockingTransport) Recv() (wire.Envelope, error) {
43 <-t.closed
44 return wire.Envelope{}, io.ErrClosedPipe
45}
46
47func (t *handshakeBlockingTransport) Close() error {
48 t.closeOnce.Do(func() { close(t.closed) })
49 return nil
50}
51
52func TestBoundedHandshakeCompletionCancellationDoesNotDoubleReceive(t *testing.T) {
53 ctx := &completionCancellationContext{}
54 tr := &closeTrackingTransport{}
55 done := make(chan error, 1)
56 go func() {
57 done <- boundedHandshakeOperation(ctx, tr, func() error {
58 ctx.canceled = true
59 return nil
60 })
61 }()
62
63 select {
64 case err := <-done:
65 require.ErrorIs(t, err, context.Canceled)
66 case <-time.After(testWaitTimeout):
67 t.Fatal("handshake completion/cancellation race deadlocked")
68 }
69}
70
71type completionCancellationContext struct{ canceled bool }
72
73func (c *completionCancellationContext) Deadline() (time.Time, bool) { return time.Time{}, false }
74func (c *completionCancellationContext) Done() <-chan struct{} { return nil }
75func (c *completionCancellationContext) Err() error {
76 if c.canceled {
77 return context.Canceled
78 }
79 return nil
80}
81func (*completionCancellationContext) Value(any) any { return nil }
82
83func TestBoundedHandshakeCancellationDoesNotWaitForUninterruptibleOperation(t *testing.T) {
84 ctx, cancel := context.WithCancel(context.Background())
85 defer cancel()
86 tr := &closeTrackingTransport{}
87 started := make(chan struct{})
88 release := make(chan struct{})
89 operationDone := make(chan struct{})
90 result := make(chan error, 1)
91 go func() {
92 result <- boundedHandshakeOperation(ctx, tr, func() error {
93 close(started)
94 <-release
95 close(operationDone)
96 return nil
97 })
98 }()
99
100 <-started
101 cancel()
102 select {
103 case err := <-result:
104 require.ErrorIs(t, err, context.Canceled)
105 case <-time.After(testWaitTimeout):
106 t.Fatal("handshake cancellation waited for the transport operation")
107 }
108 close(release)
109 awaitTestCompletion(t, operationDone, "handshake operation did not finish")
110}
111
112func TestHandshakeFirstPaintCancellationDoesNotRacePaintResult(t *testing.T) {
113 pty, releasePTY := newBlockingPTY(t)
114 defer releasePTY()
115 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
116 ctx, cancel := context.WithCancel(context.Background())
117 defer cancel()
118
119 paintStarted := make(chan struct{})
120 releasePaint := make(chan struct{})
121 paintDone := make(chan struct{})
122 d.afterAttachmentEffectAdmitted = func(attachmentCapability) {
123 close(paintStarted)
124 <-ctx.Done()
125 <-releasePaint
126 close(paintDone)
127 }
128
129 tr := newHandshakeBlockingTransport(false)
130 handshakeDone := make(chan struct{})
131 go func() {
132 d.handleHelloFrameWithContext(ctx, make(chan struct{}), func() {}, func() {}, tr,
133 mustHello(protocol.IntentNew, "paint-cancel", domain.Size{Cols: 80, Rows: 24}))
134 close(handshakeDone)
135 }()
136
137 awaitTestCompletion(t, paintStarted, "first paint did not start")
138 cancel()
139 awaitTestCompletion(t, tr.closed, "handshake cancellation did not close the transport")
140 select {
141 case <-paintDone:
142 t.Fatal("first paint completed before cancellation returned")
143 default:
144 }
145 close(releasePaint)
146 awaitTestCompletion(t, paintDone, "delayed first paint did not finish")
147 awaitTestCompletion(t, handshakeDone, "handshake cancellation did not finish")
148}
149
150func TestHandshakeFailureAfterWelcomeDoesNotReportExplicitDetach(t *testing.T) {
151 pty, release := newBlockingPTY(t)
152 defer release()
153 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
154 tr := &closeTrackingTransport{}
155 sess, ac, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "welcome-failed", Size: domain.Size{Cols: 80, Rows: 24}, TermEnv: "xterm-256color"}, tr)
156 require.NoError(t, err)
157
158 d.failHandshakeAttachment(sess, ac, tr, true)
159
160 var detached *protocol.Detached
161 for _, frame := range tr.Sends() {
162 message, decodeErr := sessionwire.DecodeServerEnvelope(frame.Payload)
163 require.NoError(t, decodeErr)
164 if value, ok := message.(protocol.Detached); ok {
165 detached = &value
166 break
167 }
168 }
169 require.NotNil(t, detached, "the client needs a lifecycle response after Welcome")
170 require.Equal(t, protocol.ReasonServerShutdown, detached.Reason, "a failed destination handshake must return to the picker")
171}
172
173func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {
174 clock := &signalClock{timers: make(chan *signalTimer, 1)}
175 d := newTestDaemon(t, nil, clock)
176 tr := newHandshakeBlockingTransport(false)
177 done := make(chan struct{})
178 go func() {
179 d.handleConn(tr)
180 close(done)
181 }()
182
183 timer := <-clock.timers
184 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
185 timer.ch <- time.Time{}
186 awaitTestCompletion(t, done, "handshake receive did not stop after timeout")
187 requireClosedHandshakeTransport(t, tr)
188 d.mu.Lock()
189 require.Empty(t, d.sessions)
190 require.Empty(t, d.resume.parked)
191 d.mu.Unlock()
192}
193
194func TestFailedResumeHandshakeRestoresParkedCredential(t *testing.T) {
195 pty, release := newBlockingPTY(t)
196 defer release()
197 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
198 oldTransport := newHandshakeBlockingTransport(false)
199 sess, ac, err := d.route(helloResumeCapable(protocol.IntentNew, "resume-timeout", 0), oldTransport)
200 require.NoError(t, err)
201 token := ac.resumeToken
202 d.clientGone(sess, ac, oldTransport, false)
203 require.Empty(t, sess.snapshotAttachments())
204
205 clock := &signalClock{timers: make(chan *signalTimer, 1)}
206 d.clock = clock
207 resumeTransport := newHandshakeBlockingTransport(true)
208 done := make(chan struct{})
209 go func() {
210 d.handleHelloFrame(resumeTransport, mustClientEnvelope(helloResumeCapable(protocol.IntentResume, sess.name, token)))
211 close(done)
212 }()
213
214 timer := <-clock.timers
215 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
216 <-resumeTransport.welcome
217 timer.ch <- time.Time{}
218 awaitTestCompletion(t, done, "failed resume handshake did not finish")
219 requireClosedHandshakeTransport(t, resumeTransport)
220 d.mu.Lock()
221 parked := d.resume.parked[token]
222 d.mu.Unlock()
223 require.NotNil(t, parked, "failed pre-claim resume must restore the parked credential")
224 require.Same(t, ac, parked.ac)
225 require.Equal(t, token, ac.resumeToken)
226 require.True(t, ac.parked)
227 require.Empty(t, sess.snapshotAttachments())
228}
229
230func TestHandshakeTimeoutClosesBlockedWelcomeSend(t *testing.T) {
231 pty, release := newBlockingPTY(t)
232 defer release()
233 clock := &signalClock{timers: make(chan *signalTimer, 4)}
234 d := newTestDaemon(t, newFactory(t, pty), clock)
235 tr := newHandshakeBlockingTransport(true)
236 done := make(chan struct{})
237 go func() {
238 d.handleHelloFrame(tr, mustHello(protocol.IntentNew, "timeout-send", domain.Size{Cols: 80, Rows: 24}))
239 close(done)
240 }()
241
242 timer := <-clock.timers
243 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
244 <-tr.welcome
245 timer.ch <- time.Time{}
246 awaitTestCompletion(t, done, "handshake send did not stop after timeout")
247 requireClosedHandshakeTransport(t, tr)
248 d.mu.Lock()
249 require.Empty(t, d.sessions, "a timed-out newly-created handshake must not leave an empty session")
250 require.Empty(t, d.resume.parked)
251 d.mu.Unlock()
252}
253
254func requireClosedHandshakeTransport(t *testing.T, tr *handshakeBlockingTransport) {
255 t.Helper()
256 select {
257 case <-tr.closed:
258 default:
259 t.Fatal("handshake timeout did not close transport")
260 }
261}
262
263func TestHandshakeTimeoutCancelsRouteRestoreWait(t *testing.T) {
264 clock := &signalClock{timers: make(chan *signalTimer, 1)}
265 d := newTestDaemon(t, nil, clock)
266 restoreDone := make(chan struct{})
267 d.mu.Lock()
268 d.inactive["restoring"] = inactiveSession{name: "restoring", restoreDone: restoreDone}
269 d.mu.Unlock()
270
271 tr := newHandshakeBlockingTransport(false)
272 done := make(chan struct{})
273 go func() {
274 d.handleHelloFrame(tr, mustHello(protocol.IntentAttach, "restoring", domain.Size{Cols: 80, Rows: 24}))
275 close(done)
276 }()
277
278 timer := <-clock.timers
279 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
280 timer.ch <- time.Time{}
281 awaitTestCompletion(t, done, "route restore wait did not stop after handshake timeout")
282 requireClosedHandshakeTransport(t, tr)
283 d.mu.Lock()
284 require.Empty(t, d.sessions)
285 require.Empty(t, d.resume.parked)
286 d.mu.Unlock()
287}
288
289func TestHandshakeTimeoutRemovesRestoredEmptySession(t *testing.T) {
290 pty, release := newBlockingPTY(t)
291 defer release()
292 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
293 d.mu.Lock()
294 d.inactive["restored"] = inactiveSession{name: "restored", cwd: "/tmp", state: protocol.SessionDown}
295 d.mu.Unlock()
296
297 clock := &signalClock{timers: make(chan *signalTimer, 4)}
298 d.clock = clock
299 tr := newHandshakeBlockingTransport(true)
300 done := make(chan struct{})
301 go func() {
302 d.handleHelloFrame(tr, mustHello(protocol.IntentAttach, "restored", domain.Size{Cols: 80, Rows: 24}))
303 close(done)
304 }()
305
306 timer := <-clock.timers
307 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
308 <-tr.welcome
309 timer.ch <- time.Time{}
310 awaitTestCompletion(t, done, "timed-out restored attachment cleanup did not finish")
311 requireClosedHandshakeTransport(t, tr)
312 d.mu.Lock()
313 require.Empty(t, d.sessions)
314 _, retained := d.inactive["restored"]
315 d.mu.Unlock()
316 require.True(t, retained, "failed attach must retain the stopped session authority")
317}
318
319func TestHandshakeTimeoutPreservesUnrelatedAttachment(t *testing.T) {
320 pty, release := newBlockingPTY(t)
321 defer release()
322 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
323 oldTransport := &closeTrackingTransport{}
324 sess, old, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "shared", Size: domain.Size{Cols: 80, Rows: 24}}, oldTransport)
325 require.NoError(t, err)
326
327 clock := &signalClock{timers: make(chan *signalTimer, 1)}
328 d.clock = clock
329 tr := newHandshakeBlockingTransport(true)
330 done := make(chan struct{})
331 go func() {
332 d.handleHelloFrame(tr, mustHello(protocol.IntentAttach, "shared", domain.Size{Cols: 80, Rows: 24}))
333 close(done)
334 }()
335
336 timer := <-clock.timers
337 require.Equal(t, protocol.HandshakeTimeout, timer.duration)
338 <-tr.welcome
339 timer.ch <- time.Time{}
340 awaitTestCompletion(t, done, "timed-out attachment cleanup did not finish")
341 requireClosedHandshakeTransport(t, tr)
342 require.False(t, oldTransport.Closed(), "cleanup closed an unrelated attachment transport")
343 require.Same(t, sess, old.currentAttachmentSession())
344 require.Equal(t, []*attachedClient{old}, sess.snapshotAttachments())
345 d.mu.Lock()
346 require.Len(t, d.sessions, 1)
347 require.Empty(t, d.resume.parked)
348 d.mu.Unlock()
349 _ = d.killSession(sess, protocol.ReasonServerShutdown, false)
350}
351
352func TestSuccessfulRouteOutlivesStoppedHandshakeContext(t *testing.T) {
353 pty, release := newBlockingPTY(t)
354 defer release()
355 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
356 ctx, cancel := context.WithCancel(context.Background())
357 sess, ac, err := d.routeWithContext(ctx, protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "lifetime", Size: domain.Size{Cols: 80, Rows: 24}}, &closeTrackingTransport{})
358 require.NoError(t, err)
359 cancel()
360 select {
361 case <-sess.ctx.Done():
362 t.Fatal("session lifetime was parented to the handshake context")
363 default:
364 }
365 require.NotNil(t, ac)
366 _ = d.killSession(sess, protocol.ReasonServerShutdown, false)
367}