detached 9e04e97

fix(snapshot): make maintenance traversal fair

commit
9e04e97f33366f56b9686a01052a8f48ef05065e
author
brice · · 2026-07-21 20:57 UTC
committer
brice · · 2026-07-21 20:57 UTC
parent
b1b3fa9
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +418 -121
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance.go +321 -121
@@ -107,22 +107,33 @@ func pendingQuarantine(dir, key string) (bool, error) {
107107 }
108108 }
109109
110-// Maintain reaps a bounded amount of stale state. Every scan uses ReadDir with
111-// a fixed batch, and a quarantine tree is peeled one entry at a time so an
112-// interrupted pass can continue after another call or restart.
113-func (r *Repository) Maintain(ctx context.Context) error {
110+// Maintain reaps a bounded amount of stale state. Continuation handles are
111+// repository-owned, so a stable prefix cannot starve entries later in a large
112+// directory. Cancellation and errors discard those handles: a restarted
113+// repository (and the next fresh cycle) always begins at a safe boundary.
114+func (r *Repository) Maintain(ctx context.Context) (err error) {
115+ r.maintenanceMu.Lock()
116+ defer r.maintenanceMu.Unlock()
114117 if err := ctx.Err(); err != nil {
118+ r.resetMaintenance()
115119 return err
116120 }
121+ defer func() {
122+ if err != nil {
123+ r.resetMaintenance()
124+ }
125+ }()
126+ if r.maintenanceCursors == nil {
127+ r.maintenanceCursors = make(map[string]*maintenanceCursor)
128+ r.maintenanceSessions = make(map[string]*sessionMaintenance)
129+ }
130+
117131 budget := maintenanceBatch
118- if err := r.removeTemps(ctx, r.dir, &budget); err != nil {
132+ if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil || budget == 0 {
119133 return err
120134 }
121- if budget == 0 {
122- return nil
123- }
124135 sessions := filepath.Join(r.dir, repositorySessionsDir)
125- entries, err := readDirBatch(sessions, maintenanceBatch)
136+ entries, _, err := r.readMaintenanceDir(sessions, maintenanceBatch, "sessions")
126137 if errors.Is(err, os.ErrNotExist) {
127138 return nil
128139 }
@@ -157,11 +168,7 @@ func (r *Repository) Maintain(ctx context.Context) error {
157168 }
158169 lock := r.sessionLock(entry.Name())
159170 lock.Lock()
160- if err := ctx.Err(); err != nil {
161- lock.Unlock()
162- return err
163- }
164- err := r.maintainSession(ctx, entry.Name(), &budget)
171+ err = r.maintainSession(ctx, entry.Name(), &budget)
165172 lock.Unlock()
166173 if err != nil {
167174 return err
@@ -170,8 +177,68 @@ func (r *Repository) Maintain(ctx context.Context) error {
170177 return nil
171178 }
172179
180+type maintenanceCursor struct{ file *os.File }
181+
182+type manifestMaintenance struct {
183+ refs map[ports.SnapshotDigest]codec.ObjectRef
184+ complete bool
185+}
186+
187+type sessionMaintenance struct {
188+ token string
189+ marked map[uint64]manifestMaintenance
190+ uncertain bool
191+ markDone bool
192+ manifestQueue []uint64
193+ sweepQueue []string
194+ sweepRootDone bool
195+}
196+
197+func (r *Repository) resetMaintenance() {
198+ for _, cursor := range r.maintenanceCursors {
199+ _ = cursor.file.Close()
200+ }
201+ r.maintenanceCursors = nil
202+ r.maintenanceSessions = nil
203+}
204+
205+// readMaintenanceDir advances one named directory cursor. It closes and
206+// forgets the handle at EOF, so the following call begins a fresh cycle.
207+func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {
208+ id := purpose + "\x00" + dir
209+ cursor := r.maintenanceCursors[id]
210+ if cursor == nil {
211+ file, err := os.Open(dir)
212+ if err != nil {
213+ return nil, false, err
214+ }
215+ cursor = &maintenanceCursor{file: file}
216+ r.maintenanceCursors[id] = cursor
217+ }
218+ entries, err := cursor.file.ReadDir(n)
219+ if err != nil && !errors.Is(err, io.EOF) {
220+ delete(r.maintenanceCursors, id)
221+ closeErr := cursor.file.Close()
222+ if closeErr != nil {
223+ return nil, false, closeErr
224+ }
225+ return nil, false, err
226+ }
227+ if !errors.Is(err, io.EOF) && len(entries) == n {
228+ return entries, false, nil
229+ }
230+ delete(r.maintenanceCursors, id)
231+ if closeErr := cursor.file.Close(); closeErr != nil {
232+ return nil, false, closeErr
233+ }
234+ return entries, true, nil
235+}
236+
173237 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
174238
239+// readDirBatch is used only for a mutating quarantine walk, where every
240+// successful step removes the entry it selected. General maintenance scans use
241+// repository-owned cursors above.
175242 func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
176243 f, err := os.Open(dir)
177244 if err != nil {
@@ -221,6 +288,8 @@ func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *in
221288 *budget--
222289 return true, nil
223290 }
291+ // A quarantine tree contains no live entries. Reopening from its start is
292+ // safe here because each successful step removes that first descendant.
224293 entries, err := readDirBatch(path, 1)
225294 if err != nil {
226295 return false, err
@@ -242,11 +311,16 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
242311 if err := ctx.Err(); err != nil {
243312 return err
244313 }
245- if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil {
314+ state, err := r.sessionMaintenanceState(key)
315+ if err != nil {
246316 return err
247317 }
318+ if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
319+ return err
320+ }
321+
248322 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
249- shards, err := readDirBatch(objectRoot, maintenanceBatch)
323+ shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
250324 if err != nil && !errors.Is(err, os.ErrNotExist) {
251325 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
252326 }
@@ -255,160 +329,286 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
255329 return nil
256330 }
257331 if shard.IsDir() {
258- if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget); err != nil {
332+ if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget, "object-temps:"+key+":"+shard.Name()); err != nil {
259333 return err
260334 }
261335 }
262336 }
337+ if *budget == 0 {
338+ return nil
339+ }
340+ if !state.markDone {
341+ if err := r.markSession(ctx, key, state); err != nil {
342+ return err
343+ }
344+ if !state.markDone {
345+ return nil
346+ }
347+ }
348+ if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
349+ return err
350+ }
351+ if state.uncertain {
352+ // An unreadable or invalid manifest may reference any existing blob. Do
353+ // not sweep in this cycle; retry marking from a fresh directory pass.
354+ r.clearSessionMaintenance(key)
355+ return nil
356+ }
357+ if err := r.sweepSession(ctx, key, state, budget); err != nil {
358+ return err
359+ }
360+ return nil
361+}
362+
363+func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
364+ token, exists, err := r.maintenanceToken(key)
365+ if err != nil {
366+ return nil, err
367+ }
368+ if !exists {
369+ r.clearSessionMaintenance(key)
370+ return &sessionMaintenance{token: "missing", marked: make(map[uint64]manifestMaintenance)}, nil
371+ }
372+ state := r.maintenanceSessions[key]
373+ if state == nil || state.token != token {
374+ r.clearSessionMaintenance(key)
375+ state = &sessionMaintenance{token: token, marked: make(map[uint64]manifestMaintenance)}
376+ r.maintenanceSessions[key] = state
377+ }
378+ return state, nil
379+}
380+
381+// maintenanceToken is the publication boundary. Publish and Delete share the
382+// per-session lock with maintenance, and a changed HEAD on a later call makes
383+// us discard partial marking/sweeping before deleting anything else.
384+func (r *Repository) maintenanceToken(key string) (string, bool, error) {
385+ data, exists, err := readOptionalBounded(r.headPath(key))
386+ if errors.Is(err, os.ErrNotExist) {
387+ return "", false, nil
388+ }
389+ if err != nil {
390+ return "", false, err
391+ }
392+ if !exists {
393+ return "", false, nil
394+ }
395+ return string(data), true, nil
396+}
397+
398+func (r *Repository) clearSessionMaintenance(key string) {
399+ delete(r.maintenanceSessions, key)
400+ prefix := "\x00" + filepath.Clean(r.sessionPath(key))
401+ for id, cursor := range r.maintenanceCursors {
402+ if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
403+ _ = cursor.file.Close()
404+ delete(r.maintenanceCursors, id)
405+ }
406+ }
407+}
263408
264- generations, err := r.generationNumbersBatch(key)
409+func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
410+ dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
411+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
412+ if errors.Is(err, os.ErrNotExist) {
413+ state.markDone = true
414+ return nil
415+ }
265416 if err != nil {
266417 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
267418 }
268- type manifestState struct {
269- generation uint64
270- refs map[ports.SnapshotDigest]codec.ObjectRef
271- complete bool
272- }
273- states := make([]manifestState, 0, len(generations))
274- complete := 0
275- for _, generation := range generations {
419+ for _, entry := range entries {
276420 if err := ctx.Err(); err != nil {
277421 return err
278422 }
279- state := manifestState{generation: generation}
423+ if entry.IsDir() {
424+ continue
425+ }
426+ generation, ok := parseGenerationFilename(entry.Name())
427+ if !ok {
428+ continue
429+ }
430+ item := manifestMaintenance{}
280431 data, err := readBounded(r.manifestPath(key, generation))
281- if err == nil {
282- manifest, decodeErr := codec.UnmarshalManifest(data)
283- if decodeErr == nil && manifest.Generation == generation && sessionKey(manifest.Name) == key {
284- state.refs = manifestRefs(manifest)
285- if state.refs != nil && withinGenerationBudget(len(data), state.refs) {
286- _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
287- state.complete = loadErr == nil
288- }
289- }
432+ if err != nil {
433+ state.uncertain = true
434+ state.marked[generation] = item
435+ continue
436+ }
437+ manifest, err := codec.UnmarshalManifest(data)
438+ if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
439+ state.uncertain = true
440+ state.marked[generation] = item
441+ continue
442+ }
443+ item.refs = manifestRefs(manifest)
444+ if item.refs == nil || !withinGenerationBudget(len(data), item.refs) {
445+ state.uncertain = true
446+ state.marked[generation] = item
447+ continue
290448 }
291- states = append(states, state)
449+ // Valid but incomplete generations are marked too. This protects blobs
450+ // written before a failed or in-flight publication reaches HEAD.
451+ _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
452+ item.complete = loadErr == nil
453+ state.marked[generation] = item
292454 }
455+ if done {
456+ state.markDone = true
457+ state.manifestQueue = retainedManifestQueue(state.marked)
458+ }
459+ return nil
460+}
293461
294- retained := make(map[uint64]bool, 2)
295- for _, state := range states {
296- if state.complete && complete < 2 {
297- retained[state.generation] = true
298- complete++
462+func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
463+ complete := make([]uint64, 0, len(marked))
464+ for generation, item := range marked {
465+ if item.complete {
466+ complete = append(complete, generation)
299467 }
300468 }
301- referenced := make(map[ports.SnapshotDigest]struct{})
302- for _, state := range states {
303- keep := retained[state.generation] || !state.complete
304- if !keep && *budget > 0 {
305- path := r.manifestPath(key, state.generation)
306- if err := ctx.Err(); err != nil {
307- return err
308- }
309- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
310- return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
311- }
312- if err := ctx.Err(); err != nil {
313- return err
469+ for i := range complete {
470+ for j := i + 1; j < len(complete); j++ {
471+ if complete[j] > complete[i] {
472+ complete[i], complete[j] = complete[j], complete[i]
314473 }
315- if err := r.syncDirectory(filepath.Dir(path)); err != nil {
316- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
317- }
318- *budget--
319- continue
320474 }
321- for digest := range state.refs {
322- referenced[digest] = struct{}{}
475+ }
476+ if len(complete) <= 2 {
477+ return nil
478+ }
479+ return complete[2:]
480+}
481+
482+func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
483+ for len(state.manifestQueue) != 0 && *budget > 0 {
484+ if err := ctx.Err(); err != nil {
485+ return err
323486 }
487+ generation := state.manifestQueue[0]
488+ path := r.manifestPath(key, generation)
489+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
490+ return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
491+ }
492+ if err := ctx.Err(); err != nil {
493+ return err
494+ }
495+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
496+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
497+ }
498+ state.manifestQueue = state.manifestQueue[1:]
499+ *budget--
324500 }
325- return r.removeUnreferencedObjects(ctx, key, referenced, budget)
501+ return nil
326502 }
327503
328-func (r *Repository) generationNumbersBatch(key string) ([]uint64, error) {
329- entries, err := readDirBatch(filepath.Join(r.sessionPath(key), repositoryGenerations), maintenanceBatch)
330- if errors.Is(err, os.ErrNotExist) {
331- return nil, nil
504+func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
505+ referenced := retainedReferences(state.marked)
506+ root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
507+ if !state.sweepRootDone {
508+ entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
509+ if errors.Is(err, os.ErrNotExist) {
510+ state.sweepRootDone = true
511+ } else if err != nil {
512+ return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
513+ } else {
514+ for _, entry := range entries {
515+ if entry.IsDir() {
516+ state.sweepQueue = append(state.sweepQueue, entry.Name())
517+ }
518+ }
519+ state.sweepRootDone = done
520+ }
332521 }
333- if err != nil {
334- return nil, err
522+ if len(state.sweepQueue) != 0 && *budget > 0 {
523+ shard := state.sweepQueue[0]
524+ done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
525+ if err != nil {
526+ return err
527+ }
528+ if done {
529+ state.sweepQueue = state.sweepQueue[1:]
530+ }
335531 }
336- numbers := make([]uint64, 0, len(entries))
337- for _, entry := range entries {
338- if !entry.IsDir() {
339- if generation, ok := parseGenerationFilename(entry.Name()); ok {
340- numbers = append(numbers, generation)
532+ if state.sweepRootDone && len(state.sweepQueue) == 0 {
533+ r.clearSessionMaintenance(key)
534+ }
535+ return nil
536+}
537+
538+func retainedReferences(marked map[uint64]manifestMaintenance) map[ports.SnapshotDigest]struct{} {
539+ complete := make([]uint64, 0, len(marked))
540+ for generation, item := range marked {
541+ if item.complete {
542+ complete = append(complete, generation)
543+ }
544+ }
545+ for i := range complete {
546+ for j := i + 1; j < len(complete); j++ {
547+ if complete[j] > complete[i] {
548+ complete[i], complete[j] = complete[j], complete[i]
341549 }
342550 }
343551 }
344- // Generation file names are zero-padded, but keep the explicit comparison
345- // in case the representation changes.
346- for i := range numbers {
347- for j := i + 1; j < len(numbers); j++ {
348- if numbers[j] > numbers[i] {
349- numbers[i], numbers[j] = numbers[j], numbers[i]
552+ keep := make(map[uint64]bool, 2)
553+ for _, generation := range complete {
554+ if len(keep) == 2 {
555+ break
556+ }
557+ keep[generation] = true
558+ }
559+ references := make(map[ports.SnapshotDigest]struct{})
560+ for generation, item := range marked {
561+ if !item.complete || keep[generation] {
562+ for digest := range item.refs {
563+ references[digest] = struct{}{}
350564 }
351565 }
352566 }
353- return numbers, nil
567+ return references
354568 }
355569
356-func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error {
357- root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
358- shards, err := readDirBatch(root, maintenanceBatch)
570+func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
571+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
359572 if errors.Is(err, os.ErrNotExist) {
360- return nil
573+ return true, nil
361574 }
362575 if err != nil {
363- return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
576+ return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
364577 }
365- for _, shard := range shards {
578+ for _, object := range entries {
366579 if *budget == 0 {
367- return nil
580+ break
581+ }
582+ if err := ctx.Err(); err != nil {
583+ return false, err
368584 }
369- if !shard.IsDir() {
585+ digest, ok := parseObjectDigest(object.Name())
586+ if !ok || object.IsDir() {
370587 continue
371588 }
372- dir := filepath.Join(root, shard.Name())
373- objects, err := readDirBatch(dir, maintenanceBatch)
374- if err != nil {
375- return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
589+ if _, used := referenced[digest]; used {
590+ continue
376591 }
377- for _, object := range objects {
378- if *budget == 0 {
379- return nil
380- }
381- if err := ctx.Err(); err != nil {
382- return err
383- }
384- digest, ok := parseObjectDigest(object.Name())
385- if !ok || object.IsDir() {
386- continue
387- }
388- if _, used := referenced[digest]; used {
389- continue
390- }
391- path := filepath.Join(dir, object.Name())
392- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
393- return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
394- }
395- if err := ctx.Err(); err != nil {
396- return err
397- }
398- if err := r.syncDirectory(dir); err != nil {
399- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
400- }
401- *budget--
592+ path := filepath.Join(dir, object.Name())
593+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
594+ return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
595+ }
596+ if err := ctx.Err(); err != nil {
597+ return false, err
402598 }
599+ if err := r.syncDirectory(dir); err != nil {
600+ return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
601+ }
602+ *budget--
403603 }
404- return nil
604+ return done, nil
405605 }
406606
407-func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) error {
607+func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
408608 if *budget == 0 {
409609 return nil
410610 }
411- entries, err := readDirBatch(dir, maintenanceBatch)
611+ entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
412612 if errors.Is(err, os.ErrNotExist) {
413613 return nil
414614 }
@@ -107,22 +107,33 @@ func pendingQuarantine(dir, key string) (bool, error) {
107 }107 }
108 }108 }
109 109
110-// Maintain reaps a bounded amount of stale state. Every scan uses ReadDir with110+// Maintain reaps a bounded amount of stale state. Continuation handles are
111-// a fixed batch, and a quarantine tree is peeled one entry at a time so an111+// repository-owned, so a stable prefix cannot starve entries later in a large
112-// interrupted pass can continue after another call or restart.112+// directory. Cancellation and errors discard those handles: a restarted
113-func (r *Repository) Maintain(ctx context.Context) error {113+// repository (and the next fresh cycle) always begins at a safe boundary.
114+func (r *Repository) Maintain(ctx context.Context) (err error) {
115+ r.maintenanceMu.Lock()
116+ defer r.maintenanceMu.Unlock()
114 if err := ctx.Err(); err != nil {117 if err := ctx.Err(); err != nil {
118+ r.resetMaintenance()
115 return err119 return err
116 }120 }
121+ defer func() {
122+ if err != nil {
123+ r.resetMaintenance()
124+ }
125+ }()
126+ if r.maintenanceCursors == nil {
127+ r.maintenanceCursors = make(map[string]*maintenanceCursor)
128+ r.maintenanceSessions = make(map[string]*sessionMaintenance)
129+ }
130+
117 budget := maintenanceBatch131 budget := maintenanceBatch
118- if err := r.removeTemps(ctx, r.dir, &budget); err != nil {132+ if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil || budget == 0 {
119 return err133 return err
120 }134 }
121- if budget == 0 {
122- return nil
123- }
124 sessions := filepath.Join(r.dir, repositorySessionsDir)135 sessions := filepath.Join(r.dir, repositorySessionsDir)
125- entries, err := readDirBatch(sessions, maintenanceBatch)136+ entries, _, err := r.readMaintenanceDir(sessions, maintenanceBatch, "sessions")
126 if errors.Is(err, os.ErrNotExist) {137 if errors.Is(err, os.ErrNotExist) {
127 return nil138 return nil
128 }139 }
@@ -157,11 +168,7 @@ func (r *Repository) Maintain(ctx context.Context) error {
157 }168 }
158 lock := r.sessionLock(entry.Name())169 lock := r.sessionLock(entry.Name())
159 lock.Lock()170 lock.Lock()
160- if err := ctx.Err(); err != nil {171+ err = r.maintainSession(ctx, entry.Name(), &budget)
161- lock.Unlock()
162- return err
163- }
164- err := r.maintainSession(ctx, entry.Name(), &budget)
165 lock.Unlock()172 lock.Unlock()
166 if err != nil {173 if err != nil {
167 return err174 return err
@@ -170,8 +177,68 @@ func (r *Repository) Maintain(ctx context.Context) error {
170 return nil177 return nil
171 }178 }
172 179
180+type maintenanceCursor struct{ file *os.File }
181+
182+type manifestMaintenance struct {
183+ refs map[ports.SnapshotDigest]codec.ObjectRef
184+ complete bool
185+}
186+
187+type sessionMaintenance struct {
188+ token string
189+ marked map[uint64]manifestMaintenance
190+ uncertain bool
191+ markDone bool
192+ manifestQueue []uint64
193+ sweepQueue []string
194+ sweepRootDone bool
195+}
196+
197+func (r *Repository) resetMaintenance() {
198+ for _, cursor := range r.maintenanceCursors {
199+ _ = cursor.file.Close()
200+ }
201+ r.maintenanceCursors = nil
202+ r.maintenanceSessions = nil
203+}
204+
205+// readMaintenanceDir advances one named directory cursor. It closes and
206+// forgets the handle at EOF, so the following call begins a fresh cycle.
207+func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {
208+ id := purpose + "\x00" + dir
209+ cursor := r.maintenanceCursors[id]
210+ if cursor == nil {
211+ file, err := os.Open(dir)
212+ if err != nil {
213+ return nil, false, err
214+ }
215+ cursor = &maintenanceCursor{file: file}
216+ r.maintenanceCursors[id] = cursor
217+ }
218+ entries, err := cursor.file.ReadDir(n)
219+ if err != nil && !errors.Is(err, io.EOF) {
220+ delete(r.maintenanceCursors, id)
221+ closeErr := cursor.file.Close()
222+ if closeErr != nil {
223+ return nil, false, closeErr
224+ }
225+ return nil, false, err
226+ }
227+ if !errors.Is(err, io.EOF) && len(entries) == n {
228+ return entries, false, nil
229+ }
230+ delete(r.maintenanceCursors, id)
231+ if closeErr := cursor.file.Close(); closeErr != nil {
232+ return nil, false, closeErr
233+ }
234+ return entries, true, nil
235+}
236+
173 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }237 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
174 238
239+// readDirBatch is used only for a mutating quarantine walk, where every
240+// successful step removes the entry it selected. General maintenance scans use
241+// repository-owned cursors above.
175 func readDirBatch(dir string, n int) ([]os.DirEntry, error) {242 func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
176 f, err := os.Open(dir)243 f, err := os.Open(dir)
177 if err != nil {244 if err != nil {
@@ -221,6 +288,8 @@ func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *in
221 *budget--288 *budget--
222 return true, nil289 return true, nil
223 }290 }
291+ // A quarantine tree contains no live entries. Reopening from its start is
292+ // safe here because each successful step removes that first descendant.
224 entries, err := readDirBatch(path, 1)293 entries, err := readDirBatch(path, 1)
225 if err != nil {294 if err != nil {
226 return false, err295 return false, err
@@ -242,11 +311,16 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
242 if err := ctx.Err(); err != nil {311 if err := ctx.Err(); err != nil {
243 return err312 return err
244 }313 }
245- if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil {314+ state, err := r.sessionMaintenanceState(key)
315+ if err != nil {
246 return err316 return err
247 }317 }
318+ if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
319+ return err
320+ }
321+
248 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)322 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
249- shards, err := readDirBatch(objectRoot, maintenanceBatch)323+ shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
250 if err != nil && !errors.Is(err, os.ErrNotExist) {324 if err != nil && !errors.Is(err, os.ErrNotExist) {
251 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))325 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
252 }326 }
@@ -255,160 +329,286 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
255 return nil329 return nil
256 }330 }
257 if shard.IsDir() {331 if shard.IsDir() {
258- if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget); err != nil {332+ if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget, "object-temps:"+key+":"+shard.Name()); err != nil {
259 return err333 return err
260 }334 }
261 }335 }
262 }336 }
337+ if *budget == 0 {
338+ return nil
339+ }
340+ if !state.markDone {
341+ if err := r.markSession(ctx, key, state); err != nil {
342+ return err
343+ }
344+ if !state.markDone {
345+ return nil
346+ }
347+ }
348+ if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
349+ return err
350+ }
351+ if state.uncertain {
352+ // An unreadable or invalid manifest may reference any existing blob. Do
353+ // not sweep in this cycle; retry marking from a fresh directory pass.
354+ r.clearSessionMaintenance(key)
355+ return nil
356+ }
357+ if err := r.sweepSession(ctx, key, state, budget); err != nil {
358+ return err
359+ }
360+ return nil
361+}
362+
363+func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
364+ token, exists, err := r.maintenanceToken(key)
365+ if err != nil {
366+ return nil, err
367+ }
368+ if !exists {
369+ r.clearSessionMaintenance(key)
370+ return &sessionMaintenance{token: "missing", marked: make(map[uint64]manifestMaintenance)}, nil
371+ }
372+ state := r.maintenanceSessions[key]
373+ if state == nil || state.token != token {
374+ r.clearSessionMaintenance(key)
375+ state = &sessionMaintenance{token: token, marked: make(map[uint64]manifestMaintenance)}
376+ r.maintenanceSessions[key] = state
377+ }
378+ return state, nil
379+}
380+
381+// maintenanceToken is the publication boundary. Publish and Delete share the
382+// per-session lock with maintenance, and a changed HEAD on a later call makes
383+// us discard partial marking/sweeping before deleting anything else.
384+func (r *Repository) maintenanceToken(key string) (string, bool, error) {
385+ data, exists, err := readOptionalBounded(r.headPath(key))
386+ if errors.Is(err, os.ErrNotExist) {
387+ return "", false, nil
388+ }
389+ if err != nil {
390+ return "", false, err
391+ }
392+ if !exists {
393+ return "", false, nil
394+ }
395+ return string(data), true, nil
396+}
397+
398+func (r *Repository) clearSessionMaintenance(key string) {
399+ delete(r.maintenanceSessions, key)
400+ prefix := "\x00" + filepath.Clean(r.sessionPath(key))
401+ for id, cursor := range r.maintenanceCursors {
402+ if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
403+ _ = cursor.file.Close()
404+ delete(r.maintenanceCursors, id)
405+ }
406+ }
407+}
263 408
264- generations, err := r.generationNumbersBatch(key)409+func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
410+ dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
411+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
412+ if errors.Is(err, os.ErrNotExist) {
413+ state.markDone = true
414+ return nil
415+ }
265 if err != nil {416 if err != nil {
266 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))417 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
267 }418 }
268- type manifestState struct {419+ for _, entry := range entries {
269- generation uint64
270- refs map[ports.SnapshotDigest]codec.ObjectRef
271- complete bool
272- }
273- states := make([]manifestState, 0, len(generations))
274- complete := 0
275- for _, generation := range generations {
276 if err := ctx.Err(); err != nil {420 if err := ctx.Err(); err != nil {
277 return err421 return err
278 }422 }
279- state := manifestState{generation: generation}423+ if entry.IsDir() {
424+ continue
425+ }
426+ generation, ok := parseGenerationFilename(entry.Name())
427+ if !ok {
428+ continue
429+ }
430+ item := manifestMaintenance{}
280 data, err := readBounded(r.manifestPath(key, generation))431 data, err := readBounded(r.manifestPath(key, generation))
281- if err == nil {432+ if err != nil {
282- manifest, decodeErr := codec.UnmarshalManifest(data)433+ state.uncertain = true
283- if decodeErr == nil && manifest.Generation == generation && sessionKey(manifest.Name) == key {434+ state.marked[generation] = item
284- state.refs = manifestRefs(manifest)435+ continue
285- if state.refs != nil && withinGenerationBudget(len(data), state.refs) {436+ }
286- _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)437+ manifest, err := codec.UnmarshalManifest(data)
287- state.complete = loadErr == nil438+ if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
288- }439+ state.uncertain = true
289- }440+ state.marked[generation] = item
441+ continue
442+ }
443+ item.refs = manifestRefs(manifest)
444+ if item.refs == nil || !withinGenerationBudget(len(data), item.refs) {
445+ state.uncertain = true
446+ state.marked[generation] = item
447+ continue
290 }448 }
291- states = append(states, state)449+ // Valid but incomplete generations are marked too. This protects blobs
450+ // written before a failed or in-flight publication reaches HEAD.
451+ _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
452+ item.complete = loadErr == nil
453+ state.marked[generation] = item
292 }454 }
455+ if done {
456+ state.markDone = true
457+ state.manifestQueue = retainedManifestQueue(state.marked)
458+ }
459+ return nil
460+}
293 461
294- retained := make(map[uint64]bool, 2)462+func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
295- for _, state := range states {463+ complete := make([]uint64, 0, len(marked))
296- if state.complete && complete < 2 {464+ for generation, item := range marked {
297- retained[state.generation] = true465+ if item.complete {
298- complete++466+ complete = append(complete, generation)
299 }467 }
300 }468 }
301- referenced := make(map[ports.SnapshotDigest]struct{})469+ for i := range complete {
302- for _, state := range states {470+ for j := i + 1; j < len(complete); j++ {
303- keep := retained[state.generation] || !state.complete471+ if complete[j] > complete[i] {
304- if !keep && *budget > 0 {472+ complete[i], complete[j] = complete[j], complete[i]
305- path := r.manifestPath(key, state.generation)
306- if err := ctx.Err(); err != nil {
307- return err
308- }
309- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
310- return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
311- }
312- if err := ctx.Err(); err != nil {
313- return err
314 }473 }
315- if err := r.syncDirectory(filepath.Dir(path)); err != nil {
316- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
317- }
318- *budget--
319- continue
320 }474 }
321- for digest := range state.refs {475+ }
322- referenced[digest] = struct{}{}476+ if len(complete) <= 2 {
477+ return nil
478+ }
479+ return complete[2:]
480+}
481+
482+func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
483+ for len(state.manifestQueue) != 0 && *budget > 0 {
484+ if err := ctx.Err(); err != nil {
485+ return err
323 }486 }
487+ generation := state.manifestQueue[0]
488+ path := r.manifestPath(key, generation)
489+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
490+ return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
491+ }
492+ if err := ctx.Err(); err != nil {
493+ return err
494+ }
495+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
496+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
497+ }
498+ state.manifestQueue = state.manifestQueue[1:]
499+ *budget--
324 }500 }
325- return r.removeUnreferencedObjects(ctx, key, referenced, budget)501+ return nil
326 }502 }
327 503
328-func (r *Repository) generationNumbersBatch(key string) ([]uint64, error) {504+func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
329- entries, err := readDirBatch(filepath.Join(r.sessionPath(key), repositoryGenerations), maintenanceBatch)505+ referenced := retainedReferences(state.marked)
330- if errors.Is(err, os.ErrNotExist) {506+ root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
331- return nil, nil507+ if !state.sweepRootDone {
508+ entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
509+ if errors.Is(err, os.ErrNotExist) {
510+ state.sweepRootDone = true
511+ } else if err != nil {
512+ return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
513+ } else {
514+ for _, entry := range entries {
515+ if entry.IsDir() {
516+ state.sweepQueue = append(state.sweepQueue, entry.Name())
517+ }
518+ }
519+ state.sweepRootDone = done
520+ }
332 }521 }
333- if err != nil {522+ if len(state.sweepQueue) != 0 && *budget > 0 {
334- return nil, err523+ shard := state.sweepQueue[0]
524+ done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
525+ if err != nil {
526+ return err
527+ }
528+ if done {
529+ state.sweepQueue = state.sweepQueue[1:]
530+ }
335 }531 }
336- numbers := make([]uint64, 0, len(entries))532+ if state.sweepRootDone && len(state.sweepQueue) == 0 {
337- for _, entry := range entries {533+ r.clearSessionMaintenance(key)
338- if !entry.IsDir() {534+ }
339- if generation, ok := parseGenerationFilename(entry.Name()); ok {535+ return nil
340- numbers = append(numbers, generation)536+}
537+
538+func retainedReferences(marked map[uint64]manifestMaintenance) map[ports.SnapshotDigest]struct{} {
539+ complete := make([]uint64, 0, len(marked))
540+ for generation, item := range marked {
541+ if item.complete {
542+ complete = append(complete, generation)
543+ }
544+ }
545+ for i := range complete {
546+ for j := i + 1; j < len(complete); j++ {
547+ if complete[j] > complete[i] {
548+ complete[i], complete[j] = complete[j], complete[i]
341 }549 }
342 }550 }
343 }551 }
344- // Generation file names are zero-padded, but keep the explicit comparison552+ keep := make(map[uint64]bool, 2)
345- // in case the representation changes.553+ for _, generation := range complete {
346- for i := range numbers {554+ if len(keep) == 2 {
347- for j := i + 1; j < len(numbers); j++ {555+ break
348- if numbers[j] > numbers[i] {556+ }
349- numbers[i], numbers[j] = numbers[j], numbers[i]557+ keep[generation] = true
558+ }
559+ references := make(map[ports.SnapshotDigest]struct{})
560+ for generation, item := range marked {
561+ if !item.complete || keep[generation] {
562+ for digest := range item.refs {
563+ references[digest] = struct{}{}
350 }564 }
351 }565 }
352 }566 }
353- return numbers, nil567+ return references
354 }568 }
355 569
356-func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error {570+func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
357- root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)571+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
358- shards, err := readDirBatch(root, maintenanceBatch)
359 if errors.Is(err, os.ErrNotExist) {572 if errors.Is(err, os.ErrNotExist) {
360- return nil573+ return true, nil
361 }574 }
362 if err != nil {575 if err != nil {
363- return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))576+ return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
364 }577 }
365- for _, shard := range shards {578+ for _, object := range entries {
366 if *budget == 0 {579 if *budget == 0 {
367- return nil580+ break
581+ }
582+ if err := ctx.Err(); err != nil {
583+ return false, err
368 }584 }
369- if !shard.IsDir() {585+ digest, ok := parseObjectDigest(object.Name())
586+ if !ok || object.IsDir() {
370 continue587 continue
371 }588 }
372- dir := filepath.Join(root, shard.Name())589+ if _, used := referenced[digest]; used {
373- objects, err := readDirBatch(dir, maintenanceBatch)590+ continue
374- if err != nil {
375- return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
376 }591 }
377- for _, object := range objects {592+ path := filepath.Join(dir, object.Name())
378- if *budget == 0 {593+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
379- return nil594+ return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
380- }595+ }
381- if err := ctx.Err(); err != nil {596+ if err := ctx.Err(); err != nil {
382- return err597+ return false, err
383- }
384- digest, ok := parseObjectDigest(object.Name())
385- if !ok || object.IsDir() {
386- continue
387- }
388- if _, used := referenced[digest]; used {
389- continue
390- }
391- path := filepath.Join(dir, object.Name())
392- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
393- return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
394- }
395- if err := ctx.Err(); err != nil {
396- return err
397- }
398- if err := r.syncDirectory(dir); err != nil {
399- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
400- }
401- *budget--
402 }598 }
599+ if err := r.syncDirectory(dir); err != nil {
600+ return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
601+ }
602+ *budget--
403 }603 }
404- return nil604+ return done, nil
405 }605 }
406 606
407-func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) error {607+func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
408 if *budget == 0 {608 if *budget == 0 {
409 return nil609 return nil
410 }610 }
411- entries, err := readDirBatch(dir, maintenanceBatch)611+ entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
412 if errors.Is(err, os.ErrNotExist) {612 if errors.Is(err, os.ErrNotExist) {
413 return nil613 return nil
414 }614 }
M internal/adapters/snapshot/maintenance_test.go +90 -0
@@ -2,10 +2,15 @@ package snapshot
22
33 import (
44 "context"
5+ "crypto/sha256"
56 "errors"
7+ "fmt"
68 "os"
79 "path/filepath"
10+ "strings"
811 "testing"
12+
13+ "github.com/bnema/vev/internal/ports"
914 )
1015
1116 func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {
@@ -268,6 +273,91 @@ func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
268273 }
269274 }
270275
276+func TestRepositoryMaintainDoesNotStarveLaterTemporaryEntries(t *testing.T) {
277+ repo := NewRepository(privateDir(t))
278+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
279+ t.Fatal(err)
280+ }
281+ // The live entries precede the stale entries in directory insertion order.
282+ // A fresh ReadDir from the start on every call would never reach the latter.
283+ for i := 0; i < maintenanceBatch+1; i++ {
284+ if err := os.WriteFile(filepath.Join(repo.dir, "live-"+string(rune('a'+i))), []byte("live"), 0o600); err != nil {
285+ t.Fatal(err)
286+ }
287+ }
288+ for i := 0; i < maintenanceBatch+1; i++ {
289+ if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-later-"+string(rune('a'+i))), []byte("stale"), 0o600); err != nil {
290+ t.Fatal(err)
291+ }
292+ }
293+ for i := 0; i < 4; i++ {
294+ if err := repo.Maintain(context.Background()); err != nil {
295+ t.Fatal(err)
296+ }
297+ }
298+ entries, err := os.ReadDir(repo.dir)
299+ if err != nil {
300+ t.Fatal(err)
301+ }
302+ for _, entry := range entries {
303+ if strings.HasPrefix(entry.Name(), ".tmp-") {
304+ t.Fatalf("stale entry %q was starved", entry.Name())
305+ }
306+ }
307+}
308+
309+func TestRepositoryMaintainMarksLargeGenerationSetBeforeSweeping(t *testing.T) {
310+ repo := NewRepository(privateDir(t))
311+ var newest ports.SnapshotPublication
312+ for generation := uint64(1); generation <= maintenanceBatch+2; generation++ {
313+ publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))
314+ if err := repo.Publish(context.Background(), publication); err != nil {
315+ t.Fatal(err)
316+ }
317+ newest = publication
318+ }
319+ key := sessionKey("named")
320+ stale := sha256.Sum256([]byte("stale object"))
321+ stalePath := repo.objectPath(key, stale)
322+ if err := os.MkdirAll(filepath.Dir(stalePath), 0o700); err != nil {
323+ t.Fatal(err)
324+ }
325+ if err := os.WriteFile(stalePath, []byte("stale object"), 0o600); err != nil {
326+ t.Fatal(err)
327+ }
328+
329+ // The first call has classified only one manifest batch. It must not sweep
330+ // even an otherwise stale object, nor a blob owned by a later manifest.
331+ if err := repo.Maintain(context.Background()); err != nil {
332+ t.Fatal(err)
333+ }
334+ if _, err := os.Lstat(stalePath); err != nil {
335+ t.Fatalf("stale object swept before complete mark pass: %v", err)
336+ }
337+ for _, object := range newest.Objects {
338+ if _, err := os.Lstat(repo.objectPath(key, object.Digest)); err != nil {
339+ t.Fatalf("object referenced by not-yet-classified manifest removed: %v", err)
340+ }
341+ }
342+
343+ for i := 0; i < 600; i++ {
344+ if err := repo.Maintain(context.Background()); err != nil {
345+ t.Fatal(err)
346+ }
347+ if _, err := os.Lstat(stalePath); errors.Is(err, os.ErrNotExist) {
348+ break
349+ }
350+ if i == 599 {
351+ t.Fatal("stale object was not eventually swept")
352+ }
353+ }
354+ for _, generation := range []uint64{maintenanceBatch + 1, maintenanceBatch + 2} {
355+ if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil {
356+ t.Fatalf("newest complete manifest %d removed: %v", generation, err)
357+ }
358+ }
359+}
360+
271361 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
272362 repo := NewRepository(privateDir(t))
273363 if err := os.Mkdir(repo.dir, 0o700); err != nil {
@@ -2,10 +2,15 @@ package snapshot
2 2
3 import (3 import (
4 "context"4 "context"
5+ "crypto/sha256"
5 "errors"6 "errors"
7+ "fmt"
6 "os"8 "os"
7 "path/filepath"9 "path/filepath"
10+ "strings"
8 "testing"11 "testing"
12+
13+ "github.com/bnema/vev/internal/ports"
9 )14 )
10 15
11 func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {16 func TestRepositoryDeleteQuarantinesCanonicalSessionBeforeCleanup(t *testing.T) {
@@ -268,6 +273,91 @@ func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
268 }273 }
269 }274 }
270 275
276+func TestRepositoryMaintainDoesNotStarveLaterTemporaryEntries(t *testing.T) {
277+ repo := NewRepository(privateDir(t))
278+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
279+ t.Fatal(err)
280+ }
281+ // The live entries precede the stale entries in directory insertion order.
282+ // A fresh ReadDir from the start on every call would never reach the latter.
283+ for i := 0; i < maintenanceBatch+1; i++ {
284+ if err := os.WriteFile(filepath.Join(repo.dir, "live-"+string(rune('a'+i))), []byte("live"), 0o600); err != nil {
285+ t.Fatal(err)
286+ }
287+ }
288+ for i := 0; i < maintenanceBatch+1; i++ {
289+ if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-later-"+string(rune('a'+i))), []byte("stale"), 0o600); err != nil {
290+ t.Fatal(err)
291+ }
292+ }
293+ for i := 0; i < 4; i++ {
294+ if err := repo.Maintain(context.Background()); err != nil {
295+ t.Fatal(err)
296+ }
297+ }
298+ entries, err := os.ReadDir(repo.dir)
299+ if err != nil {
300+ t.Fatal(err)
301+ }
302+ for _, entry := range entries {
303+ if strings.HasPrefix(entry.Name(), ".tmp-") {
304+ t.Fatalf("stale entry %q was starved", entry.Name())
305+ }
306+ }
307+}
308+
309+func TestRepositoryMaintainMarksLargeGenerationSetBeforeSweeping(t *testing.T) {
310+ repo := NewRepository(privateDir(t))
311+ var newest ports.SnapshotPublication
312+ for generation := uint64(1); generation <= maintenanceBatch+2; generation++ {
313+ publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))
314+ if err := repo.Publish(context.Background(), publication); err != nil {
315+ t.Fatal(err)
316+ }
317+ newest = publication
318+ }
319+ key := sessionKey("named")
320+ stale := sha256.Sum256([]byte("stale object"))
321+ stalePath := repo.objectPath(key, stale)
322+ if err := os.MkdirAll(filepath.Dir(stalePath), 0o700); err != nil {
323+ t.Fatal(err)
324+ }
325+ if err := os.WriteFile(stalePath, []byte("stale object"), 0o600); err != nil {
326+ t.Fatal(err)
327+ }
328+
329+ // The first call has classified only one manifest batch. It must not sweep
330+ // even an otherwise stale object, nor a blob owned by a later manifest.
331+ if err := repo.Maintain(context.Background()); err != nil {
332+ t.Fatal(err)
333+ }
334+ if _, err := os.Lstat(stalePath); err != nil {
335+ t.Fatalf("stale object swept before complete mark pass: %v", err)
336+ }
337+ for _, object := range newest.Objects {
338+ if _, err := os.Lstat(repo.objectPath(key, object.Digest)); err != nil {
339+ t.Fatalf("object referenced by not-yet-classified manifest removed: %v", err)
340+ }
341+ }
342+
343+ for i := 0; i < 600; i++ {
344+ if err := repo.Maintain(context.Background()); err != nil {
345+ t.Fatal(err)
346+ }
347+ if _, err := os.Lstat(stalePath); errors.Is(err, os.ErrNotExist) {
348+ break
349+ }
350+ if i == 599 {
351+ t.Fatal("stale object was not eventually swept")
352+ }
353+ }
354+ for _, generation := range []uint64{maintenanceBatch + 1, maintenanceBatch + 2} {
355+ if _, err := os.Lstat(repo.manifestPath(key, generation)); err != nil {
356+ t.Fatalf("newest complete manifest %d removed: %v", generation, err)
357+ }
358+ }
359+}
360+
271 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {361 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
272 repo := NewRepository(privateDir(t))362 repo := NewRepository(privateDir(t))
273 if err := os.Mkdir(repo.dir, 0o700); err != nil {363 if err := os.Mkdir(repo.dir, 0o700); err != nil {
M internal/adapters/snapshot/repository.go +7 -0
@@ -25,6 +25,13 @@ type Repository struct {
2525 dir string
2626 locks sync.Map // map[string]*sync.Mutex
2727 hooks repositoryHooks
28+
29+ // maintenanceMu owns continuation state for bounded maintenance. Keeping
30+ // directory handles on the repository makes successive calls advance rather
31+ // than repeatedly reading the first batch of a large directory.
32+ maintenanceMu sync.Mutex
33+ maintenanceCursors map[string]*maintenanceCursor
34+ maintenanceSessions map[string]*sessionMaintenance
2835 }
2936
3037 // repositoryHooks makes each persistence boundary fault-injectable. Hooks run
@@ -25,6 +25,13 @@ type Repository struct {
25 dir string25 dir string
26 locks sync.Map // map[string]*sync.Mutex26 locks sync.Map // map[string]*sync.Mutex
27 hooks repositoryHooks27 hooks repositoryHooks
28+
29+ // maintenanceMu owns continuation state for bounded maintenance. Keeping
30+ // directory handles on the repository makes successive calls advance rather
31+ // than repeatedly reading the first batch of a large directory.
32+ maintenanceMu sync.Mutex
33+ maintenanceCursors map[string]*maintenanceCursor
34+ maintenanceSessions map[string]*sessionMaintenance
28 }35 }
29 36
30 // repositoryHooks makes each persistence boundary fault-injectable. Hooks run37 // repositoryHooks makes each persistence boundary fault-injectable. Hooks run

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close