1package app
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "log/slog"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strings"
13 "syscall"
14 "time"
15
16 "github.com/bnema/vev/internal/adapters/ipc"
17 "github.com/bnema/vev/internal/adapters/lifecycle"
18 "github.com/bnema/vev/internal/platform"
19 "github.com/bnema/vev/internal/ports"
20 "github.com/bnema/vev/pkg/safedir"
21)
22
23// ErrDaemonUnreachable is returned when the daemon socket never becomes
24// dialable within the retry budget, despite a spawn attempt.
25var ErrDaemonUnreachable = errors.New("vev: daemon did not become reachable")
26
27// spawnLockName is the mkdir-based lock directory guarding daemon spawn, so
28// concurrent first-clients elect a single spawner instead of racing to
29// re-exec multiple daemons.
30const spawnLockName = "spawn.lock"
31
32// staleLockAge bounds how long a spawn lock may persist before it is treated
33// as abandoned (spawner crashed mid-flight) and forcibly taken over.
34const staleLockAge = 10 * time.Second
35
36// dialFunc dials the daemon socket in dir. Injected so tests can drive the
37// spawn/backoff logic without a real socket.
38type dialFunc func(ctx context.Context, dir string) (ports.Transport, error)
39
40// spawnFunc launches a detached daemon process. Injected so tests never
41// re-exec a real binary.
42type spawnFunc func() error
43
44type lifecycleProbe interface {
45 TryAcquire(string) (lifecycleOwnership, error)
46}
47
48type osLifecycleProbe struct{}
49
50func (osLifecycleProbe) TryAcquire(runtimeDir string) (lifecycleOwnership, error) {
51 return lifecycle.TryAcquire(runtimeDir)
52}
53
54var daemonLifecycleProbe lifecycleProbe = osLifecycleProbe{}
55
56// backoffConfig parameterises retry-dial timing so tests can shrink it.
57type backoffConfig struct {
58 initial time.Duration
59 max time.Duration
60 total time.Duration
61}
62
63// defaultBackoff is the production retry schedule: 10ms, 20ms, 40ms... capped
64// at 500ms, for roughly 5s total.
65var defaultBackoff = backoffConfig{
66 initial: 10 * time.Millisecond,
67 max: 500 * time.Millisecond,
68 total: 5 * time.Second,
69}
70
71func retryAttempts[T any](ctx context.Context, cfg backoffConfig, attempt func() (T, bool, error)) (T, error) {
72 deadline := time.Now().Add(cfg.total)
73 backoff := cfg.initial
74 if backoff <= 0 {
75 backoff = time.Millisecond
76 }
77 for {
78 if err := ctx.Err(); err != nil {
79 var zero T
80 return zero, err
81 }
82 result, done, err := attempt()
83 if err != nil {
84 var zero T
85 return zero, err
86 }
87 if done {
88 return result, nil
89 }
90 if !time.Now().Before(deadline) {
91 var zero T
92 return zero, ErrDaemonUnreachable
93 }
94 if err := waitBackoff(ctx, backoff); err != nil {
95 var zero T
96 return zero, err
97 }
98 if backoff *= 2; backoff > cfg.max {
99 backoff = cfg.max
100 }
101 }
102}
103
104func waitForLifecycleAvailability(ctx context.Context, runtimeDir string, cfg backoffConfig) (lifecycleOwnership, error) {
105 return retryAttempts(ctx, cfg, func() (lifecycleOwnership, bool, error) {
106 owner, err := daemonLifecycleProbe.TryAcquire(runtimeDir)
107 if err == nil {
108 return owner, true, nil
109 }
110 if !errors.Is(err, lifecycle.ErrBusy) {
111 return nil, false, err
112 }
113 return nil, false, nil
114 })
115}
116
117type daemonOrLifecycle struct {
118 transport ports.Transport
119 owner lifecycleOwnership
120}
121
122func waitForDaemonOrLifecycle(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, lifecycleOwnership, error) {
123 result, err := retryAttempts(ctx, cfg, func() (daemonOrLifecycle, bool, error) {
124 if transport, err := dial(ctx, dir); err == nil {
125 return daemonOrLifecycle{transport: transport}, true, nil
126 }
127 owner, err := daemonLifecycleProbe.TryAcquire(dir)
128 if err == nil {
129 return daemonOrLifecycle{owner: owner}, true, nil
130 }
131 if !errors.Is(err, lifecycle.ErrBusy) {
132 return daemonOrLifecycle{}, false, err
133 }
134 return daemonOrLifecycle{}, false, nil
135 })
136 return result.transport, result.owner, err
137}
138
139func waitBackoff(ctx context.Context, duration time.Duration) error {
140 timer := time.NewTimer(duration)
141 defer timer.Stop()
142 select {
143 case <-ctx.Done():
144 return ctx.Err()
145 case <-timer.C:
146 return nil
147 }
148}
149
150// ensureDaemonWithLifecycle never elects a spawner while another lifecycle
151// owner may still be initializing or tearing down durable state.
152func ensureDaemonWithLifecycle(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
153 if err := ctx.Err(); err != nil {
154 return nil, err
155 }
156 if transport, err := dial(ctx, dir); err == nil {
157 return transport, nil
158 }
159 transport, owner, err := waitForDaemonOrLifecycle(ctx, dir, dial, cfg)
160 if err != nil {
161 return nil, err
162 }
163 if transport != nil {
164 return transport, nil
165 }
166 if err := owner.Release(); err != nil {
167 return nil, fmt.Errorf("vev: release lifecycle spawn probe: %w", err)
168 }
169 return ensureDaemon(ctx, dir, dial, spawn, cfg)
170}
171
172// ensureDaemon returns a transport to a running daemon, spawning one if
173// necessary after ensureDaemonWithLifecycle has established lifecycle
174// availability. Tests also exercise this lower-level spawn election directly.
175func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
176 if err := ctx.Err(); err != nil {
177 return nil, err
178 }
179 if t, err := dial(ctx, dir); err == nil {
180 slog.Debug("daemon already reachable", "socket_dir", dir)
181 return t, nil
182 }
183
184 release, acquired, err := acquireSpawnLock(dir)
185 if err != nil {
186 return nil, fmt.Errorf("vev: acquiring spawn lock: %w", err)
187 }
188 if acquired {
189 // We won the election: spawn, and hold the lock until the socket is
190 // dialable (or spawn fails) so late-arriving clients keep waiting
191 // rather than spawning a second daemon.
192 defer release()
193 slog.Info("spawning daemon", "socket_dir", dir)
194 if err := ctx.Err(); err != nil {
195 return nil, err
196 }
197 if err := spawn(); err != nil {
198 slog.Error("daemon spawn failed", "err", err)
199 return nil, fmt.Errorf("vev: spawning daemon: %w", err)
200 }
201 } else {
202 slog.Debug("waiting for daemon spawned by another process", "socket_dir", dir)
203 }
204
205 return retryDial(ctx, dir, dial, cfg)
206}
207
208// acquireSpawnLock attempts to create the spawn-lock directory. It returns
209// acquired=true with a release closure when it wins, acquired=false (with a
210// no-op release) when another process holds a fresh lock, and takes over a
211// lock older than staleLockAge (crash resilience).
212func acquireSpawnLock(dir string) (release func(), acquired bool, err error) {
213 if err := safedir.EnsurePrivate(dir); err != nil {
214 return nil, false, err
215 }
216 path := filepath.Join(dir, spawnLockName)
217
218 if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
219 return func() { _ = os.Remove(path) }, true, nil
220 } else if !errors.Is(mkErr, os.ErrExist) {
221 return nil, false, mkErr
222 }
223
224 // Lock exists: take it over only if it looks abandoned.
225 if info, statErr := os.Stat(path); statErr == nil && time.Since(info.ModTime()) > staleLockAge {
226 age := time.Since(info.ModTime())
227 slog.Warn("taking over stale daemon spawn lock", "path", path, "age", age)
228 _ = os.Remove(path)
229 if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
230 return func() { _ = os.Remove(path) }, true, nil
231 }
232 }
233
234 // A live peer is spawning; wait for it via retry-dial.
235 return func() {}, false, nil
236}
237
238// retryDial dials repeatedly with exponential backoff until the daemon
239// answers, the context is cancelled, or the total budget is exhausted.
240func retryDial(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, error) {
241 transport, err := retryAttempts(ctx, cfg, func() (ports.Transport, bool, error) {
242 transport, err := dial(ctx, dir)
243 return transport, err == nil, nil
244 })
245 if errors.Is(err, ErrDaemonUnreachable) {
246 slog.Error("daemon did not become reachable before retry budget expired", "socket_dir", dir, "budget", cfg.total)
247 }
248 return transport, err
249}
250
251// realDial is the production dialer.
252func realDial(ctx context.Context, dir string) (ports.Transport, error) {
253 return ipc.DialContext(ctx, dir)
254}
255
256// realSpawn re-execs this binary through a short-lived launcher. Waiting for
257// that launcher to exit ensures the daemon has been re-parented before the
258// client continues, keeping process-tree termination from reaching it.
259func realSpawn() error {
260 exePath, err := selfExePath()
261 if err != nil {
262 return fmt.Errorf("resolving executable path: %w", err)
263 }
264
265 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
266 if err != nil {
267 return fmt.Errorf("opening %s: %w", os.DevNull, err)
268 }
269 defer func() { _ = devNull.Close() }()
270
271 var stderr bytes.Buffer
272 cmd := exec.Command(exePath, "--daemon-launcher")
273 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
274 cmd.Dir = platform.DirOrHome("")
275 cmd.Stdin = devNull
276 cmd.Stdout = devNull
277 cmd.Stderr = &stderr
278 if err := cmd.Run(); err != nil {
279 if detail := bytes.TrimSpace(stderr.Bytes()); len(detail) > 0 {
280 return fmt.Errorf("%w: %s", err, detail)
281 }
282 return err
283 }
284 slog.Info("daemon process detached")
285 return nil
286}
287
288func withoutPerformanceTraceEnv(env []string) []string {
289 filtered := make([]string, 0, len(env))
290 for _, entry := range env {
291 name, _, _ := strings.Cut(entry, "=")
292 switch name {
293 case "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN":
294 continue
295 }
296 filtered = append(filtered, entry)
297 }
298 return filtered
299}
300
301// runDaemonLauncher is the intermediate half of the double-fork. It starts
302// the long-lived daemon in a new session and exits immediately; realSpawn
303// waits for this exit so the daemon is no longer descended from the client.
304func runDaemonLauncher() error {
305 exePath, err := selfExePath()
306 if err != nil {
307 return fmt.Errorf("resolving executable path: %w", err)
308 }
309
310 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
311 if err != nil {
312 return fmt.Errorf("opening %s: %w", os.DevNull, err)
313 }
314 defer func() { _ = devNull.Close() }()
315
316 cmd := exec.Command(exePath, "--daemon")
317 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
318 cmd.Dir = platform.DirOrHome("")
319 cmd.Stdin = devNull
320 cmd.Stdout = devNull
321 cmd.Stderr = devNull
322 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
323 if err := cmd.Start(); err != nil {
324 return err
325 }
326 return cmd.Process.Release()
327}