feat(snapshot): rate-limit checkpoints

commit: 0abd08bc88b9be58c7907d21492de68529fdc1d3
author: brice · 3mo ago · 2026-07-21 21:50 UTC
parent: 025da44
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +62 -22

M internal/usecase/daemon/snapshot.go +62 -22


@@ -43,17 +43,24 @@ type snapshotCaptureTab struct {
43	43		}
44	44
45	45		type snapshotCapturePane struct {
46		-	id layout.PaneID
47		-	stableID string
48		-	cwd string
	46	+	id layout.PaneID
	47	+	stableID string
	48	+	cwd string
	49	+	// history is retained for the v3 compatibility encoder. Incremental
	50	+	// captures keep sealed identities and a copied tail separately so capture
	51	+	// never rotates a live mutable tail merely to persist it.
49	52		history vt.HistoryView
	53	+	sealed vt.HistorySnapshotView
	54	+	tail vt.HistoryView
50	55		visible []byte
51	56		visibleErr error
52	57		process *snapcodec.Process
53	58		}
54	59
55	60		const (
56		-	snapshotInterval = 30 * time.Second
	61	+	// Routine checkpoints are rate limited from completion; forced teardown
	62	+	// checkpoints use the same bounded worker but never consume this window.
	63	+	snapshotInterval = 2 * time.Minute
57	64		snapshotFinalFlushTimeout = time.Second
58	65		)
59	66

@@ -82,7 +89,14 @@ func (d *Daemon) restoreSnapshots(ctx context.Context) {
82	89		}
83	90		}
84	91		}
85		-	if !d.snapsEnabled \|\| d.snaps == nil {
	92	+	if !d.snapsEnabled {
	93	+	return
	94	+	}
	95	+	if d.snapshotRepository != nil {
	96	+	d.restoreIncrementalSnapshots(ctx)
	97	+	return
	98	+	}
	99	+	if d.snaps == nil {
86	100		return
87	101		}
88	102		blobs, err := d.snaps.Load()

@@ -139,7 +153,11 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err
139	153		}
140	154		d.mu.Unlock()
141	155
142		-	sctx, cancel := context.WithCancel(d.serveCtx)
	156	+	parent := d.serveCtx
	157	+	if parent == nil {
	158	+	parent = context.Background()
	159	+	}
	160	+	sctx, cancel := context.WithCancel(parent)
143	161		opened := make([]*tab, 0, len(snap.Tabs))
144	162		// Snapshot restore runs before any client Hello is available. Start panes
145	163		// from the daemon environment captured at startup; the next attach replaces

@@ -318,7 +336,7 @@ func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {
318	336		}
319	337		p.mu.Lock()
320	338		defer p.mu.Unlock()
321		-	screen, err := vt.NewScreenWithRestoredHistory(p.screen.Frame.Width, p.screen.Frame.Height, vt.HistoryConfig{MaxRows: defaultScrollbackRows}, snap.SealedChunks, snap.Tail)
	339	+	screen, err := vt.NewScreenWithRestoredHistory(p.screen.Frame.Width, p.screen.Frame.Height, vt.HistoryConfig{MaxRows: defaultScrollbackRows, MaxCells: defaultScrollbackCells}, snap.SealedChunks, snap.Tail)
322	340		if err != nil {
323	341		return fmt.Errorf("snapshot history: %w", err)
324	342		}

@@ -362,7 +380,7 @@ func (d Daemon) scheduleFinalSnapshot(sess session) bool {
362	380		}
363	381
364	382		func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bool {
365		-	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil \|\| !sess.snapEligible.Load() {
	383	+	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| (d.snapshotRepository == nil && d.snaps == nil) \|\| !sess.snapEligible.Load() {
366	384		return true
367	385		}
368	386		sess.snapshotMu.Lock()

@@ -496,10 +514,15 @@ func (d Daemon) captureSnapshotState(sess session, generation uint64) (*snapsh
496	514		paneCapture := snapshotCapturePane{
497	515		id: p.id,
498	516		stableID: p.stableID,
499		-	history: p.history.SealAndView(),
500	517		visible: visible,
501	518		visibleErr: visibleErr,
502	519		}
	520	+	if d.snapshotRepository != nil {
	521	+	paneCapture.sealed = p.history.SnapshotView()
	522	+	paneCapture.tail = paneCapture.sealed.Tail()
	523	+	} else {
	524	+	paneCapture.history = p.history.SealAndView()
	525	+	}
503	526		p.mu.Unlock()
504	527		paneCapture.cwd = fallbackCwd
505	528		if d.procCwd != nil && pid > 0 {

@@ -601,9 +624,20 @@ func (d *Daemon) startSnapshotEncodeWorker() {
601	624		d.finishSnapshotCapture(capture, false)
602	625		return false
603	626		}
604		-	data, err := d.encodeSnapshotCapture(capture)
605		-	if err == nil && workerCtx.Err() == nil {
606		-	err = d.snaps.Write(capture.name, data)
	627	+	var err error
	628	+	if d.snapshotRepository != nil {
	629	+	publication, publicationErr := d.incrementalPublication(capture)
	630	+	if publicationErr != nil {
	631	+	err = publicationErr
	632	+	} else if workerCtx.Err() == nil {
	633	+	err = d.snapshotRepository.Publish(workerCtx, publication)
	634	+	}
	635	+	} else {
	636	+	data, encodeErr := d.encodeSnapshotCapture(capture)
	637	+	err = encodeErr
	638	+	if err == nil && workerCtx.Err() == nil {
	639	+	err = d.snaps.Write(capture.name, data)
	640	+	}
607	641		}
608	642		d.clearSnapshotWorkerInFlight(workerID, capture)
609	643		if err != nil && workerCtx.Err() == nil {

@@ -781,15 +815,7 @@ func (d *Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error)
781	815		}
782	816
783	817		func (d *Daemon) snapshotSaver(ctx context.Context) {
784		-	t := d.clock.NewTimer(snapshotInterval)
785		-	defer t.Stop()
786		-	for {
787		-	select {
788		-	case <-ctx.Done():
789		-	return
790		-	case <-t.C():
791		-	}
792		-
	818	+	checkpoint := func() {
793	819		d.mu.Lock()
794	820		sessions := d.sessionsSnapshotLocked()
795	821		d.mu.Unlock()

@@ -798,7 +824,21 @@ func (d *Daemon) snapshotSaver(ctx context.Context) {
798	824		d.scheduleSnapshot(sess)
799	825		}
800	826		}
801		-	t.Reset(snapshotInterval)
	827	+	}
	828	+	// A session born dirty must not wait through a full interval before its
	829	+	// first crash-safe checkpoint. Producers remain non-blocking because each
	830	+	// schedule call only attempts the bounded handoff.
	831	+	checkpoint()
	832	+	t := d.clock.NewTimer(snapshotInterval)
	833	+	defer t.Stop()
	834	+	for {
	835	+	select {
	836	+	case <-ctx.Done():
	837	+	return
	838	+	case <-t.C():
	839	+	checkpoint()
	840	+	t.Reset(snapshotInterval)
	841	+	}
802	842		}
803	843		}
804	844

 }
 type snapshotCapturePane struct {
-	id         layout.PaneID
+	id       layout.PaneID
-	stableID   string
+	stableID string
-	cwd        string
+	cwd      string
+	// history is retained for the v3 compatibility encoder. Incremental
+	// captures keep sealed identities and a copied tail separately so capture
+	// never rotates a live mutable tail merely to persist it.
 	history    vt.HistoryView
+	sealed     vt.HistorySnapshotView
+	tail       vt.HistoryView
 	visible    []byte
 	visibleErr error
 	process    *snapcodec.Process
 }
 const (
-	snapshotInterval          = 30 * time.Second
+	// Routine checkpoints are rate limited from completion; forced teardown
+	// checkpoints use the same bounded worker but never consume this window.
+	snapshotInterval          = 2 * time.Minute
 	snapshotFinalFlushTimeout = time.Second
 )
 			}
 		}
 	}
-	if !d.snapsEnabled || d.snaps == nil {
+	if !d.snapsEnabled {
+		return
+	}
+	if d.snapshotRepository != nil {
+		d.restoreIncrementalSnapshots(ctx)
+		return
+	}
+	if d.snaps == nil {
 		return
 	}
 	blobs, err := d.snaps.Load()
 	}
 	d.mu.Unlock()
-	sctx, cancel := context.WithCancel(d.serveCtx)
+	parent := d.serveCtx
+	if parent == nil {
+		parent = context.Background()
+	}
+	sctx, cancel := context.WithCancel(parent)
 	opened := make([]*tab, 0, len(snap.Tabs))
 	// Snapshot restore runs before any client Hello is available. Start panes
 	// from the daemon environment captured at startup; the next attach replaces
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	screen, err := vt.NewScreenWithRestoredHistory(p.screen.Frame.Width, p.screen.Frame.Height, vt.HistoryConfig{MaxRows: defaultScrollbackRows}, snap.SealedChunks, snap.Tail)
+	screen, err := vt.NewScreenWithRestoredHistory(p.screen.Frame.Width, p.screen.Frame.Height, vt.HistoryConfig{MaxRows: defaultScrollbackRows, MaxCells: defaultScrollbackCells}, snap.SealedChunks, snap.Tail)
 	if err != nil {
 		return fmt.Errorf("snapshot history: %w", err)
 	}
 }
 func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bool {
-	if d == nil || sess == nil || !d.snapsEnabled || d.snaps == nil || !sess.snapEligible.Load() {
+	if d == nil || sess == nil || !d.snapsEnabled || (d.snapshotRepository == nil && d.snaps == nil) || !sess.snapEligible.Load() {
 		return true
 	}
 	sess.snapshotMu.Lock()
 			paneCapture := snapshotCapturePane{
 				id:         p.id,
 				stableID:   p.stableID,
-				history:    p.history.SealAndView(),
 				visible:    visible,
 				visibleErr: visibleErr,
 			}
+			if d.snapshotRepository != nil {
+				paneCapture.sealed = p.history.SnapshotView()
+				paneCapture.tail = paneCapture.sealed.Tail()
+			} else {
+				paneCapture.history = p.history.SealAndView()
+			}
 			p.mu.Unlock()
 			paneCapture.cwd = fallbackCwd
 			if d.procCwd != nil && pid > 0 {
 				d.finishSnapshotCapture(capture, false)
 				return false
 			}
-			data, err := d.encodeSnapshotCapture(capture)
+			var err error
-			if err == nil && workerCtx.Err() == nil {
+			if d.snapshotRepository != nil {
-				err = d.snaps.Write(capture.name, data)
+				publication, publicationErr := d.incrementalPublication(capture)
+				if publicationErr != nil {
+					err = publicationErr
+				} else if workerCtx.Err() == nil {
+					err = d.snapshotRepository.Publish(workerCtx, publication)
+				}
+			} else {
+				data, encodeErr := d.encodeSnapshotCapture(capture)
+				err = encodeErr
+				if err == nil && workerCtx.Err() == nil {
+					err = d.snaps.Write(capture.name, data)
+				}
 			}
 			d.clearSnapshotWorkerInFlight(workerID, capture)
 			if err != nil && workerCtx.Err() == nil {
 }
 func (d *Daemon) snapshotSaver(ctx context.Context) {
-	t := d.clock.NewTimer(snapshotInterval)
+	checkpoint := func() {
-	defer t.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-t.C():
-		}
 		d.mu.Lock()
 		sessions := d.sessionsSnapshotLocked()
 		d.mu.Unlock()
 				d.scheduleSnapshot(sess)
 			}
 		}
-		t.Reset(snapshotInterval)
+	}
+	// A session born dirty must not wait through a full interval before its
+	// first crash-safe checkpoint. Producers remain non-blocking because each
+	// schedule call only attempts the bounded handoff.
+	checkpoint()
+	t := d.clock.NewTimer(snapshotInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C():
+			checkpoint()
+			t.Reset(snapshotInterval)
+		}
 	}
 }