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