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// SnapshotIncarnations lists the canonical incarnation directories.
17func (r *Repository) SnapshotIncarnations(ctx context.Context) ([]domain.IncarnationID, error) {
18 if err := ctx.Err(); err != nil {
19 return nil, err
20 }
21 entries, err := r.readGarbageCollectionDirectory(filepath.Join(r.dir, repositorySessionsDir))
22 if errors.Is(err, os.ErrNotExist) {
23 return nil, nil
24 }
25 if err != nil {
26 return nil, fmt.Errorf("read snapshot incarnations: %w", safeFilesystemError(err))
27 }
28 ids := make([]domain.IncarnationID, 0, len(entries))
29 for _, entry := range entries {
30 var id domain.IncarnationID
31 if entry.IsDir() && id.UnmarshalText([]byte(entry.Name())) == nil {
32 ids = append(ids, id)
33 }
34 }
35 return ids, nil
36}
37
38// CollectIncarnationGarbage applies the retention policy to one incarnation.
39// keep must come from a catalogue that loaded and validated successfully: only
40// then is a nil keep known to mark an orphan.
41func (r *Repository) CollectIncarnationGarbage(ctx context.Context, id domain.IncarnationID, keep *domain.CheckpointRef) error {
42 if err := ctx.Err(); err != nil {
43 return err
44 }
45 key, err := incarnationKey(id)
46 if err != nil {
47 return err
48 }
49 lock := r.lockSession(key)
50 defer r.unlockSession(lock)
51 if keep == nil {
52 if err := r.removeIncarnationLocked(id); err != nil {
53 return fmt.Errorf("remove orphan snapshot incarnation %s: %w", id.String(), safeFilesystemError(err))
54 }
55 if err := r.syncDirectory(filepath.Join(r.dir, repositorySessionsDir)); err != nil {
56 return fmt.Errorf("sync snapshot sessions directory: %w", safeFilesystemError(err))
57 }
58 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-incarnation")
59 return nil
60 }
61 return r.pruneGenerationsLocked(ctx, id, *keep)
62}
63
64// pruneGenerationsLocked keeps the committed generation and its immediate
65// predecessor. Everything else, including a forward orphan newer than the
66// catalogue commit, is removed before unreferenced objects are swept. The
67// caller holds the incarnation's session lock.
68func (r *Repository) pruneGenerationsLocked(ctx context.Context, id domain.IncarnationID, committed domain.CheckpointRef) error {
69 generationsDir := filepath.Join(r.sessionPath(id), repositoryGenerations)
70 entries, err := r.readGarbageCollectionDirectory(generationsDir)
71 if errors.Is(err, os.ErrNotExist) {
72 return nil
73 }
74 if err != nil {
75 return fmt.Errorf("read snapshot generations for %s: %w", id.String(), safeFilesystemError(err))
76 }
77
78 kept := make(map[uint64]struct{}, 2)
79 if committed.Generation > 0 {
80 kept[committed.Generation] = struct{}{}
81 }
82 if committed.Generation > 1 {
83 kept[committed.Generation-1] = struct{}{}
84 }
85 references, err := r.referencesForGenerations(id, entries, kept, committed)
86 if err != nil {
87 return fmt.Errorf("mark snapshot objects for %s: %w", id.String(), err)
88 }
89 // Reconcile HEAD before removing generations, so an interrupted pass never
90 // leaves HEAD naming a removed manifest.
91 if err := r.reconcileHeadForRetention(id, committed); err != nil {
92 return err
93 }
94
95 // Generation removal runs to completion: it is bounded by the retained
96 // window plus any backlog, and a half-pruned directory buys nothing.
97 removed := 0
98 for _, entry := range entries {
99 generation, canonical := parseGenerationFilename(entry.Name())
100 if !canonical || entry.IsDir() {
101 continue
102 }
103 if _, retain := kept[generation]; retain {
104 continue
105 }
106 if err := r.remove(filepath.Join(generationsDir, entry.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
107 return fmt.Errorf("remove snapshot generation %d for %s: %w", generation, id.String(), safeFilesystemError(err))
108 }
109 removed++
110 }
111 if removed > 0 {
112 if err := r.syncDirectory(generationsDir); err != nil {
113 return fmt.Errorf("sync snapshot generations for %s: %w", id.String(), safeFilesystemError(err))
114 }
115 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
116 }
117 return r.sweepUnreferencedObjects(ctx, id, references)
118}
119
120// reconcileHeadForRetention makes HEAD name the committed checkpoint. GC runs
121// under the mutation fence, so no publication is in flight and any other HEAD
122// (missing, malformed, a forward orphan, an older generation, or a digest
123// mismatch) is damage: a live session's next publication must find its
124// committed parent. Without a commit, no generation is retained and HEAD is
125// removed so a crash-published HEAD cannot make a retry conflict with an orphan.
126func (r *Repository) reconcileHeadForRetention(id domain.IncarnationID, committed domain.CheckpointRef) error {
127 if committed.Generation > 0 {
128 generation, digest, err := r.readHead(id)
129 if err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
130 return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
131 }
132 if err == nil && generation == committed.Generation && digest == committed.ManifestDigest {
133 return nil
134 }
135 // referencesForGenerations already validated the committed manifest digest.
136 if err := r.writeMutable(r.headPath(id), marshalHead(committed.Generation, committed.ManifestDigest)); err != nil {
137 return fmt.Errorf("rewind snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
138 }
139 return nil
140 }
141 if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
142 return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
143 }
144 if err := r.syncDirectory(r.sessionPath(id)); err != nil {
145 return fmt.Errorf("sync snapshot incarnation for %s: %w", id.String(), safeFilesystemError(err))
146 }
147 return nil
148}
149
150func (r *Repository) referencesForGenerations(id domain.IncarnationID, entries []os.DirEntry, keep map[uint64]struct{}, committed domain.CheckpointRef) (map[ports.SnapshotDigest]struct{}, error) {
151 references := make(map[ports.SnapshotDigest]struct{})
152 committedFound := committed.Generation == 0
153 for _, entry := range entries {
154 generation, canonical := parseGenerationFilename(entry.Name())
155 if !canonical || entry.IsDir() {
156 continue
157 }
158 if _, retained := keep[generation]; !retained {
159 continue
160 }
161 data, err := r.readBounded(r.manifestPath(id, generation))
162 if err != nil {
163 return nil, err
164 }
165 manifest, err := codec.UnmarshalManifest(data)
166 if err != nil || manifest.IncarnationID != id || manifest.Generation != generation {
167 return nil, fmt.Errorf("invalid retained manifest generation %d", generation)
168 }
169 if generation == committed.Generation {
170 committedFound = true
171 if sha256.Sum256(data) != committed.ManifestDigest {
172 return nil, fmt.Errorf("committed manifest digest mismatch")
173 }
174 }
175 refs := manifestRefs(manifest)
176 if refs == nil || !withinGenerationBudget(len(data), refs) {
177 return nil, fmt.Errorf("invalid retained manifest references generation %d", generation)
178 }
179 for digest := range refs {
180 references[digest] = struct{}{}
181 }
182 }
183 if !committedFound {
184 return nil, fmt.Errorf("committed manifest generation %d is missing", committed.Generation)
185 }
186 return references, nil
187}
188
189type safePathError struct {
190 op string
191 cause error
192}
193
194func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
195func (e safePathError) Unwrap() error { return e.cause }
196
197// safeFilesystemError preserves error matching without exposing repository
198// paths in terminal-facing errors.
199func safeFilesystemError(err error) error {
200 var linkError *os.LinkError
201 if errors.As(err, &linkError) {
202 return safePathError{op: linkError.Op, cause: linkError.Err}
203 }
204 var pathError *os.PathError
205 if errors.As(err, &pathError) {
206 return safePathError{op: pathError.Op, cause: pathError.Err}
207 }
208 return err
209}
210
211func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.DirEntry, err error) {
212 directory, err := r.openDirectory(path)
213 if err != nil {
214 return nil, err
215 }
216 return readAndCloseDirectory(directory)
217}
218
219// readRootDirectory lists rel through an already pinned repository root, so a
220// sweep over thousands of objects does not reopen and revalidate the root for
221// every shard.
222func readRootDirectory(root *os.Root, rel string) ([]os.DirEntry, error) {
223 directory, err := openRootDirectory(root, rel)
224 if err != nil {
225 return nil, err
226 }
227 return readAndCloseDirectory(directory)
228}
229
230func readAndCloseDirectory(directory *os.File) (entries []os.DirEntry, err error) {
231 defer func() {
232 if closeErr := directory.Close(); closeErr != nil {
233 err = errors.Join(err, closeErr)
234 }
235 }()
236 return directory.ReadDir(-1)
237}
238
239func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) (err error) {
240 objectsRoot, ok := r.repositoryRelative(filepath.Join(r.sessionPath(id), repositoryObjectsDir))
241 if !ok {
242 return fmt.Errorf("snapshot path outside repository")
243 }
244 root, err := r.openRoot()
245 if err != nil {
246 return fmt.Errorf("open snapshot root for %s: %w", id.String(), safeFilesystemError(err))
247 }
248 defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
249 shards, err := readRootDirectory(root, objectsRoot)
250 if errors.Is(err, os.ErrNotExist) {
251 return nil
252 }
253 if err != nil {
254 return fmt.Errorf("read snapshot objects for %s: %w", id.String(), safeFilesystemError(err))
255 }
256 removed := 0
257 for _, shard := range shards {
258 if err := ctx.Err(); err != nil {
259 return err
260 }
261 if !shard.IsDir() || len(shard.Name()) != 2 {
262 continue
263 }
264 dir := filepath.Join(objectsRoot, shard.Name())
265 objects, err := readRootDirectory(root, dir)
266 if err != nil {
267 return fmt.Errorf("read snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
268 }
269 // Object unlinks are deliberately not fsynced per shard. Objects are
270 // immutable and content-addressed: an unlink lost to a crash only brings
271 // back an unreferenced object, which the next pass collects again, and a
272 // later publication revalidates any existing object before reuse.
273 for _, object := range objects {
274 digest, canonical := parseObjectDigest(object.Name())
275 if !canonical || object.IsDir() {
276 continue
277 }
278 if _, retained := references[digest]; retained {
279 continue
280 }
281 if err := root.Remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
282 return fmt.Errorf("remove snapshot object for %s: %w", id.String(), safeFilesystemError(err))
283 }
284 removed++
285 }
286 }
287 if removed > 0 {
288 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-objects", "removed", removed)
289 }
290 return nil
291}