1package daemon
2
3import (
4 "fmt"
5
6 vt "github.com/bnema/vev-vt"
7 "github.com/bnema/vev/internal/domain"
8 "github.com/bnema/vev/internal/ports"
9 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
10)
11
12// snapshotChunkCacheLimit is deliberately small: history chunks remain owned
13// by VT and each named-session cache only retains their encoded VEVO form.
14const snapshotChunkCacheLimit = 16 << 20
15
16type snapshotChunkCacheEntry struct {
17 object ports.SnapshotObject
18}
19
20// snapshotChunkCache is keyed by immutable HistoryChunk identity. Its owner
21// must hold the named session's snapshotMu for every operation. Encoded bytes
22// remain cache-owned; publications receive a copy only until a successful
23// repository publication has made the immutable object available by reference.
24type snapshotChunkCache struct {
25 limit int
26 used int
27 byPtr map[*vt.HistoryChunk]snapshotChunkCacheEntry
28 persisted map[*vt.HistoryChunk]snapcodec.ObjectRef
29 // marshalChunk makes cache admission measurable without coupling the cache
30 // policy test to VT's wire encoding implementation.
31 marshalChunk func(*vt.HistoryChunk) ([]byte, error)
32 copyObject func(ports.SnapshotObject) ports.SnapshotObject
33}
34
35func newSnapshotChunkCache(limit int) *snapshotChunkCache {
36 return &snapshotChunkCache{
37 limit: limit,
38 byPtr: make(map[*vt.HistoryChunk]snapshotChunkCacheEntry),
39 persisted: make(map[*vt.HistoryChunk]snapcodec.ObjectRef),
40 marshalChunk: vt.MarshalHistoryChunk,
41 copyObject: copySnapshotObject,
42 }
43}
44
45// objectLocked returns a complete manifest reference and, when the repository
46// does not already retain it, an independently owned object to supply. Caller
47// holds the owning session's snapshotMu.
48func (c *snapshotChunkCache) objectLocked(chunk *vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
49 if chunk == nil {
50 return snapcodec.ObjectRef{}, nil, fmt.Errorf("snapshot: nil history chunk")
51 }
52 if ref, ok := c.persisted[chunk]; ok {
53 return ref, nil, nil
54 }
55 if entry, ok := c.byPtr[chunk]; ok {
56 object := c.copyObject(entry.object)
57 return objectRef(snapcodec.HistoryChunk, object), &object, nil
58 }
59
60 payload, err := c.marshalChunk(chunk)
61 if err != nil {
62 return snapcodec.ObjectRef{}, nil, err
63 }
64 object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, payload)
65 if err != nil {
66 return snapcodec.ObjectRef{}, nil, err
67 }
68 ref := objectRef(snapcodec.HistoryChunk, object)
69
70 // An oldest-to-newest snapshot scan can exceed the cache. LRU would admit
71 // every early miss and evict the later working set, causing the next scan to
72 // encode every immutable chunk again. Once full, retain the warmed set and
73 // return later misses uncached. Pruning makes space when a retained chunk is
74 // no longer reachable by a queued or in-flight capture.
75 if len(object.Data) > c.limit || c.used > c.limit-len(object.Data) {
76 return ref, &object, nil
77 }
78 stored := c.copyObject(object)
79 c.byPtr[chunk] = snapshotChunkCacheEntry{object: stored}
80 c.used += len(stored.Data)
81 return ref, &object, nil
82}
83
84// pruneLocked releases cached encodings which no currently queued or in-flight
85// capture can use. Caller holds the owning session's snapshotMu.
86func (c *snapshotChunkCache) pruneLocked(referenced map[*vt.HistoryChunk]struct{}) {
87 for chunk, entry := range c.byPtr {
88 if _, keep := referenced[chunk]; !keep {
89 delete(c.byPtr, chunk)
90 c.used -= len(entry.object.Data)
91 }
92 }
93 for chunk := range c.persisted {
94 if _, keep := referenced[chunk]; !keep {
95 delete(c.persisted, chunk)
96 }
97 }
98}
99
100func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject {
101 object.Data = append([]byte(nil), object.Data...)
102 return object
103}
104
105// incrementalPublication converts one immutable capture into complete VEVO
106// objects plus its VEVM manifest. Encoding runs after pane locks are released.
107// Sealed history uses this capture's named-session pointer identity cache.
108// Tails and recovery transcripts are generation-specific copied mutable state
109// and are intentionally never admitted to that cache.
110func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.SnapshotPublication, error) {
111 if capture == nil || capture.session == nil || capture.name == "" {
112 return ports.SnapshotPublication{}, fmt.Errorf("snapshot: empty capture")
113 }
114 if err := prepareSnapshotChunkCache(capture); err != nil {
115 return ports.SnapshotPublication{}, err
116 }
117 manifest := snapcodec.Manifest{Generation: capture.generation, IncarnationID: capture.incarnation, ParentCheckpoint: capture.parentCheckpoint, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.defaultTab, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))}
118 objects := make([]ports.SnapshotObject, 0)
119 capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef)
120 for _, tab := range capture.tabs {
121 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))}
122 for _, pane := range tab.panes {
123 // transcript was copied under pane.mu. Marshal it here in the worker,
124 // after all pane and session locks have been released.
125 transcript, err := pane.transcript.Marshal()
126 if err != nil {
127 return ports.SnapshotPublication{}, fmt.Errorf("snapshot recovery transcript: %w", err)
128 }
129 outPane := snapcodec.ManifestPane{ID: pane.id, StableID: pane.stableID, Cwd: pane.cwd, Process: pane.process}
130 for i := 0; i < pane.sealed.ChunkCount(); i++ {
131 ref, object, err := snapshotChunkObject(capture, pane.sealed.Chunk(i))
132 if err != nil {
133 return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err)
134 }
135 outPane.Sealed = append(outPane.Sealed, ref)
136 capture.sealedRefs[pane.sealed.Chunk(i)] = ref
137 if object != nil {
138 objects = append(objects, *object)
139 }
140 }
141 tail, err := marshalSnapshotTail(pane.tail)
142 if err != nil {
143 return ports.SnapshotPublication{}, err
144 }
145 tailObject, err := snapcodec.MarshalObject(snapcodec.HistoryTail, tail)
146 if err != nil {
147 return ports.SnapshotPublication{}, err
148 }
149 transcriptObject, err := snapcodec.MarshalObject(snapcodec.RecoveryTranscript, transcript)
150 if err != nil {
151 return ports.SnapshotPublication{}, err
152 }
153 outPane.Tail = objectRef(snapcodec.HistoryTail, tailObject)
154 outPane.Transcript = objectRef(snapcodec.RecoveryTranscript, transcriptObject)
155 objects = append(objects, tailObject, transcriptObject)
156 outTab.Panes = append(outTab.Panes, outPane)
157 }
158 manifest.Tabs = append(manifest.Tabs, outTab)
159 }
160 encoded, err := snapcodec.MarshalManifest(manifest)
161 if err != nil {
162 return ports.SnapshotPublication{}, err
163 }
164 capture.checkpoint = domain.CheckpointRef{Generation: capture.generation, ManifestDigest: snapcodec.ManifestDigest(encoded)}
165 return ports.SnapshotPublication{IncarnationID: capture.incarnation, Name: capture.name, Generation: capture.generation, ParentCheckpoint: capture.parentCheckpoint, Manifest: encoded, Objects: objects}, nil
166}
167
168// marshalSnapshotTail selects the canonical empty-tail encoding before any
169// general history encoding. An empty tail is represented by its dedicated
170// canonical blob; a non-empty copied tail retains its rows.
171func marshalSnapshotTail(tail vt.HistoryView) ([]byte, error) {
172 return vt.MarshalHistory(tail)
173}
174
175func prepareSnapshotChunkCache(capture *snapshotCapture) error {
176 sess := capture.session
177 sess.snapshotMu.Lock()
178 defer sess.snapshotMu.Unlock()
179 if sess.snapshotChunkCache == nil {
180 if sess.name == "" || sess.ephemeral {
181 return fmt.Errorf("snapshot: chunk cache unavailable")
182 }
183 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
184 }
185 pruneSnapshotChunkCacheLocked(sess, capture)
186 return nil
187}
188
189func snapshotChunkObject(capture *snapshotCapture, chunk *vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
190 sess := capture.session
191 sess.snapshotMu.Lock()
192 defer sess.snapshotMu.Unlock()
193 return sess.snapshotChunkCache.objectLocked(chunk)
194}
195
196// markSnapshotCaptureObjectsPublished records that the immutable sealed
197// history referenced by capture is available in this session's repository
198// lineage. It is called only after Publish succeeds.
199func markSnapshotCaptureObjectsPublished(capture *snapshotCapture) {
200 if capture == nil || capture.session == nil {
201 return
202 }
203 sess := capture.session
204 sess.snapshotMu.Lock()
205 defer sess.snapshotMu.Unlock()
206 cache := sess.snapshotChunkCache
207 if cache == nil {
208 return
209 }
210 for chunk, ref := range capture.sealedRefs {
211 if chunk == nil {
212 continue
213 }
214 cache.persisted[chunk] = ref
215 if entry, ok := cache.byPtr[chunk]; ok {
216 delete(cache.byPtr, chunk)
217 cache.used -= len(entry.object.Data)
218 if cache.used < 0 {
219 cache.used = 0
220 }
221 }
222 }
223}
224
225// pruneSnapshotChunkCacheLocked retains only chunks referenced by captures that
226// can still be published. Caller holds sess.snapshotMu.
227func pruneSnapshotChunkCacheLocked(sess *session, capture *snapshotCapture) {
228 if sess.snapshotChunkCache == nil {
229 return
230 }
231 referenced := make(map[*vt.HistoryChunk]struct{})
232 collectSnapshotCaptureChunks(referenced, capture)
233 collectSnapshotCaptureChunks(referenced, sess.snapshotQueuedCapture)
234 collectSnapshotCaptureChunks(referenced, sess.snapshotInFlightCapture)
235 sess.snapshotChunkCache.pruneLocked(referenced)
236}
237
238func collectSnapshotCaptureChunks(referenced map[*vt.HistoryChunk]struct{}, capture *snapshotCapture) {
239 if capture == nil {
240 return
241 }
242 for _, tab := range capture.tabs {
243 for _, pane := range tab.panes {
244 for i := 0; i < pane.sealed.ChunkCount(); i++ {
245 if chunk := pane.sealed.Chunk(i); chunk != nil {
246 referenced[chunk] = struct{}{}
247 }
248 }
249 }
250 }
251}
252
253func objectRef(kind snapcodec.ObjectKind, object ports.SnapshotObject) snapcodec.ObjectRef {
254 return snapcodec.ObjectRef{Kind: kind, Digest: object.Digest, Size: uint32(len(object.Data))}
255}