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/domain/layout"
13 "github.com/bnema/vev/internal/protocol"
14 snapcodec "github.com/bnema/vev/internal/snapshotcodec"
15)
16
17func (d *Daemon) closeRestoreDone() {
18 d.mu.Lock()
19 for name, entry := range d.inactive {
20 if entry.restoreDone == nil {
21 continue
22 }
23 select {
24 case <-entry.restoreDone:
25 continue
26 default:
27 }
28 entry.state = protocol.SessionBroken
29 if entry.record.DegradedReason == "" {
30 entry.record.DegradedReason = "restore unavailable"
31 }
32 d.inactive[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, epoch uint64) 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)
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, epoch)
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. Pane terminal identity is stable and does
169 // not depend on the daemon's startup environment.
170 restoreEnv := copyEnvironment(d.baseEnv)
171 allowlist := d.restoreProcessAllowlistSnapshot()
172 stoppedTabNames := d.restoredSessionTabNames(snap.Name)
173 opened := make([]*tab, 0, len(snap.Tabs))
174 for index, tabSnap := range snap.Tabs {
175 if err := ctx.Err(); err != nil {
176 closeRestoredTabs(opened)
177 return nil, err
178 }
179 tabName := ""
180 if index < len(stoppedTabNames) {
181 tabName = stoppedTabNames[index]
182 }
183 tb, err := d.restoreSnapshotTab(ctx, sctx, snap.Name, tabName, tabSnap, restoreEnv, allowlist)
184 if err != nil {
185 closeRestoredTabs(opened)
186 return nil, err
187 }
188 opened = append(opened, tb)
189 }
190 return opened, nil
191}
192
193// closeRestoredTabs releases every PTY and pane context still owned by a
194// failed restore. Registered sessions transfer this ownership to the daemon.
195func closeRestoredTabs(tabs []*tab) {
196 for _, tb := range tabs {
197 tb.closeAllPanes()
198 }
199}
200
201func (d *Daemon) restoredSessionTabNames(name string) []string {
202 d.mu.Lock()
203 defer d.mu.Unlock()
204 return append([]string(nil), d.inactive[name].tabNames...)
205}
206
207func (d *Daemon) restoreSnapshotTab(ctx, sctx context.Context, sessionName, tabName string, tabSnap snapcodec.Tab, restoreEnv []string, allowlist map[string]struct{}) (*tab, error) {
208 tbSize := domain.Size{Cols: int(tabSnap.Cols), Rows: int(tabSnap.Rows)}
209 if !tbSize.Valid() {
210 return nil, fmt.Errorf("snapshot: invalid tab size")
211 }
212 placements, ok := layout.Solve(tabSnap.Tree.Root, domain.Rect{Width: tbSize.Cols, Height: tbSize.Rows})
213 if !ok {
214 return nil, fmt.Errorf("snapshot: unsolvable layout")
215 }
216 placementByPane := make(map[layout.PaneID]domain.Rect, len(placements))
217 for _, placement := range placements {
218 placementByPane[placement.ID] = placement.Content
219 }
220 tabStableID := tabSnap.StableID
221 if tabStableID == "" {
222 var err error
223 tabStableID, err = newStableID("t")
224 if err != nil {
225 return nil, fmt.Errorf("snapshot: generating tab identity: %w", err)
226 }
227 }
228 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}
229 if tb.nextPaneID <= 0 {
230 tb.nextPaneID = 1
231 }
232 tb.ctx, tb.cancel = context.WithCancel(sctx)
233 for _, paneSnap := range tabSnap.Panes {
234 if err := ctx.Err(); err != nil {
235 tb.closeAllPanes()
236 return nil, err
237 }
238 contentRect, ok := placementByPane[paneSnap.ID]
239 if !ok {
240 tb.closeAllPanes()
241 return nil, fmt.Errorf("snapshot: missing pane placement")
242 }
243 p, err := d.restoreSnapshotPane(ctx, sessionName, tabStableID, paneSnap, contentRect, tbSize, restoreEnv, allowlist, tb.ctx)
244 if err != nil {
245 tb.closeAllPanes()
246 return nil, err
247 }
248 tb.panes[paneSnap.ID] = p
249 }
250 return tb, nil
251}
252
253func (d *Daemon) restoreSnapshotPane(ctx context.Context, sessionName, tabStableID string, paneSnap snapcodec.Pane, contentRect domain.Rect, tabSize domain.Size, restoreEnv []string, allowlist map[string]struct{}, tabCtx context.Context) (*pane, error) {
254 paneStableID := paneSnap.StableID
255 if paneStableID == "" {
256 var err error
257 paneStableID, err = newStableID("p")
258 if err != nil {
259 return nil, fmt.Errorf("snapshot: generating pane identity: %w", err)
260 }
261 }
262 launch := d.shellLaunch(restoreEnv)
263 lifetime := d.newPaneProcessLifetime(ctx, tabCtx)
264 ptyGeometry := domain.Geometry{Size: restorePTYSize(contentRect, tabSize)}
265 pty, err := d.ptys.Open(lifetime.ctx, launch.command, launch.args, childEnvFrom(restoreEnv, sessionName, tabStableID, paneStableID), paneSnap.Cwd, ptyGeometry)
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, ptyGeometry.Size, launch.title, d.currentHistoryConfig())
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) *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, env: copyEnvironment(d.baseEnv), envProvisional: true, 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.inactive[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.
316// The epoch revalidation makes publication a true admitted transition: if a
317// KillAll purge began and is waiting on this admitted restore, the session is
318// not published, so it cannot resurrect a record the purge is about to remove.
319func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *session, epoch uint64) (bool, error) {
320 if sess.incarnation == (domain.IncarnationID{}) {
321 d.mu.Lock()
322 stopped := d.inactive[sess.name]
323 d.mu.Unlock()
324 sess.incarnation = stopped.incarnation
325 if sess.incarnation == (domain.IncarnationID{}) {
326 var err error
327 sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
328 if err != nil {
329 return false, fmt.Errorf("snapshot: generate durable identity: %w", err)
330 }
331 }
332 }
333 // Restoration is catalogue-authorized. Missing or unreadable authority is a
334 // hard failure.
335 if d.persistEnabled {
336 record, ok, err := d.catalogueRecord(sess.name)
337 if err != nil {
338 return false, err
339 }
340 if !ok {
341 return false, errors.New("snapshot: restored session is absent from catalogue")
342 }
343 if record.IncarnationID != sess.incarnation {
344 return false, errors.New("snapshot: restored session incarnation differs from catalogue")
345 }
346 }
347 d.mu.Lock()
348 defer d.mu.Unlock()
349 if d.purgeSupersededLocked(epoch) || d.findByNameLocked(sess.name) != nil || ctx.Err() != nil {
350 return false, nil
351 }
352 sess.id = domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
353 d.nextID++
354 stopped := d.inactive[sess.name]
355 sess.mu.Lock()
356 sess.restoreDone = stopped.restoreDone
357 sess.mu.Unlock()
358 if !d.registerSessionLocked(sess) {
359 return false, errors.New("daemon: session registry rejected restored session")
360 }
361 // The stopped entry is the durable authority and restore barrier until the
362 // runtime identity has actually been published. A registry rejection must
363 // leave that exact entry available for attach/recovery retries.
364 delete(d.inactive, sess.name)
365 sess.snapDirty.Store(false)
366 return true, nil
367}
368
369func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {
370 if len(snap.Tail) == 0 {
371 return fmt.Errorf("snapshot history: missing tail blob")
372 }
373 if len(snap.Transcript) == 0 {
374 return fmt.Errorf("snapshot recovery transcript: missing transcript blob")
375 }
376 p.mu.Lock()
377 defer p.mu.Unlock()
378 screen, err := vt.NewScreenWithRecoveryTranscript(p.screen.Columns(), p.screen.Rows(), p.history.Limits(), snap.SealedChunks, snap.Tail, snap.Transcript)
379 if err != nil {
380 return fmt.Errorf("snapshot recovery: %w", err)
381 }
382 p.screen = screen
383 p.history = screen.History()
384 return nil
385}
386
387func restorePTYSize(contentRect domain.Rect, tabSize domain.Size) domain.Size {
388 cols := contentRect.Width
389 if cols <= 0 {
390 cols = tabSize.Cols
391 }
392 if cols < layout.MinPaneCols {
393 cols = layout.MinPaneCols
394 }
395 rows := contentRect.Height
396 if rows <= 0 {
397 rows = min(max(tabSize.Rows, layout.MinPaneRows), 24)
398 }
399 if rows < layout.MinPaneRows {
400 rows = layout.MinPaneRows
401 }
402 return domain.Size{Cols: cols, Rows: rows}
403}
404
405// scheduleSnapshot captures a session only once while an immutable capture is
406// queued or being written. It never waits for the bounded worker queue.