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/daemonmux"
17 "github.com/bnema/vev/internal/adapters/ipc"
18 "github.com/bnema/vev/internal/adapters/lifecycle"
19 "github.com/bnema/vev/internal/platform"
20 "github.com/bnema/vev/internal/protocol/wire"
21 "github.com/bnema/vev/pkg/safedir"
22)
23
24// ErrDaemonUnreachable is returned when the daemon socket never becomes
25// dialable within the retry budget, despite a spawn attempt.
26var ErrDaemonUnreachable = errors.New("daemon did not become reachable")
27
28// ErrLifecycleHeldByOther reports that another process kept the daemon
29// lifecycle lock for the whole start budget without publishing the carriage,
30// for example a vev daemon from an older build. Spawning cannot recover this.
31var ErrLifecycleHeldByOther = errors.New("a vev daemon holds the lifecycle lock but did not answer within the start budget; if it is from an older build, run `vev kill --all` and retry")
32
33// spawnLockName is the mkdir-based lock directory guarding daemon spawn, so
34// concurrent first-clients elect a single spawner instead of racing to
35// re-exec multiple daemons.
36const spawnLockName = "spawn.lock"
37
38// staleLockAge bounds how long a spawn lock may persist before it is treated
39// as abandoned (spawner crashed mid-flight) and forcibly taken over.
40const staleLockAge = 10 * time.Second
41
42// dialFunc dials the daemon socket in dir. Injected so tests can drive the
43// spawn/backoff logic without a real socket.
44type dialFunc func(ctx context.Context, dir string) (wire.Transport, error)
45
46// spawnFunc launches a detached daemon process. Injected so tests never
47// re-exec a real binary.
48type spawnFunc func() error
49
50type lifecycleProbe interface {
51 TryAcquire(string) (lifecycleOwnership, error)
52}
53
54type osLifecycleProbe struct{}
55
56func (osLifecycleProbe) TryAcquire(runtimeDir string) (lifecycleOwnership, error) {
57 return lifecycle.TryAcquire(runtimeDir)
58}
59
60var daemonLifecycleProbe lifecycleProbe = osLifecycleProbe{}
61
62// backoffConfig parameterises retry-dial timing so tests can shrink it.
63type backoffConfig struct {
64 initial time.Duration
65 max time.Duration
66 total time.Duration
67}
68
69// defaultBackoff is the production retry schedule: 10ms, 20ms, 40ms... capped
70// at 500ms, for roughly 5s total.
71var defaultBackoff = backoffConfig{
72 initial: 10 * time.Millisecond,
73 max: 500 * time.Millisecond,
74 total: 5 * time.Second,
75}
76
77func retryAttempts[T any](ctx context.Context, cfg backoffConfig, attempt func() (T, bool, error)) (T, error) {
78 deadline := time.Now().Add(cfg.total)
79 backoff := cfg.initial
80 if backoff <= 0 {
81 backoff = time.Millisecond
82 }
83 for {
84 if err := ctx.Err(); err != nil {
85 var zero T
86 return zero, err
87 }
88 result, done, err := attempt()
89 if err != nil {
90 var zero T
91 return zero, err
92 }
93 if done {
94 return result, nil
95 }
96 if !time.Now().Before(deadline) {
97 var zero T
98 return zero, ErrDaemonUnreachable
99 }
100 if err := waitBackoff(ctx, backoff); err != nil {
101 var zero T
102 return zero, err
103 }
104 if backoff *= 2; backoff > cfg.max {
105 backoff = cfg.max
106 }
107 }
108}
109
110// waitForTargetOrLifecycle is the shared election primitive behind every
111// connect-or-spawn path: it repeatedly dials the target and probes lifecycle
112// ownership in lockDir until one of the two is available, a non-contention
113// failure is reported, or the caller's context or the retry budget ends. It is
114// generic over the dial result so the session transport and the daemonmux
115// carriage obey exactly one election discipline instead of two that could drift
116// apart. The dial closure and lockDir stay separate because the daemon's
117// lifecycle lock lives in its runtime directory while a session transport and
118// a daemonmux carriage are two different endpoints of the same daemon.
119func waitForTargetOrLifecycle[T any](ctx context.Context, lockDir string, dial func(ctx context.Context) (T, error), cfg backoffConfig) (T, lifecycleOwnership, error) {
120 type targetOrLifecycle struct {
121 target T
122 owner lifecycleOwnership
123 }
124 result, err := retryAttempts(ctx, cfg, func() (targetOrLifecycle, bool, error) {
125 if target, err := dial(ctx); err == nil {
126 return targetOrLifecycle{target: target}, true, nil
127 }
128 owner, err := daemonLifecycleProbe.TryAcquire(lockDir)
129 if err == nil {
130 return targetOrLifecycle{owner: owner}, true, nil
131 }
132 if !errors.Is(err, lifecycle.ErrBusy) {
133 return targetOrLifecycle{}, false, err
134 }
135 return targetOrLifecycle{}, false, nil
136 })
137 if errors.Is(err, ErrDaemonUnreachable) {
138 // retryAttempts only runs out of budget after busy-lock attempts: the
139 // lock stayed held and the target never answered for the whole budget.
140 err = fmt.Errorf("%w (%w)", ErrLifecycleHeldByOther, err)
141 }
142 return result.target, result.owner, err
143}
144
145func waitBackoff(ctx context.Context, duration time.Duration) error {
146 timer := time.NewTimer(duration)
147 defer timer.Stop()
148 select {
149 case <-ctx.Done():
150 return ctx.Err()
151 case <-timer.C:
152 return nil
153 }
154}
155
156// ensureDaemonWithLifecycle never elects a spawner while another lifecycle
157// owner may still be initializing or tearing down durable state. It is the
158// session-transport spelling of the shared election primitive.
159func ensureDaemonWithLifecycle(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (wire.Transport, error) {
160 return ensureTargetWithLifecycle(ctx, dir, func(ctx context.Context) (wire.Transport, error) { return dial(ctx, dir) }, spawn, cfg)
161}
162
163// ensureMuxDaemon returns a raw framed carriage to a running daemon, electing
164// exactly one spawner under the same lifecycle and spawn-lock discipline the
165// public session socket already uses. lockDir owns the daemon's lifecycle and
166// spawn locks; carriage is the private daemonmux endpoint the dial closure
167// reaches. Its readiness proof is the carriage itself: it never dials a
168// session, opens a session control stream, or otherwise asks the daemon a
169// session question.
170func ensureMuxDaemon(ctx context.Context, lockDir, carriage string, dial func(ctx context.Context, carriage string) (daemonmux.RawFramedTransport, error), spawn spawnFunc, cfg backoffConfig) (daemonmux.RawFramedTransport, error) {
171 return ensureTargetWithLifecycle(ctx, lockDir, func(ctx context.Context) (daemonmux.RawFramedTransport, error) { return dial(ctx, carriage) }, spawn, cfg)
172}
173
174// ensureTargetWithLifecycle is the shared connect-or-spawn election: one dial,
175// then lifecycle availability, then exactly one elected spawner, then a bounded
176// redial of the same target. It never spawns while another lifecycle owner may
177// still be initializing or tearing down durable state, and it returns the
178// dialed value unchanged.
179func ensureTargetWithLifecycle[T any](ctx context.Context, lockDir string, dial func(ctx context.Context) (T, error), spawn spawnFunc, cfg backoffConfig) (T, error) {
180 var zero T
181 if err := ctx.Err(); err != nil {
182 return zero, err
183 }
184 if target, err := dial(ctx); err == nil {
185 return target, nil
186 }
187 target, owner, err := waitForTargetOrLifecycle(ctx, lockDir, dial, cfg)
188 if err != nil {
189 return zero, err
190 }
191 if any(target) != nil {
192 // The target is an interface (a session transport or a raw mux
193 // carriage), so this is the same nil test the typed spelling used: a
194 // non-nil interface value means the dial succeeded.
195 return target, nil
196 }
197 if err := owner.Release(); err != nil {
198 return zero, fmt.Errorf("release lifecycle spawn probe: %w", err)
199 }
200 return ensureTarget(ctx, lockDir, dial, spawn, cfg)
201}
202
203// ensureTarget is the shared spawn election: it dials once, then takes the
204// spawn lock in lockDir, and only the elected winner spawns. Every dialer —
205// the session transport and the daemonmux carriage — therefore observes the
206// same single-spawner guarantee.
207func ensureTarget[T any](ctx context.Context, lockDir string, dial func(ctx context.Context) (T, error), spawn spawnFunc, cfg backoffConfig) (T, error) {
208 var zero T
209 if err := ctx.Err(); err != nil {
210 return zero, err
211 }
212 if t, err := dial(ctx); err == nil {
213 slog.Debug("daemon already reachable", "socket_dir", lockDir)
214 return t, nil
215 }
216
217 release, acquired, err := acquireSpawnLock(lockDir)
218 if err != nil {
219 return zero, fmt.Errorf("acquiring spawn lock: %w", err)
220 }
221 if acquired {
222 // We won the election: spawn, and hold the lock until the socket is
223 // dialable (or spawn fails) so late-arriving clients keep waiting
224 // rather than spawning a second daemon.
225 defer release()
226 slog.Info("spawning daemon", "socket_dir", lockDir)
227 if err := ctx.Err(); err != nil {
228 return zero, err
229 }
230 if err := spawn(); err != nil {
231 slog.Error("daemon spawn failed", "err", err)
232 return zero, fmt.Errorf("spawning daemon: %w", err)
233 }
234 } else {
235 slog.Debug("waiting for daemon spawned by another process", "socket_dir", lockDir)
236 }
237
238 return retryTarget(ctx, lockDir, dial, cfg)
239}
240
241// acquireSpawnLock attempts to create the spawn-lock directory. It returns
242// acquired=true with a release closure when it wins, acquired=false (with a
243// no-op release) when another process holds a fresh lock, and takes over a
244// lock older than staleLockAge (crash resilience).
245func acquireSpawnLock(dir string) (release func(), acquired bool, err error) {
246 if err := safedir.EnsurePrivate(dir); err != nil {
247 return nil, false, err
248 }
249 path := filepath.Join(dir, spawnLockName)
250
251 if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
252 return func() { _ = os.Remove(path) }, true, nil
253 } else if !errors.Is(mkErr, os.ErrExist) {
254 return nil, false, mkErr
255 }
256
257 // Lock exists: take it over only if it looks abandoned.
258 if info, statErr := os.Stat(path); statErr == nil && time.Since(info.ModTime()) > staleLockAge {
259 age := time.Since(info.ModTime())
260 slog.Warn("taking over stale daemon spawn lock", "path", path, "age", age)
261 _ = os.Remove(path)
262 if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
263 return func() { _ = os.Remove(path) }, true, nil
264 }
265 }
266
267 // A live peer is spawning; wait for it via retry-dial.
268 return func() {}, false, nil
269}
270
271// retryTarget is the shared bounded redial: it repeats the same dial with
272// exponential backoff until it answers, the caller's context ends, or the total
273// budget is exhausted. It is generic so the daemonmux carriage retries under
274// exactly the budget the session transport uses.
275func retryTarget[T any](ctx context.Context, lockDir string, dial func(ctx context.Context) (T, error), cfg backoffConfig) (T, error) {
276 var lastDialErr error
277 target, err := retryAttempts(ctx, cfg, func() (T, bool, error) {
278 target, dialErr := dial(ctx)
279 if dialErr != nil {
280 lastDialErr = dialErr
281 }
282 return target, dialErr == nil, nil
283 })
284 if errors.Is(err, ErrDaemonUnreachable) {
285 slog.Error("daemon did not become reachable before retry budget expired", "socket_dir", lockDir, "budget", cfg.total, "last_dial_error", lastDialErr)
286 if lastDialErr != nil {
287 err = fmt.Errorf("%w: last dial: %v", ErrDaemonUnreachable, lastDialErr)
288 }
289 }
290 return target, err
291}
292
293// realDial is the production dialer.
294func realDial(ctx context.Context, dir string) (wire.Transport, error) {
295 return ipc.DialContext(ctx, dir)
296}
297
298// realSpawn re-execs this binary through a short-lived launcher. Waiting for
299// that launcher to exit ensures the daemon has been re-parented before the
300// client continues, keeping process-tree termination from reaching it.
301func realSpawn() error {
302 exePath, err := selfExePath()
303 if err != nil {
304 return fmt.Errorf("resolving executable path: %w", err)
305 }
306
307 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
308 if err != nil {
309 return fmt.Errorf("opening %s: %w", os.DevNull, err)
310 }
311 defer func() { _ = devNull.Close() }()
312
313 var stderr bytes.Buffer
314 cmd := exec.Command(exePath, "--daemon-launcher")
315 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
316 cmd.Dir = platform.DirOrHome("")
317 cmd.Stdin = devNull
318 cmd.Stdout = devNull
319 cmd.Stderr = &stderr
320 if err := cmd.Run(); err != nil {
321 if detail := bytes.TrimSpace(stderr.Bytes()); len(detail) > 0 {
322 return fmt.Errorf("%w: %s", err, detail)
323 }
324 return err
325 }
326 slog.Info("daemon process detached")
327 return nil
328}
329
330func withoutPerformanceTraceEnv(env []string) []string {
331 filtered := make([]string, 0, len(env))
332 for _, entry := range env {
333 name, _, _ := strings.Cut(entry, "=")
334 switch name {
335 case "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN":
336 continue
337 }
338 filtered = append(filtered, entry)
339 }
340 return filtered
341}
342
343// runDaemonLauncher is the intermediate half of the double-fork. It starts
344// the long-lived daemon in a new session and exits immediately; realSpawn
345// waits for this exit so the daemon is no longer descended from the client.
346func runDaemonLauncher() error {
347 exePath, err := selfExePath()
348 if err != nil {
349 return fmt.Errorf("resolving executable path: %w", err)
350 }
351
352 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
353 if err != nil {
354 return fmt.Errorf("opening %s: %w", os.DevNull, err)
355 }
356 defer func() { _ = devNull.Close() }()
357
358 cmd := exec.Command(exePath, "--daemon")
359 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
360 cmd.Dir = platform.DirOrHome("")
361 cmd.Stdin = devNull
362 cmd.Stdout = devNull
363 cmd.Stderr = devNull
364 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
365 if err := cmd.Start(); err != nil {
366 return err
367 }
368 return cmd.Process.Release()
369}