1package snapshot
2
3import (
4 "context"
5 "crypto/sha256"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 codec "github.com/bnema/vev/internal/usecase/snapshot"
14)
15
16// CollectGarbage applies the complete snapshot retention policy. The keep map
17// must come from a catalogue that loaded and validated successfully: only then
18// is an incarnation absent from keep known to be an orphan.
19func (r *Repository) CollectGarbage(ctx context.Context, keep map[domain.IncarnationID]domain.CheckpointRef) error {
20 if err := ctx.Err(); err != nil {
21 return err
22 }
23 sessions := filepath.Join(r.dir, repositorySessionsDir)
24 entries, err := r.readGarbageCollectionDirectory(sessions)
25 if errors.Is(err, os.ErrNotExist) {
26 return nil
27 }
28 if err != nil {
29 return fmt.Errorf("read snapshot incarnations: %w", safeFilesystemError(err))
30 }
31
32 var collected error
33 for _, entry := range entries {
34 if err := ctx.Err(); err != nil {
35 return errors.Join(collected, err)
36 }
37 if !entry.IsDir() {
38 continue
39 }
40 var id domain.IncarnationID
41 if err := id.UnmarshalText([]byte(entry.Name())); err != nil {
42 continue
43 }
44 committed, known := keep[id]
45 if !known {
46 key, keyErr := incarnationKey(id)
47 if keyErr != nil {
48 collected = errors.Join(collected, keyErr)
49 continue
50 }
51 lock := r.lockSession(key)
52 _, known = keep[id]
53 if !known {
54 err = r.removeIncarnationLocked(id)
55 }
56 r.unlockSession(lock)
57 if err != nil {
58 collected = errors.Join(collected, fmt.Errorf("remove orphan snapshot incarnation %s: %w", id.String(), safeFilesystemError(err)))
59 continue
60 }
61 if !known {
62 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-incarnation")
63 }
64 continue
65 }
66 if err := r.pruneGenerations(ctx, id, committed); err != nil {
67 collected = errors.Join(collected, err)
68 }
69 }
70 if err := r.syncDirectory(sessions); err != nil {
71 collected = errors.Join(collected, fmt.Errorf("sync snapshot sessions directory: %w", safeFilesystemError(err)))
72 }
73 return collected
74}
75
76// pruneGenerations keeps the committed generation and its immediate
77// predecessor. Everything else, including a forward orphan newer than the
78// catalogue commit, is removed before unreferenced objects are swept.
79func (r *Repository) pruneGenerations(ctx context.Context, id domain.IncarnationID, committed domain.CheckpointRef) error {
80 key, err := incarnationKey(id)
81 if err != nil {
82 return err
83 }
84 lock := r.lockSession(key)
85 defer r.unlockSession(lock)
86
87 generationsDir := filepath.Join(r.sessionPath(id), repositoryGenerations)
88 entries, err := r.readGarbageCollectionDirectory(generationsDir)
89 if errors.Is(err, os.ErrNotExist) {
90 return nil
91 }
92 if err != nil {
93 return fmt.Errorf("read snapshot generations for %s: %w", id.String(), safeFilesystemError(err))
94 }
95
96 kept := make(map[uint64]struct{}, 2)
97 if committed.Generation > 0 {
98 kept[committed.Generation] = struct{}{}
99 }
100 if committed.Generation > 1 {
101 kept[committed.Generation-1] = struct{}{}
102 }
103 references, err := r.referencesForGenerations(id, entries, kept, committed)
104 if err != nil {
105 return fmt.Errorf("mark snapshot objects for %s: %w", id.String(), err)
106 }
107
108 removed := 0
109 for _, entry := range entries {
110 if err := ctx.Err(); err != nil {
111 return err
112 }
113 generation, canonical := parseGenerationFilename(entry.Name())
114 if !canonical || entry.IsDir() {
115 continue
116 }
117 if _, retain := kept[generation]; retain {
118 continue
119 }
120 if err := r.remove(filepath.Join(generationsDir, entry.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
121 return fmt.Errorf("remove snapshot generation %d for %s: %w", generation, id.String(), safeFilesystemError(err))
122 }
123 removed++
124 }
125 if removed > 0 {
126 if err := r.syncDirectory(generationsDir); err != nil {
127 return fmt.Errorf("sync snapshot generations for %s: %w", id.String(), safeFilesystemError(err))
128 }
129 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
130 }
131 if err := r.removeHeadForDiscardedGenerations(id, kept); err != nil {
132 return err
133 }
134 return r.sweepUnreferencedObjects(ctx, id, references)
135}
136
137// removeHeadForDiscardedGenerations removes a repository HEAD only when its
138// generation was removed by the retention policy. In particular, an
139// incarnation with no committed checkpoint retains no generations, so its
140// crash-published HEAD cannot make a retry conflict with an orphan.
141func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}) error {
142 removeHead := len(kept) == 0
143 if !removeHead {
144 generation, _, err := r.readHead(id)
145 if errors.Is(err, os.ErrNotExist) {
146 return nil
147 }
148 if err != nil {
149 return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
150 }
151 _, retained := kept[generation]
152 removeHead = !retained
153 }
154 if !removeHead {
155 return nil
156 }
157 if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
158 return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
159 }
160 if err := r.syncDirectory(r.sessionPath(id)); err != nil {
161 return fmt.Errorf("sync snapshot incarnation for %s: %w", id.String(), safeFilesystemError(err))
162 }
163 return nil
164}
165
166func (r *Repository) referencesForGenerations(id domain.IncarnationID, entries []os.DirEntry, keep map[uint64]struct{}, committed domain.CheckpointRef) (map[ports.SnapshotDigest]struct{}, error) {
167 references := make(map[ports.SnapshotDigest]struct{})
168 committedFound := committed.Generation == 0
169 for _, entry := range entries {
170 generation, canonical := parseGenerationFilename(entry.Name())
171 if !canonical || entry.IsDir() {
172 continue
173 }
174 if _, retained := keep[generation]; !retained {
175 continue
176 }
177 data, err := r.readBounded(r.manifestPath(id, generation))
178 if err != nil {
179 return nil, err
180 }
181 manifest, err := codec.UnmarshalManifest(data)
182 if err != nil || manifest.IncarnationID != id || manifest.Generation != generation {
183 return nil, fmt.Errorf("invalid retained manifest generation %d", generation)
184 }
185 if generation == committed.Generation {
186 committedFound = true
187 if sha256.Sum256(data) != committed.ManifestDigest {
188 return nil, fmt.Errorf("committed manifest digest mismatch")
189 }
190 }
191 refs := manifestRefs(manifest)
192 if refs == nil || !withinGenerationBudget(len(data), refs) {
193 return nil, fmt.Errorf("invalid retained manifest references generation %d", generation)
194 }
195 for digest := range refs {
196 references[digest] = struct{}{}
197 }
198 }
199 if !committedFound {
200 return nil, fmt.Errorf("committed manifest generation %d is missing", committed.Generation)
201 }
202 return references, nil
203}
204
205type safePathError struct {
206 op string
207 cause error
208}
209
210func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
211func (e safePathError) Unwrap() error { return e.cause }
212
213// safeFilesystemError preserves error matching without exposing repository
214// paths in terminal-facing errors.
215func safeFilesystemError(err error) error {
216 var linkError *os.LinkError
217 if errors.As(err, &linkError) {
218 return safePathError{op: linkError.Op, cause: linkError.Err}
219 }
220 var pathError *os.PathError
221 if errors.As(err, &pathError) {
222 return safePathError{op: pathError.Op, cause: pathError.Err}
223 }
224 return err
225}
226
227func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.DirEntry, err error) {
228 directory, err := r.openDirectory(path)
229 if err != nil {
230 return nil, err
231 }
232 defer func() {
233 if closeErr := directory.Close(); closeErr != nil {
234 err = errors.Join(err, closeErr)
235 }
236 }()
237 return directory.ReadDir(-1)
238}
239
240func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) error {
241 objectsRoot := filepath.Join(r.sessionPath(id), repositoryObjectsDir)
242 shards, err := r.readGarbageCollectionDirectory(objectsRoot)
243 if errors.Is(err, os.ErrNotExist) {
244 return nil
245 }
246 if err != nil {
247 return fmt.Errorf("read snapshot objects for %s: %w", id.String(), safeFilesystemError(err))
248 }
249 removed := 0
250 for _, shard := range shards {
251 if err := ctx.Err(); err != nil {
252 return err
253 }
254 if !shard.IsDir() || len(shard.Name()) != 2 {
255 continue
256 }
257 dir := filepath.Join(objectsRoot, shard.Name())
258 objects, err := r.readGarbageCollectionDirectory(dir)
259 if err != nil {
260 return fmt.Errorf("read snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
261 }
262 shardRemoved := false
263 for _, object := range objects {
264 digest, canonical := parseObjectDigest(object.Name())
265 if !canonical || object.IsDir() {
266 continue
267 }
268 if _, retained := references[digest]; retained {
269 continue
270 }
271 if err := r.remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
272 return fmt.Errorf("remove snapshot object for %s: %w", id.String(), safeFilesystemError(err))
273 }
274 removed++
275 shardRemoved = true
276 }
277 if shardRemoved {
278 if err := r.syncDirectory(dir); err != nil {
279 return fmt.Errorf("sync snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
280 }
281 }
282 }
283 if removed > 0 {
284 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-objects", "removed", removed)
285 }
286 return nil
287}