1//go:build linux
2
3package app
4
5import (
6 "context"
7 "crypto/rand"
8 "errors"
9 "io"
10 "log/slog"
11 "net"
12 "os"
13 "path/filepath"
14 "strings"
15 "sync"
16 "sync/atomic"
17 "testing"
18 "time"
19
20 "github.com/stretchr/testify/require"
21
22 vt "github.com/bnema/vev-vt"
23 "github.com/bnema/vev/internal/adapters/clock"
24 "github.com/bnema/vev/internal/adapters/ipc"
25 "github.com/bnema/vev/internal/adapters/lifecycle"
26 "github.com/bnema/vev/internal/adapters/pty"
27 "github.com/bnema/vev/internal/adapters/snapshot"
28 "github.com/bnema/vev/internal/domain"
29 "github.com/bnema/vev/internal/persist"
30 "github.com/bnema/vev/internal/ports"
31 "github.com/bnema/vev/internal/usecase/daemon"
32 "github.com/bnema/vev/internal/usecase/recovery"
33)
34
35// These integration tests drive the real daemon over a real unix socket with a
36// real PTY and the real VT/renderer pipeline. They live in internal/app (which
37// is allowed to import adapters) because the layering guard forbids
38// internal/usecase from importing internal/adapters, even in tests.
39
40func discardLog() *slog.Logger {
41 return slog.New(slog.NewTextHandler(io.Discard, nil))
42}
43
44// startDaemon binds a real listener in a temp dir and serves in the background.
45// The returned served channel receives Serve's result exactly once.
46func startDaemon(t *testing.T, opts ...daemon.Option) (dir string, served <-chan error) {
47 t.Helper()
48 return startDaemonInDir(t, filepath.Join(t.TempDir(), "vev"), opts...)
49}
50
51func startDaemonInDir(t *testing.T, dir string, opts ...daemon.Option) (string, <-chan error) {
52 t.Helper()
53 ln, err := ipc.Listen(dir)
54 require.NoError(t, err)
55
56 d := daemon.New(pty.NewFactory(), clock.New(), discardLog(), opts...)
57 ctx, cancel := context.WithCancel(context.Background())
58 ch := make(chan error, 1)
59 go func() { ch <- d.Serve(ctx, ln) }()
60 t.Cleanup(cancel)
61 return dir, ch
62}
63
64// pump wires a background Recv pump to a transport so tests can await frames
65// with a timeout (ports.Transport has no read deadline).
66type pump struct{ ch chan ports.Frame }
67
68func recvPump(tr ports.Transport) *pump {
69 p := &pump{ch: make(chan ports.Frame, 128)}
70 go func() {
71 for {
72 f, err := tr.Recv()
73 if err != nil {
74 close(p.ch)
75 return
76 }
77 p.ch <- f
78 }
79 }()
80 return p
81}
82
83// attach dials, handshakes, and returns the transport plus its frame pump.
84func attach(t *testing.T, dir string, intent uint8, name string, sz domain.Size) (ports.Transport, *pump) {
85 t.Helper()
86 return attachWithEnvironment(t, dir, intent, name, sz, nil)
87}
88
89func attachWithEnvironment(t *testing.T, dir string, intent uint8, name string, sz domain.Size, env []string) (ports.Transport, *pump) {
90 t.Helper()
91 tr, err := ipc.DialContext(context.Background(), dir)
92 require.NoError(t, err)
93 hello := ports.Hello{Version: ports.ProtocolVersion, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color", TrueColor: true, Env: env}
94 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}))
95 p := recvPump(tr)
96 select {
97 case f, ok := <-p.ch:
98 require.True(t, ok, "connection closed before welcome")
99 require.Equal(t, ports.MsgWelcome, f.Type)
100 case <-time.After(3 * time.Second):
101 t.Fatal("timed out waiting for welcome")
102 }
103 return tr, p
104}
105
106func listRemoteSessions(t *testing.T, dir string) ports.Sessions {
107 t.Helper()
108 tr, err := ipc.DialContext(context.Background(), dir)
109 require.NoError(t, err)
110 defer func() { _ = tr.Close() }()
111 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})}))
112 f, err := tr.Recv()
113 require.NoError(t, err)
114 require.Equal(t, ports.MsgSessions, f.Type)
115 sessions, err := ports.UnmarshalSessions(f.Payload)
116 require.NoError(t, err)
117 return sessions
118}
119
120func killAll(dir string) error {
121 tr, err := ipc.DialContext(context.Background(), dir)
122 if err != nil {
123 return err
124 }
125 defer func() { _ = tr.Close() }()
126 if err := tr.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
127 return err
128 }
129 _, err = tr.Recv()
130 if errors.Is(err, io.EOF) {
131 return nil
132 }
133 return err
134}
135
136// awaitText decodes MsgOutput frames into a fresh VT screen and returns once
137// the reconstructed grid contains want.
138func awaitText(t *testing.T, p *pump, sz domain.Size, want string) {
139 t.Helper()
140 _ = awaitScreenText(t, p, sz, want)
141}
142
143// awaitScreenText is like awaitText, but returns the reconstructed screen text
144// at the point the wanted text appears so callers can make additional checks.
145func awaitScreenText(t *testing.T, p *pump, sz domain.Size, want string) string {
146 t.Helper()
147 screen := vt.NewScreen(sz.Cols, sz.Rows)
148 timeout := time.After(5 * time.Second)
149 for {
150 select {
151 case f, ok := <-p.ch:
152 if !ok {
153 t.Fatalf("connection closed before %q appeared; screen=%q", want, screenText(screen))
154 }
155 if f.Type == ports.MsgOutput {
156 o, err := ports.UnmarshalOutput(f.Payload)
157 require.NoError(t, err)
158 screen.Write(o.Data)
159 text := screenText(screen)
160 if strings.Contains(text, want) {
161 return text
162 }
163 }
164 case <-timeout:
165 t.Fatalf("timed out waiting for %q; screen=%q", want, screenText(screen))
166 }
167 }
168}
169
170func assertNoTextAfterInput(t *testing.T, p *pump, sz domain.Size, absent string) {
171 t.Helper()
172 screen := vt.NewScreen(sz.Cols, sz.Rows)
173 timeout := time.After(300 * time.Millisecond)
174 for {
175 select {
176 case f, ok := <-p.ch:
177 if !ok {
178 return
179 }
180 if f.Type == ports.MsgOutput {
181 o, err := ports.UnmarshalOutput(f.Payload)
182 require.NoError(t, err)
183 screen.Write(o.Data)
184 require.NotContains(t, screenText(screen), absent)
185 }
186 case <-timeout:
187 return
188 }
189 }
190}
191
192func screenText(s *vt.Screen) string {
193 var b strings.Builder
194 for y := range s.Frame.Height {
195 for x := range s.Frame.Width {
196 r := s.Frame.At(x, y).Rune
197 if r == 0 {
198 r = ' '
199 }
200 b.WriteRune(r)
201 }
202 }
203 return b.String()
204}
205
206func shellFixture(t *testing.T, label string) string {
207 t.Helper()
208 path := filepath.Join(t.TempDir(), "shell")
209 require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND="+label+"\\n'\nexec /bin/sh\n"), 0o700))
210 return path
211}
212
213func assertChildEnvironment(t *testing.T, tr ports.Transport, p *pump, sz domain.Size, wantTestEnv, wantShell, wantRuntimeDir, wantWayland string) {
214 t.Helper()
215 command := "printf '\\033[2J\\033[H'; printf 'VEV_TEST_ENV=%s SHELL=%s XDG_RUNTIME_DIR=%s WAYLAND_DISPLAY=%s TERM=%s COLORTERM=%s TERM_PROGRAM=%s VEV_PREFIX=%.24s\\n' \"$VEV_TEST_ENV\" \"${SHELL##*/}\" \"$XDG_RUNTIME_DIR\" \"$WAYLAND_DISPLAY\" \"$TERM\" \"$COLORTERM\" \"$TERM_PROGRAM\" \"$VEV\"\n"
216 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte(command)})}))
217 text := awaitScreenText(t, p, sz, "TERM_PROGRAM=vev")
218 for _, want := range []string{
219 "VEV_TEST_ENV=" + wantTestEnv,
220 "SHELL=" + filepath.Base(wantShell),
221 "XDG_RUNTIME_DIR=" + wantRuntimeDir,
222 "WAYLAND_DISPLAY=" + wantWayland,
223 "TERM=xterm-direct",
224 "COLORTERM=truecolor",
225 "TERM_PROGRAM=vev",
226 "VEV_PREFIX=session=environment,tab=",
227 } {
228 require.Contains(t, text, want)
229 }
230}
231
232func TestIntegration_AttachEnvironmentRefreshesFuturePTYChildren(t *testing.T) {
233 sz := domain.Size{Cols: 80, Rows: 24}
234 dir, _ := startDaemon(t)
235 firstShell := shellFixture(t, "first")
236 secondShell := shellFixture(t, "second")
237 firstEnv := []string{
238 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
239 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
240 }
241
242 tr1, p1 := attachWithEnvironment(t, dir, ports.IntentNew, "environment", sz, firstEnv)
243 defer func() { _ = tr1.Close() }()
244 awaitText(t, p1, sz, "SHELL_COMMAND=first")
245 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
246
247 secondEnv := []string{
248 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
249 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
250 }
251 tr2, p2 := attachWithEnvironment(t, dir, ports.IntentAttach, "environment", sz, secondEnv)
252 defer func() { _ = tr2.Close() }()
253
254 // The first shell was already running, so it retains its original environment.
255 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
256
257 require.NoError(t, tr2.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
258 awaitText(t, p2, sz, "Commands")
259 require.NoError(t, tr2.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("CNT\r")})}))
260 awaitText(t, p2, sz, "SHELL_COMMAND=second")
261 assertChildEnvironment(t, tr2, p2, sz, "second", secondShell, "/run/second", "wayland-second")
262}
263
264func TestIntegration_TwoAttachmentsReceiveSharedMutationPTYOutput(t *testing.T) {
265 sz := domain.Size{Cols: 80, Rows: 24}
266 dir, _ := startDaemon(t)
267 firstShell := shellFixture(t, "first")
268 const sharedOutput = "SECOND_SHARED_PTY_OUTPUT"
269 secondShell := filepath.Join(t.TempDir(), "second-shell")
270 require.NoError(t, os.WriteFile(secondShell, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND=second\\n'\nIFS= read -r _\nprintf '"+sharedOutput+"\\n'\nexec /bin/sh\n"), 0o700))
271 firstEnv := []string{
272 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
273 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
274 }
275 tr1, p1 := attachWithEnvironment(t, dir, ports.IntentNew, "environment", sz, firstEnv)
276 defer func() { _ = tr1.Close() }()
277 awaitText(t, p1, sz, "SHELL_COMMAND=first")
278 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
279
280 secondEnv := []string{
281 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
282 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
283 }
284 tr2, p2 := attachWithEnvironment(t, dir, ports.IntentAttach, "environment", sz, secondEnv)
285 defer func() { _ = tr2.Close() }()
286 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
287
288 // The second attachment mutates shared tab state. Its newly opened PTY must
289 // also be rendered to that attachment, not only to the coordinator primary.
290 require.NoError(t, tr2.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
291 awaitText(t, p2, sz, "Commands")
292 require.NoError(t, tr2.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("CNT\r")})}))
293 awaitText(t, p2, sz, "SHELL_COMMAND=second")
294
295 // Each attachment owns its selected tab. Move the first attachment to the
296 // new PTY before releasing its gate, so the fresh PTY output must reach both
297 // views rather than only being recovered by a later first paint.
298 require.NoError(t, tr1.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b2")})}))
299 awaitText(t, p1, sz, "SHELL_COMMAND=second")
300 require.NoError(t, tr1.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("go\n")})}))
301 awaitText(t, p2, sz, sharedOutput)
302 awaitText(t, p1, sz, sharedOutput)
303}
304
305func TestIntegration_AttachFirstOutput(t *testing.T) {
306 sz := domain.Size{Cols: 80, Rows: 24}
307 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf HELLO; sleep 30"}))
308
309 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
310 defer func() { _ = tr.Close() }()
311
312 awaitText(t, p, sz, "HELLO")
313}
314
315func TestIntegration_InputRoundtrip(t *testing.T) {
316 sz := domain.Size{Cols: 80, Rows: 24}
317 dir, _ := startDaemon(t, daemon.WithShell("/bin/cat", nil))
318
319 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
320 defer func() { _ = tr.Close() }()
321
322 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("PINGPONG\n")})}))
323 awaitText(t, p, sz, "PINGPONG")
324}
325
326func TestIntegration_CommandPaletteCreatesTab(t *testing.T) {
327 sz := domain.Size{Cols: 80, Rows: 24}
328 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
329
330 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
331 defer func() { _ = tr.Close() }()
332 awaitText(t, p, sz, "READY")
333
334 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
335 awaitText(t, p, sz, "Commands")
336
337 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("CNT\r")})}))
338 // Tab labels are enriched with the focused pane's title; with no process
339 // inspector wired in this test, that title falls back to the shell's
340 // basename ("sh").
341 text := awaitScreenText(t, p, sz, " 1 (sh) 2 (sh) ")
342 require.Contains(t, text, " 1 (sh) 2 (sh) ")
343}
344
345func TestIntegration_CommandPaletteRenamesEphemeralSession(t *testing.T) {
346 sz := domain.Size{Cols: 80, Rows: 24}
347 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
348
349 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
350 defer func() { _ = tr.Close() }()
351 text := awaitScreenText(t, p, sz, "READY")
352 require.Contains(t, text, " 0* ")
353
354 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
355 awaitText(t, p, sz, "Commands")
356
357 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("RNS\r")})}))
358 awaitText(t, p, sz, "Rename session")
359
360 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x7fwork\r")})}))
361 text = awaitScreenText(t, p, sz, " work ")
362 require.NotContains(t, text, "work*")
363
364 listTr, err := ipc.DialContext(context.Background(), dir)
365 require.NoError(t, err)
366 defer func() { _ = listTr.Close() }()
367 require.NoError(t, listTr.Send(ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})}))
368 f, err := listTr.Recv()
369 require.NoError(t, err)
370 require.Equal(t, ports.MsgSessions, f.Type)
371 sessions, err := ports.UnmarshalSessions(f.Payload)
372 require.NoError(t, err)
373 require.NotEmpty(t, sessions.Sessions)
374 require.Equal(t, "work", sessions.Sessions[0].Name)
375 require.False(t, sessions.Sessions[0].Ephemeral)
376}
377
378func TestIntegration_AltCWithoutPaletteDoesNotCreateTab(t *testing.T) {
379 sz := domain.Size{Cols: 80, Rows: 24}
380 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
381
382 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
383 defer func() { _ = tr.Close() }()
384 text := awaitScreenText(t, p, sz, "READY")
385 require.NotContains(t, text, " 1 2 ")
386
387 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1bc")})}))
388 assertNoTextAfterInput(t, p, sz, " 1 2 ")
389}
390
391func TestIntegration_EphemeralSurvivesDetachAndReattaches(t *testing.T) {
392 sz := domain.Size{Cols: 80, Rows: 24}
393 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
394
395 tr1, p1 := attach(t, dir, ports.IntentEphemeral, "", sz)
396 awaitText(t, p1, sz, "MARKER")
397 require.NoError(t, tr1.Close())
398
399 select {
400 case err := <-served:
401 t.Fatalf("daemon shut down while an ephemeral session was alive: %v", err)
402 case <-time.After(300 * time.Millisecond):
403 }
404
405 tr2, p2 := attach(t, dir, ports.IntentAttach, "0", sz)
406 defer func() { _ = tr2.Close() }()
407 awaitText(t, p2, sz, "MARKER")
408
409 require.NoError(t, killAll(dir))
410 select {
411 case err := <-served:
412 require.NoError(t, err)
413 case <-time.After(5 * time.Second):
414 t.Fatal("daemon did not shut down after kill all")
415 }
416}
417
418func TestIntegration_EphemeralNotListedAfterDaemonRestart(t *testing.T) {
419 sz := domain.Size{Cols: 80, Rows: 24}
420 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
421 t.Setenv("XDG_STATE_HOME", stateRoot)
422 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
423 dir := ipc.SocketDir()
424
425 _, served := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "printf TEMP; sleep 30"}))
426 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
427 awaitText(t, p, sz, "TEMP")
428 require.NoError(t, tr.Close())
429
430 require.NoError(t, runKill(context.Background(), "", false, true))
431 select {
432 case err := <-served:
433 require.NoError(t, err)
434 case <-time.After(5 * time.Second):
435 t.Fatal("daemon did not stop after kill --daemon")
436 }
437
438 _, served2 := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
439 sessions := listRemoteSessions(t, dir)
440 require.Empty(t, sessions.Sessions)
441
442 require.NoError(t, runKill(context.Background(), "", false, true))
443 select {
444 case err := <-served2:
445 require.NoError(t, err)
446 case <-time.After(5 * time.Second):
447 t.Fatal("restarted daemon did not stop after kill --daemon")
448 }
449}
450
451func TestIntegration_NamedSurvivesReattach(t *testing.T) {
452 sz := domain.Size{Cols: 80, Rows: 24}
453 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
454
455 tr1, p1 := attach(t, dir, ports.IntentNew, "work", sz)
456 awaitText(t, p1, sz, "MARKER")
457
458 // Detach: a named session must survive.
459 require.NoError(t, tr1.Close())
460
461 // Daemon must still be running (session alive).
462 select {
463 case err := <-served:
464 t.Fatalf("daemon shut down while a named session was alive: %v", err)
465 case <-time.After(300 * time.Millisecond):
466 }
467
468 // Re-attach: the first paint must reproduce the retained screen state.
469 tr2, p2 := attach(t, dir, ports.IntentAttach, "work", sz)
470 defer func() { _ = tr2.Close() }()
471 awaitText(t, p2, sz, "MARKER")
472}
473
474func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
475 dir := filepath.Join(t.TempDir(), "runtime")
476 owner, err := lifecycle.TryAcquire(dir)
477 require.NoError(t, err)
478
479 ready := make(chan struct{})
480 transport := &integrationTransport{}
481 dial := func(context.Context, string) (ports.Transport, error) {
482 select {
483 case <-ready:
484 return transport, nil
485 default:
486 return nil, os.ErrNotExist
487 }
488 }
489 var spawns atomic.Int32
490 cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
491
492 const clients = 4
493 errs := make(chan error, clients)
494 var wg sync.WaitGroup
495 for range clients {
496 wg.Go(func() {
497 _, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
498 spawns.Add(1)
499 return nil
500 }, cfg)
501 errs <- err
502 })
503 }
504 require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
505 close(ready)
506 require.NoError(t, owner.Release())
507 wg.Wait()
508 close(errs)
509 for err := range errs {
510 require.NoError(t, err)
511 }
512 require.Zero(t, spawns.Load())
513}
514
515func TestListWaitsForLifecycleOwner(t *testing.T) {
516 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
517 t.Setenv("XDG_STATE_HOME", stateRoot)
518 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
519 p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
520 require.NoError(t, p.Close())
521 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
522 require.NoError(t, err)
523
524 done := make(chan error, 1)
525 go func() { done <- runList(context.Background(), command{kind: kindList}) }()
526 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
527 require.NoError(t, owner.Release())
528 require.NoError(t, <-done)
529}
530
531func TestOfflineKillWaitsForLifecycleOwner(t *testing.T) {
532 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
533 t.Setenv("XDG_STATE_HOME", stateRoot)
534 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
535 p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
536 now := time.Now().UnixNano()
537 require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, CreatedAt: now, UpdatedAt: now}))
538 require.NoError(t, p.Close())
539 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
540 require.NoError(t, err)
541
542 done := make(chan error, 1)
543 go func() { done <- runKill(context.Background(), "named", false, false) }()
544 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
545 require.NoError(t, owner.Release())
546 require.NoError(t, <-done)
547}
548
549func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
550 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
551 t.Setenv("XDG_STATE_HOME", stateRoot)
552 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
553 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
554 require.NoError(t, err)
555 _, served := startDaemonInDir(t, ipc.SocketDir())
556
557 done := make(chan error, 1)
558 go func() { done <- requestDaemonStop(context.Background()) }()
559 select {
560 case err := <-served:
561 require.NoError(t, err)
562 case <-time.After(time.Second):
563 t.Fatal("daemon did not stop")
564 }
565 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
566 require.NoError(t, owner.Release())
567 require.NoError(t, <-done)
568}
569
570func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
571 runtimeDir := filepath.Join(t.TempDir(), "runtime")
572 stateDir := filepath.Join(t.TempDir(), "state")
573 catalogue := &lifecycleObservedCatalogue{
574 Catalogue: newTestPersister(t, stateDir),
575 closed: make(chan struct{}),
576 }
577 require.NoError(t, catalogue.Create(domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}))
578 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
579 shutdownClock := newLifecycleShutdownClock(t)
580 maintenanceRepository := &lifecycleBlockingMaintenanceRepository{
581 SnapshotRepository: repository,
582 entered: make(chan struct{}),
583 release: make(chan struct{}),
584 returned: make(chan struct{}),
585 }
586 ctx, cancel := context.WithCancel(context.Background())
587 t.Cleanup(cancel)
588 var releaseOnce sync.Once
589 releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
590 t.Cleanup(releaseMaintenance)
591
592 callbackReturned := make(chan struct{})
593 wrapperReturned := make(chan error, 1)
594 go func() {
595 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
596 defer close(callbackReturned)
597 d := daemon.New(
598 pty.NewFactory(),
599 shutdownClock,
600 discardLog(),
601 daemon.WithSnapshotRepository(repository),
602 daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
603 daemon.WithCatalogue(catalogue, nil),
604 )
605 if err := d.CollectStartupGarbage(ctx); err != nil {
606 return err
607 }
608 listener, err := ipc.Listen(runtimeDir)
609 if err != nil {
610 return err
611 }
612 return d.Serve(ctx, listener)
613 })
614 }()
615
616 awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")
617 cancel()
618 select {
619 case <-callbackReturned:
620 t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
621 case <-time.After(100 * time.Millisecond):
622 }
623 assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
624 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
625 _, err := lifecycle.TryAcquire(runtimeDir)
626 require.ErrorIs(t, err, lifecycle.ErrBusy)
627
628 releaseMaintenance()
629 awaitLifecycleStage(t, maintenanceRepository.returned, "maintenance repository return")
630 awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
631 awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
632 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
633 owner, err := lifecycle.TryAcquire(runtimeDir)
634 require.NoError(t, err)
635 require.NoError(t, owner.Release())
636}
637
638type lifecycleBlockingMaintenanceRepository struct {
639 ports.SnapshotRepository
640 entered chan struct{}
641 release chan struct{}
642 returned chan struct{}
643 enterOnce sync.Once
644 returnOnce sync.Once
645}
646
647func (r *lifecycleBlockingMaintenanceRepository) CollectGarbage(context.Context, map[domain.IncarnationID]domain.CheckpointRef) error {
648 r.enterOnce.Do(func() { close(r.entered) })
649 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
650 r.returnOnce.Do(func() { close(r.returned) })
651 return nil
652}
653
654type lifecycleObservedCatalogue struct {
655 ports.Catalogue
656 closed chan struct{}
657 closeOnce sync.Once
658}
659
660func (c *lifecycleObservedCatalogue) Close() error {
661 err := c.Catalogue.Close()
662 c.closeOnce.Do(func() { close(c.closed) })
663 return err
664}
665
666func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
667 result := runBlockedSnapshotWriterShutdown(t)
668 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
669 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
670 _, err := lifecycle.TryAcquire(result.runtimeDir)
671 require.ErrorIs(t, err, lifecycle.ErrBusy)
672
673 result.releaseWriter()
674 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
675 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
676 owner, err := lifecycle.TryAcquire(result.runtimeDir)
677 require.NoError(t, err)
678 require.NoError(t, owner.Release())
679}
680
681func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
682 result := runBlockedSnapshotWriterShutdown(t)
683 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
684
685 result.releaseWriter()
686 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
687 awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
688 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
689}
690
691func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
692 result := runBlockedSnapshotWriterShutdown(t)
693 assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
694 _, err := lifecycle.TryAcquire(result.runtimeDir)
695 require.ErrorIs(t, err, lifecycle.ErrBusy)
696
697 result.releaseWriter()
698 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
699}
700
701// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
702// records. Lifecycle ownership must therefore outlive it: releasing the flock
703// while a restoration repository call is still running would let a second
704// daemon mutate the same durable state.
705func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
706 stateDir := filepath.Join(t.TempDir(), "state")
707 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
708 checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
709
710 runtimeDir := filepath.Join(t.TempDir(), "runtime")
711 blocking := &lifecycleBlockingRestore{
712 SnapshotRepository: repository,
713 entered: make(chan struct{}),
714 release: make(chan struct{}),
715 returned: make(chan struct{}),
716 }
717 var releaseOnce sync.Once
718 releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
719 t.Cleanup(releaseRestore)
720
721 shutdownClock := newLifecycleShutdownClock(t)
722 ctx, cancel := context.WithCancel(context.Background())
723 t.Cleanup(cancel)
724 listenerReady := make(chan struct{})
725 callbackReturned := make(chan struct{})
726 wrapperReturned := make(chan error, 1)
727 go func() {
728 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
729 defer close(callbackReturned)
730 opened, err := persist.OpenOrCreate(stateDir)
731 if err != nil {
732 return err
733 }
734 listener, err := ipc.Listen(runtimeDir)
735 if err != nil {
736 return errors.Join(err, opened.Catalogue.Close())
737 }
738 close(listenerReady)
739 d := daemon.New(
740 pty.NewFactory(),
741 shutdownClock,
742 discardLog(),
743 daemon.WithShell("/bin/cat", nil),
744 daemon.WithCatalogue(opened.Catalogue, opened.Records),
745 daemon.WithSnapshotRepository(blocking),
746 )
747 return d.Serve(ctx, listener)
748 })
749 }()
750
751 awaitLifecycleStage(t, listenerReady, "daemon listener")
752 awaitLifecycleStage(t, blocking.entered, "restoration repository call")
753 require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
754
755 cancel()
756 shutdownClock.nextTimer(t).fire()
757 select {
758 case <-callbackReturned:
759 t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
760 case <-time.After(100 * time.Millisecond):
761 }
762 assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
763 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
764 _, err := lifecycle.TryAcquire(runtimeDir)
765 require.ErrorIs(t, err, lifecycle.ErrBusy)
766
767 releaseRestore()
768 awaitLifecycleStage(t, blocking.returned, "restoration repository return")
769 awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
770 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
771 owner, err := lifecycle.TryAcquire(runtimeDir)
772 require.NoError(t, err)
773 require.NoError(t, owner.Release())
774}
775
776// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
777// repository until one named session owns a committed checkpoint, then shuts it
778// down. The returned name is restorable by any later daemon on the same state.
779func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
780 t.Helper()
781 const name = "restorable"
782 runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
783 ctx, cancel := context.WithCancel(context.Background())
784 defer cancel()
785
786 opened, err := persist.OpenOrCreate(stateDir)
787 require.NoError(t, err)
788 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
789 listener, err := ipc.Listen(runtimeDir)
790 require.NoError(t, err)
791 d := daemon.New(
792 pty.NewFactory(),
793 clock.New(),
794 discardLog(),
795 daemon.WithShell("/bin/cat", nil),
796 daemon.WithCatalogue(opened.Catalogue, opened.Records),
797 daemon.WithSnapshotRepository(repository),
798 daemon.WithRecoveryCoordinator(coordinator),
799 )
800 served := make(chan error, 1)
801 go func() { served <- d.Serve(ctx, listener) }()
802
803 tr, _ := attach(t, runtimeDir, ports.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
804 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("checkpoint me\n")})}))
805 require.Eventually(t, func() bool {
806 record, ok, _ := opened.Catalogue.Record(name)
807 return ok && record.Committed != nil && record.DegradedReason == ""
808 }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
809 require.NoError(t, tr.Close())
810
811 cancel()
812 require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
813 return name
814}
815
816type lifecycleBlockingRestore struct {
817 ports.SnapshotRepository
818 entered chan struct{}
819 release chan struct{}
820 returned chan struct{}
821 enterOnce sync.Once
822 returnOnce sync.Once
823 mu sync.Mutex
824 repaired string
825}
826
827func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
828 r.mu.Lock()
829 r.repaired = name
830 r.mu.Unlock()
831 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
832}
833
834func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
835 r.enterOnce.Do(func() { close(r.entered) })
836 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
837 r.returnOnce.Do(func() { close(r.returned) })
838 return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
839}
840
841func (r *lifecycleBlockingRestore) repairedName() string {
842 r.mu.Lock()
843 defer r.mu.Unlock()
844 return r.repaired
845}
846
847type blockedSnapshotWriterShutdown struct {
848 runtimeDir string
849 callbackReturned <-chan struct{}
850 wrapperReturned <-chan error
851 publishReturned <-chan struct{}
852 releaseWriter func()
853}
854
855func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
856 t.Helper()
857 runtimeDir := filepath.Join(t.TempDir(), "runtime")
858 stateDir := filepath.Join(t.TempDir(), "state")
859 shutdownClock := newLifecycleShutdownClock(t)
860 repository := newLifecycleBlockingRepository()
861 ctx, cancel := context.WithCancel(context.Background())
862 t.Cleanup(cancel)
863
864 listenerReady := make(chan struct{})
865 listenerClosed := make(chan struct{})
866 callbackReturned := make(chan struct{})
867 wrapperReturned := make(chan error, 1)
868 go func() {
869 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
870 defer close(callbackReturned)
871 listener, err := ipc.Listen(runtimeDir)
872 if err != nil {
873 return err
874 }
875 observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
876 close(listenerReady)
877 d := daemon.New(
878 pty.NewFactory(),
879 shutdownClock,
880 discardLog(),
881 daemon.WithShell("/bin/cat", nil),
882 daemon.WithSnapshotRepository(repository),
883 )
884 return d.Serve(ctx, observed)
885 })
886 }()
887
888 awaitLifecycleStage(t, listenerReady, "daemon listener")
889 tr, _ := attach(t, runtimeDir, ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
890 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("dirty state\n")})}))
891 awaitLifecycleStage(t, repository.entered, "snapshot publication")
892 require.NoError(t, tr.Close())
893 require.Eventually(t, func() bool {
894 sessions := listRemoteSessions(t, runtimeDir)
895 return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
896 }, time.Second, 10*time.Millisecond)
897 cancel()
898 awaitLifecycleStage(t, listenerClosed, "listener close")
899
900 shutdownClock.nextTimer(t).fire()
901 select {
902 case <-callbackReturned:
903 t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
904 case <-time.After(100 * time.Millisecond):
905 }
906 assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
907 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
908
909 var releaseOnce sync.Once
910 releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
911 t.Cleanup(releaseWriter)
912 return blockedSnapshotWriterShutdown{
913 runtimeDir: runtimeDir,
914 callbackReturned: callbackReturned,
915 wrapperReturned: wrapperReturned,
916 publishReturned: repository.returned,
917 releaseWriter: releaseWriter,
918 }
919}
920
921type lifecycleBlockingRepository struct {
922 ports.SnapshotRepository
923 entered chan struct{}
924 release chan struct{}
925 returned chan struct{}
926 enterOnce sync.Once
927 returnOnce sync.Once
928}
929
930func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
931 return &lifecycleBlockingRepository{
932 entered: make(chan struct{}),
933 release: make(chan struct{}),
934 returned: make(chan struct{}),
935 }
936}
937
938func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
939 r.enterOnce.Do(func() { close(r.entered) })
940 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
941 r.returnOnce.Do(func() { close(r.returned) })
942 return nil
943}
944
945type lifecycleShutdownClock struct {
946 t *testing.T
947 base ports.Clock
948 timers chan *lifecycleManualTimer
949}
950
951func newLifecycleShutdownClock(t *testing.T) *lifecycleShutdownClock {
952 t.Helper()
953 return &lifecycleShutdownClock{t: t, base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
954}
955
956func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
957func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
958 if delay != daemon.SnapshotShutdownTimeout() {
959 return c.base.NewTimer(delay)
960 }
961 timer := &lifecycleManualTimer{t: c.t, ch: make(chan time.Time, 1)}
962 select {
963 case c.timers <- timer:
964 default:
965 c.t.Errorf("unexpected additional snapshot shutdown timer with delay %s", delay)
966 timer.fire()
967 }
968 return timer
969}
970
971func (c *lifecycleShutdownClock) nextTimer(t *testing.T) *lifecycleManualTimer {
972 t.Helper()
973 select {
974 case timer := <-c.timers:
975 return timer
976 case <-time.After(3 * time.Second):
977 t.Fatal("timed out waiting for shutdown deadline timer")
978 return nil
979 }
980}
981
982type lifecycleManualTimer struct {
983 t *testing.T
984 ch chan time.Time
985 stopped atomic.Bool
986}
987
988func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
989func (t *lifecycleManualTimer) Reset(delay time.Duration) bool {
990 if delay != daemon.SnapshotShutdownTimeout() {
991 t.t.Errorf("snapshot shutdown timer reset with delay %s, want %s", delay, daemon.SnapshotShutdownTimeout())
992 }
993 t.stopped.Store(false)
994 return false
995}
996func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
997func (t *lifecycleManualTimer) fire() {
998 if t.stopped.Load() {
999 return
1000 }
1001 select {
1002 case t.ch <- time.Now():
1003 default:
1004 }
1005}
1006
1007func TestLifecycleManualTimerRepeatedFireIsNonBlocking(t *testing.T) {
1008 timer := &lifecycleManualTimer{t: t, ch: make(chan time.Time, 1)}
1009 timer.fire()
1010
1011 returned := make(chan struct{})
1012 go func() {
1013 timer.fire()
1014 close(returned)
1015 }()
1016
1017 select {
1018 case <-returned:
1019 case <-time.After(time.Second):
1020 t.Fatal("repeated manual timer fire blocked on a pending tick")
1021 }
1022}
1023
1024type lifecycleObservedListener struct {
1025 ports.Listener
1026 closed chan struct{}
1027 once sync.Once
1028}
1029
1030func (l *lifecycleObservedListener) Close() error {
1031 err := l.Listener.Close()
1032 l.once.Do(func() { close(l.closed) })
1033 return err
1034}
1035
1036func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
1037 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1038 stateDir := filepath.Join(t.TempDir(), "state")
1039 oldOwner, err := lifecycle.TryAcquire(runtimeDir)
1040 require.NoError(t, err)
1041 oldOpened, err := persist.OpenOrCreate(stateDir)
1042 require.NoError(t, err)
1043 oldListener, err := ipc.Listen(runtimeDir)
1044 require.NoError(t, err)
1045
1046 socketCloseEntered := make(chan struct{})
1047 allowSocketClose := make(chan struct{})
1048 socketClosed := make(chan struct{})
1049 catalogueCloseEntered := make(chan struct{})
1050 allowCatalogueClose := make(chan struct{})
1051 catalogueClosed := make(chan struct{})
1052 ownerReleaseEntered := make(chan struct{})
1053 allowOwnerRelease := make(chan struct{})
1054 ownerReleased := make(chan struct{})
1055 teardownEvents := make(chan string, 3)
1056
1057 controlledListener := &lifecycleControlledListener{
1058 Listener: oldListener,
1059 entered: socketCloseEntered,
1060 proceed: allowSocketClose,
1061 closed: socketClosed,
1062 events: teardownEvents,
1063 }
1064 controlledCatalogue := &lifecycleControlledCatalogue{
1065 Catalogue: oldOpened.Catalogue,
1066 entered: catalogueCloseEntered,
1067 proceed: allowCatalogueClose,
1068 closed: catalogueClosed,
1069 events: teardownEvents,
1070 }
1071 oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
1072 oldCtx, stopOld := context.WithCancel(context.Background())
1073 oldDone := make(chan error, 1)
1074 go func() {
1075 serveErr := oldDaemon.Serve(oldCtx, controlledListener)
1076 teardownEvents <- "owner-release"
1077 close(ownerReleaseEntered)
1078 <-allowOwnerRelease
1079 releaseErr := oldOwner.Release()
1080 close(ownerReleased)
1081 oldDone <- errors.Join(serveErr, releaseErr)
1082 }()
1083
1084 newAcquireAttempted := make(chan struct{})
1085 newDurableOpen := make(chan struct{})
1086 newListen := make(chan struct{})
1087 startupEvents := make(chan string, 3)
1088 newDone := make(chan error, 1)
1089 go func() {
1090 newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
1091 close(newDurableOpen)
1092 opened, err := persist.OpenOrCreate(stateDir)
1093 if err != nil {
1094 return err
1095 }
1096 startupEvents <- "durable-open"
1097
1098 close(newListen)
1099 listener, err := ipc.Listen(runtimeDir)
1100 if err != nil {
1101 return errors.Join(err, opened.Catalogue.Close())
1102 }
1103 startupEvents <- "listen"
1104 return errors.Join(listener.Close(), opened.Catalogue.Close())
1105 }, lifecycleStartupDeps{
1106 ensurePrivate: func(string) error { return nil },
1107 acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
1108 close(newAcquireAttempted)
1109 owner, err := lifecycle.Acquire(ctx, dir, retry)
1110 if err != nil {
1111 return nil, err
1112 }
1113 <-ownerReleased
1114 startupEvents <- "owner-acquired"
1115 return owner, nil
1116 },
1117 })
1118 }()
1119
1120 awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
1121 assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
1122 assertLifecycleStagePending(t, newListen, "new listen before old teardown")
1123
1124 stopOld()
1125 awaitLifecycleStage(t, socketCloseEntered, "old socket close")
1126 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
1127 close(allowSocketClose)
1128 awaitLifecycleStage(t, socketClosed, "old socket closed")
1129
1130 awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
1131 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
1132 close(allowCatalogueClose)
1133 awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
1134
1135 awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
1136 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
1137 close(allowOwnerRelease)
1138 awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
1139
1140 require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
1141 require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
1142 require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
1143 <-teardownEvents,
1144 <-teardownEvents,
1145 <-teardownEvents,
1146 })
1147 require.Equal(t, []string{"owner-acquired", "durable-open", "listen"}, []string{
1148 <-startupEvents,
1149 <-startupEvents,
1150 <-startupEvents,
1151 })
1152}
1153
1154func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
1155 t.Helper()
1156 select {
1157 case <-stage:
1158 case <-time.After(3 * time.Second):
1159 t.Fatalf("timed out waiting for %s", name)
1160 }
1161}
1162
1163func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
1164 t.Helper()
1165 select {
1166 case err := <-result:
1167 return err
1168 case <-time.After(3 * time.Second):
1169 t.Fatalf("timed out waiting for %s", name)
1170 return nil
1171 }
1172}
1173
1174func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
1175 t.Helper()
1176 select {
1177 case <-stage:
1178 t.Fatalf("unexpected %s", name)
1179 default:
1180 }
1181}
1182
1183func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
1184 t.Helper()
1185 select {
1186 case err := <-result:
1187 t.Fatalf("unexpected %s: %v", name, err)
1188 default:
1189 }
1190}
1191
1192func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
1193 t.Helper()
1194 assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
1195 assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
1196}
1197
1198type lifecycleControlledListener struct {
1199 ports.Listener
1200 entered chan struct{}
1201 proceed chan struct{}
1202 closed chan struct{}
1203 events chan<- string
1204 once sync.Once
1205 err error
1206}
1207
1208func (l *lifecycleControlledListener) Close() error {
1209 l.once.Do(func() {
1210 l.events <- "socket-close"
1211 close(l.entered)
1212 <-l.proceed
1213 l.err = l.Listener.Close()
1214 close(l.closed)
1215 })
1216 return l.err
1217}
1218
1219type lifecycleControlledCatalogue struct {
1220 ports.Catalogue
1221 entered chan struct{}
1222 proceed chan struct{}
1223 closed chan struct{}
1224 events chan<- string
1225 once sync.Once
1226 err error
1227}
1228
1229func (c *lifecycleControlledCatalogue) Close() error {
1230 c.once.Do(func() {
1231 c.events <- "catalogue-close"
1232 close(c.entered)
1233 <-c.proceed
1234 c.err = c.Catalogue.Close()
1235 close(c.closed)
1236 })
1237 return c.err
1238}
1239
1240type integrationTransport struct{}
1241
1242func (*integrationTransport) Send(ports.Frame) error { return nil }
1243func (*integrationTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
1244func (*integrationTransport) Close() error { return nil }
1245func (*integrationTransport) LocalAddr() net.Addr { return nil }
1246func (*integrationTransport) RemoteAddr() net.Addr { return nil }
1247
1248func TestIntegration_KillAllShutsDownDaemon(t *testing.T) {
1249 sz := domain.Size{Cols: 80, Rows: 24}
1250 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
1251
1252 tr1, _ := attach(t, dir, ports.IntentNew, "one", sz)
1253 require.NoError(t, tr1.Close())
1254 tr2, _ := attach(t, dir, ports.IntentNew, "two", sz)
1255 require.NoError(t, tr2.Close())
1256
1257 killTr, err := ipc.DialContext(context.Background(), dir)
1258 require.NoError(t, err)
1259 defer func() { _ = killTr.Close() }()
1260 require.NoError(t, killTr.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}))
1261
1262 _, err = killTr.Recv()
1263 require.ErrorIs(t, err, io.EOF)
1264 var serveErr error
1265 require.Eventually(t, func() bool {
1266 select {
1267 case serveErr = <-served:
1268 return true
1269 default:
1270 return false
1271 }
1272 }, 15*time.Second, 10*time.Millisecond, "daemon did not shut down after kill all")
1273 require.NoError(t, serveErr)
1274}