1package daemon
2
3import (
4 "context"
5 "crypto/rand"
6 "errors"
7 "fmt"
8 "math"
9
10 vt "github.com/bnema/vev-vt"
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/usecase/layout"
14 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
15)
16
17func (d *Daemon) closeRestoreDone() {
18 d.mu.Lock()
19 for name, entry := range d.stopped {
20 if entry.restoreDone == nil {
21 continue
22 }
23 select {
24 case <-entry.restoreDone:
25 continue
26 default:
27 }
28 entry.state = ports.SessionBroken
29 if entry.record.DegradedReason == "" {
30 entry.record.DegradedReason = "restore unavailable"
31 }
32 d.stopped[name] = entry
33 closeRuntimeRestoreDoneLocked(entry.restoreDone)
34 }
35 d.mu.Unlock()
36 d.restoreOnce.Do(func() { close(d.restoreDone) })
37}
38
39// startSnapshotRestoration launches catalogue-driven restoration as a durable
40// writer. Restoration reconciles committed checkpoints, so lifecycle ownership
41// must outlive it exactly as it outlives the snapshot and maintenance workers. The completion
42// channel is registered before the goroutine starts, so shutdown can never
43// begin waiting while this writer is still unregistered.
44func (d *Daemon) startSnapshotRestoration() {
45 if d == nil {
46 return
47 }
48 d.snapshotWorkerMu.Lock()
49 if d.restoreWorkerDone != nil {
50 d.snapshotWorkerMu.Unlock()
51 return
52 }
53 done := make(chan struct{})
54 d.restoreWorkerDone = done
55 d.snapshotWorkerMu.Unlock()
56 d.sessWg.Go(func() {
57 defer close(done)
58 d.restoreSnapshots(d.serveCtx)
59 })
60}
61
62func (d *Daemon) restoreSnapshots(ctx context.Context) {
63 if d == nil {
64 return
65 }
66 defer d.logStartupRecoveryCounts(0)
67 defer d.closeRestoreDone()
68 if ns := d.noticeStore; ns != nil {
69 claimed, err := ns.Claim()
70 if err != nil {
71 d.log.Warn("claiming pending notices failed", "err", err)
72 } else {
73 for _, n := range claimed {
74 d.notices.record(n)
75 d.notices.queueGlobal(n)
76 }
77 // Do not discard the claim until every notice has been recorded and
78 // queued. An Ack failure is safe: the claim is replayed at startup.
79 if err := ns.Ack(); err != nil {
80 d.log.Warn("acknowledging pending notices failed", "err", err)
81 }
82 }
83 }
84 if !d.snapsEnabled {
85 return
86 }
87 if d.snapshotRepository != nil {
88 d.restoreIncrementalSnapshots(ctx)
89 }
90
91}
92
93func validateRestoreSessionSnapshot(snap snapcodec.Session) error {
94 if len(snap.Tabs) == 0 {
95 if snap.Active != 0 {
96 return fmt.Errorf("snapshot: active tab out of range")
97 }
98 } else if int(snap.Active) >= len(snap.Tabs) {
99 return fmt.Errorf("snapshot: active tab out of range")
100 }
101 if snap.Name == "" {
102 return fmt.Errorf("snapshot: empty session name")
103 }
104 if snap.CreatedAt > math.MaxInt64 {
105 return fmt.Errorf("snapshot: created_at overflows int64")
106 }
107 return nil
108}
109
110// restoreSession owns the restored tabs until registration succeeds. Every
111// unsuccessful path closes their PTYs and cancels their shared session context.
112func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64, checkpoint domain.CheckpointRef) error {
113 if err := ctx.Err(); err != nil {
114 return err
115 }
116 if err := validateRestoreSessionSnapshot(snap); err != nil {
117 return err
118 }
119 if d.restoredSessionAlreadyExists(snap.Name) {
120 return nil
121 }
122
123 parent := d.serveCtx
124 if parent == nil {
125 parent = context.Background()
126 }
127 sctx, cancel := context.WithCancel(parent)
128 opened, err := d.restoreSnapshotTabs(ctx, sctx, snap)
129 if err != nil {
130 cancel()
131 return err
132 }
133 if len(opened) == 0 {
134 cancel()
135 return nil
136 }
137 ownsOpened := true
138 defer func() {
139 if ownsOpened {
140 closeRestoredTabs(opened)
141 cancel()
142 }
143 }()
144
145 sess := d.newRestoredSession(snap, sctx, cancel, opened, terminalEnvFromEnvironment(d.baseEnv))
146 // The loaded manifest is the repository head for this name. Future dirty
147 // checkpoints must continue from it rather than reuse generation one.
148 sess.snapshotPublishedGeneration = repositoryGeneration
149 sess.snapshotPublishedCheckpoint = &checkpoint
150 registered, err := d.persistAndRegisterRestoredSession(ctx, sess)
151 if err != nil || !registered {
152 return err
153 }
154 ownsOpened = false
155 for _, tb := range opened {
156 d.startTabGoroutines(sess, tb)
157 }
158 return nil
159}
160
161func (d *Daemon) restoredSessionAlreadyExists(name string) bool {
162 d.mu.Lock()
163 defer d.mu.Unlock()
164 return d.closing || d.findByNameLocked(name) != nil
165}
166
167func (d *Daemon) restoreSnapshotTabs(ctx, sctx context.Context, snap snapcodec.Session) ([]*tab, error) {
168 // Restore runs before client Hello. Use the daemon's startup environment,
169 // including its terminal capability, until an attach supplies a fresher one.
170 restoreEnv := copyEnvironment(d.baseEnv)
171 restoreTerm := terminalEnvFromEnvironment(restoreEnv)
172 allowlist := d.restoreProcessAllowlistSnapshot()
173 stoppedTabNames := d.restoredSessionTabNames(snap.Name)
174 opened := make([]*tab, 0, len(snap.Tabs))
175 for index, tabSnap := range snap.Tabs {
176 if err := ctx.Err(); err != nil {
177 closeRestoredTabs(opened)
178 return nil, err
179 }
180 tabName := ""
181 if index < len(stoppedTabNames) {
182 tabName = stoppedTabNames[index]
183 }
184 tb, err := d.restoreSnapshotTab(ctx, sctx, snap.Name, tabName, tabSnap, restoreEnv, restoreTerm, allowlist)
185 if err != nil {
186 closeRestoredTabs(opened)
187 return nil, err
188 }
189 opened = append(opened, tb)
190 }
191 return opened, nil
192}
193
194// closeRestoredTabs releases every PTY and pane context still owned by a
195// failed restore. Registered sessions transfer this ownership to the daemon.
196func closeRestoredTabs(tabs []*tab) {
197 for _, tb := range tabs {
198 tb.closeAllPanes()
199 }
200}
201
202func (d *Daemon) restoredSessionTabNames(name string) []string {
203 d.mu.Lock()
204 defer d.mu.Unlock()
205 return append([]string(nil), d.stopped[name].tabNames...)
206}
207
208func (d *Daemon) restoreSnapshotTab(ctx, sctx context.Context, sessionName, tabName string, tabSnap snapcodec.Tab, restoreEnv []string, restoreTerm terminalEnv, allowlist map[string]struct{}) (*tab, error) {
209 tbSize := domain.Size{Cols: int(tabSnap.Cols), Rows: int(tabSnap.Rows)}
210 if !tbSize.Valid() {
211 return nil, fmt.Errorf("snapshot: invalid tab size")
212 }
213 placements, ok := layout.Solve(tabSnap.Tree.Root, domain.Rect{Width: tbSize.Cols, Height: tbSize.Rows})
214 if !ok {
215 return nil, fmt.Errorf("snapshot: unsolvable layout")
216 }
217 placementByPane := make(map[layout.PaneID]domain.Rect, len(placements))
218 for _, placement := range placements {
219 placementByPane[placement.ID] = placement.Content
220 }
221 tabStableID := tabSnap.StableID
222 if tabStableID == "" {
223 var err error
224 tabStableID, err = newStableID("t")
225 if err != nil {
226 return nil, fmt.Errorf("snapshot: generating tab identity: %w", err)
227 }
228 }
229 tb := &tab{stableID: tabStableID, name: tabName, tree: tabSnap.Tree.Clone(), panes: make(map[layout.PaneID]*pane, len(tabSnap.Panes)), nextPaneID: int(tabSnap.NextPaneID), size: tbSize}
230 if tb.nextPaneID <= 0 {
231 tb.nextPaneID = 1
232 }
233 tb.ctx, tb.cancel = context.WithCancel(sctx)
234 for _, paneSnap := range tabSnap.Panes {
235 if err := ctx.Err(); err != nil {
236 tb.closeAllPanes()
237 return nil, err
238 }
239 contentRect, ok := placementByPane[paneSnap.ID]
240 if !ok {
241 tb.closeAllPanes()
242 return nil, fmt.Errorf("snapshot: missing pane placement")
243 }
244 p, err := d.restoreSnapshotPane(ctx, sessionName, tabStableID, paneSnap, contentRect, tbSize, restoreEnv, restoreTerm, allowlist, tb.ctx)
245 if err != nil {
246 tb.closeAllPanes()
247 return nil, err
248 }
249 tb.panes[paneSnap.ID] = p
250 }
251 return tb, nil
252}
253
254func (d *Daemon) restoreSnapshotPane(ctx context.Context, sessionName, tabStableID string, paneSnap snapcodec.Pane, contentRect domain.Rect, tabSize domain.Size, restoreEnv []string, restoreTerm terminalEnv, allowlist map[string]struct{}, tabCtx context.Context) (*pane, error) {
255 paneStableID := paneSnap.StableID
256 if paneStableID == "" {
257 var err error
258 paneStableID, err = newStableID("p")
259 if err != nil {
260 return nil, fmt.Errorf("snapshot: generating pane identity: %w", err)
261 }
262 }
263 launch := d.shellLaunch(restoreEnv)
264 lifetime := d.newPaneProcessLifetime(ctx, tabCtx)
265 pty, err := d.ptys.Open(lifetime.ctx, launch.command, launch.args, childEnvFrom(restoreEnv, sessionName, tabStableID, paneStableID, restoreTerm), paneSnap.Cwd, restorePTYSize(contentRect, tabSize))
266 if err != nil {
267 lifetime.abort()
268 if pty != nil {
269 _ = pty.Close()
270 }
271 return nil, err
272 }
273 p := newPaneWithStableIDAndTitle(paneSnap.ID, paneStableID, pty, restorePTYSize(contentRect, tabSize), launch.title)
274 if !lifetime.publish(p) {
275 _ = pty.Close()
276 return nil, lifetime.ctx.Err()
277 }
278 p.rect = contentRect
279 if err := restorePaneTerminal(p, paneSnap); err != nil {
280 closePaneProcess(p)
281 return nil, err
282 }
283 if decision := planProcessRestore(paneSnap.Process, allowlist); decision.Restore {
284 if _, err := pty.Write([]byte(decision.Command + "\n")); err != nil {
285 d.log.Warn("writing snapshot restore command failed", "err", err, "session", sessionName, "pane", paneSnap.ID)
286 d.NotifyGlobal(domain.NoticeWarn, domain.NoticeAutoResume, "couldn't restore the running program in session "+sessionName, err)
287 }
288 }
289 return p, nil
290}
291
292func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context, cancel context.CancelFunc, tabs []*tab, term terminalEnv) *session {
293 // Restored sessions keep ordered tabs; attachment repair deterministically
294 // selects the first tab instead of restoring an interactive client view.
295 sess := &session{sessionCore: sessionCore{name: snap.Name, createdAt: int64(snap.CreatedAt)}, ctx: sctx, cancel: cancel, tabs: tabs, terminal: term, env: copyEnvironment(d.baseEnv), snapshotWake: d.snapshotWake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)}
296 // Restored tabs remain private until persistAndRegisterRestoredSession.
297 // Initialize owners now so registration and reader startup cannot expose an
298 // ownerless pane.
299 for _, tb := range tabs {
300 publishTiledPaneOwners(sess, tb)
301 }
302 sess.snapEligible.Store(true)
303 if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 {
304 sess.cwd = snap.Tabs[0].Panes[0].Cwd
305 }
306 d.mu.Lock()
307 if stopped, ok := d.stopped[snap.Name]; ok {
308 sess.mruAt.Store(stopped.lastUsedSeq)
309 }
310 d.mu.Unlock()
311 return sess
312}
313
314// persistAndRegisterRestoredSession atomically transfers a successfully
315// persisted session to the daemon. The caller retains cleanup on false/error.
316func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *session) (bool, error) {
317 if sess.incarnation == (domain.IncarnationID{}) {
318 d.mu.Lock()
319 stopped := d.stopped[sess.name]
320 d.mu.Unlock()
321 sess.incarnation = stopped.incarnation
322 if sess.incarnation == (domain.IncarnationID{}) {
323 var err error
324 sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
325 if err != nil {
326 return false, fmt.Errorf("snapshot: generate durable identity: %w", err)
327 }
328 }
329 }
330 // Restoration is catalogue-authorized. Missing or unreadable authority is a
331 // hard failure.
332 if d.persistEnabled {
333 record, ok, err := d.catalogueRecord(sess.name)
334 if err != nil {
335 return false, err
336 }
337 if !ok {
338 return false, errors.New("snapshot: restored session is absent from catalogue")
339 }
340 if record.IncarnationID != sess.incarnation {
341 return false, errors.New("snapshot: restored session incarnation differs from catalogue")
342 }
343 }
344 d.mu.Lock()
345 defer d.mu.Unlock()
346 if d.closing || d.findByNameLocked(sess.name) != nil || ctx.Err() != nil {
347 return false, nil
348 }
349 sess.id = domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
350 d.nextID++
351 stopped := d.stopped[sess.name]
352 sess.mu.Lock()
353 sess.restoreDone = stopped.restoreDone
354 sess.mu.Unlock()
355 if !d.registerSessionLocked(sess) {
356 return false, errors.New("daemon: session registry rejected restored session")
357 }
358 // The stopped entry is the durable authority and restore barrier until the
359 // runtime identity has actually been published. A registry rejection must
360 // leave that exact entry available for attach/recovery retries.
361 delete(d.stopped, sess.name)
362 sess.snapDirty.Store(false)
363 return true, nil
364}
365
366func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {
367 if len(snap.Tail) == 0 {
368 return fmt.Errorf("snapshot history: missing tail blob")
369 }
370 if len(snap.Transcript) == 0 {
371 return fmt.Errorf("snapshot recovery transcript: missing transcript blob")
372 }
373 p.mu.Lock()
374 defer p.mu.Unlock()
375 screen, err := vt.NewScreenWithRecoveryTranscript(p.screen.Frame.Width, p.screen.Frame.Height, vt.HistoryConfig{MaxRows: defaultScrollbackRows, MaxCells: defaultScrollbackCells}, snap.SealedChunks, snap.Tail, snap.Transcript)
376 if err != nil {
377 return fmt.Errorf("snapshot recovery: %w", err)
378 }
379 p.screen = screen
380 p.history = screen.History()
381 return nil
382}
383
384func restorePTYSize(contentRect domain.Rect, tabSize domain.Size) domain.Size {
385 cols := contentRect.Width
386 if cols <= 0 {
387 cols = tabSize.Cols
388 }
389 if cols < layout.MinPaneCols {
390 cols = layout.MinPaneCols
391 }
392 rows := contentRect.Height
393 if rows <= 0 {
394 rows = min(max(tabSize.Rows, layout.MinPaneRows), 24)
395 }
396 if rows < layout.MinPaneRows {
397 rows = layout.MinPaneRows
398 }
399 return domain.Size{Cols: cols, Rows: rows}
400}
401
402// scheduleSnapshot captures a session only once while an immutable capture is
403// queued or being written. It never waits for the bounded worker queue.