1package app
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "io/fs"
10 "net"
11 "os"
12 "os/exec"
13 "path/filepath"
14 "syscall"
15 "time"
16
17 "github.com/bnema/vev/internal/adapters/brokerconfig"
18 "github.com/bnema/vev/internal/adapters/brokeripc"
19 "github.com/bnema/vev/internal/adapters/lifecycle"
20 "github.com/bnema/vev/internal/platform"
21 "github.com/bnema/vev/internal/ports"
22 "github.com/bnema/vev/pkg/safedir"
23)
24
25// Detached connect-or-spawn and status.
26//
27// `_broker-launcher` is the intermediate half of a double fork: it starts one
28// hidden `_broker-serve` in a new session and exits immediately, exactly like
29// the daemon launcher, so the broker is reparented before the status caller
30// continues. `_broker-status` is the client-facing probe: without --ensure it
31// is dial-only and renders the observable state, and with --ensure it
32// coordinates a race-safe connect-or-spawn.
33//
34// Both commands are hidden from public help and change no ordinary command,
35// path, or factory. They operate only inside the operator-supplied offline
36// root; a root that overlaps the production runtime or state directories is
37// refused before anything is created.
38//
39// Race safety rests on two exclusive descriptor-backed locks that are never
40// stolen by age and whose inodes are never unlinked:
41//
42// - the broker lifetime lock (`layout.Runtime/lifecycle.lock`) is held by the
43// foreground `_broker-serve` for its whole run; a status probe uses it only
44// to detect that a broker already owns the sandbox and therefore must not
45// be duplicated.
46// - the spawn-election lock (`layout.Spawn/lifecycle.lock`) elects exactly
47// one status caller to spawn. Holding it across the readiness wait keeps
48// late callers waiting on the winner instead of racing a second spawn; if
49// the winner dies, the kernel releases the lock, so a waiter can take over
50// without ever unlinking or age-stealing the lock file.
51//
52// The elected spawner holds the election until the endpoint answers a valid
53// broker registration and publishes its initial snapshot, or until the one
54// absolute overall deadline expires. A live-but-incompatible endpoint and a
55// stalled handshake are diagnosed without spawning.
56
57// productionBrokerLauncherCommand is the hidden intermediate that starts one
58// detached broker serve process and exits.
59const productionBrokerLauncherCommand = "_broker-production-launcher"
60
61// Bounds for the detached connect-or-spawn path. Every wait is bounded by the
62// one absolute overall deadline, and the first dial is bounded on its own so a
63// stalled endpoint cannot consume the whole budget before classification.
64const (
65 brokerStatusDefaultTimeout = 10 * time.Second
66 brokerStatusProbeBudget = 2 * time.Second
67 brokerStatusInitialBackoff = 5 * time.Millisecond
68 brokerStatusMaxBackoff = 50 * time.Millisecond
69)
70
71var (
72 // errBrokerAbsent reports that no broker is bound at the endpoint, so a
73 // spawn may recover it. A missing socket and a stale socket left by a dead
74 // owner are both absence.
75 errBrokerAbsent = errors.New("broker endpoint is absent")
76 // errBrokerIncompatible reports that the endpoint exists but is not a
77 // compatible broker (a foreign path, a live endpoint that fails the broker
78 // handshake, or a handshake that stalls past its bound). It is never an
79 // invitation to spawn.
80 errBrokerIncompatible = errors.New("broker endpoint is not a compatible broker")
81 // errBrokerNotReady reports that --ensure did not observe a ready broker
82 // before the overall deadline.
83 errBrokerNotReady = errors.New("broker did not become ready")
84)
85
86// brokerStatusReport is the bounded JSON status document. It carries exactly
87// the ready/offline state, the broker endpoint, and the broker's epoch,
88// revision, and host count. Epoch, revision, and host count are zero when the
89// broker is offline.
90type brokerStatusReport struct {
91 Status string `json:"status"`
92 Endpoint string `json:"endpoint"`
93 Epoch ports.BrokerEpoch `json:"epoch"`
94 Revision ports.BrokerRevision `json:"revision"`
95 HostCount int `json:"host_count"`
96}
97
98// brokerStatusLock is one held exclusive election lock.
99type brokerStatusLock interface {
100 Release() error
101}
102
103// brokerStatusDeps are the injectable seams of `_broker-status`. Every seam has
104// a production default; tests substitute them to drive the election, deadline,
105// and classification deterministically.
106type brokerStatusDeps struct {
107 now func() time.Time
108 wait func(context.Context, time.Duration) error
109 probe func(context.Context, string) (brokerStatusReport, error)
110 acquireElection func(string) (brokerStatusLock, error)
111 lifetimeOwner func(string) (bool, error)
112 spawn func(ctx context.Context, root string, grace time.Duration, graceSet bool) error
113 stdout io.Writer
114}
115
116// defaultBrokerStatusDeps returns the production seams.
117func defaultBrokerStatusDeps() brokerStatusDeps {
118 return brokerStatusDeps{
119 now: time.Now,
120 wait: waitBackoff,
121 probe: probeBrokerStatus,
122 acquireElection: func(dir string) (brokerStatusLock, error) {
123 return lifecycle.TryAcquire(dir)
124 },
125 lifetimeOwner: func(runtimeDir string) (bool, error) {
126 owner, err := lifecycle.TryAcquire(runtimeDir)
127 if err != nil {
128 if errors.Is(err, lifecycle.ErrBusy) {
129 return true, nil
130 }
131 return false, err
132 }
133 return false, owner.Release()
134 },
135 spawn: func(ctx context.Context, _ string, _ time.Duration, _ bool) error {
136 return spawnProductionBrokerLauncher(ctx)
137 },
138 stdout: os.Stdout,
139 }
140}
141
142// parseProductionBrokerLauncherArgs parses the argument-free hidden launcher entry.
143func parseProductionBrokerLauncherArgs(args []string) (command, error) {
144 if len(args) != 0 {
145 return command{}, usagef("`%s` does not accept arguments", productionBrokerLauncherCommand)
146 }
147 return command{kind: kindProductionBrokerLauncher}, nil
148}
149
150// runProductionBrokerLauncherCommand starts one detached broker serve process.
151func runProductionBrokerLauncherCommand(context.Context) error {
152 return startDetachedBrokerServe([]string{productionBrokerServeCommand}, filepath.Join(productionBrokerLayout().Log, "vev-broker-crash.log"))
153}
154
155// startDetachedBrokerServe starts one `_broker-serve` in a new session and
156// releases it, so the broker outlives the short-lived launcher. It mirrors the
157// daemon launcher's double-fork boundary: no ambient variable is dropped beyond
158// the performance-trace inputs, stdio is /dev/null, and the process is
159// reparented immediately.
160func startDetachedBrokerServe(args []string, crashPath string) error {
161 exePath, err := selfExePath()
162 if err != nil {
163 return fmt.Errorf("resolving executable path: %w", err)
164 }
165 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
166 if err != nil {
167 return fmt.Errorf("opening %s: %w", os.DevNull, err)
168 }
169 defer func() { _ = devNull.Close() }()
170
171 cmd := exec.Command(exePath, args...)
172 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
173 cmd.Dir = platform.DirOrHome("")
174 cmd.Stdin = devNull
175 cmd.Stdout = devNull
176 if crashPath == "" {
177 cmd.Stderr = devNull
178 } else {
179 if err := safedir.EnsurePrivate(filepath.Dir(crashPath)); err != nil {
180 return fmt.Errorf("secure broker crash log directory: %w", err)
181 }
182 if err := os.Rename(crashPath, crashPath+".prev"); err != nil && !errors.Is(err, os.ErrNotExist) {
183 return fmt.Errorf("preserve previous broker crash log: %w", err)
184 }
185 crashLog, err := os.OpenFile(crashPath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0o600)
186 if err != nil {
187 return fmt.Errorf("open broker crash log: %w", err)
188 }
189 defer func() { _ = crashLog.Close() }()
190 cmd.Stderr = crashLog
191 }
192 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
193 if err := cmd.Start(); err != nil {
194 return fmt.Errorf("starting detached broker: %w", err)
195 }
196 return cmd.Process.Release()
197}
198
199// spawnBrokerLauncher runs the hidden launcher, bounded by ctx, and waits for
200// it to exit, so the broker has been reparented before the caller retries the
201// endpoint. exec.CommandContext is deliberate: the overall status deadline or a
202// signal kills a launcher that is still waiting, while killing the launcher
203// after it has started the detached `_broker-serve` does not touch that serve --
204// it is in its own session and was released, so it survives.
205func spawnProductionBrokerLauncher(ctx context.Context) error {
206 return runBrokerLauncherProcess(ctx, []string{productionBrokerLauncherCommand})
207}
208
209func runBrokerLauncherProcess(ctx context.Context, args []string) error {
210 exePath, err := selfExePath()
211 if err != nil {
212 return fmt.Errorf("resolving executable path: %w", err)
213 }
214 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
215 if err != nil {
216 return fmt.Errorf("opening %s: %w", os.DevNull, err)
217 }
218 defer func() { _ = devNull.Close() }()
219
220 var stderr bytes.Buffer
221 cmd := exec.CommandContext(ctx, exePath, args...)
222 cmd.Env = withoutPerformanceTraceEnv(os.Environ())
223 cmd.Dir = platform.DirOrHome("")
224 cmd.Stdin = devNull
225 cmd.Stdout = devNull
226 cmd.Stderr = &stderr
227 if err := cmd.Run(); err != nil {
228 if detail := bytes.TrimSpace(stderr.Bytes()); len(detail) > 0 {
229 return fmt.Errorf("launching broker: %w: %s", err, detail)
230 }
231 return fmt.Errorf("launching broker: %w", err)
232 }
233 return nil
234}
235
236// brokerStatusRequest carries one ensure run's fixed inputs.
237type brokerStatusRequest struct {
238 layout brokerconfig.Layout
239 socketPath string
240 deadline time.Time
241 timeout time.Duration
242 root string
243 grace time.Duration
244 graceSet bool
245}
246
247// ensureBrokerReady connects to an already-ready broker or elects one spawner
248// and holds the election until the endpoint answers a valid broker registration
249// and publishes its initial snapshot, or until the absolute deadline expires.
250func ensureBrokerReady(ctx context.Context, req brokerStatusRequest, deps brokerStatusDeps) (brokerStatusReport, error) {
251 ctx, cancel := context.WithDeadline(ctx, req.deadline)
252 defer cancel()
253
254 // Secure the sandbox paths before any lock is taken: the spawn-election
255 // directory lives beneath the runtime directory, and both must be private
256 // and symlink-free before they can hold an exclusive lock.
257 if err := safedir.EnsurePrivate(req.layout.Spawn); err != nil {
258 return brokerStatusReport{}, fmt.Errorf("secure broker sandbox spawn directory: %w", err)
259 }
260 if err := req.layout.VerifyCreated(); err != nil {
261 return brokerStatusReport{}, fmt.Errorf("verify broker sandbox paths: %w", err)
262 }
263
264 var held brokerStatusLock
265 defer func() {
266 if held != nil {
267 _ = held.Release()
268 }
269 }()
270 spawned := false
271 backoff := brokerStatusInitialBackoff
272
273 for {
274 if err := ctx.Err(); err != nil {
275 return brokerStatusReport{}, err
276 }
277 if !deps.now().Before(req.deadline) {
278 return brokerStatusReport{}, fmt.Errorf("%w: endpoint %s did not answer Register within %s", errBrokerNotReady, req.socketPath, req.timeout)
279 }
280 report, err := probeWithin(ctx, deps, req.socketPath, req.deadline)
281 if err == nil {
282 return report, nil
283 }
284 if ctx.Err() != nil {
285 return brokerStatusReport{}, ctx.Err()
286 }
287 // A probe that finished after the deadline is a bounded failure, never
288 // an absence that would justify one more spawn.
289 if !deps.now().Before(req.deadline) {
290 return brokerStatusReport{}, fmt.Errorf("%w: endpoint %s did not answer Register within %s", errBrokerNotReady, req.socketPath, req.timeout)
291 }
292 if errors.Is(err, brokeripc.ErrBrokerRetired) {
293 return brokerStatusReport{}, err
294 }
295 if !errors.Is(err, errBrokerAbsent) {
296 return brokerStatusReport{}, fmt.Errorf("broker endpoint %s is live but incompatible: %w; run \"vev kill --all\" and retry", req.socketPath, err)
297 }
298
299 if held == nil {
300 lock, err := deps.acquireElection(req.layout.Spawn)
301 switch {
302 case err == nil:
303 held = lock
304 // Re-dial after winning the election: the previous holder may
305 // already have published readiness.
306 report, err := probeWithin(ctx, deps, req.socketPath, req.deadline)
307 if err == nil {
308 return report, nil
309 }
310 if ctx.Err() != nil {
311 return brokerStatusReport{}, ctx.Err()
312 }
313 if errors.Is(err, brokeripc.ErrBrokerRetired) {
314 return brokerStatusReport{}, err
315 }
316 if !errors.Is(err, errBrokerAbsent) {
317 return brokerStatusReport{}, fmt.Errorf("broker endpoint %s is live but incompatible: %w; run \"vev kill --all\" and retry", req.socketPath, err)
318 }
319 case errors.Is(err, lifecycle.ErrBusy):
320 // Another caller is elected and spawning. Only the elected
321 // spawner ever probes the broker lifetime lock, so a waiter can
322 // never briefly steal ownership from the broker it is waiting for.
323 if err := deps.wait(ctx, backoff); err != nil {
324 return brokerStatusReport{}, err
325 }
326 backoff *= 2
327 if backoff > brokerStatusMaxBackoff {
328 backoff = brokerStatusMaxBackoff
329 }
330 continue
331 default:
332 return brokerStatusReport{}, fmt.Errorf("elect broker spawner: %w", err)
333 }
334 }
335 if held != nil && !spawned {
336 busy, err := deps.lifetimeOwner(req.layout.Runtime)
337 if err != nil {
338 return brokerStatusReport{}, fmt.Errorf("inspect broker lifetime owner: %w", err)
339 }
340 if !busy {
341 if err := deps.spawn(ctx, req.root, req.grace, req.graceSet); err != nil {
342 return brokerStatusReport{}, err
343 }
344 spawned = true
345 }
346 }
347
348 if err := deps.wait(ctx, backoff); err != nil {
349 return brokerStatusReport{}, err
350 }
351 backoff *= 2
352 if backoff > brokerStatusMaxBackoff {
353 backoff = brokerStatusMaxBackoff
354 }
355 }
356}
357
358// probeWithin bounds one probe by the smaller of the probe budget and the
359// remaining overall time.
360func probeWithin(ctx context.Context, deps brokerStatusDeps, socketPath string, deadline time.Time) (brokerStatusReport, error) {
361 budget := brokerStatusProbeBudget
362 if remaining := deadline.Sub(deps.now()); remaining < budget {
363 budget = remaining
364 }
365 if budget <= 0 {
366 return brokerStatusReport{}, context.DeadlineExceeded
367 }
368 probeCtx, cancel := context.WithTimeout(ctx, budget)
369 defer cancel()
370 return deps.probe(probeCtx, socketPath)
371}
372
373// probeBrokerStatus is the production probe: it classifies endpoint absence
374// first, dials with the bounded broker setup, then waits for the broker to
375// register the connection and publish its initial snapshot.
376func probeBrokerStatus(ctx context.Context, socketPath string) (brokerStatusReport, error) {
377 info, err := os.Lstat(socketPath)
378 switch {
379 case errors.Is(err, fs.ErrNotExist):
380 return brokerStatusReport{}, absentError(socketPath)
381 case err != nil:
382 return brokerStatusReport{}, fmt.Errorf("inspect broker endpoint %s: %w", socketPath, err)
383 case info.Mode()&os.ModeSocket == 0:
384 return brokerStatusReport{}, incompatibleError(socketPath, errors.New("path is not a Unix socket"))
385 }
386
387 service, err := brokeripc.Dial(ctx, socketPath, brokeripc.Config{})
388 if errors.Is(err, brokeripc.ErrBrokerRetired) {
389 return brokerStatusReport{}, err
390 }
391 if err != nil {
392 if backendAbsent(err) {
393 return brokerStatusReport{}, absentError(socketPath)
394 }
395 if ctx.Err() != nil {
396 return brokerStatusReport{}, ctx.Err()
397 }
398 // The endpoint may have retired between the stat and the dial; re-check
399 // so a broker that just shut down is classified as absent, not as an
400 // incompatible endpoint that must never be respawned.
401 if _, statErr := os.Lstat(socketPath); errors.Is(statErr, fs.ErrNotExist) {
402 return brokerStatusReport{}, absentError(socketPath)
403 }
404 return brokerStatusReport{}, incompatibleError(socketPath, err)
405 }
406 defer func() { _ = service.Close() }()
407
408 sub, err := service.Subscribe()
409 if err != nil {
410 return brokerStatusReport{}, incompatibleError(socketPath, err)
411 }
412 defer sub.Close()
413 changed := sub.Changed()
414 for {
415 if snapshot := service.Snapshot(); snapshot.Revision != 0 {
416 return readyBrokerStatus(socketPath, snapshot), nil
417 }
418 select {
419 case <-ctx.Done():
420 return brokerStatusReport{}, ctx.Err()
421 case _, ok := <-changed:
422 if ok {
423 continue
424 }
425 if snapshot := service.Snapshot(); snapshot.Revision != 0 {
426 return readyBrokerStatus(socketPath, snapshot), nil
427 }
428 return brokerStatusReport{}, incompatibleError(socketPath, errors.New("connection ended before the initial snapshot"))
429 }
430 }
431}
432
433// readyBrokerStatus projects one committed snapshot into the ready report.
434// HostCount counts configured remote hosts; the local daemon observation is
435// not a configured host.
436func readyBrokerStatus(socketPath string, snapshot ports.BrokerSnapshot) brokerStatusReport {
437 hosts := 0
438 for _, daemon := range snapshot.Daemons {
439 if !daemon.Local {
440 hosts++
441 }
442 }
443 return brokerStatusReport{
444 Status: "ready",
445 Endpoint: socketPath,
446 Epoch: snapshot.Epoch,
447 Revision: snapshot.Revision,
448 HostCount: hosts,
449 }
450}
451
452// backendAbsent reports whether one dial failure means no broker is bound:
453// a missing socket or a stale socket whose dead owner left the pathname behind.
454// Any other dial failure is a live endpoint that must not be respawned.
455func backendAbsent(err error) bool {
456 var opErr *net.OpError
457 if !errors.As(err, &opErr) {
458 return false
459 }
460 return errors.Is(err, fs.ErrNotExist) || errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ECONNREFUSED)
461}
462
463// absentError wraps one recoverable-absence diagnosis.
464func absentError(socketPath string) error {
465 return fmt.Errorf("%w: %s", errBrokerAbsent, socketPath)
466}
467
468// incompatibleError wraps one live-but-incompatible diagnosis.
469func incompatibleError(socketPath string, cause error) error {
470 return fmt.Errorf("%w: %s: %w", errBrokerIncompatible, socketPath, cause)
471}