feat(app): guard daemon state with lifecycle ownership
- commit
- 9d96051160779636a021bce38a434c667c6416f4
- author
- brice · · 2026-07-25 19:40 UTC
- parent
- e4c6c0f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +415 -27
A internal/adapters/lifecycle/flock.go +103 -0
| @@ -0,0 +1,103 @@ | |||
| 1 | + | // Package lifecycle provides exclusive process ownership of vev's durable | |
| 2 | + | // state for the complete daemon lifetime. | |
| 3 | + | package lifecycle | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | "fmt" | |
| 9 | + | "os" | |
| 10 | + | "path/filepath" | |
| 11 | + | "sync" | |
| 12 | + | "syscall" | |
| 13 | + | "time" | |
| 14 | + | ||
| 15 | + | "github.com/bnema/vev/pkg/safedir" | |
| 16 | + | ) | |
| 17 | + | ||
| 18 | + | // ErrBusy reports that another process currently owns durable session state. | |
| 19 | + | var ErrBusy = errors.New("durable session state is busy") | |
| 20 | + | ||
| 21 | + | // Owner holds the lifecycle lock descriptor open until Release. | |
| 22 | + | type Owner struct { | |
| 23 | + | file *os.File | |
| 24 | + | path string | |
| 25 | + | once sync.Once | |
| 26 | + | err error | |
| 27 | + | } | |
| 28 | + | ||
| 29 | + | // Path returns the fixed lifecycle lock path in runtimeDir. | |
| 30 | + | func Path(runtimeDir string) string { | |
| 31 | + | return filepath.Join(runtimeDir, "lifecycle.lock") | |
| 32 | + | } | |
| 33 | + | ||
| 34 | + | // TryAcquire attempts to take lifecycle ownership without waiting. | |
| 35 | + | func TryAcquire(runtimeDir string) (*Owner, error) { | |
| 36 | + | if err := safedir.EnsurePrivate(runtimeDir); err != nil { | |
| 37 | + | return nil, fmt.Errorf("lifecycle: securing runtime directory: %w", err) | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | path := Path(runtimeDir) | |
| 41 | + | file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) | |
| 42 | + | if err != nil { | |
| 43 | + | return nil, fmt.Errorf("lifecycle: opening %s: %w", path, err) | |
| 44 | + | } | |
| 45 | + | if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { | |
| 46 | + | closeErr := file.Close() | |
| 47 | + | if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) { | |
| 48 | + | return nil, errors.Join(ErrBusy, closeErr) | |
| 49 | + | } | |
| 50 | + | return nil, errors.Join(fmt.Errorf("lifecycle: locking %s: %w", path, err), closeErr) | |
| 51 | + | } | |
| 52 | + | return &Owner{file: file, path: path}, nil | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | // Acquire waits until lifecycle ownership is available or ctx is cancelled. | |
| 56 | + | // Only lock contention is retried; filesystem and lock errors fail closed. | |
| 57 | + | func Acquire(ctx context.Context, runtimeDir string, retry time.Duration) (*Owner, error) { | |
| 58 | + | if retry <= 0 { | |
| 59 | + | retry = time.Millisecond | |
| 60 | + | } | |
| 61 | + | for { | |
| 62 | + | if err := ctx.Err(); err != nil { | |
| 63 | + | return nil, err | |
| 64 | + | } | |
| 65 | + | owner, err := TryAcquire(runtimeDir) | |
| 66 | + | if err == nil { | |
| 67 | + | return owner, nil | |
| 68 | + | } | |
| 69 | + | if !errors.Is(err, ErrBusy) { | |
| 70 | + | return nil, err | |
| 71 | + | } | |
| 72 | + | ||
| 73 | + | timer := time.NewTimer(retry) | |
| 74 | + | select { | |
| 75 | + | case <-ctx.Done(): | |
| 76 | + | timer.Stop() | |
| 77 | + | return nil, ctx.Err() | |
| 78 | + | case <-timer.C: | |
| 79 | + | } | |
| 80 | + | } | |
| 81 | + | } | |
| 82 | + | ||
| 83 | + | // Release unlocks and closes the owner exactly once. | |
| 84 | + | func (o *Owner) Release() error { | |
| 85 | + | if o == nil { | |
| 86 | + | return nil | |
| 87 | + | } | |
| 88 | + | o.once.Do(func() { | |
| 89 | + | if o.file == nil { | |
| 90 | + | return | |
| 91 | + | } | |
| 92 | + | unlockErr := syscall.Flock(int(o.file.Fd()), syscall.LOCK_UN) | |
| 93 | + | closeErr := o.file.Close() | |
| 94 | + | if unlockErr != nil { | |
| 95 | + | unlockErr = fmt.Errorf("lifecycle: unlocking %s: %w", o.path, unlockErr) | |
| 96 | + | } | |
| 97 | + | if closeErr != nil { | |
| 98 | + | closeErr = fmt.Errorf("lifecycle: closing %s: %w", o.path, closeErr) | |
| 99 | + | } | |
| 100 | + | o.err = errors.Join(unlockErr, closeErr) | |
| 101 | + | }) | |
| 102 | + | return o.err | |
| 103 | + | } | |
| @@ -0,0 +1,103 @@ | |||||
| 1 | + | // Package lifecycle provides exclusive process ownership of vev's durable | |||
| 2 | + | // state for the complete daemon lifetime. | |||
| 3 | + | package lifecycle | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | "fmt" | |||
| 9 | + | "os" | |||
| 10 | + | "path/filepath" | |||
| 11 | + | "sync" | |||
| 12 | + | "syscall" | |||
| 13 | + | "time" | |||
| 14 | + | ||||
| 15 | + | "github.com/bnema/vev/pkg/safedir" | |||
| 16 | + | ) | |||
| 17 | + | ||||
| 18 | + | // ErrBusy reports that another process currently owns durable session state. | |||
| 19 | + | var ErrBusy = errors.New("durable session state is busy") | |||
| 20 | + | ||||
| 21 | + | // Owner holds the lifecycle lock descriptor open until Release. | |||
| 22 | + | type Owner struct { | |||
| 23 | + | file *os.File | |||
| 24 | + | path string | |||
| 25 | + | once sync.Once | |||
| 26 | + | err error | |||
| 27 | + | } | |||
| 28 | + | ||||
| 29 | + | // Path returns the fixed lifecycle lock path in runtimeDir. | |||
| 30 | + | func Path(runtimeDir string) string { | |||
| 31 | + | return filepath.Join(runtimeDir, "lifecycle.lock") | |||
| 32 | + | } | |||
| 33 | + | ||||
| 34 | + | // TryAcquire attempts to take lifecycle ownership without waiting. | |||
| 35 | + | func TryAcquire(runtimeDir string) (*Owner, error) { | |||
| 36 | + | if err := safedir.EnsurePrivate(runtimeDir); err != nil { | |||
| 37 | + | return nil, fmt.Errorf("lifecycle: securing runtime directory: %w", err) | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | path := Path(runtimeDir) | |||
| 41 | + | file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) | |||
| 42 | + | if err != nil { | |||
| 43 | + | return nil, fmt.Errorf("lifecycle: opening %s: %w", path, err) | |||
| 44 | + | } | |||
| 45 | + | if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { | |||
| 46 | + | closeErr := file.Close() | |||
| 47 | + | if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) { | |||
| 48 | + | return nil, errors.Join(ErrBusy, closeErr) | |||
| 49 | + | } | |||
| 50 | + | return nil, errors.Join(fmt.Errorf("lifecycle: locking %s: %w", path, err), closeErr) | |||
| 51 | + | } | |||
| 52 | + | return &Owner{file: file, path: path}, nil | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | // Acquire waits until lifecycle ownership is available or ctx is cancelled. | |||
| 56 | + | // Only lock contention is retried; filesystem and lock errors fail closed. | |||
| 57 | + | func Acquire(ctx context.Context, runtimeDir string, retry time.Duration) (*Owner, error) { | |||
| 58 | + | if retry <= 0 { | |||
| 59 | + | retry = time.Millisecond | |||
| 60 | + | } | |||
| 61 | + | for { | |||
| 62 | + | if err := ctx.Err(); err != nil { | |||
| 63 | + | return nil, err | |||
| 64 | + | } | |||
| 65 | + | owner, err := TryAcquire(runtimeDir) | |||
| 66 | + | if err == nil { | |||
| 67 | + | return owner, nil | |||
| 68 | + | } | |||
| 69 | + | if !errors.Is(err, ErrBusy) { | |||
| 70 | + | return nil, err | |||
| 71 | + | } | |||
| 72 | + | ||||
| 73 | + | timer := time.NewTimer(retry) | |||
| 74 | + | select { | |||
| 75 | + | case <-ctx.Done(): | |||
| 76 | + | timer.Stop() | |||
| 77 | + | return nil, ctx.Err() | |||
| 78 | + | case <-timer.C: | |||
| 79 | + | } | |||
| 80 | + | } | |||
| 81 | + | } | |||
| 82 | + | ||||
| 83 | + | // Release unlocks and closes the owner exactly once. | |||
| 84 | + | func (o *Owner) Release() error { | |||
| 85 | + | if o == nil { | |||
| 86 | + | return nil | |||
| 87 | + | } | |||
| 88 | + | o.once.Do(func() { | |||
| 89 | + | if o.file == nil { | |||
| 90 | + | return | |||
| 91 | + | } | |||
| 92 | + | unlockErr := syscall.Flock(int(o.file.Fd()), syscall.LOCK_UN) | |||
| 93 | + | closeErr := o.file.Close() | |||
| 94 | + | if unlockErr != nil { | |||
| 95 | + | unlockErr = fmt.Errorf("lifecycle: unlocking %s: %w", o.path, unlockErr) | |||
| 96 | + | } | |||
| 97 | + | if closeErr != nil { | |||
| 98 | + | closeErr = fmt.Errorf("lifecycle: closing %s: %w", o.path, closeErr) | |||
| 99 | + | } | |||
| 100 | + | o.err = errors.Join(unlockErr, closeErr) | |||
| 101 | + | }) | |||
| 102 | + | return o.err | |||
| 103 | + | } | |||
A internal/adapters/lifecycle/flock_test.go +89 -0
| @@ -0,0 +1,89 @@ | |||
| 1 | + | package lifecycle | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "bufio" | |
| 5 | + | "context" | |
| 6 | + | "errors" | |
| 7 | + | "fmt" | |
| 8 | + | "io" | |
| 9 | + | "os" | |
| 10 | + | "os/exec" | |
| 11 | + | "path/filepath" | |
| 12 | + | "testing" | |
| 13 | + | "time" | |
| 14 | + | ||
| 15 | + | "github.com/stretchr/testify/require" | |
| 16 | + | ) | |
| 17 | + | ||
| 18 | + | func TestLifecycleLockProcessHelper(t *testing.T) { | |
| 19 | + | if os.Getenv("VEV_LIFECYCLE_LOCK_HELPER") != "1" { | |
| 20 | + | t.Skip("helper process") | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | owner, err := TryAcquire(os.Getenv("VEV_LIFECYCLE_LOCK_DIR")) | |
| 24 | + | require.NoError(t, err) | |
| 25 | + | defer func() { require.NoError(t, owner.Release()) }() | |
| 26 | + | ||
| 27 | + | _, err = fmt.Fprintln(os.Stdout, "locked") | |
| 28 | + | require.NoError(t, err) | |
| 29 | + | _, err = io.Copy(io.Discard, os.Stdin) | |
| 30 | + | require.NoError(t, err) | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | func TestLifecycleLock(t *testing.T) { | |
| 34 | + | t.Run("process contention and reacquire", testLifecycleLockProcessContentionAndReacquire) | |
| 35 | + | t.Run("acquire cancellation", testLifecycleLockAcquireCancellation) | |
| 36 | + | t.Run("unavailable runtime directory", testLifecycleLockUnavailableRuntimeDirectory) | |
| 37 | + | } | |
| 38 | + | ||
| 39 | + | func testLifecycleLockProcessContentionAndReacquire(t *testing.T) { | |
| 40 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |
| 41 | + | cmd := exec.Command(os.Args[0], "-test.run=^TestLifecycleLockProcessHelper$") | |
| 42 | + | cmd.Env = append(os.Environ(), | |
| 43 | + | "VEV_LIFECYCLE_LOCK_HELPER=1", | |
| 44 | + | "VEV_LIFECYCLE_LOCK_DIR="+runtimeDir, | |
| 45 | + | ) | |
| 46 | + | stdin, err := cmd.StdinPipe() | |
| 47 | + | require.NoError(t, err) | |
| 48 | + | stdout, err := cmd.StdoutPipe() | |
| 49 | + | require.NoError(t, err) | |
| 50 | + | cmd.Stderr = os.Stderr | |
| 51 | + | require.NoError(t, cmd.Start()) | |
| 52 | + | ||
| 53 | + | line, err := bufio.NewReader(stdout).ReadString('\n') | |
| 54 | + | require.NoError(t, err) | |
| 55 | + | require.Equal(t, "locked\n", line) | |
| 56 | + | ||
| 57 | + | _, err = TryAcquire(runtimeDir) | |
| 58 | + | require.ErrorIs(t, err, ErrBusy) | |
| 59 | + | ||
| 60 | + | require.NoError(t, stdin.Close()) | |
| 61 | + | require.NoError(t, cmd.Wait()) | |
| 62 | + | ||
| 63 | + | owner, err := TryAcquire(runtimeDir) | |
| 64 | + | require.NoError(t, err) | |
| 65 | + | require.NoError(t, owner.Release()) | |
| 66 | + | require.NoError(t, owner.Release(), "release is idempotent") | |
| 67 | + | } | |
| 68 | + | ||
| 69 | + | func testLifecycleLockAcquireCancellation(t *testing.T) { | |
| 70 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |
| 71 | + | owner, err := TryAcquire(runtimeDir) | |
| 72 | + | require.NoError(t, err) | |
| 73 | + | defer func() { require.NoError(t, owner.Release()) }() | |
| 74 | + | ||
| 75 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 76 | + | cancel() | |
| 77 | + | ||
| 78 | + | _, err = Acquire(ctx, runtimeDir, time.Millisecond) | |
| 79 | + | require.ErrorIs(t, err, context.Canceled) | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | func testLifecycleLockUnavailableRuntimeDirectory(t *testing.T) { | |
| 83 | + | runtimePath := t.TempDir() + "/not-a-directory" | |
| 84 | + | require.NoError(t, os.WriteFile(runtimePath, []byte("occupied"), 0o600)) | |
| 85 | + | ||
| 86 | + | _, err := TryAcquire(runtimePath) | |
| 87 | + | require.Error(t, err) | |
| 88 | + | require.False(t, errors.Is(err, ErrBusy)) | |
| 89 | + | } | |
| @@ -0,0 +1,89 @@ | |||||
| 1 | + | package lifecycle | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "bufio" | |||
| 5 | + | "context" | |||
| 6 | + | "errors" | |||
| 7 | + | "fmt" | |||
| 8 | + | "io" | |||
| 9 | + | "os" | |||
| 10 | + | "os/exec" | |||
| 11 | + | "path/filepath" | |||
| 12 | + | "testing" | |||
| 13 | + | "time" | |||
| 14 | + | ||||
| 15 | + | "github.com/stretchr/testify/require" | |||
| 16 | + | ) | |||
| 17 | + | ||||
| 18 | + | func TestLifecycleLockProcessHelper(t *testing.T) { | |||
| 19 | + | if os.Getenv("VEV_LIFECYCLE_LOCK_HELPER") != "1" { | |||
| 20 | + | t.Skip("helper process") | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | owner, err := TryAcquire(os.Getenv("VEV_LIFECYCLE_LOCK_DIR")) | |||
| 24 | + | require.NoError(t, err) | |||
| 25 | + | defer func() { require.NoError(t, owner.Release()) }() | |||
| 26 | + | ||||
| 27 | + | _, err = fmt.Fprintln(os.Stdout, "locked") | |||
| 28 | + | require.NoError(t, err) | |||
| 29 | + | _, err = io.Copy(io.Discard, os.Stdin) | |||
| 30 | + | require.NoError(t, err) | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | func TestLifecycleLock(t *testing.T) { | |||
| 34 | + | t.Run("process contention and reacquire", testLifecycleLockProcessContentionAndReacquire) | |||
| 35 | + | t.Run("acquire cancellation", testLifecycleLockAcquireCancellation) | |||
| 36 | + | t.Run("unavailable runtime directory", testLifecycleLockUnavailableRuntimeDirectory) | |||
| 37 | + | } | |||
| 38 | + | ||||
| 39 | + | func testLifecycleLockProcessContentionAndReacquire(t *testing.T) { | |||
| 40 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |||
| 41 | + | cmd := exec.Command(os.Args[0], "-test.run=^TestLifecycleLockProcessHelper$") | |||
| 42 | + | cmd.Env = append(os.Environ(), | |||
| 43 | + | "VEV_LIFECYCLE_LOCK_HELPER=1", | |||
| 44 | + | "VEV_LIFECYCLE_LOCK_DIR="+runtimeDir, | |||
| 45 | + | ) | |||
| 46 | + | stdin, err := cmd.StdinPipe() | |||
| 47 | + | require.NoError(t, err) | |||
| 48 | + | stdout, err := cmd.StdoutPipe() | |||
| 49 | + | require.NoError(t, err) | |||
| 50 | + | cmd.Stderr = os.Stderr | |||
| 51 | + | require.NoError(t, cmd.Start()) | |||
| 52 | + | ||||
| 53 | + | line, err := bufio.NewReader(stdout).ReadString('\n') | |||
| 54 | + | require.NoError(t, err) | |||
| 55 | + | require.Equal(t, "locked\n", line) | |||
| 56 | + | ||||
| 57 | + | _, err = TryAcquire(runtimeDir) | |||
| 58 | + | require.ErrorIs(t, err, ErrBusy) | |||
| 59 | + | ||||
| 60 | + | require.NoError(t, stdin.Close()) | |||
| 61 | + | require.NoError(t, cmd.Wait()) | |||
| 62 | + | ||||
| 63 | + | owner, err := TryAcquire(runtimeDir) | |||
| 64 | + | require.NoError(t, err) | |||
| 65 | + | require.NoError(t, owner.Release()) | |||
| 66 | + | require.NoError(t, owner.Release(), "release is idempotent") | |||
| 67 | + | } | |||
| 68 | + | ||||
| 69 | + | func testLifecycleLockAcquireCancellation(t *testing.T) { | |||
| 70 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |||
| 71 | + | owner, err := TryAcquire(runtimeDir) | |||
| 72 | + | require.NoError(t, err) | |||
| 73 | + | defer func() { require.NoError(t, owner.Release()) }() | |||
| 74 | + | ||||
| 75 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 76 | + | cancel() | |||
| 77 | + | ||||
| 78 | + | _, err = Acquire(ctx, runtimeDir, time.Millisecond) | |||
| 79 | + | require.ErrorIs(t, err, context.Canceled) | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | func testLifecycleLockUnavailableRuntimeDirectory(t *testing.T) { | |||
| 83 | + | runtimePath := t.TempDir() + "/not-a-directory" | |||
| 84 | + | require.NoError(t, os.WriteFile(runtimePath, []byte("occupied"), 0o600)) | |||
| 85 | + | ||||
| 86 | + | _, err := TryAcquire(runtimePath) | |||
| 87 | + | require.Error(t, err) | |||
| 88 | + | require.False(t, errors.Is(err, ErrBusy)) | |||
| 89 | + | } | |||
M internal/app/run.go +87 -27
| @@ -32,6 +32,7 @@ import ( | |||
| 32 | 32 | "github.com/bnema/vev/internal/adapters/config" | |
| 33 | 33 | "github.com/bnema/vev/internal/adapters/dgram" | |
| 34 | 34 | "github.com/bnema/vev/internal/adapters/ipc" | |
| 35 | + | "github.com/bnema/vev/internal/adapters/lifecycle" | |
| 35 | 36 | "github.com/bnema/vev/internal/adapters/noticefile" | |
| 36 | 37 | "github.com/bnema/vev/internal/adapters/observability" | |
| 37 | 38 | "github.com/bnema/vev/internal/adapters/pty" | |
| @@ -50,6 +51,7 @@ import ( | |||
| 50 | 51 | "github.com/bnema/vev/internal/usecase/daemon" | |
| 51 | 52 | pdgram "github.com/bnema/vev/pkg/dgram" | |
| 52 | 53 | "github.com/bnema/vev/pkg/kv" | |
| 54 | + | "github.com/bnema/vev/pkg/safedir" | |
| 53 | 55 | ) | |
| 54 | 56 | ||
| 55 | 57 | // cmdKind identifies which sub-command the CLI parsed. | |
| @@ -410,11 +412,58 @@ func pprofAddrIsLoopback(addr string) bool { | |||
| 410 | 412 | return ip != nil && ip.IsLoopback() | |
| 411 | 413 | } | |
| 412 | 414 | ||
| 415 | + | type ownedDaemonStart func(context.Context) error | |
| 416 | + | ||
| 417 | + | type lifecycleOwnership interface { | |
| 418 | + | Release() error | |
| 419 | + | } | |
| 420 | + | ||
| 421 | + | type lifecycleStartupDeps struct { | |
| 422 | + | ensurePrivate func(string) error | |
| 423 | + | acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error) | |
| 424 | + | } | |
| 425 | + | ||
| 426 | + | const lifecycleAcquireRetry = 25 * time.Millisecond | |
| 427 | + | ||
| 428 | + | func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart) error { | |
| 429 | + | return runWithLifecycleOwnerDeps(ctx, runtimeDir, stateDir, start, lifecycleStartupDeps{ | |
| 430 | + | ensurePrivate: safedir.EnsurePrivate, | |
| 431 | + | acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) { | |
| 432 | + | return lifecycle.Acquire(ctx, runtimeDir, retry) | |
| 433 | + | }, | |
| 434 | + | }) | |
| 435 | + | } | |
| 436 | + | ||
| 437 | + | func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) { | |
| 438 | + | if err := deps.ensurePrivate(runtimeDir); err != nil { | |
| 439 | + | return fmt.Errorf("vev: secure runtime directory: %w", err) | |
| 440 | + | } | |
| 441 | + | if err := deps.ensurePrivate(stateDir); err != nil { | |
| 442 | + | return fmt.Errorf("vev: secure state directory: %w", err) | |
| 443 | + | } | |
| 444 | + | owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry) | |
| 445 | + | if err != nil { | |
| 446 | + | return fmt.Errorf("vev: acquire lifecycle ownership: %w", err) | |
| 447 | + | } | |
| 448 | + | defer func() { | |
| 449 | + | retErr = errors.Join(retErr, owner.Release()) | |
| 450 | + | }() | |
| 451 | + | if start == nil { | |
| 452 | + | return errors.New("vev: nil owned daemon startup") | |
| 453 | + | } | |
| 454 | + | return start(ctx) | |
| 455 | + | } | |
| 456 | + | ||
| 413 | 457 | // runDaemon runs the daemon in the foreground (the hidden --daemon path, | |
| 414 | - | // entered by an auto-spawned child): it sets up logging, binds the socket, | |
| 415 | - | // constructs the daemon, and serves until the last session exits or a | |
| 416 | - | // termination signal arrives (graceful shutdown notifies attached clients). | |
| 417 | - | func runDaemon() (retErr error) { | |
| 458 | + | // entered by an auto-spawned child) while holding exclusive lifecycle | |
| 459 | + | // ownership through complete teardown. | |
| 460 | + | func runDaemon() error { | |
| 461 | + | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) | |
| 462 | + | defer stop() | |
| 463 | + | return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned) | |
| 464 | + | } | |
| 465 | + | ||
| 466 | + | func runDaemonOwned(ctx context.Context) (retErr error) { | |
| 418 | 467 | log, logCloser, err := configureLogging(logging.Daemon, true) | |
| 419 | 468 | if err != nil { | |
| 420 | 469 | return err | |
| @@ -429,17 +478,6 @@ func runDaemon() (retErr error) { | |||
| 429 | 478 | if observerCloser != nil { | |
| 430 | 479 | defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }() | |
| 431 | 480 | } | |
| 432 | - | // Every accepted local connection shares this process's serialized sink. | |
| 433 | - | ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) | |
| 434 | - | if err != nil { | |
| 435 | - | log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err) | |
| 436 | - | return fmt.Errorf("vev: daemon listen: %w", err) | |
| 437 | - | } | |
| 438 | - | defer func() { _ = ln.Close() }() | |
| 439 | - | ||
| 440 | - | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) | |
| 441 | - | defer stop() | |
| 442 | - | ||
| 443 | 481 | if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" { | |
| 444 | 482 | if !pprofAddrIsLoopback(addr) { | |
| 445 | 483 | log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr) | |
| @@ -452,7 +490,6 @@ func runDaemon() (retErr error) { | |||
| 452 | 490 | log.Info("pprof enabled", "addr", addr) | |
| 453 | 491 | } | |
| 454 | 492 | ||
| 455 | - | log.Info("daemon starting", "socket", ln.Addr()) | |
| 456 | 493 | daemonOpts := []daemon.Option(nil) | |
| 457 | 494 | if observer != nil { | |
| 458 | 495 | daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer)) | |
| @@ -471,19 +508,42 @@ func runDaemon() (retErr error) { | |||
| 471 | 508 | daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository)) | |
| 472 | 509 | daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir()))) | |
| 473 | 510 | storePath := persist.StorePath(platform.StateDir()) | |
| 474 | - | var storeErr error | |
| 475 | - | if store, err := kv.Open(storePath); err != nil { | |
| 476 | - | log.Warn("opening session store failed; persistence disabled", "path", storePath, "err", err) | |
| 477 | - | storeErr = err | |
| 478 | - | } else { | |
| 479 | - | log.Info("session persistence enabled", "path", storePath) | |
| 480 | - | daemonOpts = append(daemonOpts, daemon.WithStore(store)) | |
| 511 | + | catalogueCandidate := false | |
| 512 | + | for _, candidate := range []string{storePath, storePath + ".next", storePath + ".prev"} { | |
| 513 | + | if _, statErr := os.Stat(candidate); statErr == nil { | |
| 514 | + | catalogueCandidate = true | |
| 515 | + | } else if !errors.Is(statErr, os.ErrNotExist) { | |
| 516 | + | return fmt.Errorf("vev: inspect durable session catalogue %s: %w", candidate, statErr) | |
| 517 | + | } | |
| 481 | 518 | } | |
| 482 | - | d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...) | |
| 483 | - | if storeErr != nil { | |
| 484 | - | d.NotifyGlobal(domain.NoticeWarn, domain.NoticePersistDisabled, | |
| 485 | - | "session persistence is disabled; sessions will not survive daemon restarts", storeErr) | |
| 519 | + | if !catalogueCandidate { | |
| 520 | + | return fmt.Errorf("vev: durable session catalogue unavailable at %s", storePath) | |
| 486 | 521 | } | |
| 522 | + | store, err := kv.Open(storePath) | |
| 523 | + | if err != nil { | |
| 524 | + | log.Error("opening session store failed", "path", storePath, "err", err) | |
| 525 | + | return fmt.Errorf("vev: open durable session catalogue %s: %w", storePath, err) | |
| 526 | + | } | |
| 527 | + | if _, err := persist.New(store).LoadAll(); err != nil { | |
| 528 | + | closeErr := store.Close() | |
| 529 | + | log.Error("validating session catalogue failed", "path", storePath, "err", err) | |
| 530 | + | return errors.Join(fmt.Errorf("vev: validate durable session catalogue %s: %w", storePath, err), closeErr) | |
| 531 | + | } | |
| 532 | + | log.Info("session persistence enabled", "path", storePath) | |
| 533 | + | daemonOpts = append(daemonOpts, daemon.WithStore(store)) | |
| 534 | + | ||
| 535 | + | // Socket publication follows lifecycle acquisition and strict catalogue | |
| 536 | + | // validation. Any earlier failure leaves no apparently healthy daemon. | |
| 537 | + | ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) | |
| 538 | + | if err != nil { | |
| 539 | + | closeErr := store.Close() | |
| 540 | + | log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err) | |
| 541 | + | return errors.Join(fmt.Errorf("vev: daemon listen: %w", err), closeErr) | |
| 542 | + | } | |
| 543 | + | defer func() { _ = ln.Close() }() | |
| 544 | + | log.Info("daemon starting", "socket", ln.Addr()) | |
| 545 | + | ||
| 546 | + | d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...) | |
| 487 | 547 | watchCtx, stopWatch := context.WithCancel(ctx) | |
| 488 | 548 | defer stopWatch() | |
| 489 | 549 | go func() { | |
| @@ -32,6 +32,7 @@ import ( | |||||
| 32 | "github.com/bnema/vev/internal/adapters/config" | 32 | "github.com/bnema/vev/internal/adapters/config" | ||
| 33 | "github.com/bnema/vev/internal/adapters/dgram" | 33 | "github.com/bnema/vev/internal/adapters/dgram" | ||
| 34 | "github.com/bnema/vev/internal/adapters/ipc" | 34 | "github.com/bnema/vev/internal/adapters/ipc" | ||
| 35 | + | "github.com/bnema/vev/internal/adapters/lifecycle" | |||
| 35 | "github.com/bnema/vev/internal/adapters/noticefile" | 36 | "github.com/bnema/vev/internal/adapters/noticefile" | ||
| 36 | "github.com/bnema/vev/internal/adapters/observability" | 37 | "github.com/bnema/vev/internal/adapters/observability" | ||
| 37 | "github.com/bnema/vev/internal/adapters/pty" | 38 | "github.com/bnema/vev/internal/adapters/pty" | ||
| @@ -50,6 +51,7 @@ import ( | |||||
| 50 | "github.com/bnema/vev/internal/usecase/daemon" | 51 | "github.com/bnema/vev/internal/usecase/daemon" | ||
| 51 | pdgram "github.com/bnema/vev/pkg/dgram" | 52 | pdgram "github.com/bnema/vev/pkg/dgram" | ||
| 52 | "github.com/bnema/vev/pkg/kv" | 53 | "github.com/bnema/vev/pkg/kv" | ||
| 54 | + | "github.com/bnema/vev/pkg/safedir" | |||
| 53 | ) | 55 | ) | ||
| 54 | 56 | ||||
| 55 | // cmdKind identifies which sub-command the CLI parsed. | 57 | // cmdKind identifies which sub-command the CLI parsed. | ||
| @@ -410,11 +412,58 @@ func pprofAddrIsLoopback(addr string) bool { | |||||
| 410 | return ip != nil && ip.IsLoopback() | 412 | return ip != nil && ip.IsLoopback() | ||
| 411 | } | 413 | } | ||
| 412 | 414 | ||||
| 415 | + | type ownedDaemonStart func(context.Context) error | |||
| 416 | + | ||||
| 417 | + | type lifecycleOwnership interface { | |||
| 418 | + | Release() error | |||
| 419 | + | } | |||
| 420 | + | ||||
| 421 | + | type lifecycleStartupDeps struct { | |||
| 422 | + | ensurePrivate func(string) error | |||
| 423 | + | acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error) | |||
| 424 | + | } | |||
| 425 | + | ||||
| 426 | + | const lifecycleAcquireRetry = 25 * time.Millisecond | |||
| 427 | + | ||||
| 428 | + | func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart) error { | |||
| 429 | + | return runWithLifecycleOwnerDeps(ctx, runtimeDir, stateDir, start, lifecycleStartupDeps{ | |||
| 430 | + | ensurePrivate: safedir.EnsurePrivate, | |||
| 431 | + | acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) { | |||
| 432 | + | return lifecycle.Acquire(ctx, runtimeDir, retry) | |||
| 433 | + | }, | |||
| 434 | + | }) | |||
| 435 | + | } | |||
| 436 | + | ||||
| 437 | + | func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) { | |||
| 438 | + | if err := deps.ensurePrivate(runtimeDir); err != nil { | |||
| 439 | + | return fmt.Errorf("vev: secure runtime directory: %w", err) | |||
| 440 | + | } | |||
| 441 | + | if err := deps.ensurePrivate(stateDir); err != nil { | |||
| 442 | + | return fmt.Errorf("vev: secure state directory: %w", err) | |||
| 443 | + | } | |||
| 444 | + | owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry) | |||
| 445 | + | if err != nil { | |||
| 446 | + | return fmt.Errorf("vev: acquire lifecycle ownership: %w", err) | |||
| 447 | + | } | |||
| 448 | + | defer func() { | |||
| 449 | + | retErr = errors.Join(retErr, owner.Release()) | |||
| 450 | + | }() | |||
| 451 | + | if start == nil { | |||
| 452 | + | return errors.New("vev: nil owned daemon startup") | |||
| 453 | + | } | |||
| 454 | + | return start(ctx) | |||
| 455 | + | } | |||
| 456 | + | ||||
| 413 | // runDaemon runs the daemon in the foreground (the hidden --daemon path, | 457 | // runDaemon runs the daemon in the foreground (the hidden --daemon path, | ||
| 414 | - | // entered by an auto-spawned child): it sets up logging, binds the socket, | 458 | + | // entered by an auto-spawned child) while holding exclusive lifecycle |
| 415 | - | // constructs the daemon, and serves until the last session exits or a | 459 | + | // ownership through complete teardown. |
| 416 | - | // termination signal arrives (graceful shutdown notifies attached clients). | 460 | + | func runDaemon() error { |
| 417 | - | func runDaemon() (retErr error) { | 461 | + | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) |
| 462 | + | defer stop() | |||
| 463 | + | return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned) | |||
| 464 | + | } | |||
| 465 | + | ||||
| 466 | + | func runDaemonOwned(ctx context.Context) (retErr error) { | |||
| 418 | log, logCloser, err := configureLogging(logging.Daemon, true) | 467 | log, logCloser, err := configureLogging(logging.Daemon, true) | ||
| 419 | if err != nil { | 468 | if err != nil { | ||
| 420 | return err | 469 | return err | ||
| @@ -429,17 +478,6 @@ func runDaemon() (retErr error) { | |||||
| 429 | if observerCloser != nil { | 478 | if observerCloser != nil { | ||
| 430 | defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }() | 479 | defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }() | ||
| 431 | } | 480 | } | ||
| 432 | - | // Every accepted local connection shares this process's serialized sink. | |||
| 433 | - | ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) | |||
| 434 | - | if err != nil { | |||
| 435 | - | log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err) | |||
| 436 | - | return fmt.Errorf("vev: daemon listen: %w", err) | |||
| 437 | - | } | |||
| 438 | - | defer func() { _ = ln.Close() }() | |||
| 439 | - | ||||
| 440 | - | ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) | |||
| 441 | - | defer stop() | |||
| 442 | - | ||||
| 443 | if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" { | 481 | if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" { | ||
| 444 | if !pprofAddrIsLoopback(addr) { | 482 | if !pprofAddrIsLoopback(addr) { | ||
| 445 | log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr) | 483 | log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr) | ||
| @@ -452,7 +490,6 @@ func runDaemon() (retErr error) { | |||||
| 452 | log.Info("pprof enabled", "addr", addr) | 490 | log.Info("pprof enabled", "addr", addr) | ||
| 453 | } | 491 | } | ||
| 454 | 492 | ||||
| 455 | - | log.Info("daemon starting", "socket", ln.Addr()) | |||
| 456 | daemonOpts := []daemon.Option(nil) | 493 | daemonOpts := []daemon.Option(nil) | ||
| 457 | if observer != nil { | 494 | if observer != nil { | ||
| 458 | daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer)) | 495 | daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer)) | ||
| @@ -471,19 +508,42 @@ func runDaemon() (retErr error) { | |||||
| 471 | daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository)) | 508 | daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository)) | ||
| 472 | daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir()))) | 509 | daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir()))) | ||
| 473 | storePath := persist.StorePath(platform.StateDir()) | 510 | storePath := persist.StorePath(platform.StateDir()) | ||
| 474 | - | var storeErr error | 511 | + | catalogueCandidate := false |
| 475 | - | if store, err := kv.Open(storePath); err != nil { | 512 | + | for _, candidate := range []string{storePath, storePath + ".next", storePath + ".prev"} { |
| 476 | - | log.Warn("opening session store failed; persistence disabled", "path", storePath, "err", err) | 513 | + | if _, statErr := os.Stat(candidate); statErr == nil { |
| 477 | - | storeErr = err | 514 | + | catalogueCandidate = true |
| 478 | - | } else { | 515 | + | } else if !errors.Is(statErr, os.ErrNotExist) { |
| 479 | - | log.Info("session persistence enabled", "path", storePath) | 516 | + | return fmt.Errorf("vev: inspect durable session catalogue %s: %w", candidate, statErr) |
| 480 | - | daemonOpts = append(daemonOpts, daemon.WithStore(store)) | 517 | + | } |
| 481 | } | 518 | } | ||
| 482 | - | d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...) | 519 | + | if !catalogueCandidate { |
| 483 | - | if storeErr != nil { | 520 | + | return fmt.Errorf("vev: durable session catalogue unavailable at %s", storePath) |
| 484 | - | d.NotifyGlobal(domain.NoticeWarn, domain.NoticePersistDisabled, | |||
| 485 | - | "session persistence is disabled; sessions will not survive daemon restarts", storeErr) | |||
| 486 | } | 521 | } | ||
| 522 | + | store, err := kv.Open(storePath) | |||
| 523 | + | if err != nil { | |||
| 524 | + | log.Error("opening session store failed", "path", storePath, "err", err) | |||
| 525 | + | return fmt.Errorf("vev: open durable session catalogue %s: %w", storePath, err) | |||
| 526 | + | } | |||
| 527 | + | if _, err := persist.New(store).LoadAll(); err != nil { | |||
| 528 | + | closeErr := store.Close() | |||
| 529 | + | log.Error("validating session catalogue failed", "path", storePath, "err", err) | |||
| 530 | + | return errors.Join(fmt.Errorf("vev: validate durable session catalogue %s: %w", storePath, err), closeErr) | |||
| 531 | + | } | |||
| 532 | + | log.Info("session persistence enabled", "path", storePath) | |||
| 533 | + | daemonOpts = append(daemonOpts, daemon.WithStore(store)) | |||
| 534 | + | ||||
| 535 | + | // Socket publication follows lifecycle acquisition and strict catalogue | |||
| 536 | + | // validation. Any earlier failure leaves no apparently healthy daemon. | |||
| 537 | + | ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) | |||
| 538 | + | if err != nil { | |||
| 539 | + | closeErr := store.Close() | |||
| 540 | + | log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err) | |||
| 541 | + | return errors.Join(fmt.Errorf("vev: daemon listen: %w", err), closeErr) | |||
| 542 | + | } | |||
| 543 | + | defer func() { _ = ln.Close() }() | |||
| 544 | + | log.Info("daemon starting", "socket", ln.Addr()) | |||
| 545 | + | ||||
| 546 | + | d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...) | |||
| 487 | watchCtx, stopWatch := context.WithCancel(ctx) | 547 | watchCtx, stopWatch := context.WithCancel(ctx) | ||
| 488 | defer stopWatch() | 548 | defer stopWatch() | ||
| 489 | go func() { | 549 | go func() { | ||
M internal/app/run_test.go +136 -0
| @@ -13,6 +13,7 @@ import ( | |||
| 13 | 13 | "testing" | |
| 14 | 14 | "time" | |
| 15 | 15 | ||
| 16 | + | "github.com/bnema/vev/internal/adapters/lifecycle" | |
| 16 | 17 | "github.com/bnema/vev/internal/adapters/snapshot" | |
| 17 | 18 | "github.com/bnema/vev/internal/domain" | |
| 18 | 19 | "github.com/bnema/vev/internal/persist" | |
| @@ -20,10 +21,145 @@ import ( | |||
| 20 | 21 | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 21 | 22 | "github.com/bnema/vev/internal/usecase/client" | |
| 22 | 23 | "github.com/bnema/vev/internal/usecase/confirm" | |
| 24 | + | "github.com/bnema/vev/pkg/kv" | |
| 23 | 25 | "github.com/stretchr/testify/mock" | |
| 24 | 26 | "github.com/stretchr/testify/require" | |
| 25 | 27 | ) | |
| 26 | 28 | ||
| 29 | + | type fakeLifecycleOwnership struct { | |
| 30 | + | release func() error | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | func (o fakeLifecycleOwnership) Release() error { return o.release() } | |
| 34 | + | ||
| 35 | + | func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) { | |
| 36 | + | t.Run("ownership prefix is exact", func(t *testing.T) { | |
| 37 | + | var events []string | |
| 38 | + | owner := fakeLifecycleOwnership{release: func() error { | |
| 39 | + | events = append(events, "unlock") | |
| 40 | + | return nil | |
| 41 | + | }} | |
| 42 | + | deps := lifecycleStartupDeps{ | |
| 43 | + | ensurePrivate: func(path string) error { | |
| 44 | + | if path == "runtime" { | |
| 45 | + | events = append(events, "ensure-runtime") | |
| 46 | + | } else { | |
| 47 | + | events = append(events, "ensure-state") | |
| 48 | + | } | |
| 49 | + | return nil | |
| 50 | + | }, | |
| 51 | + | acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) { | |
| 52 | + | events = append(events, "lock") | |
| 53 | + | return owner, nil | |
| 54 | + | }, | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | err := runWithLifecycleOwnerDeps(context.Background(), "runtime", "state", func(context.Context) error { | |
| 58 | + | events = append(events, "durable-open", "listen", "serve") | |
| 59 | + | return nil | |
| 60 | + | }, deps) | |
| 61 | + | require.NoError(t, err) | |
| 62 | + | require.Equal(t, []string{"ensure-runtime", "ensure-state", "lock"}, events[:3]) | |
| 63 | + | require.Equal(t, []string{"ensure-runtime", "ensure-state", "lock", "durable-open", "listen", "serve", "unlock"}, events) | |
| 64 | + | }) | |
| 65 | + | ||
| 66 | + | t.Run("callback observes held lock and failure releases it", func(t *testing.T) { | |
| 67 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |
| 68 | + | stateDir := filepath.Join(t.TempDir(), "state") | |
| 69 | + | startErr := errors.New("catalogue corrupt") | |
| 70 | + | ||
| 71 | + | err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, func(context.Context) error { | |
| 72 | + | _, lockErr := lifecycle.TryAcquire(runtimeDir) | |
| 73 | + | require.ErrorIs(t, lockErr, lifecycle.ErrBusy) | |
| 74 | + | return startErr | |
| 75 | + | }) | |
| 76 | + | require.ErrorIs(t, err, startErr) | |
| 77 | + | ||
| 78 | + | owner, err := lifecycle.TryAcquire(runtimeDir) | |
| 79 | + | require.NoError(t, err, "callback failure must release lifecycle ownership") | |
| 80 | + | require.NoError(t, owner.Release()) | |
| 81 | + | }) | |
| 82 | + | ||
| 83 | + | t.Run("busy and unavailable ownership fail before startup", func(t *testing.T) { | |
| 84 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |
| 85 | + | stateDir := filepath.Join(t.TempDir(), "state") | |
| 86 | + | owner, err := lifecycle.TryAcquire(runtimeDir) | |
| 87 | + | require.NoError(t, err) | |
| 88 | + | defer func() { require.NoError(t, owner.Release()) }() | |
| 89 | + | ||
| 90 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 91 | + | cancel() | |
| 92 | + | started := false | |
| 93 | + | err = runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(context.Context) error { | |
| 94 | + | started = true | |
| 95 | + | return nil | |
| 96 | + | }) | |
| 97 | + | require.ErrorIs(t, err, context.Canceled) | |
| 98 | + | require.False(t, started) | |
| 99 | + | ||
| 100 | + | badState := filepath.Join(t.TempDir(), "state-file") | |
| 101 | + | require.NoError(t, os.WriteFile(badState, []byte("unavailable"), 0o600)) | |
| 102 | + | err = runWithLifecycleOwner(context.Background(), filepath.Join(t.TempDir(), "other-runtime"), badState, func(context.Context) error { | |
| 103 | + | started = true | |
| 104 | + | return nil | |
| 105 | + | }) | |
| 106 | + | require.Error(t, err) | |
| 107 | + | require.False(t, started) | |
| 108 | + | }) | |
| 109 | + | ||
| 110 | + | t.Run("corrupt catalogue fails before socket publication", func(t *testing.T) { | |
| 111 | + | runtimeRoot, stateRoot := t.TempDir(), t.TempDir() | |
| 112 | + | runtimeDir := filepath.Join(runtimeRoot, "vev") | |
| 113 | + | stateDir := filepath.Join(stateRoot, "vev") | |
| 114 | + | require.NoError(t, os.Mkdir(stateDir, 0o700)) | |
| 115 | + | store, err := kv.Open(persist.StorePath(stateDir)) | |
| 116 | + | require.NoError(t, err) | |
| 117 | + | require.NoError(t, store.Set([]byte("work"), []byte("malformed catalogue value"))) | |
| 118 | + | require.NoError(t, store.Close()) | |
| 119 | + | ||
| 120 | + | t.Setenv("XDG_RUNTIME_DIR", runtimeRoot) | |
| 121 | + | t.Setenv("XDG_STATE_HOME", stateRoot) | |
| 122 | + | err = runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, runDaemonOwned) | |
| 123 | + | require.Error(t, err) | |
| 124 | + | require.Contains(t, err.Error(), "validate durable session catalogue") | |
| 125 | + | _, statErr := os.Stat(filepath.Join(runtimeDir, "daemon.sock")) | |
| 126 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |
| 127 | + | }) | |
| 128 | + | ||
| 129 | + | t.Run("unavailable catalogue fails without creating state or socket", func(t *testing.T) { | |
| 130 | + | runtimeRoot, stateRoot := t.TempDir(), t.TempDir() | |
| 131 | + | runtimeDir := filepath.Join(runtimeRoot, "vev") | |
| 132 | + | stateDir := filepath.Join(stateRoot, "vev") | |
| 133 | + | t.Setenv("XDG_RUNTIME_DIR", runtimeRoot) | |
| 134 | + | t.Setenv("XDG_STATE_HOME", stateRoot) | |
| 135 | + | ||
| 136 | + | err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, runDaemonOwned) | |
| 137 | + | require.Error(t, err) | |
| 138 | + | require.Contains(t, err.Error(), "durable session catalogue unavailable") | |
| 139 | + | _, statErr := os.Stat(persist.StorePath(stateDir)) | |
| 140 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |
| 141 | + | _, statErr = os.Stat(filepath.Join(runtimeDir, "daemon.sock")) | |
| 142 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |
| 143 | + | }) | |
| 144 | + | ||
| 145 | + | t.Run("callback and release errors are joined", func(t *testing.T) { | |
| 146 | + | startErr := errors.New("catalogue unavailable") | |
| 147 | + | releaseErr := errors.New("unlock failed") | |
| 148 | + | deps := lifecycleStartupDeps{ | |
| 149 | + | ensurePrivate: func(string) error { return nil }, | |
| 150 | + | acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) { | |
| 151 | + | return fakeLifecycleOwnership{release: func() error { return releaseErr }}, nil | |
| 152 | + | }, | |
| 153 | + | } | |
| 154 | + | ||
| 155 | + | err := runWithLifecycleOwnerDeps(context.Background(), "runtime", "state", func(context.Context) error { | |
| 156 | + | return startErr | |
| 157 | + | }, deps) | |
| 158 | + | require.ErrorIs(t, err, startErr) | |
| 159 | + | require.ErrorIs(t, err, releaseErr) | |
| 160 | + | }) | |
| 161 | + | } | |
| 162 | + | ||
| 27 | 163 | // scriptRecv makes a transport yield frames in order, then wait for done | |
| 28 | 164 | // before returning EOF. The shared done channel coordinates proxy readers | |
| 29 | 165 | // without relying on scheduling or sleeps. | |
| @@ -13,6 +13,7 @@ import ( | |||||
| 13 | "testing" | 13 | "testing" | ||
| 14 | "time" | 14 | "time" | ||
| 15 | 15 | ||||
| 16 | + | "github.com/bnema/vev/internal/adapters/lifecycle" | |||
| 16 | "github.com/bnema/vev/internal/adapters/snapshot" | 17 | "github.com/bnema/vev/internal/adapters/snapshot" | ||
| 17 | "github.com/bnema/vev/internal/domain" | 18 | "github.com/bnema/vev/internal/domain" | ||
| 18 | "github.com/bnema/vev/internal/persist" | 19 | "github.com/bnema/vev/internal/persist" | ||
| @@ -20,10 +21,145 @@ import ( | |||||
| 20 | portsmocks "github.com/bnema/vev/internal/ports/mocks" | 21 | portsmocks "github.com/bnema/vev/internal/ports/mocks" | ||
| 21 | "github.com/bnema/vev/internal/usecase/client" | 22 | "github.com/bnema/vev/internal/usecase/client" | ||
| 22 | "github.com/bnema/vev/internal/usecase/confirm" | 23 | "github.com/bnema/vev/internal/usecase/confirm" | ||
| 24 | + | "github.com/bnema/vev/pkg/kv" | |||
| 23 | "github.com/stretchr/testify/mock" | 25 | "github.com/stretchr/testify/mock" | ||
| 24 | "github.com/stretchr/testify/require" | 26 | "github.com/stretchr/testify/require" | ||
| 25 | ) | 27 | ) | ||
| 26 | 28 | ||||
| 29 | + | type fakeLifecycleOwnership struct { | |||
| 30 | + | release func() error | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | func (o fakeLifecycleOwnership) Release() error { return o.release() } | |||
| 34 | + | ||||
| 35 | + | func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) { | |||
| 36 | + | t.Run("ownership prefix is exact", func(t *testing.T) { | |||
| 37 | + | var events []string | |||
| 38 | + | owner := fakeLifecycleOwnership{release: func() error { | |||
| 39 | + | events = append(events, "unlock") | |||
| 40 | + | return nil | |||
| 41 | + | }} | |||
| 42 | + | deps := lifecycleStartupDeps{ | |||
| 43 | + | ensurePrivate: func(path string) error { | |||
| 44 | + | if path == "runtime" { | |||
| 45 | + | events = append(events, "ensure-runtime") | |||
| 46 | + | } else { | |||
| 47 | + | events = append(events, "ensure-state") | |||
| 48 | + | } | |||
| 49 | + | return nil | |||
| 50 | + | }, | |||
| 51 | + | acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) { | |||
| 52 | + | events = append(events, "lock") | |||
| 53 | + | return owner, nil | |||
| 54 | + | }, | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | err := runWithLifecycleOwnerDeps(context.Background(), "runtime", "state", func(context.Context) error { | |||
| 58 | + | events = append(events, "durable-open", "listen", "serve") | |||
| 59 | + | return nil | |||
| 60 | + | }, deps) | |||
| 61 | + | require.NoError(t, err) | |||
| 62 | + | require.Equal(t, []string{"ensure-runtime", "ensure-state", "lock"}, events[:3]) | |||
| 63 | + | require.Equal(t, []string{"ensure-runtime", "ensure-state", "lock", "durable-open", "listen", "serve", "unlock"}, events) | |||
| 64 | + | }) | |||
| 65 | + | ||||
| 66 | + | t.Run("callback observes held lock and failure releases it", func(t *testing.T) { | |||
| 67 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |||
| 68 | + | stateDir := filepath.Join(t.TempDir(), "state") | |||
| 69 | + | startErr := errors.New("catalogue corrupt") | |||
| 70 | + | ||||
| 71 | + | err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, func(context.Context) error { | |||
| 72 | + | _, lockErr := lifecycle.TryAcquire(runtimeDir) | |||
| 73 | + | require.ErrorIs(t, lockErr, lifecycle.ErrBusy) | |||
| 74 | + | return startErr | |||
| 75 | + | }) | |||
| 76 | + | require.ErrorIs(t, err, startErr) | |||
| 77 | + | ||||
| 78 | + | owner, err := lifecycle.TryAcquire(runtimeDir) | |||
| 79 | + | require.NoError(t, err, "callback failure must release lifecycle ownership") | |||
| 80 | + | require.NoError(t, owner.Release()) | |||
| 81 | + | }) | |||
| 82 | + | ||||
| 83 | + | t.Run("busy and unavailable ownership fail before startup", func(t *testing.T) { | |||
| 84 | + | runtimeDir := filepath.Join(t.TempDir(), "runtime") | |||
| 85 | + | stateDir := filepath.Join(t.TempDir(), "state") | |||
| 86 | + | owner, err := lifecycle.TryAcquire(runtimeDir) | |||
| 87 | + | require.NoError(t, err) | |||
| 88 | + | defer func() { require.NoError(t, owner.Release()) }() | |||
| 89 | + | ||||
| 90 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 91 | + | cancel() | |||
| 92 | + | started := false | |||
| 93 | + | err = runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(context.Context) error { | |||
| 94 | + | started = true | |||
| 95 | + | return nil | |||
| 96 | + | }) | |||
| 97 | + | require.ErrorIs(t, err, context.Canceled) | |||
| 98 | + | require.False(t, started) | |||
| 99 | + | ||||
| 100 | + | badState := filepath.Join(t.TempDir(), "state-file") | |||
| 101 | + | require.NoError(t, os.WriteFile(badState, []byte("unavailable"), 0o600)) | |||
| 102 | + | err = runWithLifecycleOwner(context.Background(), filepath.Join(t.TempDir(), "other-runtime"), badState, func(context.Context) error { | |||
| 103 | + | started = true | |||
| 104 | + | return nil | |||
| 105 | + | }) | |||
| 106 | + | require.Error(t, err) | |||
| 107 | + | require.False(t, started) | |||
| 108 | + | }) | |||
| 109 | + | ||||
| 110 | + | t.Run("corrupt catalogue fails before socket publication", func(t *testing.T) { | |||
| 111 | + | runtimeRoot, stateRoot := t.TempDir(), t.TempDir() | |||
| 112 | + | runtimeDir := filepath.Join(runtimeRoot, "vev") | |||
| 113 | + | stateDir := filepath.Join(stateRoot, "vev") | |||
| 114 | + | require.NoError(t, os.Mkdir(stateDir, 0o700)) | |||
| 115 | + | store, err := kv.Open(persist.StorePath(stateDir)) | |||
| 116 | + | require.NoError(t, err) | |||
| 117 | + | require.NoError(t, store.Set([]byte("work"), []byte("malformed catalogue value"))) | |||
| 118 | + | require.NoError(t, store.Close()) | |||
| 119 | + | ||||
| 120 | + | t.Setenv("XDG_RUNTIME_DIR", runtimeRoot) | |||
| 121 | + | t.Setenv("XDG_STATE_HOME", stateRoot) | |||
| 122 | + | err = runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, runDaemonOwned) | |||
| 123 | + | require.Error(t, err) | |||
| 124 | + | require.Contains(t, err.Error(), "validate durable session catalogue") | |||
| 125 | + | _, statErr := os.Stat(filepath.Join(runtimeDir, "daemon.sock")) | |||
| 126 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |||
| 127 | + | }) | |||
| 128 | + | ||||
| 129 | + | t.Run("unavailable catalogue fails without creating state or socket", func(t *testing.T) { | |||
| 130 | + | runtimeRoot, stateRoot := t.TempDir(), t.TempDir() | |||
| 131 | + | runtimeDir := filepath.Join(runtimeRoot, "vev") | |||
| 132 | + | stateDir := filepath.Join(stateRoot, "vev") | |||
| 133 | + | t.Setenv("XDG_RUNTIME_DIR", runtimeRoot) | |||
| 134 | + | t.Setenv("XDG_STATE_HOME", stateRoot) | |||
| 135 | + | ||||
| 136 | + | err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, runDaemonOwned) | |||
| 137 | + | require.Error(t, err) | |||
| 138 | + | require.Contains(t, err.Error(), "durable session catalogue unavailable") | |||
| 139 | + | _, statErr := os.Stat(persist.StorePath(stateDir)) | |||
| 140 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |||
| 141 | + | _, statErr = os.Stat(filepath.Join(runtimeDir, "daemon.sock")) | |||
| 142 | + | require.ErrorIs(t, statErr, os.ErrNotExist) | |||
| 143 | + | }) | |||
| 144 | + | ||||
| 145 | + | t.Run("callback and release errors are joined", func(t *testing.T) { | |||
| 146 | + | startErr := errors.New("catalogue unavailable") | |||
| 147 | + | releaseErr := errors.New("unlock failed") | |||
| 148 | + | deps := lifecycleStartupDeps{ | |||
| 149 | + | ensurePrivate: func(string) error { return nil }, | |||
| 150 | + | acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) { | |||
| 151 | + | return fakeLifecycleOwnership{release: func() error { return releaseErr }}, nil | |||
| 152 | + | }, | |||
| 153 | + | } | |||
| 154 | + | ||||
| 155 | + | err := runWithLifecycleOwnerDeps(context.Background(), "runtime", "state", func(context.Context) error { | |||
| 156 | + | return startErr | |||
| 157 | + | }, deps) | |||
| 158 | + | require.ErrorIs(t, err, startErr) | |||
| 159 | + | require.ErrorIs(t, err, releaseErr) | |||
| 160 | + | }) | |||
| 161 | + | } | |||
| 162 | + | ||||
| 27 | // scriptRecv makes a transport yield frames in order, then wait for done | 163 | // scriptRecv makes a transport yield frames in order, then wait for done | ||
| 28 | // before returning EOF. The shared done channel coordinates proxy readers | 164 | // before returning EOF. The shared done channel coordinates proxy readers | ||
| 29 | // without relying on scheduling or sleeps. | 165 | // without relying on scheduling or sleeps. | ||