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 "github.com/bnema/vev/internal/adapters/clock"
23 "github.com/bnema/vev/internal/adapters/ipc"
24 "github.com/bnema/vev/internal/adapters/lifecycle"
25 "github.com/bnema/vev/internal/adapters/pty"
26 "github.com/bnema/vev/internal/adapters/snapshot"
27 "github.com/bnema/vev/internal/domain"
28 "github.com/bnema/vev/internal/persist"
29 "github.com/bnema/vev/internal/ports"
30 "github.com/bnema/vev/internal/usecase/daemon"
31 "github.com/bnema/vev/internal/usecase/recovery"
32 "github.com/bnema/vev/pkg/vt"
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_AttachFirstOutput(t *testing.T) {
265 sz := domain.Size{Cols: 80, Rows: 24}
266 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf HELLO; sleep 30"}))
267
268 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
269 defer func() { _ = tr.Close() }()
270
271 awaitText(t, p, sz, "HELLO")
272}
273
274func TestIntegration_InputRoundtrip(t *testing.T) {
275 sz := domain.Size{Cols: 80, Rows: 24}
276 dir, _ := startDaemon(t, daemon.WithShell("/bin/cat", nil))
277
278 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
279 defer func() { _ = tr.Close() }()
280
281 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("PINGPONG\n")})}))
282 awaitText(t, p, sz, "PINGPONG")
283}
284
285func TestIntegration_CommandPaletteCreatesTab(t *testing.T) {
286 sz := domain.Size{Cols: 80, Rows: 24}
287 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
288
289 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
290 defer func() { _ = tr.Close() }()
291 awaitText(t, p, sz, "READY")
292
293 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
294 awaitText(t, p, sz, "Commands")
295
296 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("CNT\r")})}))
297 // Tab labels are enriched with the focused pane's title; with no process
298 // inspector wired in this test, that title falls back to the shell's
299 // basename ("sh").
300 text := awaitScreenText(t, p, sz, " 1 (sh) 2 (sh) ")
301 require.Contains(t, text, " 1 (sh) 2 (sh) ")
302}
303
304func TestIntegration_CommandPaletteRenamesEphemeralSession(t *testing.T) {
305 sz := domain.Size{Cols: 80, Rows: 24}
306 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
307
308 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
309 defer func() { _ = tr.Close() }()
310 text := awaitScreenText(t, p, sz, "READY")
311 require.Contains(t, text, " 0* ")
312
313 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1b ")})}))
314 awaitText(t, p, sz, "Commands")
315
316 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("RNS\r")})}))
317 awaitText(t, p, sz, "Rename session")
318
319 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x7fwork\r")})}))
320 text = awaitScreenText(t, p, sz, " work ")
321 require.NotContains(t, text, "work*")
322
323 listTr, err := ipc.DialContext(context.Background(), dir)
324 require.NoError(t, err)
325 defer func() { _ = listTr.Close() }()
326 require.NoError(t, listTr.Send(ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})}))
327 f, err := listTr.Recv()
328 require.NoError(t, err)
329 require.Equal(t, ports.MsgSessions, f.Type)
330 sessions, err := ports.UnmarshalSessions(f.Payload)
331 require.NoError(t, err)
332 require.Len(t, sessions.Sessions, 1)
333 require.Equal(t, "work", sessions.Sessions[0].Name)
334 require.False(t, sessions.Sessions[0].Ephemeral)
335}
336
337func TestIntegration_AltCWithoutPaletteDoesNotCreateTab(t *testing.T) {
338 sz := domain.Size{Cols: 80, Rows: 24}
339 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
340
341 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
342 defer func() { _ = tr.Close() }()
343 text := awaitScreenText(t, p, sz, "READY")
344 require.NotContains(t, text, " 1 2 ")
345
346 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("\x1bc")})}))
347 assertNoTextAfterInput(t, p, sz, " 1 2 ")
348}
349
350func TestIntegration_EphemeralSurvivesDetachAndReattaches(t *testing.T) {
351 sz := domain.Size{Cols: 80, Rows: 24}
352 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
353
354 tr1, p1 := attach(t, dir, ports.IntentEphemeral, "", sz)
355 awaitText(t, p1, sz, "MARKER")
356 require.NoError(t, tr1.Close())
357
358 select {
359 case err := <-served:
360 t.Fatalf("daemon shut down while an ephemeral session was alive: %v", err)
361 case <-time.After(300 * time.Millisecond):
362 }
363
364 tr2, p2 := attach(t, dir, ports.IntentAttach, "0", sz)
365 defer func() { _ = tr2.Close() }()
366 awaitText(t, p2, sz, "MARKER")
367
368 require.NoError(t, killAll(dir))
369 select {
370 case err := <-served:
371 require.NoError(t, err)
372 case <-time.After(5 * time.Second):
373 t.Fatal("daemon did not shut down after kill all")
374 }
375}
376
377func TestIntegration_EphemeralNotListedAfterDaemonRestart(t *testing.T) {
378 sz := domain.Size{Cols: 80, Rows: 24}
379 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
380 t.Setenv("XDG_STATE_HOME", stateRoot)
381 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
382 dir := ipc.SocketDir()
383
384 _, served := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "printf TEMP; sleep 30"}))
385 tr, p := attach(t, dir, ports.IntentEphemeral, "", sz)
386 awaitText(t, p, sz, "TEMP")
387 require.NoError(t, tr.Close())
388
389 require.NoError(t, runKill(context.Background(), "", false, true))
390 select {
391 case err := <-served:
392 require.NoError(t, err)
393 case <-time.After(5 * time.Second):
394 t.Fatal("daemon did not stop after kill --daemon")
395 }
396
397 _, served2 := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
398 sessions := listRemoteSessions(t, dir)
399 require.Empty(t, sessions.Sessions)
400
401 require.NoError(t, runKill(context.Background(), "", false, true))
402 select {
403 case err := <-served2:
404 require.NoError(t, err)
405 case <-time.After(5 * time.Second):
406 t.Fatal("restarted daemon did not stop after kill --daemon")
407 }
408}
409
410func TestIntegration_NamedSurvivesReattach(t *testing.T) {
411 sz := domain.Size{Cols: 80, Rows: 24}
412 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
413
414 tr1, p1 := attach(t, dir, ports.IntentNew, "work", sz)
415 awaitText(t, p1, sz, "MARKER")
416
417 // Detach: a named session must survive.
418 require.NoError(t, tr1.Close())
419
420 // Daemon must still be running (session alive).
421 select {
422 case err := <-served:
423 t.Fatalf("daemon shut down while a named session was alive: %v", err)
424 case <-time.After(300 * time.Millisecond):
425 }
426
427 // Re-attach: the first paint must reproduce the retained screen state.
428 tr2, p2 := attach(t, dir, ports.IntentAttach, "work", sz)
429 defer func() { _ = tr2.Close() }()
430 awaitText(t, p2, sz, "MARKER")
431}
432
433func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
434 dir := filepath.Join(t.TempDir(), "runtime")
435 owner, err := lifecycle.TryAcquire(dir)
436 require.NoError(t, err)
437
438 ready := make(chan struct{})
439 transport := &integrationTransport{}
440 dial := func(context.Context, string) (ports.Transport, error) {
441 select {
442 case <-ready:
443 return transport, nil
444 default:
445 return nil, os.ErrNotExist
446 }
447 }
448 var spawns atomic.Int32
449 cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
450
451 const clients = 4
452 errs := make(chan error, clients)
453 var wg sync.WaitGroup
454 for range clients {
455 wg.Go(func() {
456 _, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
457 spawns.Add(1)
458 return nil
459 }, cfg)
460 errs <- err
461 })
462 }
463 require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
464 close(ready)
465 require.NoError(t, owner.Release())
466 wg.Wait()
467 close(errs)
468 for err := range errs {
469 require.NoError(t, err)
470 }
471 require.Zero(t, spawns.Load())
472}
473
474func TestListWaitsForLifecycleOwner(t *testing.T) {
475 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
476 t.Setenv("XDG_STATE_HOME", stateRoot)
477 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
478 p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
479 require.NoError(t, p.Close())
480 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
481 require.NoError(t, err)
482
483 done := make(chan error, 1)
484 go func() { done <- runList(context.Background()) }()
485 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
486 require.NoError(t, owner.Release())
487 require.NoError(t, <-done)
488}
489
490func TestOfflineKillWaitsForLifecycleOwner(t *testing.T) {
491 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
492 t.Setenv("XDG_STATE_HOME", stateRoot)
493 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
494 p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
495 now := time.Now().UnixNano()
496 require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, CreatedAt: now, UpdatedAt: now}))
497 require.NoError(t, p.Close())
498 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
499 require.NoError(t, err)
500
501 done := make(chan error, 1)
502 go func() { done <- runKill(context.Background(), "named", false, false) }()
503 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
504 require.NoError(t, owner.Release())
505 require.NoError(t, <-done)
506}
507
508func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
509 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
510 t.Setenv("XDG_STATE_HOME", stateRoot)
511 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
512 owner, err := lifecycle.TryAcquire(ipc.SocketDir())
513 require.NoError(t, err)
514 _, served := startDaemonInDir(t, ipc.SocketDir())
515
516 done := make(chan error, 1)
517 go func() { done <- requestDaemonStop(context.Background()) }()
518 select {
519 case err := <-served:
520 require.NoError(t, err)
521 case <-time.After(time.Second):
522 t.Fatal("daemon did not stop")
523 }
524 require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
525 require.NoError(t, owner.Release())
526 require.NoError(t, <-done)
527}
528
529func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
530 runtimeDir := filepath.Join(t.TempDir(), "runtime")
531 stateDir := filepath.Join(t.TempDir(), "state")
532 catalogue := &lifecycleObservedCatalogue{
533 Catalogue: newTestPersister(t, stateDir),
534 closed: make(chan struct{}),
535 }
536 require.NoError(t, catalogue.Create(domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}))
537 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
538 shutdownClock := newLifecycleShutdownClock(t)
539 maintenanceRepository := &lifecycleBlockingMaintenanceRepository{
540 SnapshotRepository: repository,
541 entered: make(chan struct{}),
542 release: make(chan struct{}),
543 returned: make(chan struct{}),
544 }
545 ctx, cancel := context.WithCancel(context.Background())
546 t.Cleanup(cancel)
547 var releaseOnce sync.Once
548 releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
549 t.Cleanup(releaseMaintenance)
550
551 callbackReturned := make(chan struct{})
552 wrapperReturned := make(chan error, 1)
553 go func() {
554 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
555 defer close(callbackReturned)
556 d := daemon.New(
557 pty.NewFactory(),
558 shutdownClock,
559 discardLog(),
560 daemon.WithSnapshotRepository(repository),
561 daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
562 daemon.WithCatalogue(catalogue, nil),
563 )
564 if err := d.CollectStartupGarbage(ctx); err != nil {
565 return err
566 }
567 listener, err := ipc.Listen(runtimeDir)
568 if err != nil {
569 return err
570 }
571 return d.Serve(ctx, listener)
572 })
573 }()
574
575 awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")
576 cancel()
577 select {
578 case <-callbackReturned:
579 t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
580 case <-time.After(100 * time.Millisecond):
581 }
582 assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
583 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
584 _, err := lifecycle.TryAcquire(runtimeDir)
585 require.ErrorIs(t, err, lifecycle.ErrBusy)
586
587 releaseMaintenance()
588 awaitLifecycleStage(t, maintenanceRepository.returned, "maintenance repository return")
589 awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
590 awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
591 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
592 owner, err := lifecycle.TryAcquire(runtimeDir)
593 require.NoError(t, err)
594 require.NoError(t, owner.Release())
595}
596
597type lifecycleBlockingMaintenanceRepository struct {
598 ports.SnapshotRepository
599 entered chan struct{}
600 release chan struct{}
601 returned chan struct{}
602 enterOnce sync.Once
603 returnOnce sync.Once
604}
605
606func (r *lifecycleBlockingMaintenanceRepository) CollectGarbage(context.Context, map[domain.IncarnationID]domain.CheckpointRef) error {
607 r.enterOnce.Do(func() { close(r.entered) })
608 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
609 r.returnOnce.Do(func() { close(r.returned) })
610 return nil
611}
612
613type lifecycleObservedCatalogue struct {
614 ports.Catalogue
615 closed chan struct{}
616 closeOnce sync.Once
617}
618
619func (c *lifecycleObservedCatalogue) Close() error {
620 err := c.Catalogue.Close()
621 c.closeOnce.Do(func() { close(c.closed) })
622 return err
623}
624
625func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
626 result := runBlockedSnapshotWriterShutdown(t)
627 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
628 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
629 _, err := lifecycle.TryAcquire(result.runtimeDir)
630 require.ErrorIs(t, err, lifecycle.ErrBusy)
631
632 result.releaseWriter()
633 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
634 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
635 owner, err := lifecycle.TryAcquire(result.runtimeDir)
636 require.NoError(t, err)
637 require.NoError(t, owner.Release())
638}
639
640func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
641 result := runBlockedSnapshotWriterShutdown(t)
642 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
643
644 result.releaseWriter()
645 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
646 awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
647 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
648}
649
650func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
651 result := runBlockedSnapshotWriterShutdown(t)
652 assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
653 _, err := lifecycle.TryAcquire(result.runtimeDir)
654 require.ErrorIs(t, err, lifecycle.ErrBusy)
655
656 result.releaseWriter()
657 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
658}
659
660// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
661// records. Lifecycle ownership must therefore outlive it: releasing the flock
662// while a restoration repository call is still running would let a second
663// daemon mutate the same durable state.
664func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
665 stateDir := filepath.Join(t.TempDir(), "state")
666 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
667 checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
668
669 runtimeDir := filepath.Join(t.TempDir(), "runtime")
670 blocking := &lifecycleBlockingRestore{
671 SnapshotRepository: repository,
672 entered: make(chan struct{}),
673 release: make(chan struct{}),
674 returned: make(chan struct{}),
675 }
676 var releaseOnce sync.Once
677 releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
678 t.Cleanup(releaseRestore)
679
680 shutdownClock := newLifecycleShutdownClock(t)
681 ctx, cancel := context.WithCancel(context.Background())
682 t.Cleanup(cancel)
683 listenerReady := make(chan struct{})
684 callbackReturned := make(chan struct{})
685 wrapperReturned := make(chan error, 1)
686 go func() {
687 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
688 defer close(callbackReturned)
689 opened, err := persist.OpenOrCreate(stateDir)
690 if err != nil {
691 return err
692 }
693 listener, err := ipc.Listen(runtimeDir)
694 if err != nil {
695 return errors.Join(err, opened.Catalogue.Close())
696 }
697 close(listenerReady)
698 d := daemon.New(
699 pty.NewFactory(),
700 shutdownClock,
701 discardLog(),
702 daemon.WithShell("/bin/cat", nil),
703 daemon.WithCatalogue(opened.Catalogue, opened.Records),
704 daemon.WithSnapshotRepository(blocking),
705 )
706 return d.Serve(ctx, listener)
707 })
708 }()
709
710 awaitLifecycleStage(t, listenerReady, "daemon listener")
711 awaitLifecycleStage(t, blocking.entered, "restoration repository call")
712 require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
713
714 cancel()
715 shutdownClock.nextTimer(t).fire()
716 select {
717 case <-callbackReturned:
718 t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
719 case <-time.After(100 * time.Millisecond):
720 }
721 assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
722 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
723 _, err := lifecycle.TryAcquire(runtimeDir)
724 require.ErrorIs(t, err, lifecycle.ErrBusy)
725
726 releaseRestore()
727 awaitLifecycleStage(t, blocking.returned, "restoration repository return")
728 awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
729 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
730 owner, err := lifecycle.TryAcquire(runtimeDir)
731 require.NoError(t, err)
732 require.NoError(t, owner.Release())
733}
734
735// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
736// repository until one named session owns a committed checkpoint, then shuts it
737// down. The returned name is restorable by any later daemon on the same state.
738func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
739 t.Helper()
740 const name = "restorable"
741 runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
742 ctx, cancel := context.WithCancel(context.Background())
743 defer cancel()
744
745 opened, err := persist.OpenOrCreate(stateDir)
746 require.NoError(t, err)
747 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
748 listener, err := ipc.Listen(runtimeDir)
749 require.NoError(t, err)
750 d := daemon.New(
751 pty.NewFactory(),
752 clock.New(),
753 discardLog(),
754 daemon.WithShell("/bin/cat", nil),
755 daemon.WithCatalogue(opened.Catalogue, opened.Records),
756 daemon.WithSnapshotRepository(repository),
757 daemon.WithRecoveryCoordinator(coordinator),
758 )
759 served := make(chan error, 1)
760 go func() { served <- d.Serve(ctx, listener) }()
761
762 tr, _ := attach(t, runtimeDir, ports.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
763 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("checkpoint me\n")})}))
764 require.Eventually(t, func() bool {
765 record, ok, _ := opened.Catalogue.Record(name)
766 return ok && record.Committed != nil && record.DegradedReason == ""
767 }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
768 require.NoError(t, tr.Close())
769
770 cancel()
771 require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
772 return name
773}
774
775type lifecycleBlockingRestore struct {
776 ports.SnapshotRepository
777 entered chan struct{}
778 release chan struct{}
779 returned chan struct{}
780 enterOnce sync.Once
781 returnOnce sync.Once
782 mu sync.Mutex
783 repaired string
784}
785
786func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
787 r.mu.Lock()
788 r.repaired = name
789 r.mu.Unlock()
790 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
791}
792
793func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
794 r.enterOnce.Do(func() { close(r.entered) })
795 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
796 r.returnOnce.Do(func() { close(r.returned) })
797 return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
798}
799
800func (r *lifecycleBlockingRestore) repairedName() string {
801 r.mu.Lock()
802 defer r.mu.Unlock()
803 return r.repaired
804}
805
806type blockedSnapshotWriterShutdown struct {
807 runtimeDir string
808 callbackReturned <-chan struct{}
809 wrapperReturned <-chan error
810 publishReturned <-chan struct{}
811 releaseWriter func()
812}
813
814func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
815 t.Helper()
816 runtimeDir := filepath.Join(t.TempDir(), "runtime")
817 stateDir := filepath.Join(t.TempDir(), "state")
818 shutdownClock := newLifecycleShutdownClock(t)
819 repository := newLifecycleBlockingRepository()
820 ctx, cancel := context.WithCancel(context.Background())
821 t.Cleanup(cancel)
822
823 listenerReady := make(chan struct{})
824 listenerClosed := make(chan struct{})
825 callbackReturned := make(chan struct{})
826 wrapperReturned := make(chan error, 1)
827 go func() {
828 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
829 defer close(callbackReturned)
830 listener, err := ipc.Listen(runtimeDir)
831 if err != nil {
832 return err
833 }
834 observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
835 close(listenerReady)
836 d := daemon.New(
837 pty.NewFactory(),
838 shutdownClock,
839 discardLog(),
840 daemon.WithShell("/bin/cat", nil),
841 daemon.WithSnapshotRepository(repository),
842 )
843 return d.Serve(ctx, observed)
844 })
845 }()
846
847 awaitLifecycleStage(t, listenerReady, "daemon listener")
848 tr, _ := attach(t, runtimeDir, ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
849 require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("dirty state\n")})}))
850 awaitLifecycleStage(t, repository.entered, "snapshot publication")
851 require.NoError(t, tr.Close())
852 require.Eventually(t, func() bool {
853 sessions := listRemoteSessions(t, runtimeDir)
854 return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
855 }, time.Second, 10*time.Millisecond)
856 cancel()
857 awaitLifecycleStage(t, listenerClosed, "listener close")
858
859 shutdownClock.nextTimer(t).fire()
860 select {
861 case <-callbackReturned:
862 t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
863 case <-time.After(100 * time.Millisecond):
864 }
865 assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
866 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
867
868 var releaseOnce sync.Once
869 releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
870 t.Cleanup(releaseWriter)
871 return blockedSnapshotWriterShutdown{
872 runtimeDir: runtimeDir,
873 callbackReturned: callbackReturned,
874 wrapperReturned: wrapperReturned,
875 publishReturned: repository.returned,
876 releaseWriter: releaseWriter,
877 }
878}
879
880type lifecycleBlockingRepository struct {
881 ports.SnapshotRepository
882 entered chan struct{}
883 release chan struct{}
884 returned chan struct{}
885 enterOnce sync.Once
886 returnOnce sync.Once
887}
888
889func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
890 return &lifecycleBlockingRepository{
891 entered: make(chan struct{}),
892 release: make(chan struct{}),
893 returned: make(chan struct{}),
894 }
895}
896
897func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
898 r.enterOnce.Do(func() { close(r.entered) })
899 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
900 r.returnOnce.Do(func() { close(r.returned) })
901 return nil
902}
903
904type lifecycleShutdownClock struct {
905 t *testing.T
906 base ports.Clock
907 timers chan *lifecycleManualTimer
908}
909
910func newLifecycleShutdownClock(t *testing.T) *lifecycleShutdownClock {
911 t.Helper()
912 return &lifecycleShutdownClock{t: t, base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
913}
914
915func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
916func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
917 if delay != daemon.SnapshotShutdownTimeout() {
918 return c.base.NewTimer(delay)
919 }
920 timer := &lifecycleManualTimer{t: c.t, ch: make(chan time.Time, 1)}
921 select {
922 case c.timers <- timer:
923 default:
924 c.t.Errorf("unexpected additional snapshot shutdown timer with delay %s", delay)
925 timer.fire()
926 }
927 return timer
928}
929
930func (c *lifecycleShutdownClock) nextTimer(t *testing.T) *lifecycleManualTimer {
931 t.Helper()
932 select {
933 case timer := <-c.timers:
934 return timer
935 case <-time.After(3 * time.Second):
936 t.Fatal("timed out waiting for shutdown deadline timer")
937 return nil
938 }
939}
940
941type lifecycleManualTimer struct {
942 t *testing.T
943 ch chan time.Time
944 stopped atomic.Bool
945}
946
947func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
948func (t *lifecycleManualTimer) Reset(delay time.Duration) bool {
949 if delay != daemon.SnapshotShutdownTimeout() {
950 t.t.Errorf("snapshot shutdown timer reset with delay %s, want %s", delay, daemon.SnapshotShutdownTimeout())
951 }
952 t.stopped.Store(false)
953 return false
954}
955func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
956func (t *lifecycleManualTimer) fire() {
957 if t.stopped.Load() {
958 return
959 }
960 select {
961 case t.ch <- time.Now():
962 default:
963 }
964}
965
966func TestLifecycleManualTimerRepeatedFireIsNonBlocking(t *testing.T) {
967 timer := &lifecycleManualTimer{t: t, ch: make(chan time.Time, 1)}
968 timer.fire()
969
970 returned := make(chan struct{})
971 go func() {
972 timer.fire()
973 close(returned)
974 }()
975
976 select {
977 case <-returned:
978 case <-time.After(time.Second):
979 t.Fatal("repeated manual timer fire blocked on a pending tick")
980 }
981}
982
983type lifecycleObservedListener struct {
984 ports.Listener
985 closed chan struct{}
986 once sync.Once
987}
988
989func (l *lifecycleObservedListener) Close() error {
990 err := l.Listener.Close()
991 l.once.Do(func() { close(l.closed) })
992 return err
993}
994
995func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
996 runtimeDir := filepath.Join(t.TempDir(), "runtime")
997 stateDir := filepath.Join(t.TempDir(), "state")
998 oldOwner, err := lifecycle.TryAcquire(runtimeDir)
999 require.NoError(t, err)
1000 oldOpened, err := persist.OpenOrCreate(stateDir)
1001 require.NoError(t, err)
1002 oldListener, err := ipc.Listen(runtimeDir)
1003 require.NoError(t, err)
1004
1005 socketCloseEntered := make(chan struct{})
1006 allowSocketClose := make(chan struct{})
1007 socketClosed := make(chan struct{})
1008 catalogueCloseEntered := make(chan struct{})
1009 allowCatalogueClose := make(chan struct{})
1010 catalogueClosed := make(chan struct{})
1011 ownerReleaseEntered := make(chan struct{})
1012 allowOwnerRelease := make(chan struct{})
1013 ownerReleased := make(chan struct{})
1014 teardownEvents := make(chan string, 3)
1015
1016 controlledListener := &lifecycleControlledListener{
1017 Listener: oldListener,
1018 entered: socketCloseEntered,
1019 proceed: allowSocketClose,
1020 closed: socketClosed,
1021 events: teardownEvents,
1022 }
1023 controlledCatalogue := &lifecycleControlledCatalogue{
1024 Catalogue: oldOpened.Catalogue,
1025 entered: catalogueCloseEntered,
1026 proceed: allowCatalogueClose,
1027 closed: catalogueClosed,
1028 events: teardownEvents,
1029 }
1030 oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
1031 oldCtx, stopOld := context.WithCancel(context.Background())
1032 oldDone := make(chan error, 1)
1033 go func() {
1034 serveErr := oldDaemon.Serve(oldCtx, controlledListener)
1035 teardownEvents <- "owner-release"
1036 close(ownerReleaseEntered)
1037 <-allowOwnerRelease
1038 releaseErr := oldOwner.Release()
1039 close(ownerReleased)
1040 oldDone <- errors.Join(serveErr, releaseErr)
1041 }()
1042
1043 newAcquireAttempted := make(chan struct{})
1044 newDurableOpen := make(chan struct{})
1045 newListen := make(chan struct{})
1046 startupEvents := make(chan string, 3)
1047 newDone := make(chan error, 1)
1048 go func() {
1049 newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
1050 close(newDurableOpen)
1051 opened, err := persist.OpenOrCreate(stateDir)
1052 if err != nil {
1053 return err
1054 }
1055 startupEvents <- "durable-open"
1056
1057 close(newListen)
1058 listener, err := ipc.Listen(runtimeDir)
1059 if err != nil {
1060 return errors.Join(err, opened.Catalogue.Close())
1061 }
1062 startupEvents <- "listen"
1063 return errors.Join(listener.Close(), opened.Catalogue.Close())
1064 }, lifecycleStartupDeps{
1065 ensurePrivate: func(string) error { return nil },
1066 acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
1067 close(newAcquireAttempted)
1068 owner, err := lifecycle.Acquire(ctx, dir, retry)
1069 if err != nil {
1070 return nil, err
1071 }
1072 <-ownerReleased
1073 startupEvents <- "owner-acquired"
1074 return owner, nil
1075 },
1076 })
1077 }()
1078
1079 awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
1080 assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
1081 assertLifecycleStagePending(t, newListen, "new listen before old teardown")
1082
1083 stopOld()
1084 awaitLifecycleStage(t, socketCloseEntered, "old socket close")
1085 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
1086 close(allowSocketClose)
1087 awaitLifecycleStage(t, socketClosed, "old socket closed")
1088
1089 awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
1090 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
1091 close(allowCatalogueClose)
1092 awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
1093
1094 awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
1095 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
1096 close(allowOwnerRelease)
1097 awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
1098
1099 require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
1100 require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
1101 require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
1102 <-teardownEvents,
1103 <-teardownEvents,
1104 <-teardownEvents,
1105 })
1106 require.Equal(t, []string{"owner-acquired", "durable-open", "listen"}, []string{
1107 <-startupEvents,
1108 <-startupEvents,
1109 <-startupEvents,
1110 })
1111}
1112
1113func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
1114 t.Helper()
1115 select {
1116 case <-stage:
1117 case <-time.After(3 * time.Second):
1118 t.Fatalf("timed out waiting for %s", name)
1119 }
1120}
1121
1122func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
1123 t.Helper()
1124 select {
1125 case err := <-result:
1126 return err
1127 case <-time.After(3 * time.Second):
1128 t.Fatalf("timed out waiting for %s", name)
1129 return nil
1130 }
1131}
1132
1133func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
1134 t.Helper()
1135 select {
1136 case <-stage:
1137 t.Fatalf("unexpected %s", name)
1138 default:
1139 }
1140}
1141
1142func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
1143 t.Helper()
1144 select {
1145 case err := <-result:
1146 t.Fatalf("unexpected %s: %v", name, err)
1147 default:
1148 }
1149}
1150
1151func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
1152 t.Helper()
1153 assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
1154 assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
1155}
1156
1157type lifecycleControlledListener struct {
1158 ports.Listener
1159 entered chan struct{}
1160 proceed chan struct{}
1161 closed chan struct{}
1162 events chan<- string
1163 once sync.Once
1164 err error
1165}
1166
1167func (l *lifecycleControlledListener) Close() error {
1168 l.once.Do(func() {
1169 l.events <- "socket-close"
1170 close(l.entered)
1171 <-l.proceed
1172 l.err = l.Listener.Close()
1173 close(l.closed)
1174 })
1175 return l.err
1176}
1177
1178type lifecycleControlledCatalogue struct {
1179 ports.Catalogue
1180 entered chan struct{}
1181 proceed chan struct{}
1182 closed chan struct{}
1183 events chan<- string
1184 once sync.Once
1185 err error
1186}
1187
1188func (c *lifecycleControlledCatalogue) Close() error {
1189 c.once.Do(func() {
1190 c.events <- "catalogue-close"
1191 close(c.entered)
1192 <-c.proceed
1193 c.err = c.Catalogue.Close()
1194 close(c.closed)
1195 })
1196 return c.err
1197}
1198
1199type integrationTransport struct{}
1200
1201func (*integrationTransport) Send(ports.Frame) error { return nil }
1202func (*integrationTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
1203func (*integrationTransport) Close() error { return nil }
1204func (*integrationTransport) LocalAddr() net.Addr { return nil }
1205func (*integrationTransport) RemoteAddr() net.Addr { return nil }
1206
1207func TestIntegration_KillAllShutsDownDaemon(t *testing.T) {
1208 sz := domain.Size{Cols: 80, Rows: 24}
1209 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
1210
1211 tr1, _ := attach(t, dir, ports.IntentNew, "one", sz)
1212 require.NoError(t, tr1.Close())
1213 tr2, _ := attach(t, dir, ports.IntentNew, "two", sz)
1214 require.NoError(t, tr2.Close())
1215
1216 killTr, err := ipc.DialContext(context.Background(), dir)
1217 require.NoError(t, err)
1218 defer func() { _ = killTr.Close() }()
1219 require.NoError(t, killTr.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}))
1220
1221 _, err = killTr.Recv()
1222 require.ErrorIs(t, err, io.EOF)
1223 var serveErr error
1224 require.Eventually(t, func() bool {
1225 select {
1226 case serveErr = <-served:
1227 return true
1228 default:
1229 return false
1230 }
1231 }, 15*time.Second, 10*time.Millisecond, "daemon did not shut down after kill all")
1232 require.NoError(t, serveErr)
1233}