feat(snapshot): incremental captures
- commit
- 025da44f4c84854e967084e9e532bd078bd466de
- author
- brice · · 2026-07-21 21:50 UTC
- parent
- 34db907
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +259 -40
M internal/usecase/daemon/daemon.go +70 -39
| @@ -32,8 +32,13 @@ const debounceInterval = minDebounceInterval | |||
| 32 | 32 | // ptyReadBufSize is the PTY reader's read buffer. | |
| 33 | 33 | const ptyReadBufSize = 32 * 1024 | |
| 34 | 34 | ||
| 35 | - | // defaultScrollbackRows is the per-tab retained history size. | |
| 36 | - | const defaultScrollbackRows = 10_000 | |
| 35 | + | // New and restored panes are bounded by both history rows and cells. The | |
| 36 | + | // cell budget keeps a very wide terminal from retaining disproportionately | |
| 37 | + | // more scrollback than the normal 160-column case. | |
| 38 | + | const ( | |
| 39 | + | defaultScrollbackRows = 10_000 | |
| 40 | + | defaultScrollbackCells = 12_000 * 160 | |
| 41 | + | ) | |
| 37 | 42 | ||
| 38 | 43 | // detachNotifyTimeout bounds the best-effort Detached notification on the | |
| 39 | 44 | // detach/kill/shutdown paths: if a wedged client (full kernel send buffer) | |
| @@ -101,21 +106,30 @@ type Daemon struct { | |||
| 101 | 106 | beforeCopyModeRevalidate func() | |
| 102 | 107 | // beforeCopyMouseMap is a deterministic seam after an immutable copy-input | |
| 103 | 108 | // snapshot is captured and before its mapped position is applied. | |
| 104 | - | beforeCopyMouseMap func() | |
| 105 | - | ptys ports.PTYFactory | |
| 106 | - | clock ports.Clock | |
| 107 | - | log *slog.Logger | |
| 108 | - | runtimeObserver ports.RuntimeObserver | |
| 109 | - | baseEnv []string | |
| 110 | - | shell string | |
| 111 | - | shellArgs []string | |
| 112 | - | shellOverride bool | |
| 113 | - | persist *persist.Persister | |
| 114 | - | persistEnabled bool | |
| 115 | - | snaps ports.SnapshotStore | |
| 116 | - | snapsEnabled bool | |
| 117 | - | noticeStore ports.NoticeStore | |
| 118 | - | snapshotMarshal func(snapcodec.Session) ([]byte, error) | |
| 109 | + | beforeCopyMouseMap func() | |
| 110 | + | ptys ports.PTYFactory | |
| 111 | + | clock ports.Clock | |
| 112 | + | log *slog.Logger | |
| 113 | + | runtimeObserver ports.RuntimeObserver | |
| 114 | + | baseEnv []string | |
| 115 | + | shell string | |
| 116 | + | shellArgs []string | |
| 117 | + | shellOverride bool | |
| 118 | + | persist *persist.Persister | |
| 119 | + | persistEnabled bool | |
| 120 | + | // snapshotRepository is the sole production checkpoint contract. snaps is | |
| 121 | + | // retained temporarily only for the v3 test bridge; application wiring | |
| 122 | + | // always installs snapshotRepository. | |
| 123 | + | snapshotRepository ports.SnapshotRepository | |
| 124 | + | legacySnapshots ports.LegacySnapshotSource | |
| 125 | + | snaps ports.SnapshotStore | |
| 126 | + | snapsEnabled bool | |
| 127 | + | noticeStore ports.NoticeStore | |
| 128 | + | snapshotMarshal func(snapcodec.Session) ([]byte, error) | |
| 129 | + | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | |
| 130 | + | // independent of pane state and is bounded to prevent checkpoint history | |
| 131 | + | // from becoming unbounded daemon memory. | |
| 132 | + | snapshotChunkCache *snapshotChunkCache | |
| 119 | 133 | snapshotJobs chan *snapshotCapture | |
| 120 | 134 | snapshotWorkerMu sync.Mutex | |
| 121 | 135 | snapshotWorkerID uint64 | |
| @@ -232,8 +246,18 @@ func WithStore(store ports.Store) Option { | |||
| 232 | 246 | } | |
| 233 | 247 | } | |
| 234 | 248 | ||
| 235 | - | // WithSnapshotStore enables durable named session snapshots. A nil store keeps | |
| 236 | - | // the daemon in no-op snapshot mode. | |
| 249 | + | // WithSnapshotRepository enables content-addressed incremental snapshots. The | |
| 250 | + | // legacy source is read-only migration input and is never used for new writes. | |
| 251 | + | func WithSnapshotRepository(repository ports.SnapshotRepository, legacy ports.LegacySnapshotSource) Option { | |
| 252 | + | return func(d *Daemon) { | |
| 253 | + | d.snapshotRepository = repository | |
| 254 | + | d.legacySnapshots = legacy | |
| 255 | + | d.snapsEnabled = repository != nil | |
| 256 | + | } | |
| 257 | + | } | |
| 258 | + | ||
| 259 | + | // WithSnapshotStore is the v3 compatibility seam for existing embedders. New | |
| 260 | + | // application wiring must use WithSnapshotRepository. | |
| 237 | 261 | func WithSnapshotStore(store ports.SnapshotStore) Option { | |
| 238 | 262 | return func(d *Daemon) { | |
| 239 | 263 | d.snaps = store | |
| @@ -343,23 +367,24 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||
| 343 | 367 | shell = "/bin/sh" | |
| 344 | 368 | } | |
| 345 | 369 | d := &Daemon{ | |
| 346 | - | sessions: make(map[domain.SessionID]*session), | |
| 347 | - | stopped: make(map[string]stoppedSession), | |
| 348 | - | parked: make(map[uint64]*parkedAttachment), | |
| 349 | - | ptys: ptys, | |
| 350 | - | clock: clock, | |
| 351 | - | log: log, | |
| 352 | - | baseEnv: os.Environ(), | |
| 353 | - | shell: shell, | |
| 354 | - | persist: persist.New(nil), | |
| 355 | - | dirOrHome: dirOrHome, | |
| 356 | - | done: make(chan struct{}), | |
| 357 | - | restoreDone: make(chan struct{}), | |
| 358 | - | animWake: make(chan struct{}, 1), | |
| 359 | - | snapshotMarshal: snapcodec.Marshal, | |
| 360 | - | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | |
| 361 | - | notices: newNoticeCenter(), | |
| 362 | - | resumeParkGrace: defaultResumeParkGrace, | |
| 370 | + | sessions: make(map[domain.SessionID]*session), | |
| 371 | + | stopped: make(map[string]stoppedSession), | |
| 372 | + | parked: make(map[uint64]*parkedAttachment), | |
| 373 | + | ptys: ptys, | |
| 374 | + | clock: clock, | |
| 375 | + | log: log, | |
| 376 | + | baseEnv: os.Environ(), | |
| 377 | + | shell: shell, | |
| 378 | + | persist: persist.New(nil), | |
| 379 | + | dirOrHome: dirOrHome, | |
| 380 | + | done: make(chan struct{}), | |
| 381 | + | restoreDone: make(chan struct{}), | |
| 382 | + | animWake: make(chan struct{}, 1), | |
| 383 | + | snapshotMarshal: snapcodec.Marshal, | |
| 384 | + | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | |
| 385 | + | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | |
| 386 | + | notices: newNoticeCenter(), | |
| 387 | + | resumeParkGrace: defaultResumeParkGrace, | |
| 363 | 388 | barScripts: &barScriptState{ | |
| 364 | 389 | cfg: barConfigFromDomain(domain.Defaults().Bar), | |
| 365 | 390 | outputs: make(map[domain.SessionID]barScriptOutputs), | |
| @@ -622,9 +647,15 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) { | |||
| 622 | 647 | if target == nil { | |
| 623 | 648 | if stopped, ok := d.stopped[k.Name]; ok { | |
| 624 | 649 | d.mu.Unlock() | |
| 625 | - | if d.snapsEnabled && d.snaps != nil { | |
| 626 | - | if err := d.snaps.Delete(k.Name); err != nil { | |
| 627 | - | d.log.Warn("deleting stopped session snapshot failed", "err", err, "session", k.Name) | |
| 650 | + | if d.snapsEnabled { | |
| 651 | + | var deleteErr error | |
| 652 | + | if d.snapshotRepository != nil { | |
| 653 | + | deleteErr = d.snapshotRepository.Delete(context.Background(), k.Name) | |
| 654 | + | } else if d.snaps != nil { | |
| 655 | + | deleteErr = d.snaps.Delete(k.Name) | |
| 656 | + | } | |
| 657 | + | if deleteErr != nil { | |
| 658 | + | d.log.Warn("deleting stopped session snapshot failed", "err", deleteErr, "session", k.Name) | |
| 628 | 659 | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session snapshot failed")) | |
| 629 | 660 | return | |
| 630 | 661 | } | |
| @@ -32,8 +32,13 @@ const debounceInterval = minDebounceInterval | |||||
| 32 | // ptyReadBufSize is the PTY reader's read buffer. | 32 | // ptyReadBufSize is the PTY reader's read buffer. | ||
| 33 | const ptyReadBufSize = 32 * 1024 | 33 | const ptyReadBufSize = 32 * 1024 | ||
| 34 | 34 | ||||
| 35 | - | // defaultScrollbackRows is the per-tab retained history size. | 35 | + | // New and restored panes are bounded by both history rows and cells. The |
| 36 | - | const defaultScrollbackRows = 10_000 | 36 | + | // cell budget keeps a very wide terminal from retaining disproportionately |
| 37 | + | // more scrollback than the normal 160-column case. | |||
| 38 | + | const ( | |||
| 39 | + | defaultScrollbackRows = 10_000 | |||
| 40 | + | defaultScrollbackCells = 12_000 * 160 | |||
| 41 | + | ) | |||
| 37 | 42 | ||||
| 38 | // detachNotifyTimeout bounds the best-effort Detached notification on the | 43 | // detachNotifyTimeout bounds the best-effort Detached notification on the | ||
| 39 | // detach/kill/shutdown paths: if a wedged client (full kernel send buffer) | 44 | // detach/kill/shutdown paths: if a wedged client (full kernel send buffer) | ||
| @@ -101,21 +106,30 @@ type Daemon struct { | |||||
| 101 | beforeCopyModeRevalidate func() | 106 | beforeCopyModeRevalidate func() | ||
| 102 | // beforeCopyMouseMap is a deterministic seam after an immutable copy-input | 107 | // beforeCopyMouseMap is a deterministic seam after an immutable copy-input | ||
| 103 | // snapshot is captured and before its mapped position is applied. | 108 | // snapshot is captured and before its mapped position is applied. | ||
| 104 | - | beforeCopyMouseMap func() | 109 | + | beforeCopyMouseMap func() |
| 105 | - | ptys ports.PTYFactory | 110 | + | ptys ports.PTYFactory |
| 106 | - | clock ports.Clock | 111 | + | clock ports.Clock |
| 107 | - | log *slog.Logger | 112 | + | log *slog.Logger |
| 108 | - | runtimeObserver ports.RuntimeObserver | 113 | + | runtimeObserver ports.RuntimeObserver |
| 109 | - | baseEnv []string | 114 | + | baseEnv []string |
| 110 | - | shell string | 115 | + | shell string |
| 111 | - | shellArgs []string | 116 | + | shellArgs []string |
| 112 | - | shellOverride bool | 117 | + | shellOverride bool |
| 113 | - | persist *persist.Persister | 118 | + | persist *persist.Persister |
| 114 | - | persistEnabled bool | 119 | + | persistEnabled bool |
| 115 | - | snaps ports.SnapshotStore | 120 | + | // snapshotRepository is the sole production checkpoint contract. snaps is |
| 116 | - | snapsEnabled bool | 121 | + | // retained temporarily only for the v3 test bridge; application wiring |
| 117 | - | noticeStore ports.NoticeStore | 122 | + | // always installs snapshotRepository. |
| 118 | - | snapshotMarshal func(snapcodec.Session) ([]byte, error) | 123 | + | snapshotRepository ports.SnapshotRepository |
| 124 | + | legacySnapshots ports.LegacySnapshotSource | |||
| 125 | + | snaps ports.SnapshotStore | |||
| 126 | + | snapsEnabled bool | |||
| 127 | + | noticeStore ports.NoticeStore | |||
| 128 | + | snapshotMarshal func(snapcodec.Session) ([]byte, error) | |||
| 129 | + | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | |||
| 130 | + | // independent of pane state and is bounded to prevent checkpoint history | |||
| 131 | + | // from becoming unbounded daemon memory. | |||
| 132 | + | snapshotChunkCache *snapshotChunkCache | |||
| 119 | snapshotJobs chan *snapshotCapture | 133 | snapshotJobs chan *snapshotCapture | ||
| 120 | snapshotWorkerMu sync.Mutex | 134 | snapshotWorkerMu sync.Mutex | ||
| 121 | snapshotWorkerID uint64 | 135 | snapshotWorkerID uint64 | ||
| @@ -232,8 +246,18 @@ func WithStore(store ports.Store) Option { | |||||
| 232 | } | 246 | } | ||
| 233 | } | 247 | } | ||
| 234 | 248 | ||||
| 235 | - | // WithSnapshotStore enables durable named session snapshots. A nil store keeps | 249 | + | // WithSnapshotRepository enables content-addressed incremental snapshots. The |
| 236 | - | // the daemon in no-op snapshot mode. | 250 | + | // legacy source is read-only migration input and is never used for new writes. |
| 251 | + | func WithSnapshotRepository(repository ports.SnapshotRepository, legacy ports.LegacySnapshotSource) Option { | |||
| 252 | + | return func(d *Daemon) { | |||
| 253 | + | d.snapshotRepository = repository | |||
| 254 | + | d.legacySnapshots = legacy | |||
| 255 | + | d.snapsEnabled = repository != nil | |||
| 256 | + | } | |||
| 257 | + | } | |||
| 258 | + | ||||
| 259 | + | // WithSnapshotStore is the v3 compatibility seam for existing embedders. New | |||
| 260 | + | // application wiring must use WithSnapshotRepository. | |||
| 237 | func WithSnapshotStore(store ports.SnapshotStore) Option { | 261 | func WithSnapshotStore(store ports.SnapshotStore) Option { | ||
| 238 | return func(d *Daemon) { | 262 | return func(d *Daemon) { | ||
| 239 | d.snaps = store | 263 | d.snaps = store | ||
| @@ -343,23 +367,24 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||||
| 343 | shell = "/bin/sh" | 367 | shell = "/bin/sh" | ||
| 344 | } | 368 | } | ||
| 345 | d := &Daemon{ | 369 | d := &Daemon{ | ||
| 346 | - | sessions: make(map[domain.SessionID]*session), | 370 | + | sessions: make(map[domain.SessionID]*session), |
| 347 | - | stopped: make(map[string]stoppedSession), | 371 | + | stopped: make(map[string]stoppedSession), |
| 348 | - | parked: make(map[uint64]*parkedAttachment), | 372 | + | parked: make(map[uint64]*parkedAttachment), |
| 349 | - | ptys: ptys, | 373 | + | ptys: ptys, |
| 350 | - | clock: clock, | 374 | + | clock: clock, |
| 351 | - | log: log, | 375 | + | log: log, |
| 352 | - | baseEnv: os.Environ(), | 376 | + | baseEnv: os.Environ(), |
| 353 | - | shell: shell, | 377 | + | shell: shell, |
| 354 | - | persist: persist.New(nil), | 378 | + | persist: persist.New(nil), |
| 355 | - | dirOrHome: dirOrHome, | 379 | + | dirOrHome: dirOrHome, |
| 356 | - | done: make(chan struct{}), | 380 | + | done: make(chan struct{}), |
| 357 | - | restoreDone: make(chan struct{}), | 381 | + | restoreDone: make(chan struct{}), |
| 358 | - | animWake: make(chan struct{}, 1), | 382 | + | animWake: make(chan struct{}, 1), |
| 359 | - | snapshotMarshal: snapcodec.Marshal, | 383 | + | snapshotMarshal: snapcodec.Marshal, |
| 360 | - | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | 384 | + | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), |
| 361 | - | notices: newNoticeCenter(), | 385 | + | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), |
| 362 | - | resumeParkGrace: defaultResumeParkGrace, | 386 | + | notices: newNoticeCenter(), |
| 387 | + | resumeParkGrace: defaultResumeParkGrace, | |||
| 363 | barScripts: &barScriptState{ | 388 | barScripts: &barScriptState{ | ||
| 364 | cfg: barConfigFromDomain(domain.Defaults().Bar), | 389 | cfg: barConfigFromDomain(domain.Defaults().Bar), | ||
| 365 | outputs: make(map[domain.SessionID]barScriptOutputs), | 390 | outputs: make(map[domain.SessionID]barScriptOutputs), | ||
| @@ -622,9 +647,15 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) { | |||||
| 622 | if target == nil { | 647 | if target == nil { | ||
| 623 | if stopped, ok := d.stopped[k.Name]; ok { | 648 | if stopped, ok := d.stopped[k.Name]; ok { | ||
| 624 | d.mu.Unlock() | 649 | d.mu.Unlock() | ||
| 625 | - | if d.snapsEnabled && d.snaps != nil { | 650 | + | if d.snapsEnabled { |
| 626 | - | if err := d.snaps.Delete(k.Name); err != nil { | 651 | + | var deleteErr error |
| 627 | - | d.log.Warn("deleting stopped session snapshot failed", "err", err, "session", k.Name) | 652 | + | if d.snapshotRepository != nil { |
| 653 | + | deleteErr = d.snapshotRepository.Delete(context.Background(), k.Name) | |||
| 654 | + | } else if d.snaps != nil { | |||
| 655 | + | deleteErr = d.snaps.Delete(k.Name) | |||
| 656 | + | } | |||
| 657 | + | if deleteErr != nil { | |||
| 658 | + | d.log.Warn("deleting stopped session snapshot failed", "err", deleteErr, "session", k.Name) | |||
| 628 | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session snapshot failed")) | 659 | _ = tr.Send(frameError(ports.ErrInternal, "deleting stopped session snapshot failed")) | ||
| 629 | return | 660 | return | ||
| 630 | } | 661 | } | ||
M internal/usecase/daemon/pane.go +1 -1
| @@ -47,7 +47,7 @@ func newPane(id layout.PaneID, pty ports.PTY, sz domain.Size) *pane { | |||
| 47 | 47 | } | |
| 48 | 48 | ||
| 49 | 49 | func newPaneWithStableID(id layout.PaneID, stableID string, pty ports.PTY, sz domain.Size) *pane { | |
| 50 | - | screen := vt.NewScreenWithHistory(sz.Cols, sz.Rows, vt.HistoryConfig{MaxRows: defaultScrollbackRows}) | |
| 50 | + | screen := vt.NewScreenWithHistory(sz.Cols, sz.Rows, vt.HistoryConfig{MaxRows: defaultScrollbackRows, MaxCells: defaultScrollbackCells}) | |
| 51 | 51 | return &pane{ | |
| 52 | 52 | id: id, | |
| 53 | 53 | stableID: stableID, | |
| @@ -47,7 +47,7 @@ func newPane(id layout.PaneID, pty ports.PTY, sz domain.Size) *pane { | |||||
| 47 | } | 47 | } | ||
| 48 | 48 | ||||
| 49 | func newPaneWithStableID(id layout.PaneID, stableID string, pty ports.PTY, sz domain.Size) *pane { | 49 | func newPaneWithStableID(id layout.PaneID, stableID string, pty ports.PTY, sz domain.Size) *pane { | ||
| 50 | - | screen := vt.NewScreenWithHistory(sz.Cols, sz.Rows, vt.HistoryConfig{MaxRows: defaultScrollbackRows}) | 50 | + | screen := vt.NewScreenWithHistory(sz.Cols, sz.Rows, vt.HistoryConfig{MaxRows: defaultScrollbackRows, MaxCells: defaultScrollbackCells}) |
| 51 | return &pane{ | 51 | return &pane{ | ||
| 52 | id: id, | 52 | id: id, | ||
| 53 | stableID: stableID, | 53 | stableID: stableID, | ||
A internal/usecase/daemon/snapshot_incremental.go +153 -0
| @@ -0,0 +1,153 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "container/list" | |
| 5 | + | "fmt" | |
| 6 | + | "sync" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/ports" | |
| 9 | + | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 10 | + | "github.com/bnema/vev/pkg/vt" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | // snapshotChunkCacheLimit is deliberately small: history chunks remain owned | |
| 14 | + | // by VT and the cache only retains their encoded VEVO representation. | |
| 15 | + | const snapshotChunkCacheLimit = 16 << 20 | |
| 16 | + | ||
| 17 | + | type snapshotChunkCacheEntry struct { | |
| 18 | + | chunk *vt.HistoryChunk | |
| 19 | + | object ports.SnapshotObject | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | // snapshotChunkCache is keyed by immutable HistoryChunk identity. Entries | |
| 23 | + | // never expose cache-owned bytes: publication gets a fresh slice, so a | |
| 24 | + | // repository cannot mutate data used by an in-flight or later checkpoint. | |
| 25 | + | type snapshotChunkCache struct { | |
| 26 | + | mu sync.Mutex | |
| 27 | + | limit int | |
| 28 | + | used int | |
| 29 | + | byPtr map[*vt.HistoryChunk]*list.Element | |
| 30 | + | lru *list.List // front is most recently used; values are snapshotChunkCacheEntry | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | func newSnapshotChunkCache(limit int) *snapshotChunkCache { | |
| 34 | + | return &snapshotChunkCache{limit: limit, byPtr: make(map[*vt.HistoryChunk]*list.Element), lru: list.New()} | |
| 35 | + | } | |
| 36 | + | ||
| 37 | + | func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |
| 38 | + | if chunk == nil { | |
| 39 | + | return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk") | |
| 40 | + | } | |
| 41 | + | c.mu.Lock() | |
| 42 | + | if elem := c.byPtr[chunk]; elem != nil { | |
| 43 | + | c.lru.MoveToFront(elem) | |
| 44 | + | object := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |
| 45 | + | c.mu.Unlock() | |
| 46 | + | return object, nil | |
| 47 | + | } | |
| 48 | + | c.mu.Unlock() | |
| 49 | + | ||
| 50 | + | payload, err := vt.MarshalHistoryChunk(chunk) | |
| 51 | + | if err != nil { | |
| 52 | + | return ports.SnapshotObject{}, err | |
| 53 | + | } | |
| 54 | + | object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, payload) | |
| 55 | + | if err != nil { | |
| 56 | + | return ports.SnapshotObject{}, err | |
| 57 | + | } | |
| 58 | + | ||
| 59 | + | c.mu.Lock() | |
| 60 | + | // A second encoder can only occur in direct tests; retain one canonical | |
| 61 | + | // cache entry and let both publications own independent byte slices. | |
| 62 | + | if elem := c.byPtr[chunk]; elem != nil { | |
| 63 | + | c.lru.MoveToFront(elem) | |
| 64 | + | cached := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |
| 65 | + | c.mu.Unlock() | |
| 66 | + | return cached, nil | |
| 67 | + | } | |
| 68 | + | stored := copySnapshotObject(object) | |
| 69 | + | elem := c.lru.PushFront(snapshotChunkCacheEntry{chunk: chunk, object: stored}) | |
| 70 | + | c.byPtr[chunk] = elem | |
| 71 | + | c.used += len(stored.Data) | |
| 72 | + | for c.used > c.limit && c.lru.Len() > 0 { | |
| 73 | + | oldest := c.lru.Back() | |
| 74 | + | entry := oldest.Value.(snapshotChunkCacheEntry) | |
| 75 | + | delete(c.byPtr, entry.chunk) | |
| 76 | + | c.used -= len(entry.object.Data) | |
| 77 | + | c.lru.Remove(oldest) | |
| 78 | + | } | |
| 79 | + | c.mu.Unlock() | |
| 80 | + | return copySnapshotObject(object), nil | |
| 81 | + | } | |
| 82 | + | ||
| 83 | + | func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | |
| 84 | + | object.Data = append([]byte(nil), object.Data...) | |
| 85 | + | return object | |
| 86 | + | } | |
| 87 | + | ||
| 88 | + | // incrementalPublication converts one immutable capture into complete VEVO | |
| 89 | + | // objects plus its VEVM manifest. Encoding runs after pane locks are released. | |
| 90 | + | // Sealed history uses pointer identity caching; tails and visible frames are | |
| 91 | + | // intentionally not cached because they are copied mutable state. | |
| 92 | + | func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) { | |
| 93 | + | if capture == nil || capture.name == "" { | |
| 94 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture") | |
| 95 | + | } | |
| 96 | + | if d.snapshotChunkCache == nil { | |
| 97 | + | d.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 98 | + | } | |
| 99 | + | manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | |
| 100 | + | objects := make([]ports.SnapshotObject, 0) | |
| 101 | + | for _, tab := range capture.tabs { | |
| 102 | + | outTab := snapcodec.ManifestTab{StableID: tab.stableID, Cols: tab.cols, Rows: tab.rows, NextPaneID: tab.nextPaneID, Focus: tab.focus, Tree: tab.tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.panes))} | |
| 103 | + | for _, pane := range tab.panes { | |
| 104 | + | if pane.visibleErr != nil { | |
| 105 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot visible: %w", pane.visibleErr) | |
| 106 | + | } | |
| 107 | + | outPane := snapcodec.ManifestPane{ID: pane.id, StableID: pane.stableID, Cwd: pane.cwd, Process: pane.process} | |
| 108 | + | sealedCount := pane.sealed.ChunkCount() | |
| 109 | + | sealedChunk := pane.sealed.Chunk | |
| 110 | + | if sealedCount == 0 && pane.history.ChunkCount() > 0 { | |
| 111 | + | sealedCount = pane.history.ChunkCount() | |
| 112 | + | sealedChunk = pane.history.Chunk | |
| 113 | + | } | |
| 114 | + | for i := 0; i < sealedCount; i++ { | |
| 115 | + | object, err := d.snapshotChunkCache.object(sealedChunk(i)) | |
| 116 | + | if err != nil { | |
| 117 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err) | |
| 118 | + | } | |
| 119 | + | outPane.Sealed = append(outPane.Sealed, objectRef(snapcodec.HistoryChunk, object)) | |
| 120 | + | objects = append(objects, object) | |
| 121 | + | } | |
| 122 | + | tail, err := vt.MarshalHistory(pane.tail) | |
| 123 | + | if pane.tail.Len() == 0 { | |
| 124 | + | tail, err = vt.MarshalEmptyHistoryTail() | |
| 125 | + | } | |
| 126 | + | if err != nil { | |
| 127 | + | return ports.SnapshotPublication{}, err | |
| 128 | + | } | |
| 129 | + | tailObject, err := snapcodec.MarshalObject(snapcodec.HistoryTail, tail) | |
| 130 | + | if err != nil { | |
| 131 | + | return ports.SnapshotPublication{}, err | |
| 132 | + | } | |
| 133 | + | visibleObject, err := snapcodec.MarshalObject(snapcodec.Visible, pane.visible) | |
| 134 | + | if err != nil { | |
| 135 | + | return ports.SnapshotPublication{}, err | |
| 136 | + | } | |
| 137 | + | outPane.Tail = objectRef(snapcodec.HistoryTail, tailObject) | |
| 138 | + | outPane.Visible = objectRef(snapcodec.Visible, visibleObject) | |
| 139 | + | objects = append(objects, tailObject, visibleObject) | |
| 140 | + | outTab.Panes = append(outTab.Panes, outPane) | |
| 141 | + | } | |
| 142 | + | manifest.Tabs = append(manifest.Tabs, outTab) | |
| 143 | + | } | |
| 144 | + | encoded, err := snapcodec.MarshalManifest(manifest) | |
| 145 | + | if err != nil { | |
| 146 | + | return ports.SnapshotPublication{}, err | |
| 147 | + | } | |
| 148 | + | return ports.SnapshotPublication{Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | |
| 149 | + | } | |
| 150 | + | ||
| 151 | + | func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef { | |
| 152 | + | return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))} | |
| 153 | + | } | |
| @@ -0,0 +1,153 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "container/list" | |||
| 5 | + | "fmt" | |||
| 6 | + | "sync" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/ports" | |||
| 9 | + | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |||
| 10 | + | "github.com/bnema/vev/pkg/vt" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | // snapshotChunkCacheLimit is deliberately small: history chunks remain owned | |||
| 14 | + | // by VT and the cache only retains their encoded VEVO representation. | |||
| 15 | + | const snapshotChunkCacheLimit = 16 << 20 | |||
| 16 | + | ||||
| 17 | + | type snapshotChunkCacheEntry struct { | |||
| 18 | + | chunk *vt.HistoryChunk | |||
| 19 | + | object ports.SnapshotObject | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | // snapshotChunkCache is keyed by immutable HistoryChunk identity. Entries | |||
| 23 | + | // never expose cache-owned bytes: publication gets a fresh slice, so a | |||
| 24 | + | // repository cannot mutate data used by an in-flight or later checkpoint. | |||
| 25 | + | type snapshotChunkCache struct { | |||
| 26 | + | mu sync.Mutex | |||
| 27 | + | limit int | |||
| 28 | + | used int | |||
| 29 | + | byPtr map[*vt.HistoryChunk]*list.Element | |||
| 30 | + | lru *list.List // front is most recently used; values are snapshotChunkCacheEntry | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | func newSnapshotChunkCache(limit int) *snapshotChunkCache { | |||
| 34 | + | return &snapshotChunkCache{limit: limit, byPtr: make(map[*vt.HistoryChunk]*list.Element), lru: list.New()} | |||
| 35 | + | } | |||
| 36 | + | ||||
| 37 | + | func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |||
| 38 | + | if chunk == nil { | |||
| 39 | + | return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk") | |||
| 40 | + | } | |||
| 41 | + | c.mu.Lock() | |||
| 42 | + | if elem := c.byPtr[chunk]; elem != nil { | |||
| 43 | + | c.lru.MoveToFront(elem) | |||
| 44 | + | object := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |||
| 45 | + | c.mu.Unlock() | |||
| 46 | + | return object, nil | |||
| 47 | + | } | |||
| 48 | + | c.mu.Unlock() | |||
| 49 | + | ||||
| 50 | + | payload, err := vt.MarshalHistoryChunk(chunk) | |||
| 51 | + | if err != nil { | |||
| 52 | + | return ports.SnapshotObject{}, err | |||
| 53 | + | } | |||
| 54 | + | object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, payload) | |||
| 55 | + | if err != nil { | |||
| 56 | + | return ports.SnapshotObject{}, err | |||
| 57 | + | } | |||
| 58 | + | ||||
| 59 | + | c.mu.Lock() | |||
| 60 | + | // A second encoder can only occur in direct tests; retain one canonical | |||
| 61 | + | // cache entry and let both publications own independent byte slices. | |||
| 62 | + | if elem := c.byPtr[chunk]; elem != nil { | |||
| 63 | + | c.lru.MoveToFront(elem) | |||
| 64 | + | cached := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |||
| 65 | + | c.mu.Unlock() | |||
| 66 | + | return cached, nil | |||
| 67 | + | } | |||
| 68 | + | stored := copySnapshotObject(object) | |||
| 69 | + | elem := c.lru.PushFront(snapshotChunkCacheEntry{chunk: chunk, object: stored}) | |||
| 70 | + | c.byPtr[chunk] = elem | |||
| 71 | + | c.used += len(stored.Data) | |||
| 72 | + | for c.used > c.limit && c.lru.Len() > 0 { | |||
| 73 | + | oldest := c.lru.Back() | |||
| 74 | + | entry := oldest.Value.(snapshotChunkCacheEntry) | |||
| 75 | + | delete(c.byPtr, entry.chunk) | |||
| 76 | + | c.used -= len(entry.object.Data) | |||
| 77 | + | c.lru.Remove(oldest) | |||
| 78 | + | } | |||
| 79 | + | c.mu.Unlock() | |||
| 80 | + | return copySnapshotObject(object), nil | |||
| 81 | + | } | |||
| 82 | + | ||||
| 83 | + | func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | |||
| 84 | + | object.Data = append([]byte(nil), object.Data...) | |||
| 85 | + | return object | |||
| 86 | + | } | |||
| 87 | + | ||||
| 88 | + | // incrementalPublication converts one immutable capture into complete VEVO | |||
| 89 | + | // objects plus its VEVM manifest. Encoding runs after pane locks are released. | |||
| 90 | + | // Sealed history uses pointer identity caching; tails and visible frames are | |||
| 91 | + | // intentionally not cached because they are copied mutable state. | |||
| 92 | + | func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) { | |||
| 93 | + | if capture == nil || capture.name == "" { | |||
| 94 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture") | |||
| 95 | + | } | |||
| 96 | + | if d.snapshotChunkCache == nil { | |||
| 97 | + | d.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 98 | + | } | |||
| 99 | + | manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | |||
| 100 | + | objects := make([]ports.SnapshotObject, 0) | |||
| 101 | + | for _, tab := range capture.tabs { | |||
| 102 | + | outTab := snapcodec.ManifestTab{StableID: tab.stableID, Cols: tab.cols, Rows: tab.rows, NextPaneID: tab.nextPaneID, Focus: tab.focus, Tree: tab.tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.panes))} | |||
| 103 | + | for _, pane := range tab.panes { | |||
| 104 | + | if pane.visibleErr != nil { | |||
| 105 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot visible: %w", pane.visibleErr) | |||
| 106 | + | } | |||
| 107 | + | outPane := snapcodec.ManifestPane{ID: pane.id, StableID: pane.stableID, Cwd: pane.cwd, Process: pane.process} | |||
| 108 | + | sealedCount := pane.sealed.ChunkCount() | |||
| 109 | + | sealedChunk := pane.sealed.Chunk | |||
| 110 | + | if sealedCount == 0 && pane.history.ChunkCount() > 0 { | |||
| 111 | + | sealedCount = pane.history.ChunkCount() | |||
| 112 | + | sealedChunk = pane.history.Chunk | |||
| 113 | + | } | |||
| 114 | + | for i := 0; i < sealedCount; i++ { | |||
| 115 | + | object, err := d.snapshotChunkCache.object(sealedChunk(i)) | |||
| 116 | + | if err != nil { | |||
| 117 | + | return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err) | |||
| 118 | + | } | |||
| 119 | + | outPane.Sealed = append(outPane.Sealed, objectRef(snapcodec.HistoryChunk, object)) | |||
| 120 | + | objects = append(objects, object) | |||
| 121 | + | } | |||
| 122 | + | tail, err := vt.MarshalHistory(pane.tail) | |||
| 123 | + | if pane.tail.Len() == 0 { | |||
| 124 | + | tail, err = vt.MarshalEmptyHistoryTail() | |||
| 125 | + | } | |||
| 126 | + | if err != nil { | |||
| 127 | + | return ports.SnapshotPublication{}, err | |||
| 128 | + | } | |||
| 129 | + | tailObject, err := snapcodec.MarshalObject(snapcodec.HistoryTail, tail) | |||
| 130 | + | if err != nil { | |||
| 131 | + | return ports.SnapshotPublication{}, err | |||
| 132 | + | } | |||
| 133 | + | visibleObject, err := snapcodec.MarshalObject(snapcodec.Visible, pane.visible) | |||
| 134 | + | if err != nil { | |||
| 135 | + | return ports.SnapshotPublication{}, err | |||
| 136 | + | } | |||
| 137 | + | outPane.Tail = objectRef(snapcodec.HistoryTail, tailObject) | |||
| 138 | + | outPane.Visible = objectRef(snapcodec.Visible, visibleObject) | |||
| 139 | + | objects = append(objects, tailObject, visibleObject) | |||
| 140 | + | outTab.Panes = append(outTab.Panes, outPane) | |||
| 141 | + | } | |||
| 142 | + | manifest.Tabs = append(manifest.Tabs, outTab) | |||
| 143 | + | } | |||
| 144 | + | encoded, err := snapcodec.MarshalManifest(manifest) | |||
| 145 | + | if err != nil { | |||
| 146 | + | return ports.SnapshotPublication{}, err | |||
| 147 | + | } | |||
| 148 | + | return ports.SnapshotPublication{Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | |||
| 149 | + | } | |||
| 150 | + | ||||
| 151 | + | func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef { | |||
| 152 | + | return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))} | |||
| 153 | + | } | |||
A internal/usecase/daemon/snapshot_incremental_test.go +35 -0
| @@ -0,0 +1,35 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | ||
| 6 | + | "github.com/bnema/vev/pkg/renderer" | |
| 7 | + | "github.com/bnema/vev/pkg/vt" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |
| 11 | + | history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1}) | |
| 12 | + | history.Append([]renderer.Cell{renderer.BlankCell()}) | |
| 13 | + | view := history.SealAndView() | |
| 14 | + | visible, err := vt.MarshalVisible(renderer.NewFrame(1, 1)) | |
| 15 | + | if err != nil { | |
| 16 | + | t.Fatal(err) | |
| 17 | + | } | |
| 18 | + | d := New(nil, nil, nil) | |
| 19 | + | capture := &snapshotCapture{name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | |
| 20 | + | first, err := d.incrementalPublication(capture) | |
| 21 | + | if err != nil { | |
| 22 | + | t.Fatal(err) | |
| 23 | + | } | |
| 24 | + | capture.generation = 2 | |
| 25 | + | second, err := d.incrementalPublication(capture) | |
| 26 | + | if err != nil { | |
| 27 | + | t.Fatal(err) | |
| 28 | + | } | |
| 29 | + | if len(first.Objects) == 0 || len(second.Objects) == 0 { | |
| 30 | + | t.Fatal("publication omitted required objects") | |
| 31 | + | } | |
| 32 | + | if got, want := d.snapshotChunkCache.used, d.snapshotChunkCache.limit; got > want { | |
| 33 | + | t.Fatalf("cache bytes = %d, limit = %d", got, want) | |
| 34 | + | } | |
| 35 | + | } | |
| @@ -0,0 +1,35 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | ||||
| 6 | + | "github.com/bnema/vev/pkg/renderer" | |||
| 7 | + | "github.com/bnema/vev/pkg/vt" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||
| 11 | + | history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1}) | |||
| 12 | + | history.Append([]renderer.Cell{renderer.BlankCell()}) | |||
| 13 | + | view := history.SealAndView() | |||
| 14 | + | visible, err := vt.MarshalVisible(renderer.NewFrame(1, 1)) | |||
| 15 | + | if err != nil { | |||
| 16 | + | t.Fatal(err) | |||
| 17 | + | } | |||
| 18 | + | d := New(nil, nil, nil) | |||
| 19 | + | capture := &snapshotCapture{name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | |||
| 20 | + | first, err := d.incrementalPublication(capture) | |||
| 21 | + | if err != nil { | |||
| 22 | + | t.Fatal(err) | |||
| 23 | + | } | |||
| 24 | + | capture.generation = 2 | |||
| 25 | + | second, err := d.incrementalPublication(capture) | |||
| 26 | + | if err != nil { | |||
| 27 | + | t.Fatal(err) | |||
| 28 | + | } | |||
| 29 | + | if len(first.Objects) == 0 || len(second.Objects) == 0 { | |||
| 30 | + | t.Fatal("publication omitted required objects") | |||
| 31 | + | } | |||
| 32 | + | if got, want := d.snapshotChunkCache.used, d.snapshotChunkCache.limit; got > want { | |||
| 33 | + | t.Fatalf("cache bytes = %d, limit = %d", got, want) | |||
| 34 | + | } | |||
| 35 | + | } | |||