1package app
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/binary"
7 "errors"
8 "fmt"
9 "io"
10 "log/slog"
11 "os"
12 "os/signal"
13 "syscall"
14 "time"
15
16 "github.com/bnema/vev/internal/adapters/brokerconfig"
17 "github.com/bnema/vev/internal/adapters/brokeripc"
18 "github.com/bnema/vev/internal/adapters/brokerstore"
19 "github.com/bnema/vev/internal/adapters/clock"
20 "github.com/bnema/vev/internal/adapters/daemonidentity"
21 "github.com/bnema/vev/internal/adapters/daemonmux"
22 "github.com/bnema/vev/internal/adapters/ipc"
23 "github.com/bnema/vev/internal/adapters/lifecycle"
24 "github.com/bnema/vev/internal/adapters/sessionwire"
25 "github.com/bnema/vev/internal/logging"
26 "github.com/bnema/vev/internal/platform"
27 "github.com/bnema/vev/internal/ports"
28 "github.com/bnema/vev/internal/usecase/broker"
29 "github.com/bnema/vev/pkg/safedir"
30)
31
32// Offline broker foreground composition.
33//
34// `_broker-serve` is a hidden, isolated sandbox command. It takes one strict
35// offline root, derives only sandbox runtime, state, and log paths from it, and
36// composes the broker core and local IPC listener over the immutable
37// config marker. It never reads or writes production runtime, state, or config
38// and never changes an ordinary command, path, or factory.
39//
40// Nothing here spawns a detached process or daemon, reports status, or changes
41// a production path. When the sandbox configuration provisions a local binding,
42// the registry observes that existing Unix daemonmux route without opening a
43// logical stream; it owns no remote membership and never reaches SSH or QUIC.
44// The composed process is foreground: it owns one lifetime lock for its whole run, serves
45// until a signal or its parent context ends it, or until the broker idles out,
46// and then shuts down in a fixed order.
47
48const productionBrokerServeCommand = "_broker-production-serve"
49
50// brokerServePoolLimits bound the offline broker pool. They are explicit, not
51// derived from the environment, and small enough that a stalled peer is
52// refused rather than absorbed.
53var brokerServePoolLimits = broker.PoolLimits{
54 Physical: 64,
55 Clients: 2 * brokeripc.DefaultMaxClients,
56 Streams: 256,
57 StreamsPerClient: 64,
58 Warm: brokerconfig.DefaultWarmTransports,
59}
60
61// brokerPoolLimits applies the configured warm transport retention to the
62// fixed pool bounds.
63func brokerPoolLimits(config *brokerconfig.Config) broker.PoolLimits {
64 limits := brokerServePoolLimits
65 limits.Warm = config.WarmTransports()
66 limits.Idle = config.WarmIdleTimeout()
67 return limits
68}
69
70// brokerServeOptions is the parsed hidden invocation.
71type brokerServeOptions struct {
72 // idleGrace is zero when the operator did not override it; the supervisor
73 // then applies its own 5m default.
74 idleGrace time.Duration
75}
76
77// parseProductionBrokerServeArgs parses the argument-free hidden serve entry.
78func parseProductionBrokerServeArgs(args []string) (command, error) {
79 if len(args) != 0 {
80 return command{}, usagef("`%s` does not accept arguments", productionBrokerServeCommand)
81 }
82 return command{kind: kindProductionBrokerServe, brokerServe: brokerServeOptions{}}, nil
83}
84
85// brokerServeDeps are the composition seams. Production supplies the real
86// adapters; tests override the clock, the owner lock, the logger, and the
87// readiness hook so idle shutdown and cleanup are deterministic.
88type productionIdentityBinder struct {
89 delegate ports.BrokerIdentityBinder
90 stateDir string
91 policy ports.BrokerPolicy
92}
93
94func (b productionIdentityBinder) BindAuthenticatedIdentity(ctx context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
95 if !request.Fence.Local {
96 return b.delegate.BindAuthenticatedIdentity(ctx, request)
97 }
98 if request.Policy != b.policy {
99 return "", errors.New("local daemon policy authority changed")
100 }
101 identity, err := daemonidentity.Load(b.stateDir)
102 if err != nil {
103 return "", fmt.Errorf("revalidate daemon authority: %w", err)
104 }
105 if identity != request.Identity {
106 return "", errors.New("local daemon identity authority changed")
107 }
108 return identity, nil
109}
110
111type brokerServeDeps struct {
112 clock func() ports.Clock
113 acquire func(string) (lifecycleOwnership, error)
114 newLogger func(string) (*slog.Logger, io.Closer, error)
115 newEpoch func() (ports.BrokerEpoch, error)
116 listen func(string, ports.BrokerEpoch, ports.BrokerAuthority, brokeripc.Config) (ports.BrokerListener, error)
117 onReady func(socketPath string)
118}
119
120// defaultBrokerServeDeps builds the production seams.
121func defaultBrokerServeDeps() brokerServeDeps {
122 return brokerServeDeps{
123 clock: func() ports.Clock { return clock.New() },
124 acquire: func(runtimeDir string) (lifecycleOwnership, error) {
125 return lifecycle.TryAcquire(runtimeDir)
126 },
127 newLogger: sandboxLogging,
128 newEpoch: newBrokerEpoch,
129 listen: func(path string, epoch ports.BrokerEpoch, authority ports.BrokerAuthority, cfg brokeripc.Config) (ports.BrokerListener, error) {
130 return brokeripc.Listen(path, epoch, authority, cfg)
131 },
132 }
133}
134
135// runBrokerServeCommand enters the hidden foreground process: it installs
136// SIGINT/SIGTERM cancellation over the parent context and runs the sandbox.
137func runBrokerServeCommand(ctx context.Context, options brokerServeOptions) (retErr error) {
138 ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
139 defer stop()
140 observer, observerCloser, err := newPerformanceTrace(clock.New())
141 if err != nil {
142 return fmt.Errorf("performance trace: %w", err)
143 }
144 if observer != nil {
145 observer.ObserveRuntime(ports.NewRuntimeMark("broker", ports.RuntimeTransportDiagnostic, 0, true))
146 }
147 if observerCloser != nil {
148 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
149 }
150 return runBrokerServe(ctx, options, defaultBrokerServeDeps())
151}
152
153// sandboxLogging configures broker logging inside the sandbox log directory.
154// It never consults the production state directory.
155func sandboxLogging(dir string) (*slog.Logger, io.Closer, error) {
156 return logging.Setup(logging.Config{
157 Dir: dir,
158 Component: logging.Broker,
159 Level: logging.EnvLevel(),
160 MaxBytes: logging.DefaultMaxBytes,
161 })
162}
163
164// newBrokerEpoch samples one cryptographically random non-zero broker epoch.
165func newBrokerEpoch() (ports.BrokerEpoch, error) {
166 var raw [8]byte
167 if _, err := rand.Read(raw[:]); err != nil {
168 return 0, fmt.Errorf("sample broker epoch: %w", err)
169 }
170 epoch := ports.BrokerEpoch(binary.BigEndian.Uint64(raw[:]))
171 if epoch == 0 {
172 epoch = 1
173 }
174 return epoch, nil
175}
176
177// effectiveIdleGrace resolves the sandbox idle grace from an explicit flag and
178// the provisioned configuration. The provisioned value is the single
179// deterministic source: a status probe cannot read a running broker's effective
180// grace through the broker IPC protocol, so an explicit grace that conflicts
181// with a provisioned one is diagnosed here rather than silently applied.
182func effectiveIdleGrace(config *brokerconfig.Config, explicit time.Duration) (time.Duration, error) {
183 provisioned, ok := config.IdleGrace()
184 switch {
185 case explicit <= 0:
186 if ok {
187 return provisioned, nil
188 }
189 // Zero lets the supervisor apply its own built-in default.
190 return 0, nil
191 case ok && explicit != provisioned:
192 return 0, usagef("`--idle-grace` %s conflicts with the provisioned idle grace %s", explicit, provisioned)
193 default:
194 return explicit, nil
195 }
196}
197
198// runBrokerServe runs one isolated offline broker in the foreground until ctx
199// is cancelled or the broker idles out, then shuts down in the fixed order
200// listener -> pool -> registry -> store -> log -> owner lock.
201//
202// Every resource is acquired in dependency order and released in reverse on
203// any failure, so a partial setup never leaks a store lock, an owner lock, a
204// bound socket, or a supervisor goroutine. The temporary startup lease pins the
205// supervisor open from construction until the listener and its accept drain are
206// ready, so the idle timer can never fire during setup.
207func runBrokerServe(ctx context.Context, options brokerServeOptions, deps brokerServeDeps) (retErr error) {
208 ctx, cancel := context.WithCancel(ctx)
209 defer cancel()
210 if err := ensureProductionBrokerConfig(); err != nil {
211 return err
212 }
213 if err := safedir.EnsurePrivate(ipc.SocketDir()); err != nil {
214 return fmt.Errorf("secure broker runtime parent: %w", err)
215 }
216 layout := productionBrokerLayout()
217 configPath := productionBrokerConfigPath()
218 for _, dir := range []string{layout.Root, layout.Runtime, layout.State, layout.Log} {
219 if err := safedir.EnsurePrivate(dir); err != nil {
220 return fmt.Errorf("secure broker sandbox directory %s: %w", dir, err)
221 }
222 }
223 // Re-walk every component now that EnsurePrivate created them, so a
224 // component swapped for a symlink in that window fails closed.
225 if err := layout.VerifyCreated(); err != nil {
226 return fmt.Errorf("verify broker paths: %w", err)
227 }
228 // The broker is allowed to precede the daemon. Identity is therefore an
229 // optional existing authority here; acquisition binds it after an
230 // authenticated StartIfNeeded connection when the daemon is first born.
231 identity, err := daemonidentity.Load(platform.StateDir())
232 if err != nil && !errors.Is(err, os.ErrNotExist) {
233 return fmt.Errorf("load daemon authority: %w", err)
234 }
235 config, err := brokerconfig.LoadProduction(layout, configPath, identity, localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
236 if err != nil {
237 return err
238 }
239 idleGrace, err := effectiveIdleGrace(config, options.idleGrace)
240 if err != nil {
241 return err
242 }
243 owner, err := deps.acquire(layout.Runtime)
244 if err != nil {
245 return fmt.Errorf("broker sandbox ownership: %w", err)
246 }
247 defer func() { retErr = errors.Join(retErr, owner.Release()) }()
248
249 log, logCloser, err := deps.newLogger(layout.Log)
250 if err != nil {
251 return err
252 }
253 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
254
255 // The durable route schema delegates SSH trust to OpenSSH. Refuse route
256 // overrides that OpenSSH configuration must represent instead of dropping
257 // security inputs during import.
258 for _, endpoint := range config.Endpoints() {
259 registration, _ := config.Registration(endpoint)
260 if registration.Route.KnownHostsFile() != "" || registration.Route.ConnectTimeout() != 0 {
261 return errors.New("broker routes require SSH trust and timeouts in OpenSSH configuration")
262 }
263 }
264 initialHosts, err := configuredBrokerHosts(config)
265 if err != nil {
266 return err
267 }
268 store, err := brokerstore.Open(brokerstore.Options{Dir: layout.State, InitialHosts: initialHosts, InitialImportProvided: true})
269 if err != nil {
270 return fmt.Errorf("open broker sandbox store: %w", err)
271 }
272 defer func() { retErr = errors.Join(retErr, store.Close()) }()
273 if err := upgradeBrokerHostVersions(store); err != nil {
274 return err
275 }
276
277 epoch, err := deps.newEpoch()
278 if err != nil {
279 return err
280 }
281 // One clock is shared by the supervisor, registry, and pool so an injected
282 // clock (for example a manual test clock) governs every timer in the sandbox.
283 clk := deps.clock()
284 supervisor, err := broker.NewSupervisor(clk, broker.SupervisorConfig{IdleGrace: idleGrace})
285 if err != nil {
286 return err
287 }
288 // The supervisor owns the ordered drain; its deferred Close is a safety net
289 // for every error path below. A successful run closes it explicitly, and the
290 // flag suppresses the deferred duplicate join so one shutdown error is never
291 // reported twice.
292 supervisorClosed := false
293 defer func() {
294 if !supervisorClosed {
295 retErr = errors.Join(retErr, supervisor.Close())
296 }
297 }()
298
299 // The startup lease pins the broker open until the listener and its accept
300 // drain are ready. It is released exactly once, and again on any error path.
301 startupLease, err := supervisor.AdmitOperation()
302 if err != nil {
303 return err
304 }
305 defer startupLease.Release()
306
307 // Membership is imported once, then only durable remote authority is used.
308 resolver := newBrokerRoutes(config)
309 connector, err := brokerDynamicMuxConnector(config, resolver, log)
310 if err != nil {
311 return err
312 }
313 loadLocalIdentity := func() (ports.BrokerDaemonIdentity, error) {
314 identity, loadErr := daemonidentity.Load(platform.StateDir())
315 if errors.Is(loadErr, os.ErrNotExist) {
316 return "", nil
317 }
318 return identity, loadErr
319 }
320 registryConfig, err := offlineRegistryConfig(config, epoch, connector, loadLocalIdentity)
321 if err != nil {
322 return err
323 }
324 remoteProbe := &broker.RemoteProbe{Epoch: epoch, Routes: resolver, Connector: connector, Codec: sessionwire.BrokerCodec{}}
325 registry, err := broker.NewRegistryWithConfig(epoch, store, remoteProbe, clk, log, registryConfig)
326 if err != nil {
327 return err
328 }
329 resolver.Hosts = registry
330 binder := productionIdentityBinder{delegate: registry, stateDir: platform.StateDir(), policy: localDaemonPolicy()}
331 remoteProbe.Binder = binder
332 remoteProbe.Hosts = registry
333 // RegisterRunner starts immediately: finish the cyclic composition before
334 // publishing it to any goroutine. Shutdown remains listener -> pool -> registry.
335 if err := supervisor.RegisterRunner("registry", registry); err != nil {
336 return err
337 }
338 pool, err := broker.NewPool(epoch, resolver, binder, connector, clk, brokerPoolLimits(config))
339 if err != nil {
340 return err
341 }
342 // Observation borrows pooled transports rather than dialing its own.
343 remoteProbe.SharePool(pool)
344 if registryConfig.Local != nil {
345 if local, ok := registryConfig.Local.Probe.(*broker.LocalDaemonProbe); ok {
346 local.SharePool(pool)
347 }
348 }
349 if err := supervisor.RegisterCloseable("pool", pool); err != nil {
350 return err
351 }
352 authority, err := broker.NewAuthority(epoch, registry, pool, supervisor, sessionwire.BrokerCodec{})
353 if err != nil {
354 return err
355 }
356 socketPath := brokeripc.SocketPath(layout.Runtime)
357 listener, err := deps.listen(socketPath, epoch, authority, brokeripc.Config{OnRetire: cancel})
358 if err != nil {
359 return fmt.Errorf("listen on broker sandbox socket: %w", err)
360 }
361 if err := supervisor.RegisterCloseable("listener", listener); err != nil {
362 _ = listener.Close()
363 return err
364 }
365
366 started := make(chan struct{})
367 drained := make(chan struct{})
368 failed := make(chan struct{})
369 go drainBrokerAccept(listener, log, started, drained, failed)
370 <-started
371 startupLease.Release()
372 log.Info("broker_ready", "socket", socketPath, "endpoints", len(config.Endpoints()))
373 if deps.onReady != nil {
374 deps.onReady(socketPath)
375 }
376
377 select {
378 case <-ctx.Done():
379 case <-supervisor.Done():
380 case <-failed:
381 // An unexpected terminal accept failure leaves the socket bound but
382 // unable to admit work. Commit shutdown so the sandbox tears down in the
383 // fixed order instead of wedging as a bound, unserving broker.
384 }
385 // Closing the supervisor commits shutdown, cancels the registry run, and
386 // drains the listener, pool, and registry in order. The store, log, and
387 // owner lock are released by the deferred cleanup afterwards.
388 closeErr := supervisor.Close()
389 supervisorClosed = true
390 <-drained
391 log.Info("broker_stopped")
392 return closeErr
393}
394
395// drainBrokerAccept continuously drains admitted broker sessions so the
396// listener's bounded admission FIFO never stalls.
397//
398// A drained session is the listener's own connection-scoped service: closing it
399// here would end a live client, so it is deliberately left open and owned by
400// the listener. The drain stops when the listener closes or reports a terminal
401// accept failure. A listener close is the expected, orderly end of the drain and
402// is never logged; any other terminal accept failure is unexpected: it is logged
403// as an error and reported on failed so the caller commits shutdown instead of
404// leaving a bound broker that can no longer admit work.
405// configuredBrokerHosts converts the one-time configuration import into store
406// input. brokerstore consumes it only while creating its first manifest; an
407// existing (including deliberately empty) membership is never reseeded.
408func configuredBrokerHosts(config *brokerconfig.Config) ([]ports.BrokerHostRecord, error) {
409 records := make([]ports.BrokerHostRecord, 0, len(config.Endpoints()))
410 for _, endpoint := range config.Endpoints() {
411 registration, ok := config.Registration(endpoint)
412 if !ok {
413 return nil, fmt.Errorf("missing configured broker registration %q", endpoint)
414 }
415 route, err := durableBrokerRoute(registration.Route)
416 if err != nil {
417 return nil, fmt.Errorf("import configured broker host %q: %w", endpoint, err)
418 }
419 records = append(records, ports.BrokerHostRecord{Registration: registration.Registration, Pinned: true, Policy: registration.Policy, Identity: registration.Identity, Route: route})
420 }
421 return records, nil
422}
423
424func durableBrokerRoute(route brokerconfig.Route) (ports.BrokerRouteSpec, error) {
425 spec := ports.BrokerRouteSpec{Path: route.Path(), Target: route.Target(), Argv: route.Argv()}
426 switch route.Kind() {
427 case brokerconfig.RouteUnix:
428 spec.Kind = ports.BrokerRouteUnix
429 case brokerconfig.RouteSSHStdio:
430 spec.Kind = ports.BrokerRouteSSHStdio
431 case brokerconfig.RouteSSHQUIC:
432 spec.Kind = ports.BrokerRouteSSHQUIC
433 default:
434 return ports.BrokerRouteSpec{}, errors.New("unknown configured broker route kind")
435 }
436 if err := spec.Validate(); err != nil {
437 return ports.BrokerRouteSpec{}, err
438 }
439 return spec, nil
440}
441
442// offlineRegistryConfig selects the registry's observation producers from one
443// immutable configuration. A configuration that provisions a local binding
444// always composes the broker-owned local producer over the broker's already
445// owned endpoint connector, so the probe is dynamic: it derives its authority
446// from the local route, the local daemon policy, and an identity loader that is
447// re-read on every attempt, so the broker may precede the daemon and start
448// observing the moment the daemon first publishes its identity. Such a
449// composition is mutable: the broker owns host membership for its whole run,
450// including a zero-host membership, so a caller can add and remove hosts.
451//
452// A configuration that provisions no local binding is an explicitly configured
453// fixture: it is deliberately left observation-disabled, so a read-only
454// snapshot owner issues no probes, arms no timers, and never reconciles. The
455// disabled fixture's membership mode follows the endpoints it was provisioned
456// with, exactly as before.
457func offlineRegistryConfig(config *brokerconfig.Config, epoch ports.BrokerEpoch, connector ports.BrokerEndpointConnector, loaders ...broker.LocalIdentityLoader) (broker.RegistryConfig, error) {
458 if config == nil {
459 return broker.RegistryConfig{}, errors.New("offline registry requires configuration")
460 }
461 if _, ok := config.LocalBinding(); !ok {
462 mode := broker.MembershipImmutable
463 if len(config.Endpoints()) != 0 {
464 mode = broker.MembershipMutable
465 }
466 return broker.RegistryConfig{ObservationDisabled: true, MembershipMode: mode}, nil
467 }
468 local, err := brokerLocalObservation(config, epoch, connector, loaders...)
469 if err != nil {
470 return broker.RegistryConfig{}, err
471 }
472 return broker.RegistryConfig{Local: local, MembershipMode: broker.MembershipMutable}, nil
473}
474
475func drainBrokerAccept(listener ports.BrokerListener, log *slog.Logger, started chan<- struct{}, drained chan<- struct{}, failed chan<- struct{}) {
476 defer close(drained)
477 close(started)
478 for {
479 session, err := listener.Accept()
480 if err != nil {
481 if errors.Is(err, brokeripc.ErrListenerClosed) {
482 return
483 }
484 log.Error("broker_offline_accept_failed", "error", err)
485 close(failed)
486 return
487 }
488 if session == nil {
489 continue
490 }
491 log.Debug("broker_offline_connection", "connection", fmt.Sprintf("%x", session.ConnectionID()))
492 }
493}