detached 6f62b03

test(app): prove lifecycle ownership transfer order

commit
6f62b03a084917bfd3a25fc74ae12f87569b32b8
author
brice · · 2026-07-26 01:18 UTC
committer
brice · · 2026-07-26 01:19 UTC
parent
8759629
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
1 file · +205 -11
Unified
UnifiedSplit
M internal/app/daemon_integration_test.go +205 -11
@@ -4,6 +4,7 @@ package app
44
55 import (
66 "context"
7+ "crypto/rand"
78 "errors"
89 "io"
910 "log/slog"
@@ -22,10 +23,13 @@ import (
2223 "github.com/bnema/vev/internal/adapters/ipc"
2324 "github.com/bnema/vev/internal/adapters/lifecycle"
2425 "github.com/bnema/vev/internal/adapters/pty"
26+ "github.com/bnema/vev/internal/adapters/recoveryfs"
27+ "github.com/bnema/vev/internal/adapters/snapshot"
2528 "github.com/bnema/vev/internal/domain"
2629 "github.com/bnema/vev/internal/persist"
2730 "github.com/bnema/vev/internal/ports"
2831 "github.com/bnema/vev/internal/usecase/daemon"
32+ "github.com/bnema/vev/internal/usecase/recovery"
2933 "github.com/bnema/vev/pkg/vt"
3034 )
3135
@@ -526,22 +530,212 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
526530 }
527531
528532 func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
529- dir := filepath.Join(t.TempDir(), "runtime")
530- old, err := lifecycle.TryAcquire(dir)
533+ runtimeDir := filepath.Join(t.TempDir(), "runtime")
534+ stateDir := filepath.Join(t.TempDir(), "state")
535+ snapshotRepository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
536+
537+ oldOwner, err := lifecycle.TryAcquire(runtimeDir)
538+ require.NoError(t, err)
539+ oldOpened, err := persist.OpenOrMigrate(context.Background(), persist.OpenDeps{
540+ StateDir: stateDir,
541+ Random: rand.Reader,
542+ SnapshotMigration: snapshotRepository,
543+ })
531544 require.NoError(t, err)
532- var catalogueOpen, listenCalled atomic.Bool
545+ oldListener, err := ipc.Listen(runtimeDir)
546+ require.NoError(t, err)
547+
548+ socketCloseEntered := make(chan struct{})
549+ allowSocketClose := make(chan struct{})
550+ socketClosed := make(chan struct{})
551+ catalogueCloseEntered := make(chan struct{})
552+ allowCatalogueClose := make(chan struct{})
553+ catalogueClosed := make(chan struct{})
554+ ownerReleaseEntered := make(chan struct{})
555+ allowOwnerRelease := make(chan struct{})
556+ ownerReleased := make(chan struct{})
557+ teardownEvents := make(chan string, 3)
558+
559+ controlledListener := &lifecycleControlledListener{
560+ Listener: oldListener,
561+ entered: socketCloseEntered,
562+ proceed: allowSocketClose,
563+ closed: socketClosed,
564+ events: teardownEvents,
565+ }
566+ controlledCatalogue := &lifecycleControlledCatalogue{
567+ Catalogue: oldOpened.Catalogue,
568+ entered: catalogueCloseEntered,
569+ proceed: allowCatalogueClose,
570+ closed: catalogueClosed,
571+ events: teardownEvents,
572+ }
573+ oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
574+ oldCtx, stopOld := context.WithCancel(context.Background())
575+ oldDone := make(chan error, 1)
576+ go func() {
577+ serveErr := oldDaemon.Serve(oldCtx, controlledListener)
578+ teardownEvents <- "owner-release"
579+ close(ownerReleaseEntered)
580+ <-allowOwnerRelease
581+ releaseErr := oldOwner.Release()
582+ close(ownerReleased)
583+ oldDone <- errors.Join(serveErr, releaseErr)
584+ }()
585+
586+ newAcquireAttempted := make(chan struct{})
587+ newDurableOpen := make(chan struct{})
588+ newListen := make(chan struct{})
589+ startupEvents := make(chan string, 3)
533590 newDone := make(chan error, 1)
534591 go func() {
535- newDone <- runWithLifecycleOwner(context.Background(), dir, filepath.Join(t.TempDir(), "state"), func(context.Context) error {
536- catalogueOpen.Store(true)
537- listenCalled.Store(true)
538- return nil
592+ newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
593+ close(newDurableOpen)
594+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
595+ StateDir: stateDir,
596+ Random: rand.Reader,
597+ SnapshotMigration: snapshotRepository,
598+ })
599+ if err != nil {
600+ return err
601+ }
602+ coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, recoveryfs.New(stateDir), rand.Reader)
603+ if err := coordinator.Recover(ctx); err != nil {
604+ return errors.Join(err, opened.Catalogue.Close())
605+ }
606+ startupEvents <- "durable-open-recover"
607+
608+ close(newListen)
609+ listener, err := ipc.Listen(runtimeDir)
610+ if err != nil {
611+ return errors.Join(err, opened.Catalogue.Close())
612+ }
613+ startupEvents <- "listen"
614+ return errors.Join(listener.Close(), opened.Catalogue.Close())
615+ }, lifecycleStartupDeps{
616+ ensurePrivate: func(string) error { return nil },
617+ acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
618+ close(newAcquireAttempted)
619+ owner, err := lifecycle.Acquire(ctx, dir, retry)
620+ if err != nil {
621+ return nil, err
622+ }
623+ <-ownerReleased
624+ startupEvents <- "owner-acquired"
625+ return owner, nil
626+ },
539627 })
540628 }()
541- require.Never(t, func() bool { return catalogueOpen.Load() || listenCalled.Load() }, 50*time.Millisecond, time.Millisecond)
542- require.NoError(t, old.Release())
543- require.Eventually(t, catalogueOpen.Load, time.Second, time.Millisecond)
544- require.NoError(t, <-newDone)
629+
630+ awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
631+ assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
632+ assertLifecycleStagePending(t, newListen, "new listen before old teardown")
633+
634+ stopOld()
635+ awaitLifecycleStage(t, socketCloseEntered, "old socket close")
636+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
637+ close(allowSocketClose)
638+ awaitLifecycleStage(t, socketClosed, "old socket closed")
639+
640+ awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
641+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
642+ close(allowCatalogueClose)
643+ awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
644+
645+ awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
646+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
647+ close(allowOwnerRelease)
648+ awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
649+
650+ require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
651+ require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
652+ require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
653+ <-teardownEvents,
654+ <-teardownEvents,
655+ <-teardownEvents,
656+ })
657+ require.Equal(t, []string{"owner-acquired", "durable-open-recover", "listen"}, []string{
658+ <-startupEvents,
659+ <-startupEvents,
660+ <-startupEvents,
661+ })
662+}
663+
664+func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
665+ t.Helper()
666+ select {
667+ case <-stage:
668+ case <-time.After(3 * time.Second):
669+ t.Fatalf("timed out waiting for %s", name)
670+ }
671+}
672+
673+func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
674+ t.Helper()
675+ select {
676+ case err := <-result:
677+ return err
678+ case <-time.After(3 * time.Second):
679+ t.Fatalf("timed out waiting for %s", name)
680+ return nil
681+ }
682+}
683+
684+func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
685+ t.Helper()
686+ select {
687+ case <-stage:
688+ t.Fatalf("unexpected %s", name)
689+ default:
690+ }
691+}
692+
693+func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
694+ t.Helper()
695+ assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
696+ assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
697+}
698+
699+type lifecycleControlledListener struct {
700+ ports.Listener
701+ entered chan struct{}
702+ proceed chan struct{}
703+ closed chan struct{}
704+ events chan<- string
705+ once sync.Once
706+ err error
707+}
708+
709+func (l *lifecycleControlledListener) Close() error {
710+ l.once.Do(func() {
711+ l.events <- "socket-close"
712+ close(l.entered)
713+ <-l.proceed
714+ l.err = l.Listener.Close()
715+ close(l.closed)
716+ })
717+ return l.err
718+}
719+
720+type lifecycleControlledCatalogue struct {
721+ ports.Catalogue
722+ entered chan struct{}
723+ proceed chan struct{}
724+ closed chan struct{}
725+ events chan<- string
726+ once sync.Once
727+ err error
728+}
729+
730+func (c *lifecycleControlledCatalogue) Close() error {
731+ c.once.Do(func() {
732+ c.events <- "catalogue-close"
733+ close(c.entered)
734+ <-c.proceed
735+ c.err = c.Catalogue.Close()
736+ close(c.closed)
737+ })
738+ return c.err
545739 }
546740
547741 type integrationTransport struct{}
@@ -4,6 +4,7 @@ package app
4 4
5 import (5 import (
6 "context"6 "context"
7+ "crypto/rand"
7 "errors"8 "errors"
8 "io"9 "io"
9 "log/slog"10 "log/slog"
@@ -22,10 +23,13 @@ import (
22 "github.com/bnema/vev/internal/adapters/ipc"23 "github.com/bnema/vev/internal/adapters/ipc"
23 "github.com/bnema/vev/internal/adapters/lifecycle"24 "github.com/bnema/vev/internal/adapters/lifecycle"
24 "github.com/bnema/vev/internal/adapters/pty"25 "github.com/bnema/vev/internal/adapters/pty"
26+ "github.com/bnema/vev/internal/adapters/recoveryfs"
27+ "github.com/bnema/vev/internal/adapters/snapshot"
25 "github.com/bnema/vev/internal/domain"28 "github.com/bnema/vev/internal/domain"
26 "github.com/bnema/vev/internal/persist"29 "github.com/bnema/vev/internal/persist"
27 "github.com/bnema/vev/internal/ports"30 "github.com/bnema/vev/internal/ports"
28 "github.com/bnema/vev/internal/usecase/daemon"31 "github.com/bnema/vev/internal/usecase/daemon"
32+ "github.com/bnema/vev/internal/usecase/recovery"
29 "github.com/bnema/vev/pkg/vt"33 "github.com/bnema/vev/pkg/vt"
30 )34 )
31 35
@@ -526,22 +530,212 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
526 }530 }
527 531
528 func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {532 func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
529- dir := filepath.Join(t.TempDir(), "runtime")533+ runtimeDir := filepath.Join(t.TempDir(), "runtime")
530- old, err := lifecycle.TryAcquire(dir)534+ stateDir := filepath.Join(t.TempDir(), "state")
535+ snapshotRepository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
536+
537+ oldOwner, err := lifecycle.TryAcquire(runtimeDir)
538+ require.NoError(t, err)
539+ oldOpened, err := persist.OpenOrMigrate(context.Background(), persist.OpenDeps{
540+ StateDir: stateDir,
541+ Random: rand.Reader,
542+ SnapshotMigration: snapshotRepository,
543+ })
531 require.NoError(t, err)544 require.NoError(t, err)
532- var catalogueOpen, listenCalled atomic.Bool545+ oldListener, err := ipc.Listen(runtimeDir)
546+ require.NoError(t, err)
547+
548+ socketCloseEntered := make(chan struct{})
549+ allowSocketClose := make(chan struct{})
550+ socketClosed := make(chan struct{})
551+ catalogueCloseEntered := make(chan struct{})
552+ allowCatalogueClose := make(chan struct{})
553+ catalogueClosed := make(chan struct{})
554+ ownerReleaseEntered := make(chan struct{})
555+ allowOwnerRelease := make(chan struct{})
556+ ownerReleased := make(chan struct{})
557+ teardownEvents := make(chan string, 3)
558+
559+ controlledListener := &lifecycleControlledListener{
560+ Listener: oldListener,
561+ entered: socketCloseEntered,
562+ proceed: allowSocketClose,
563+ closed: socketClosed,
564+ events: teardownEvents,
565+ }
566+ controlledCatalogue := &lifecycleControlledCatalogue{
567+ Catalogue: oldOpened.Catalogue,
568+ entered: catalogueCloseEntered,
569+ proceed: allowCatalogueClose,
570+ closed: catalogueClosed,
571+ events: teardownEvents,
572+ }
573+ oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
574+ oldCtx, stopOld := context.WithCancel(context.Background())
575+ oldDone := make(chan error, 1)
576+ go func() {
577+ serveErr := oldDaemon.Serve(oldCtx, controlledListener)
578+ teardownEvents <- "owner-release"
579+ close(ownerReleaseEntered)
580+ <-allowOwnerRelease
581+ releaseErr := oldOwner.Release()
582+ close(ownerReleased)
583+ oldDone <- errors.Join(serveErr, releaseErr)
584+ }()
585+
586+ newAcquireAttempted := make(chan struct{})
587+ newDurableOpen := make(chan struct{})
588+ newListen := make(chan struct{})
589+ startupEvents := make(chan string, 3)
533 newDone := make(chan error, 1)590 newDone := make(chan error, 1)
534 go func() {591 go func() {
535- newDone <- runWithLifecycleOwner(context.Background(), dir, filepath.Join(t.TempDir(), "state"), func(context.Context) error {592+ newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
536- catalogueOpen.Store(true)593+ close(newDurableOpen)
537- listenCalled.Store(true)594+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
538- return nil595+ StateDir: stateDir,
596+ Random: rand.Reader,
597+ SnapshotMigration: snapshotRepository,
598+ })
599+ if err != nil {
600+ return err
601+ }
602+ coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, recoveryfs.New(stateDir), rand.Reader)
603+ if err := coordinator.Recover(ctx); err != nil {
604+ return errors.Join(err, opened.Catalogue.Close())
605+ }
606+ startupEvents <- "durable-open-recover"
607+
608+ close(newListen)
609+ listener, err := ipc.Listen(runtimeDir)
610+ if err != nil {
611+ return errors.Join(err, opened.Catalogue.Close())
612+ }
613+ startupEvents <- "listen"
614+ return errors.Join(listener.Close(), opened.Catalogue.Close())
615+ }, lifecycleStartupDeps{
616+ ensurePrivate: func(string) error { return nil },
617+ acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
618+ close(newAcquireAttempted)
619+ owner, err := lifecycle.Acquire(ctx, dir, retry)
620+ if err != nil {
621+ return nil, err
622+ }
623+ <-ownerReleased
624+ startupEvents <- "owner-acquired"
625+ return owner, nil
626+ },
539 })627 })
540 }()628 }()
541- require.Never(t, func() bool { return catalogueOpen.Load() || listenCalled.Load() }, 50*time.Millisecond, time.Millisecond)629+
542- require.NoError(t, old.Release())630+ awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
543- require.Eventually(t, catalogueOpen.Load, time.Second, time.Millisecond)631+ assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
544- require.NoError(t, <-newDone)632+ assertLifecycleStagePending(t, newListen, "new listen before old teardown")
633+
634+ stopOld()
635+ awaitLifecycleStage(t, socketCloseEntered, "old socket close")
636+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
637+ close(allowSocketClose)
638+ awaitLifecycleStage(t, socketClosed, "old socket closed")
639+
640+ awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
641+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
642+ close(allowCatalogueClose)
643+ awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
644+
645+ awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
646+ assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
647+ close(allowOwnerRelease)
648+ awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
649+
650+ require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
651+ require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
652+ require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
653+ <-teardownEvents,
654+ <-teardownEvents,
655+ <-teardownEvents,
656+ })
657+ require.Equal(t, []string{"owner-acquired", "durable-open-recover", "listen"}, []string{
658+ <-startupEvents,
659+ <-startupEvents,
660+ <-startupEvents,
661+ })
662+}
663+
664+func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
665+ t.Helper()
666+ select {
667+ case <-stage:
668+ case <-time.After(3 * time.Second):
669+ t.Fatalf("timed out waiting for %s", name)
670+ }
671+}
672+
673+func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
674+ t.Helper()
675+ select {
676+ case err := <-result:
677+ return err
678+ case <-time.After(3 * time.Second):
679+ t.Fatalf("timed out waiting for %s", name)
680+ return nil
681+ }
682+}
683+
684+func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
685+ t.Helper()
686+ select {
687+ case <-stage:
688+ t.Fatalf("unexpected %s", name)
689+ default:
690+ }
691+}
692+
693+func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
694+ t.Helper()
695+ assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
696+ assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
697+}
698+
699+type lifecycleControlledListener struct {
700+ ports.Listener
701+ entered chan struct{}
702+ proceed chan struct{}
703+ closed chan struct{}
704+ events chan<- string
705+ once sync.Once
706+ err error
707+}
708+
709+func (l *lifecycleControlledListener) Close() error {
710+ l.once.Do(func() {
711+ l.events <- "socket-close"
712+ close(l.entered)
713+ <-l.proceed
714+ l.err = l.Listener.Close()
715+ close(l.closed)
716+ })
717+ return l.err
718+}
719+
720+type lifecycleControlledCatalogue struct {
721+ ports.Catalogue
722+ entered chan struct{}
723+ proceed chan struct{}
724+ closed chan struct{}
725+ events chan<- string
726+ once sync.Once
727+ err error
728+}
729+
730+func (c *lifecycleControlledCatalogue) Close() error {
731+ c.once.Do(func() {
732+ c.events <- "catalogue-close"
733+ close(c.entered)
734+ <-c.proceed
735+ c.err = c.Catalogue.Close()
736+ close(c.closed)
737+ })
738+ return c.err
545 }739 }
546 740
547 type integrationTransport struct{}741 type integrationTransport struct{}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close