1package app
2
3import (
4 "context"
5 "crypto/rand"
6 "errors"
7 "fmt"
8 "log/slog"
9 "net"
10 "net/http"
11 "os"
12 "os/signal"
13 "path/filepath"
14 "strings"
15 "syscall"
16 "time"
17
18 snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot"
19
20 "github.com/bnema/vev/internal/adapters/clock"
21 "github.com/bnema/vev/internal/adapters/config"
22 "github.com/bnema/vev/internal/adapters/daemonidentity"
23 "github.com/bnema/vev/internal/adapters/daemonmux"
24 "github.com/bnema/vev/internal/adapters/ipc"
25 "github.com/bnema/vev/internal/adapters/lifecycle"
26 "github.com/bnema/vev/internal/adapters/noticefile"
27 "github.com/bnema/vev/internal/adapters/pty"
28 "github.com/bnema/vev/internal/adapters/sessionwire"
29 "github.com/bnema/vev/internal/adapters/shellcmd"
30 "github.com/bnema/vev/internal/domain"
31 "github.com/bnema/vev/internal/logging"
32 "github.com/bnema/vev/internal/persist"
33 "github.com/bnema/vev/internal/platform"
34 "github.com/bnema/vev/internal/ports"
35 "github.com/bnema/vev/internal/protocol/wire"
36 "github.com/bnema/vev/internal/usecase/daemon"
37 "github.com/bnema/vev/internal/usecase/recovery"
38 "github.com/bnema/vev/pkg/safedir"
39)
40
41// Daemon composition seams replaced by tests.
42var (
43 openCatalogue = persist.OpenOrCreate
44 listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (wire.Listener, error) {
45 return ipc.Listen(dir, ipc.WithRuntimeObserver(observer))
46 }
47)
48
49func snapshotDir() string {
50 return filepath.Join(platform.StateDir(), "snapshots")
51}
52
53func developmentTempDirOption(ensurePrivate func(string) error) (daemon.Option, error) {
54 dir, active := platform.DevelopmentEnvironmentTempDir()
55 if !active {
56 return nil, nil
57 }
58 if err := ensurePrivate(dir); err != nil {
59 return nil, fmt.Errorf("secure development temp directory: %w", err)
60 }
61 return daemon.WithTempDir(dir), nil
62}
63
64func pprofAddrIsLoopback(addr string) bool {
65 host, _, err := net.SplitHostPort(addr)
66 if err != nil || host == "" {
67 return false
68 }
69 if strings.EqualFold(host, "localhost") {
70 return true
71 }
72 ip := net.ParseIP(host)
73 return ip != nil && ip.IsLoopback()
74}
75
76type ownedDaemonStart func(context.Context) error
77
78type lifecycleOwnership interface {
79 Release() error
80}
81
82type lifecycleStartupDeps struct {
83 ensurePrivate func(string) error
84 acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)
85 log *slog.Logger
86}
87
88const lifecycleAcquireRetry = 25 * time.Millisecond
89
90func lifecycleStartupDependencies(log *slog.Logger) lifecycleStartupDeps {
91 return lifecycleStartupDeps{
92 ensurePrivate: safedir.EnsurePrivate,
93 acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) {
94 return lifecycle.Acquire(ctx, runtimeDir, retry)
95 },
96 log: log,
97 }
98}
99
100func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart) error {
101 return runWithLifecycleOwnerDeps(ctx, runtimeDir, stateDir, start, lifecycleStartupDependencies(nil))
102}
103
104func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {
105 log := deps.log
106 if log == nil {
107 log = slog.Default()
108 }
109 if err := deps.ensurePrivate(runtimeDir); err != nil {
110 return fmt.Errorf("secure runtime directory: %w", err)
111 }
112 if err := deps.ensurePrivate(stateDir); err != nil {
113 return fmt.Errorf("secure state directory: %w", err)
114 }
115 lockPath := lifecycle.Path(runtimeDir)
116 log.Info("lifecycle_owner_wait", "path", lockPath)
117 owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)
118 if err != nil {
119 log.Warn("lifecycle_owner_wait_failed", "path", lockPath, "reason_code", "acquire-failed")
120 return fmt.Errorf("acquire lifecycle ownership: %w", err)
121 }
122 log.Info("lifecycle_owner_acquired", "path", lockPath)
123 defer func() {
124 releaseErr := owner.Release()
125 if releaseErr != nil {
126 log.Error("lifecycle_owner_release_failed", "path", lockPath, "reason_code", "release-failed")
127 } else {
128 log.Info("lifecycle_owner_released", "path", lockPath)
129 }
130 retErr = errors.Join(retErr, releaseErr)
131 }()
132 if start == nil {
133 return errors.New("nil owned daemon startup")
134 }
135 return start(ctx)
136}
137
138// runDaemon runs the daemon in the foreground (the hidden --daemon path,
139// entered by an auto-spawned child) while holding exclusive lifecycle
140// ownership through complete teardown.
141func runDaemon() (retErr error) {
142 ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
143 defer stop()
144 log, logCloser, err := configureLogging(logging.Daemon, true)
145 if err != nil {
146 return err
147 }
148 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
149 return runWithLifecycleOwnerDeps(ctx, ipc.SocketDir(), platform.StateDir(), func(ctx context.Context) error {
150 return runDaemonOwnedWithLogger(ctx, log)
151 }, lifecycleStartupDependencies(log))
152}
153
154func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
155 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
156}
157
158func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
159 healthy, fresh, broken := 0, 0, 0
160 for _, record := range records {
161 switch {
162 case record.DegradedReason != "":
163 broken++
164 case record.Committed == nil:
165 fresh++
166 default:
167 healthy++
168 }
169 }
170 log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "broken", broken)
171}
172
173func runDaemonOwned(ctx context.Context) (retErr error) {
174 log, logCloser, err := configureLogging(logging.Daemon, true)
175 if err != nil {
176 return err
177 }
178 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
179 return runDaemonOwnedWithLogger(ctx, log)
180}
181
182func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
183 clk := clock.New()
184 observer, observerCloser, err := newPerformanceTrace(clk)
185 if err != nil {
186 return fmt.Errorf("performance trace: %w", err)
187 }
188 if observerCloser != nil {
189 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
190 }
191 if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" {
192 if !pprofAddrIsLoopback(addr) {
193 log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr)
194 }
195 go func() {
196 if err := http.ListenAndServe(addr, nil); err != nil && !errors.Is(err, http.ErrServerClosed) {
197 log.Error("pprof server exited", "err", err)
198 }
199 }()
200 log.Info("pprof enabled", "addr", addr)
201 }
202
203 var daemonOpts []daemon.Option
204 developmentTempOpt, err := developmentTempDirOption(safedir.EnsurePrivate)
205 if err != nil {
206 return err
207 }
208 if developmentTempOpt != nil {
209 daemonOpts = append(daemonOpts, developmentTempOpt)
210 }
211 if observer != nil {
212 daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
213 }
214 configPath := platform.ConfigPath()
215 cfg := loadConfigOrDefaults(log, configPath)
216 daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
217 daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
218 daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
219 snapshotRepository := snapshotadapter.NewRepositoryWithLogger(snapshotDir(), log)
220 daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository))
221 noticeStore := noticefile.New(platform.StateDir())
222 daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
223 stateDir := platform.StateDir()
224 storePath := persist.StorePath(stateDir)
225 opened, err := openCatalogue(stateDir)
226 if err != nil {
227 log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
228 if errors.Is(err, persist.ErrCatalogueUnreadable) {
229 return unreadableCatalogueError(stateDir)
230 }
231 return fmt.Errorf("open durable session state %s: %w", storePath, err)
232 }
233 recoveryMode := "current"
234 if opened.NewInstall {
235 recoveryMode = "new-install"
236 }
237 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
238 if opened.Migration.Performed {
239 recoveryMode = "catalogue-migrated"
240 log.Info("catalogue_migrated", "source_formats", opened.Migration.SourceFormats, "target_format", opened.Migration.TargetFormat, "records", opened.Migration.RecordCount, "backup", opened.Migration.BackupPath)
241 }
242 logCatalogueRecovery(log, opened.Records, recoveryMode)
243 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
244 daemonOpts = append(daemonOpts, daemon.WithSnapshotGarbageCollection())
245 log.Info("session persistence enabled", "path", storePath)
246 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
247
248 // Construct the catalogue-backed expected-session registry before socket
249 // publication. Phase 3 snapshot restoration remains asynchronous in Serve.
250 identity, err := daemonidentity.LoadOrCreate(stateDir)
251 if err != nil {
252 return fmt.Errorf("establish daemon identity: %w", err)
253 }
254 incarnation, err := daemonidentity.NewIncarnation()
255 if err != nil {
256 return fmt.Errorf("establish daemon incarnation: %w", err)
257 }
258 binding, err := daemonmux.NewServerBindings(identity, incarnation, []daemonmux.ServerPolicyAdmission{
259 {Policy: localDaemonPolicy(), Origin: ports.SessionOriginLocal},
260 {Policy: remoteBrokerPolicy(hostTransportQUIC), Origin: ports.SessionOriginRemote},
261 {Policy: remoteBrokerPolicy(hostTransportSSH), Origin: ports.SessionOriginRemote},
262 })
263 if err != nil {
264 return fmt.Errorf("establish daemon mux authority: %w", err)
265 }
266 aggregate := daemonmux.NewAggregateListener()
267 muxSupervisor, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 0)
268 if err != nil {
269 return err
270 }
271 defer func() { retErr = errors.Join(retErr, muxSupervisor.Close()) }()
272 muxListener, err := ipc.ListenMux(daemonmux.SocketPath(ipc.SocketDir()))
273 if err != nil {
274 return fmt.Errorf("listen daemon mux: %w", err)
275 }
276 defer func() { retErr = errors.Join(retErr, muxListener.Close()) }()
277 go func() {
278 for {
279 raw, acceptErr := muxListener.Accept()
280 if acceptErr != nil {
281 return
282 }
283 go func() { _ = muxSupervisor.Adopt(ctx, raw) }()
284 }
285 }()
286
287 // The catalogue-backed registry exists before the socket is published.
288 // Snapshot GC runs in the background once restoration finishes, so it never
289 // delays socket publication.
290 d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
291 ln, err := listenDaemon(ipc.SocketDir(), observer)
292 if err != nil {
293 closeErr := opened.Catalogue.Close()
294 log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
295 return errors.Join(fmt.Errorf("daemon listen: %w", err), closeErr)
296 }
297 defer func() { _ = ln.Close() }()
298 log.Info("daemon starting", "socket", ln.Addr())
299
300 watchCtx, stopWatch := context.WithCancel(ctx)
301 defer stopWatch()
302 go func() {
303 if err := config.Watch(watchCtx, clk, configPath, func(cfg domain.Config, warnings []domain.Warning) {
304 logConfigWarnings(log, warnings)
305 d.ApplyConfig(cfg)
306 }); err != nil && !errors.Is(err, context.Canceled) {
307 log.Warn("config watcher stopped", "path", configPath, "err", err)
308 }
309 }()
310 if err := aggregate.Register(sessionwire.NewServerListener(ln)); err != nil {
311 return fmt.Errorf("register local session listener: %w", err)
312 }
313 if err := d.Serve(ctx, aggregate); err != nil {
314 log.Error("daemon exited", "err", err)
315 return err
316 }
317 log.Info("daemon exited cleanly")
318 return nil
319}