fix(daemon): scope snapshot cache per session
- commit
- 225928f3d7b937573be88a11b29bb93d323a17db
- author
- brice · · 2026-07-21 23:49 UTC
- parent
- 106b3de
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +203 -38
M internal/usecase/daemon/daemon.go +1 -6
| @@ -132,11 +132,7 @@ type Daemon struct { | |||
| 132 | 132 | legacyImportPending map[string][]byte | |
| 133 | 133 | noticeStore ports.NoticeStore | |
| 134 | 134 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | |
| 135 | - | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | |
| 136 | - | // independent of pane state and is bounded to prevent checkpoint history | |
| 137 | - | // from becoming unbounded daemon memory. | |
| 138 | - | snapshotChunkCache *snapshotChunkCache | |
| 139 | - | snapshotJobs chan *snapshotCapture | |
| 135 | + | snapshotJobs chan *snapshotCapture | |
| 140 | 136 | // snapshotWake wakes the repository scheduler when a session becomes dirty | |
| 141 | 137 | // or an attempt completes. It is never closed and producers only send | |
| 142 | 138 | // non-blockingly. | |
| @@ -392,7 +388,6 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||
| 392 | 388 | restoreDone: make(chan struct{}), | |
| 393 | 389 | animWake: make(chan struct{}, 1), | |
| 394 | 390 | snapshotMarshal: snapcodec.Marshal, | |
| 395 | - | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | |
| 396 | 391 | legacyImportPending: make(map[string][]byte), | |
| 397 | 392 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | |
| 398 | 393 | snapshotWake: make(chan struct{}, 1), | |
| @@ -132,11 +132,7 @@ type Daemon struct { | |||||
| 132 | legacyImportPending map[string][]byte | 132 | legacyImportPending map[string][]byte | ||
| 133 | noticeStore ports.NoticeStore | 133 | noticeStore ports.NoticeStore | ||
| 134 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | 134 | snapshotMarshal func(snapcodec.Session) ([]byte, error) | ||
| 135 | - | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | 135 | + | snapshotJobs chan *snapshotCapture |
| 136 | - | // independent of pane state and is bounded to prevent checkpoint history | |||
| 137 | - | // from becoming unbounded daemon memory. | |||
| 138 | - | snapshotChunkCache *snapshotChunkCache | |||
| 139 | - | snapshotJobs chan *snapshotCapture | |||
| 140 | // snapshotWake wakes the repository scheduler when a session becomes dirty | 136 | // snapshotWake wakes the repository scheduler when a session becomes dirty | ||
| 141 | // or an attempt completes. It is never closed and producers only send | 137 | // or an attempt completes. It is never closed and producers only send | ||
| 142 | // non-blockingly. | 138 | // non-blockingly. | ||
| @@ -392,7 +388,6 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||||
| 392 | restoreDone: make(chan struct{}), | 388 | restoreDone: make(chan struct{}), | ||
| 393 | animWake: make(chan struct{}, 1), | 389 | animWake: make(chan struct{}, 1), | ||
| 394 | snapshotMarshal: snapcodec.Marshal, | 390 | snapshotMarshal: snapcodec.Marshal, | ||
| 395 | - | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | |||
| 396 | legacyImportPending: make(map[string][]byte), | 391 | legacyImportPending: make(map[string][]byte), | ||
| 397 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | 392 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | ||
| 398 | snapshotWake: make(chan struct{}, 1), | 393 | snapshotWake: make(chan struct{}, 1), | ||
M internal/usecase/daemon/session.go +20 -0
| @@ -83,6 +83,10 @@ type session struct { | |||
| 83 | 83 | // snapshotFailureSig is the stable class of the active failed publication. | |
| 84 | 84 | // It is guarded by snapshotMu and deliberately never contains error text. | |
| 85 | 85 | snapshotFailureSig string | |
| 86 | + | // snapshotChunkCache retains encoded sealed history only for this named | |
| 87 | + | // session. snapshotMu owns it so cache touches never contend with pane | |
| 88 | + | // state or mutate bytes retained by queued/in-flight publications. | |
| 89 | + | snapshotChunkCache *snapshotChunkCache | |
| 86 | 90 | snapshotChanged chan struct{} | |
| 87 | 91 | snapshotWake chan struct{} | |
| 88 | 92 | // syncGen makes synchronized-output watchdog generations unique across all | |
| @@ -230,6 +234,9 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 230 | 234 | createdAt: createdAt, | |
| 231 | 235 | snapshotWake: d.snapshotWake, | |
| 232 | 236 | } | |
| 237 | + | if !ephemeral && name != "" { | |
| 238 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 239 | + | } | |
| 233 | 240 | if lastUsedSeq > 0 { | |
| 234 | 241 | sess.mruAt.Store(lastUsedSeq) | |
| 235 | 242 | } | |
| @@ -650,6 +657,11 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||
| 650 | 657 | resumeSnapshotCoordinatorForNewIdentity(sess, quarantine) | |
| 651 | 658 | return nil | |
| 652 | 659 | } | |
| 660 | + | if wasEphemeral { | |
| 661 | + | sess.snapshotMu.Lock() | |
| 662 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 663 | + | sess.snapshotMu.Unlock() | |
| 664 | + | } | |
| 653 | 665 | markSnapshotDirty(sess) | |
| 654 | 666 | return nil | |
| 655 | 667 | } | |
| @@ -912,6 +924,14 @@ func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error { | |||
| 912 | 924 | } | |
| 913 | 925 | } | |
| 914 | 926 | } | |
| 927 | + | if !isEphemeral { | |
| 928 | + | // Joining first preserves byte ownership for a publication that is still | |
| 929 | + | // encoding while this session is being torn down. | |
| 930 | + | <-quarantineSnapshotCoordinator(sess) | |
| 931 | + | sess.snapshotMu.Lock() | |
| 932 | + | sess.snapshotChunkCache = nil | |
| 933 | + | sess.snapshotMu.Unlock() | |
| 934 | + | } | |
| 915 | 935 | d.mu.Lock() | |
| 916 | 936 | if _, ok := d.sessions[sess.id]; !ok { | |
| 917 | 937 | d.mu.Unlock() | |
| @@ -83,6 +83,10 @@ type session struct { | |||||
| 83 | // snapshotFailureSig is the stable class of the active failed publication. | 83 | // snapshotFailureSig is the stable class of the active failed publication. | ||
| 84 | // It is guarded by snapshotMu and deliberately never contains error text. | 84 | // It is guarded by snapshotMu and deliberately never contains error text. | ||
| 85 | snapshotFailureSig string | 85 | snapshotFailureSig string | ||
| 86 | + | // snapshotChunkCache retains encoded sealed history only for this named | |||
| 87 | + | // session. snapshotMu owns it so cache touches never contend with pane | |||
| 88 | + | // state or mutate bytes retained by queued/in-flight publications. | |||
| 89 | + | snapshotChunkCache *snapshotChunkCache | |||
| 86 | snapshotChanged chan struct{} | 90 | snapshotChanged chan struct{} | ||
| 87 | snapshotWake chan struct{} | 91 | snapshotWake chan struct{} | ||
| 88 | // syncGen makes synchronized-output watchdog generations unique across all | 92 | // syncGen makes synchronized-output watchdog generations unique across all | ||
| @@ -230,6 +234,9 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 230 | createdAt: createdAt, | 234 | createdAt: createdAt, | ||
| 231 | snapshotWake: d.snapshotWake, | 235 | snapshotWake: d.snapshotWake, | ||
| 232 | } | 236 | } | ||
| 237 | + | if !ephemeral && name != "" { | |||
| 238 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 239 | + | } | |||
| 233 | if lastUsedSeq > 0 { | 240 | if lastUsedSeq > 0 { | ||
| 234 | sess.mruAt.Store(lastUsedSeq) | 241 | sess.mruAt.Store(lastUsedSeq) | ||
| 235 | } | 242 | } | ||
| @@ -650,6 +657,11 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||||
| 650 | resumeSnapshotCoordinatorForNewIdentity(sess, quarantine) | 657 | resumeSnapshotCoordinatorForNewIdentity(sess, quarantine) | ||
| 651 | return nil | 658 | return nil | ||
| 652 | } | 659 | } | ||
| 660 | + | if wasEphemeral { | |||
| 661 | + | sess.snapshotMu.Lock() | |||
| 662 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 663 | + | sess.snapshotMu.Unlock() | |||
| 664 | + | } | |||
| 653 | markSnapshotDirty(sess) | 665 | markSnapshotDirty(sess) | ||
| 654 | return nil | 666 | return nil | ||
| 655 | } | 667 | } | ||
| @@ -912,6 +924,14 @@ func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error { | |||||
| 912 | } | 924 | } | ||
| 913 | } | 925 | } | ||
| 914 | } | 926 | } | ||
| 927 | + | if !isEphemeral { | |||
| 928 | + | // Joining first preserves byte ownership for a publication that is still | |||
| 929 | + | // encoding while this session is being torn down. | |||
| 930 | + | <-quarantineSnapshotCoordinator(sess) | |||
| 931 | + | sess.snapshotMu.Lock() | |||
| 932 | + | sess.snapshotChunkCache = nil | |||
| 933 | + | sess.snapshotMu.Unlock() | |||
| 934 | + | } | |||
| 915 | d.mu.Lock() | 935 | d.mu.Lock() | ||
| 916 | if _, ok := d.sessions[sess.id]; !ok { | 936 | if _, ok := d.sessions[sess.id]; !ok { | ||
| 917 | d.mu.Unlock() | 937 | d.mu.Unlock() | ||
M internal/usecase/daemon/snapshot.go +2 -1
| @@ -268,7 +268,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||
| 268 | 268 | return nil | |
| 269 | 269 | } | |
| 270 | 270 | createdAt := int64(snap.CreatedAt) | |
| 271 | - | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake} | |
| 271 | + | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)} | |
| 272 | 272 | sess.snapEligible.Store(true) | |
| 273 | 273 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | |
| 274 | 274 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | |
| @@ -392,6 +392,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||
| 392 | 392 | sess.snapshotAttempted = false | |
| 393 | 393 | sess.snapshotAttemptKind = snapshotAttemptRoutine | |
| 394 | 394 | sess.snapshotFailureSig = "" | |
| 395 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 395 | 396 | sess.snapDirty.Store(true) | |
| 396 | 397 | sess.snapEligible.Store(true) | |
| 397 | 398 | sess.signalSnapshotChangedLocked() | |
| @@ -268,7 +268,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||||
| 268 | return nil | 268 | return nil | ||
| 269 | } | 269 | } | ||
| 270 | createdAt := int64(snap.CreatedAt) | 270 | createdAt := int64(snap.CreatedAt) | ||
| 271 | - | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake} | 271 | + | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)} |
| 272 | sess.snapEligible.Store(true) | 272 | sess.snapEligible.Store(true) | ||
| 273 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | 273 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | ||
| 274 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | 274 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | ||
| @@ -392,6 +392,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||||
| 392 | sess.snapshotAttempted = false | 392 | sess.snapshotAttempted = false | ||
| 393 | sess.snapshotAttemptKind = snapshotAttemptRoutine | 393 | sess.snapshotAttemptKind = snapshotAttemptRoutine | ||
| 394 | sess.snapshotFailureSig = "" | 394 | sess.snapshotFailureSig = "" | ||
| 395 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 395 | sess.snapDirty.Store(true) | 396 | sess.snapDirty.Store(true) | ||
| 396 | sess.snapEligible.Store(true) | 397 | sess.snapEligible.Store(true) | ||
| 397 | sess.signalSnapshotChangedLocked() | 398 | sess.signalSnapshotChangedLocked() | ||
A internal/usecase/daemon/snapshot_acceptance_test.go +81 -0
| @@ -0,0 +1,81 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "bytes" | |
| 5 | + | "testing" | |
| 6 | + | ||
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/vev/pkg/renderer" | |
| 10 | + | "github.com/bnema/vev/pkg/vt" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | func TestSnapshotChunkCacheIsScopedToNamedSession(t *testing.T) { | |
| 14 | + | d := New(nil, nil, nil) | |
| 15 | + | a := newSnapshotTestSession(t, "a", false, "/a") | |
| 16 | + | b := newSnapshotTestSession(t, "b", false, "/b") | |
| 17 | + | a.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 18 | + | b.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 19 | + | ||
| 20 | + | // Eleven maximum-size chunks exceed the per-session 16 MiB cache budget. | |
| 21 | + | // The capture retains independently owned publication objects while its LRU | |
| 22 | + | // sheds the oldest entry. | |
| 23 | + | aHistory := snapshotCacheHistory(t, 11, 'a') | |
| 24 | + | aCapture := snapshotCacheCapture(a, aHistory, 1) | |
| 25 | + | aFirst, err := d.incrementalPublication(aCapture) | |
| 26 | + | require.NoError(t, err) | |
| 27 | + | require.NotEmpty(t, aFirst.Objects) | |
| 28 | + | require.LessOrEqual(t, a.snapshotChunkCache.used, a.snapshotChunkCache.limit) | |
| 29 | + | require.LessOrEqual(t, a.snapshotChunkCache.limit, snapshotChunkCacheLimit) | |
| 30 | + | require.NotContains(t, a.snapshotChunkCache.byPtr, aHistory.Chunk(0), "session A should evict its oldest entry") | |
| 31 | + | ||
| 32 | + | bHistory := snapshotCacheHistory(t, 1, 'b') | |
| 33 | + | bCapture := snapshotCacheCapture(b, bHistory, 1) | |
| 34 | + | _, err = d.incrementalPublication(bCapture) | |
| 35 | + | require.NoError(t, err) | |
| 36 | + | require.Contains(t, b.snapshotChunkCache.byPtr, bHistory.Chunk(0)) | |
| 37 | + | ||
| 38 | + | inFlightBytes := append([]byte(nil), aFirst.Objects[0].Data...) | |
| 39 | + | aSecond := snapshotCacheCapture(a, snapshotCacheHistory(t, 1, 'z'), 2) | |
| 40 | + | _, err = d.incrementalPublication(aSecond) | |
| 41 | + | require.NoError(t, err) | |
| 42 | + | require.True(t, bytes.Equal(inFlightBytes, aFirst.Objects[0].Data), "LRU eviction must not mutate queued publication bytes") | |
| 43 | + | require.LessOrEqual(t, a.snapshotChunkCache.used, snapshotChunkCacheLimit) | |
| 44 | + | ||
| 45 | + | _, err = d.incrementalPublication(snapshotCacheCapture(b, bHistory, 2)) | |
| 46 | + | require.NoError(t, err) | |
| 47 | + | require.Contains(t, b.snapshotChunkCache.byPtr, bHistory.Chunk(0), "session A eviction must not affect session B") | |
| 48 | + | require.LessOrEqual(t, b.snapshotChunkCache.used, snapshotChunkCacheLimit) | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | func snapshotCacheHistory(t *testing.T, chunks int, first rune) vt.HistoryView { | |
| 52 | + | t.Helper() | |
| 53 | + | history := vt.NewHistory(vt.HistoryConfig{MaxRows: chunks * 256, ChunkRows: 256}) | |
| 54 | + | for row := 0; row < chunks*256; row++ { | |
| 55 | + | cells := make([]renderer.Cell, 160) | |
| 56 | + | for col := range cells { | |
| 57 | + | cells[col] = renderer.Cell{Rune: first + rune((row+col)%26)} | |
| 58 | + | } | |
| 59 | + | require.NoError(t, history.Append(cells)) | |
| 60 | + | } | |
| 61 | + | return history.SealAndView() | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | func snapshotCacheCapture(sess *session, history vt.HistoryView, generation uint64) *snapshotCapture { | |
| 65 | + | return &snapshotCapture{ | |
| 66 | + | session: sess, | |
| 67 | + | name: sess.name, | |
| 68 | + | generation: generation, | |
| 69 | + | tabs: []snapshotCaptureTab{{ | |
| 70 | + | stableID: "tab", | |
| 71 | + | cols: 1, | |
| 72 | + | rows: 1, | |
| 73 | + | panes: []snapshotCapturePane{{ | |
| 74 | + | id: "pane", | |
| 75 | + | stableID: "pane", | |
| 76 | + | history: history, | |
| 77 | + | visible: vt.NewScreen(1, 1).PrimaryVisibleSnapshot(), | |
| 78 | + | }}, | |
| 79 | + | }}, | |
| 80 | + | } | |
| 81 | + | } | |
| @@ -0,0 +1,81 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "bytes" | |||
| 5 | + | "testing" | |||
| 6 | + | ||||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/vev/pkg/renderer" | |||
| 10 | + | "github.com/bnema/vev/pkg/vt" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | func TestSnapshotChunkCacheIsScopedToNamedSession(t *testing.T) { | |||
| 14 | + | d := New(nil, nil, nil) | |||
| 15 | + | a := newSnapshotTestSession(t, "a", false, "/a") | |||
| 16 | + | b := newSnapshotTestSession(t, "b", false, "/b") | |||
| 17 | + | a.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 18 | + | b.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 19 | + | ||||
| 20 | + | // Eleven maximum-size chunks exceed the per-session 16 MiB cache budget. | |||
| 21 | + | // The capture retains independently owned publication objects while its LRU | |||
| 22 | + | // sheds the oldest entry. | |||
| 23 | + | aHistory := snapshotCacheHistory(t, 11, 'a') | |||
| 24 | + | aCapture := snapshotCacheCapture(a, aHistory, 1) | |||
| 25 | + | aFirst, err := d.incrementalPublication(aCapture) | |||
| 26 | + | require.NoError(t, err) | |||
| 27 | + | require.NotEmpty(t, aFirst.Objects) | |||
| 28 | + | require.LessOrEqual(t, a.snapshotChunkCache.used, a.snapshotChunkCache.limit) | |||
| 29 | + | require.LessOrEqual(t, a.snapshotChunkCache.limit, snapshotChunkCacheLimit) | |||
| 30 | + | require.NotContains(t, a.snapshotChunkCache.byPtr, aHistory.Chunk(0), "session A should evict its oldest entry") | |||
| 31 | + | ||||
| 32 | + | bHistory := snapshotCacheHistory(t, 1, 'b') | |||
| 33 | + | bCapture := snapshotCacheCapture(b, bHistory, 1) | |||
| 34 | + | _, err = d.incrementalPublication(bCapture) | |||
| 35 | + | require.NoError(t, err) | |||
| 36 | + | require.Contains(t, b.snapshotChunkCache.byPtr, bHistory.Chunk(0)) | |||
| 37 | + | ||||
| 38 | + | inFlightBytes := append([]byte(nil), aFirst.Objects[0].Data...) | |||
| 39 | + | aSecond := snapshotCacheCapture(a, snapshotCacheHistory(t, 1, 'z'), 2) | |||
| 40 | + | _, err = d.incrementalPublication(aSecond) | |||
| 41 | + | require.NoError(t, err) | |||
| 42 | + | require.True(t, bytes.Equal(inFlightBytes, aFirst.Objects[0].Data), "LRU eviction must not mutate queued publication bytes") | |||
| 43 | + | require.LessOrEqual(t, a.snapshotChunkCache.used, snapshotChunkCacheLimit) | |||
| 44 | + | ||||
| 45 | + | _, err = d.incrementalPublication(snapshotCacheCapture(b, bHistory, 2)) | |||
| 46 | + | require.NoError(t, err) | |||
| 47 | + | require.Contains(t, b.snapshotChunkCache.byPtr, bHistory.Chunk(0), "session A eviction must not affect session B") | |||
| 48 | + | require.LessOrEqual(t, b.snapshotChunkCache.used, snapshotChunkCacheLimit) | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | func snapshotCacheHistory(t *testing.T, chunks int, first rune) vt.HistoryView { | |||
| 52 | + | t.Helper() | |||
| 53 | + | history := vt.NewHistory(vt.HistoryConfig{MaxRows: chunks * 256, ChunkRows: 256}) | |||
| 54 | + | for row := 0; row < chunks*256; row++ { | |||
| 55 | + | cells := make([]renderer.Cell, 160) | |||
| 56 | + | for col := range cells { | |||
| 57 | + | cells[col] = renderer.Cell{Rune: first + rune((row+col)%26)} | |||
| 58 | + | } | |||
| 59 | + | require.NoError(t, history.Append(cells)) | |||
| 60 | + | } | |||
| 61 | + | return history.SealAndView() | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | func snapshotCacheCapture(sess *session, history vt.HistoryView, generation uint64) *snapshotCapture { | |||
| 65 | + | return &snapshotCapture{ | |||
| 66 | + | session: sess, | |||
| 67 | + | name: sess.name, | |||
| 68 | + | generation: generation, | |||
| 69 | + | tabs: []snapshotCaptureTab{{ | |||
| 70 | + | stableID: "tab", | |||
| 71 | + | cols: 1, | |||
| 72 | + | rows: 1, | |||
| 73 | + | panes: []snapshotCapturePane{{ | |||
| 74 | + | id: "pane", | |||
| 75 | + | stableID: "pane", | |||
| 76 | + | history: history, | |||
| 77 | + | visible: vt.NewScreen(1, 1).PrimaryVisibleSnapshot(), | |||
| 78 | + | }}, | |||
| 79 | + | }}, | |||
| 80 | + | } | |||
| 81 | + | } | |||
M internal/usecase/daemon/snapshot_incremental.go +96 -29
| @@ -3,7 +3,6 @@ package daemon | |||
| 3 | 3 | import ( | |
| 4 | 4 | "container/list" | |
| 5 | 5 | "fmt" | |
| 6 | - | "sync" | |
| 7 | 6 | ||
| 8 | 7 | "github.com/bnema/vev/internal/ports" | |
| 9 | 8 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| @@ -11,7 +10,7 @@ import ( | |||
| 11 | 10 | ) | |
| 12 | 11 | ||
| 13 | 12 | // snapshotChunkCacheLimit is deliberately small: history chunks remain owned | |
| 14 | - | // by VT and the cache only retains their encoded VEVO representation. | |
| 13 | + | // by VT and each named-session cache only retains their encoded VEVO form. | |
| 15 | 14 | const snapshotChunkCacheLimit = 16 << 20 | |
| 16 | 15 | ||
| 17 | 16 | type snapshotChunkCacheEntry struct { | |
| @@ -19,11 +18,11 @@ type snapshotChunkCacheEntry struct { | |||
| 19 | 18 | object ports.SnapshotObject | |
| 20 | 19 | } | |
| 21 | 20 | ||
| 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. | |
| 21 | + | // snapshotChunkCache is keyed by immutable HistoryChunk identity. Its owner | |
| 22 | + | // must hold the named session's snapshotMu for every operation. Entries never | |
| 23 | + | // expose cache-owned bytes: a publication gets a fresh slice, so a repository | |
| 24 | + | // cannot mutate data used by an in-flight or later checkpoint. | |
| 25 | 25 | type snapshotChunkCache struct { | |
| 26 | - | mu sync.Mutex | |
| 27 | 26 | limit int | |
| 28 | 27 | used int | |
| 29 | 28 | byPtr map[*vt.HistoryChunk]*list.Element | |
| @@ -34,18 +33,16 @@ func newSnapshotChunkCache(limit int) *snapshotChunkCache { | |||
| 34 | 33 | return &snapshotChunkCache{limit: limit, byPtr: make(map[*vt.HistoryChunk]*list.Element), lru: list.New()} | |
| 35 | 34 | } | |
| 36 | 35 | ||
| 37 | - | func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |
| 36 | + | // objectLocked returns an independently owned object. Caller holds the owning | |
| 37 | + | // session's snapshotMu. | |
| 38 | + | func (c *snapshotChunkCache) objectLocked(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |
| 38 | 39 | if chunk == nil { | |
| 39 | 40 | return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk") | |
| 40 | 41 | } | |
| 41 | - | c.mu.Lock() | |
| 42 | 42 | if elem := c.byPtr[chunk]; elem != nil { | |
| 43 | 43 | c.lru.MoveToFront(elem) | |
| 44 | - | object := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |
| 45 | - | c.mu.Unlock() | |
| 46 | - | return object, nil | |
| 44 | + | return copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object), nil | |
| 47 | 45 | } | |
| 48 | - | c.mu.Unlock() | |
| 49 | 46 | ||
| 50 | 47 | payload, err := vt.MarshalHistoryChunk(chunk) | |
| 51 | 48 | if err != nil { | |
| @@ -56,28 +53,42 @@ func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObjec | |||
| 56 | 53 | return ports.SnapshotObject{}, err | |
| 57 | 54 | } | |
| 58 | 55 | ||
| 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 | |
| 56 | + | // A single chunk over the ceiling must not flush useful smaller entries. | |
| 57 | + | // It is already independently owned by this publication, so skipping the | |
| 58 | + | // cache cannot affect its lifetime. | |
| 59 | + | if len(object.Data) > c.limit { | |
| 60 | + | return object, nil | |
| 67 | 61 | } | |
| 68 | 62 | stored := copySnapshotObject(object) | |
| 69 | 63 | elem := c.lru.PushFront(snapshotChunkCacheEntry{chunk: chunk, object: stored}) | |
| 70 | 64 | c.byPtr[chunk] = elem | |
| 71 | 65 | c.used += len(stored.Data) | |
| 72 | - | for c.used > c.limit && c.lru.Len() > 0 { | |
| 66 | + | for c.used > c.limit { | |
| 73 | 67 | oldest := c.lru.Back() | |
| 68 | + | if oldest == nil { | |
| 69 | + | break | |
| 70 | + | } | |
| 74 | 71 | entry := oldest.Value.(snapshotChunkCacheEntry) | |
| 75 | 72 | delete(c.byPtr, entry.chunk) | |
| 76 | 73 | c.used -= len(entry.object.Data) | |
| 77 | 74 | c.lru.Remove(oldest) | |
| 78 | 75 | } | |
| 79 | - | c.mu.Unlock() | |
| 80 | - | return copySnapshotObject(object), nil | |
| 76 | + | return object, nil | |
| 77 | + | } | |
| 78 | + | ||
| 79 | + | // pruneLocked releases cached encodings which no currently queued or in-flight | |
| 80 | + | // capture can use. Caller holds the owning session's snapshotMu. | |
| 81 | + | func (c *snapshotChunkCache) pruneLocked(referenced map[*vt.HistoryChunk]struct{}) { | |
| 82 | + | for elem := c.lru.Back(); elem != nil; { | |
| 83 | + | previous := elem.Prev() | |
| 84 | + | entry := elem.Value.(snapshotChunkCacheEntry) | |
| 85 | + | if _, keep := referenced[entry.chunk]; !keep { | |
| 86 | + | delete(c.byPtr, entry.chunk) | |
| 87 | + | c.used -= len(entry.object.Data) | |
| 88 | + | c.lru.Remove(elem) | |
| 89 | + | } | |
| 90 | + | elem = previous | |
| 91 | + | } | |
| 81 | 92 | } | |
| 82 | 93 | ||
| 83 | 94 | func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | |
| @@ -87,14 +98,15 @@ func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | |||
| 87 | 98 | ||
| 88 | 99 | // incrementalPublication converts one immutable capture into complete VEVO | |
| 89 | 100 | // 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. | |
| 101 | + | // Sealed history uses this capture's named-session pointer identity cache; | |
| 102 | + | // tails and visible frames are intentionally not cached because they are copied | |
| 103 | + | // mutable state. | |
| 92 | 104 | func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) { | |
| 93 | - | if capture == nil || capture.name == "" { | |
| 105 | + | if capture == nil || capture.session == nil || capture.name == "" { | |
| 94 | 106 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture") | |
| 95 | 107 | } | |
| 96 | - | if d.snapshotChunkCache == nil { | |
| 97 | - | d.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 108 | + | if err := prepareSnapshotChunkCache(capture); err != nil { | |
| 109 | + | return ports.SnapshotPublication{}, err | |
| 98 | 110 | } | |
| 99 | 111 | manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | |
| 100 | 112 | objects := make([]ports.SnapshotObject, 0) | |
| @@ -115,7 +127,7 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||
| 115 | 127 | sealedChunk = pane.history.Chunk | |
| 116 | 128 | } | |
| 117 | 129 | for i := 0; i < sealedCount; i++ { | |
| 118 | - | object, err := d.snapshotChunkCache.object(sealedChunk(i)) | |
| 130 | + | object, err := snapshotChunkObject(capture, sealedChunk(i)) | |
| 119 | 131 | if err != nil { | |
| 120 | 132 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err) | |
| 121 | 133 | } | |
| @@ -151,6 +163,61 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||
| 151 | 163 | return ports.SnapshotPublication{Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | |
| 152 | 164 | } | |
| 153 | 165 | ||
| 166 | + | func prepareSnapshotChunkCache(capture *snapshotCapture) error { | |
| 167 | + | sess := capture.session | |
| 168 | + | sess.snapshotMu.Lock() | |
| 169 | + | defer sess.snapshotMu.Unlock() | |
| 170 | + | if sess.snapshotChunkCache == nil { | |
| 171 | + | if sess.name == "" || sess.ephemeral { | |
| 172 | + | return fmt.Errorf("snapshot: chunk cache unavailable") | |
| 173 | + | } | |
| 174 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| 175 | + | } | |
| 176 | + | pruneSnapshotChunkCacheLocked(sess, capture) | |
| 177 | + | return nil | |
| 178 | + | } | |
| 179 | + | ||
| 180 | + | func snapshotChunkObject(capture *snapshotCapture, chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |
| 181 | + | sess := capture.session | |
| 182 | + | sess.snapshotMu.Lock() | |
| 183 | + | defer sess.snapshotMu.Unlock() | |
| 184 | + | return sess.snapshotChunkCache.objectLocked(chunk) | |
| 185 | + | } | |
| 186 | + | ||
| 187 | + | // pruneSnapshotChunkCacheLocked retains only chunks referenced by captures that | |
| 188 | + | // can still be published. Caller holds sess.snapshotMu. | |
| 189 | + | func pruneSnapshotChunkCacheLocked(sess *session, capture *snapshotCapture) { | |
| 190 | + | if sess.snapshotChunkCache == nil { | |
| 191 | + | return | |
| 192 | + | } | |
| 193 | + | referenced := make(map[*vt.HistoryChunk]struct{}) | |
| 194 | + | collectSnapshotCaptureChunks(referenced, capture) | |
| 195 | + | collectSnapshotCaptureChunks(referenced, sess.snapshotQueuedCapture) | |
| 196 | + | collectSnapshotCaptureChunks(referenced, sess.snapshotInFlightCapture) | |
| 197 | + | sess.snapshotChunkCache.pruneLocked(referenced) | |
| 198 | + | } | |
| 199 | + | ||
| 200 | + | func collectSnapshotCaptureChunks(referenced map[*vt.HistoryChunk]struct{}, capture *snapshotCapture) { | |
| 201 | + | if capture == nil { | |
| 202 | + | return | |
| 203 | + | } | |
| 204 | + | for _, tab := range capture.tabs { | |
| 205 | + | for _, pane := range tab.panes { | |
| 206 | + | sealedCount := pane.sealed.ChunkCount() | |
| 207 | + | sealedChunk := pane.sealed.Chunk | |
| 208 | + | if sealedCount == 0 && pane.history.ChunkCount() > 0 { | |
| 209 | + | sealedCount = pane.history.ChunkCount() | |
| 210 | + | sealedChunk = pane.history.Chunk | |
| 211 | + | } | |
| 212 | + | for i := 0; i < sealedCount; i++ { | |
| 213 | + | if chunk := sealedChunk(i); chunk != nil { | |
| 214 | + | referenced[chunk] = struct{}{} | |
| 215 | + | } | |
| 216 | + | } | |
| 217 | + | } | |
| 218 | + | } | |
| 219 | + | } | |
| 220 | + | ||
| 154 | 221 | func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef { | |
| 155 | 222 | return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))} | |
| 156 | 223 | } | |
| @@ -3,7 +3,6 @@ package daemon | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "container/list" | 4 | "container/list" | ||
| 5 | "fmt" | 5 | "fmt" | ||
| 6 | - | "sync" | |||
| 7 | 6 | ||||
| 8 | "github.com/bnema/vev/internal/ports" | 7 | "github.com/bnema/vev/internal/ports" | ||
| 9 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 8 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| @@ -11,7 +10,7 @@ import ( | |||||
| 11 | ) | 10 | ) | ||
| 12 | 11 | ||||
| 13 | // snapshotChunkCacheLimit is deliberately small: history chunks remain owned | 12 | // snapshotChunkCacheLimit is deliberately small: history chunks remain owned | ||
| 14 | - | // by VT and the cache only retains their encoded VEVO representation. | 13 | + | // by VT and each named-session cache only retains their encoded VEVO form. |
| 15 | const snapshotChunkCacheLimit = 16 << 20 | 14 | const snapshotChunkCacheLimit = 16 << 20 | ||
| 16 | 15 | ||||
| 17 | type snapshotChunkCacheEntry struct { | 16 | type snapshotChunkCacheEntry struct { | ||
| @@ -19,11 +18,11 @@ type snapshotChunkCacheEntry struct { | |||||
| 19 | object ports.SnapshotObject | 18 | object ports.SnapshotObject | ||
| 20 | } | 19 | } | ||
| 21 | 20 | ||||
| 22 | - | // snapshotChunkCache is keyed by immutable HistoryChunk identity. Entries | 21 | + | // snapshotChunkCache is keyed by immutable HistoryChunk identity. Its owner |
| 23 | - | // never expose cache-owned bytes: publication gets a fresh slice, so a | 22 | + | // must hold the named session's snapshotMu for every operation. Entries never |
| 24 | - | // repository cannot mutate data used by an in-flight or later checkpoint. | 23 | + | // expose cache-owned bytes: a publication gets a fresh slice, so a repository |
| 24 | + | // cannot mutate data used by an in-flight or later checkpoint. | |||
| 25 | type snapshotChunkCache struct { | 25 | type snapshotChunkCache struct { | ||
| 26 | - | mu sync.Mutex | |||
| 27 | limit int | 26 | limit int | ||
| 28 | used int | 27 | used int | ||
| 29 | byPtr map[*vt.HistoryChunk]*list.Element | 28 | byPtr map[*vt.HistoryChunk]*list.Element | ||
| @@ -34,18 +33,16 @@ func newSnapshotChunkCache(limit int) *snapshotChunkCache { | |||||
| 34 | return &snapshotChunkCache{limit: limit, byPtr: make(map[*vt.HistoryChunk]*list.Element), lru: list.New()} | 33 | return &snapshotChunkCache{limit: limit, byPtr: make(map[*vt.HistoryChunk]*list.Element), lru: list.New()} | ||
| 35 | } | 34 | } | ||
| 36 | 35 | ||||
| 37 | - | func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | 36 | + | // objectLocked returns an independently owned object. Caller holds the owning |
| 37 | + | // session's snapshotMu. | |||
| 38 | + | func (c *snapshotChunkCache) objectLocked(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |||
| 38 | if chunk == nil { | 39 | if chunk == nil { | ||
| 39 | return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk") | 40 | return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk") | ||
| 40 | } | 41 | } | ||
| 41 | - | c.mu.Lock() | |||
| 42 | if elem := c.byPtr[chunk]; elem != nil { | 42 | if elem := c.byPtr[chunk]; elem != nil { | ||
| 43 | c.lru.MoveToFront(elem) | 43 | c.lru.MoveToFront(elem) | ||
| 44 | - | object := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | 44 | + | return copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object), nil |
| 45 | - | c.mu.Unlock() | |||
| 46 | - | return object, nil | |||
| 47 | } | 45 | } | ||
| 48 | - | c.mu.Unlock() | |||
| 49 | 46 | ||||
| 50 | payload, err := vt.MarshalHistoryChunk(chunk) | 47 | payload, err := vt.MarshalHistoryChunk(chunk) | ||
| 51 | if err != nil { | 48 | if err != nil { | ||
| @@ -56,28 +53,42 @@ func (c *snapshotChunkCache) object(chunk *vt.HistoryChunk) (ports.SnapshotObjec | |||||
| 56 | return ports.SnapshotObject{}, err | 53 | return ports.SnapshotObject{}, err | ||
| 57 | } | 54 | } | ||
| 58 | 55 | ||||
| 59 | - | c.mu.Lock() | 56 | + | // A single chunk over the ceiling must not flush useful smaller entries. |
| 60 | - | // A second encoder can only occur in direct tests; retain one canonical | 57 | + | // It is already independently owned by this publication, so skipping the |
| 61 | - | // cache entry and let both publications own independent byte slices. | 58 | + | // cache cannot affect its lifetime. |
| 62 | - | if elem := c.byPtr[chunk]; elem != nil { | 59 | + | if len(object.Data) > c.limit { |
| 63 | - | c.lru.MoveToFront(elem) | 60 | + | return object, nil |
| 64 | - | cached := copySnapshotObject(elem.Value.(snapshotChunkCacheEntry).object) | |||
| 65 | - | c.mu.Unlock() | |||
| 66 | - | return cached, nil | |||
| 67 | } | 61 | } | ||
| 68 | stored := copySnapshotObject(object) | 62 | stored := copySnapshotObject(object) | ||
| 69 | elem := c.lru.PushFront(snapshotChunkCacheEntry{chunk: chunk, object: stored}) | 63 | elem := c.lru.PushFront(snapshotChunkCacheEntry{chunk: chunk, object: stored}) | ||
| 70 | c.byPtr[chunk] = elem | 64 | c.byPtr[chunk] = elem | ||
| 71 | c.used += len(stored.Data) | 65 | c.used += len(stored.Data) | ||
| 72 | - | for c.used > c.limit && c.lru.Len() > 0 { | 66 | + | for c.used > c.limit { |
| 73 | oldest := c.lru.Back() | 67 | oldest := c.lru.Back() | ||
| 68 | + | if oldest == nil { | |||
| 69 | + | break | |||
| 70 | + | } | |||
| 74 | entry := oldest.Value.(snapshotChunkCacheEntry) | 71 | entry := oldest.Value.(snapshotChunkCacheEntry) | ||
| 75 | delete(c.byPtr, entry.chunk) | 72 | delete(c.byPtr, entry.chunk) | ||
| 76 | c.used -= len(entry.object.Data) | 73 | c.used -= len(entry.object.Data) | ||
| 77 | c.lru.Remove(oldest) | 74 | c.lru.Remove(oldest) | ||
| 78 | } | 75 | } | ||
| 79 | - | c.mu.Unlock() | 76 | + | return object, nil |
| 80 | - | return copySnapshotObject(object), nil | 77 | + | } |
| 78 | + | ||||
| 79 | + | // pruneLocked releases cached encodings which no currently queued or in-flight | |||
| 80 | + | // capture can use. Caller holds the owning session's snapshotMu. | |||
| 81 | + | func (c *snapshotChunkCache) pruneLocked(referenced map[*vt.HistoryChunk]struct{}) { | |||
| 82 | + | for elem := c.lru.Back(); elem != nil; { | |||
| 83 | + | previous := elem.Prev() | |||
| 84 | + | entry := elem.Value.(snapshotChunkCacheEntry) | |||
| 85 | + | if _, keep := referenced[entry.chunk]; !keep { | |||
| 86 | + | delete(c.byPtr, entry.chunk) | |||
| 87 | + | c.used -= len(entry.object.Data) | |||
| 88 | + | c.lru.Remove(elem) | |||
| 89 | + | } | |||
| 90 | + | elem = previous | |||
| 91 | + | } | |||
| 81 | } | 92 | } | ||
| 82 | 93 | ||||
| 83 | func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | 94 | func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | ||
| @@ -87,14 +98,15 @@ func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject { | |||||
| 87 | 98 | ||||
| 88 | // incrementalPublication converts one immutable capture into complete VEVO | 99 | // incrementalPublication converts one immutable capture into complete VEVO | ||
| 89 | // objects plus its VEVM manifest. Encoding runs after pane locks are released. | 100 | // objects plus its VEVM manifest. Encoding runs after pane locks are released. | ||
| 90 | - | // Sealed history uses pointer identity caching; tails and visible frames are | 101 | + | // Sealed history uses this capture's named-session pointer identity cache; |
| 91 | - | // intentionally not cached because they are copied mutable state. | 102 | + | // tails and visible frames are intentionally not cached because they are copied |
| 103 | + | // mutable state. | |||
| 92 | func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) { | 104 | func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) { | ||
| 93 | - | if capture == nil || capture.name == "" { | 105 | + | if capture == nil || capture.session == nil || capture.name == "" { |
| 94 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture") | 106 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture") | ||
| 95 | } | 107 | } | ||
| 96 | - | if d.snapshotChunkCache == nil { | 108 | + | if err := prepareSnapshotChunkCache(capture); err != nil { |
| 97 | - | d.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | 109 | + | return ports.SnapshotPublication{}, err |
| 98 | } | 110 | } | ||
| 99 | manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | 111 | 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) | 112 | objects := make([]ports.SnapshotObject, 0) | ||
| @@ -115,7 +127,7 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||||
| 115 | sealedChunk = pane.history.Chunk | 127 | sealedChunk = pane.history.Chunk | ||
| 116 | } | 128 | } | ||
| 117 | for i := 0; i < sealedCount; i++ { | 129 | for i := 0; i < sealedCount; i++ { | ||
| 118 | - | object, err := d.snapshotChunkCache.object(sealedChunk(i)) | 130 | + | object, err := snapshotChunkObject(capture, sealedChunk(i)) |
| 119 | if err != nil { | 131 | if err != nil { | ||
| 120 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err) | 132 | return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err) | ||
| 121 | } | 133 | } | ||
| @@ -151,6 +163,61 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||||
| 151 | return ports.SnapshotPublication{Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | 163 | return ports.SnapshotPublication{Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | ||
| 152 | } | 164 | } | ||
| 153 | 165 | ||||
| 166 | + | func prepareSnapshotChunkCache(capture *snapshotCapture) error { | |||
| 167 | + | sess := capture.session | |||
| 168 | + | sess.snapshotMu.Lock() | |||
| 169 | + | defer sess.snapshotMu.Unlock() | |||
| 170 | + | if sess.snapshotChunkCache == nil { | |||
| 171 | + | if sess.name == "" || sess.ephemeral { | |||
| 172 | + | return fmt.Errorf("snapshot: chunk cache unavailable") | |||
| 173 | + | } | |||
| 174 | + | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |||
| 175 | + | } | |||
| 176 | + | pruneSnapshotChunkCacheLocked(sess, capture) | |||
| 177 | + | return nil | |||
| 178 | + | } | |||
| 179 | + | ||||
| 180 | + | func snapshotChunkObject(capture *snapshotCapture, chunk *vt.HistoryChunk) (ports.SnapshotObject, error) { | |||
| 181 | + | sess := capture.session | |||
| 182 | + | sess.snapshotMu.Lock() | |||
| 183 | + | defer sess.snapshotMu.Unlock() | |||
| 184 | + | return sess.snapshotChunkCache.objectLocked(chunk) | |||
| 185 | + | } | |||
| 186 | + | ||||
| 187 | + | // pruneSnapshotChunkCacheLocked retains only chunks referenced by captures that | |||
| 188 | + | // can still be published. Caller holds sess.snapshotMu. | |||
| 189 | + | func pruneSnapshotChunkCacheLocked(sess *session, capture *snapshotCapture) { | |||
| 190 | + | if sess.snapshotChunkCache == nil { | |||
| 191 | + | return | |||
| 192 | + | } | |||
| 193 | + | referenced := make(map[*vt.HistoryChunk]struct{}) | |||
| 194 | + | collectSnapshotCaptureChunks(referenced, capture) | |||
| 195 | + | collectSnapshotCaptureChunks(referenced, sess.snapshotQueuedCapture) | |||
| 196 | + | collectSnapshotCaptureChunks(referenced, sess.snapshotInFlightCapture) | |||
| 197 | + | sess.snapshotChunkCache.pruneLocked(referenced) | |||
| 198 | + | } | |||
| 199 | + | ||||
| 200 | + | func collectSnapshotCaptureChunks(referenced map[*vt.HistoryChunk]struct{}, capture *snapshotCapture) { | |||
| 201 | + | if capture == nil { | |||
| 202 | + | return | |||
| 203 | + | } | |||
| 204 | + | for _, tab := range capture.tabs { | |||
| 205 | + | for _, pane := range tab.panes { | |||
| 206 | + | sealedCount := pane.sealed.ChunkCount() | |||
| 207 | + | sealedChunk := pane.sealed.Chunk | |||
| 208 | + | if sealedCount == 0 && pane.history.ChunkCount() > 0 { | |||
| 209 | + | sealedCount = pane.history.ChunkCount() | |||
| 210 | + | sealedChunk = pane.history.Chunk | |||
| 211 | + | } | |||
| 212 | + | for i := 0; i < sealedCount; i++ { | |||
| 213 | + | if chunk := sealedChunk(i); chunk != nil { | |||
| 214 | + | referenced[chunk] = struct{}{} | |||
| 215 | + | } | |||
| 216 | + | } | |||
| 217 | + | } | |||
| 218 | + | } | |||
| 219 | + | } | |||
| 220 | + | ||||
| 154 | func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef { | 221 | func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef { | ||
| 155 | return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))} | 222 | return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))} | ||
| 156 | } | 223 | } | ||
M internal/usecase/daemon/snapshot_incremental_test.go +3 -2
| @@ -41,7 +41,8 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||
| 41 | 41 | view := history.SealAndView() | |
| 42 | 42 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | |
| 43 | 43 | d := New(nil, nil, nil) | |
| 44 | - | capture := &snapshotCapture{name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | |
| 44 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 45 | + | capture := &snapshotCapture{session: sess, name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | |
| 45 | 46 | first, err := d.incrementalPublication(capture) | |
| 46 | 47 | if err != nil { | |
| 47 | 48 | t.Fatal(err) | |
| @@ -54,7 +55,7 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||
| 54 | 55 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | |
| 55 | 56 | t.Fatal("publication omitted required objects") | |
| 56 | 57 | } | |
| 57 | - | if got, want := d.snapshotChunkCache.used, d.snapshotChunkCache.limit; got > want { | |
| 58 | + | if got, want := sess.snapshotChunkCache.used, sess.snapshotChunkCache.limit; got > want { | |
| 58 | 59 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | |
| 59 | 60 | } | |
| 60 | 61 | } | |
| @@ -41,7 +41,8 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||||
| 41 | view := history.SealAndView() | 41 | view := history.SealAndView() | ||
| 42 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | 42 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | ||
| 43 | d := New(nil, nil, nil) | 43 | d := New(nil, nil, nil) | ||
| 44 | - | capture := &snapshotCapture{name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | 44 | + | sess := newSnapshotTestSession(t, "work", false, "/work") |
| 45 | + | capture := &snapshotCapture{session: sess, name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", history: view, visible: visible}}}}} | |||
| 45 | first, err := d.incrementalPublication(capture) | 46 | first, err := d.incrementalPublication(capture) | ||
| 46 | if err != nil { | 47 | if err != nil { | ||
| 47 | t.Fatal(err) | 48 | t.Fatal(err) | ||
| @@ -54,7 +55,7 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||||
| 54 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | 55 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | ||
| 55 | t.Fatal("publication omitted required objects") | 56 | t.Fatal("publication omitted required objects") | ||
| 56 | } | 57 | } | ||
| 57 | - | if got, want := d.snapshotChunkCache.used, d.snapshotChunkCache.limit; got > want { | 58 | + | if got, want := sess.snapshotChunkCache.used, sess.snapshotChunkCache.limit; got > want { |
| 58 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | 59 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | ||
| 59 | } | 60 | } | ||
| 60 | } | 61 | } | ||