fix(snapshot): preserve maintenance continuation state

commit: abf52792d11e04faae8c09e397f5b6e63c6a10c1
author: brice · 3mo ago · 2026-07-21 21:06 UTC
parent: 9e04e97
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +267 -33

M internal/adapters/snapshot/legacy.go +21 -3


@@ -78,23 +78,41 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er
78	78		}
79	79
80	80		// DeleteLegacy deletes the deterministic v3 file and synchronizes the root.
81		-	// A sync failure is returned even though the unlink may have succeeded; retrying
82		-	// is safe because an already absent legacy file is considered deleted.
	81	+	// Once an unlink succeeds, its directory sync remains pending until it succeeds;
	82	+	// this lets an absent-file retry complete the durability boundary without
	83	+	// deleting a file recreated in the meantime.
83	84		func (r *Repository) DeleteLegacy(ctx context.Context, name string) error {
84	85		if err := ctx.Err(); err != nil {
85	86		return err
86	87		}
87		-	path := filepath.Join(r.dir, filenameForName(name))
	88	+	filename := filenameForName(name)
	89	+	lock := r.sessionLock("legacy-sync:" + filename)
	90	+	lock.Lock()
	91	+	defer lock.Unlock()
	92	+	if err := ctx.Err(); err != nil {
	93	+	return err
	94	+	}
	95	+	if _, pending := r.pendingLegacySync.Load(filename); pending {
	96	+	return r.syncPendingLegacy(ctx, filename)
	97	+	}
	98	+
	99	+	path := filepath.Join(r.dir, filename)
88	100		if err := r.remove(path); errors.Is(err, os.ErrNotExist) {
89	101		return nil
90	102		} else if err != nil {
91	103		return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err))
92	104		}
	105	+	r.pendingLegacySync.Store(filename, struct{}{})
	106	+	return r.syncPendingLegacy(ctx, filename)
	107	+	}
	108	+
	109	+	func (r *Repository) syncPendingLegacy(ctx context.Context, filename string) error {
93	110		if err := ctx.Err(); err != nil {
94	111		return err
95	112		}
96	113		if err := r.syncDirectory(r.dir); err != nil {
97	114		return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err))
98	115		}
	116	+	r.pendingLegacySync.Delete(filename)
99	117		return nil
100	118		}

 }
 // DeleteLegacy deletes the deterministic v3 file and synchronizes the root.
-// A sync failure is returned even though the unlink may have succeeded; retrying
+// Once an unlink succeeds, its directory sync remains pending until it succeeds;
-// is safe because an already absent legacy file is considered deleted.
+// this lets an absent-file retry complete the durability boundary without
+// deleting a file recreated in the meantime.
 func (r *Repository) DeleteLegacy(ctx context.Context, name string) error {
 	if err := ctx.Err(); err != nil {
 		return err
 	}
-	path := filepath.Join(r.dir, filenameForName(name))
+	filename := filenameForName(name)
+	lock := r.sessionLock("legacy-sync:" + filename)
+	lock.Lock()
+	defer lock.Unlock()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if _, pending := r.pendingLegacySync.Load(filename); pending {
+		return r.syncPendingLegacy(ctx, filename)
+	}
+	path := filepath.Join(r.dir, filename)
 	if err := r.remove(path); errors.Is(err, os.ErrNotExist) {
 		return nil
 	} else if err != nil {
 		return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err))
 	}
+	r.pendingLegacySync.Store(filename, struct{}{})
+	return r.syncPendingLegacy(ctx, filename)
+}
+func (r *Repository) syncPendingLegacy(ctx context.Context, filename string) error {
 	if err := ctx.Err(); err != nil {
 		return err
 	}
 	if err := r.syncDirectory(r.dir); err != nil {
 		return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err))
 	}
+	r.pendingLegacySync.Delete(filename)
 	return nil
 }

M internal/adapters/snapshot/legacy_test.go +11 -2


@@ -81,12 +81,21 @@ func TestRepositoryDeleteLegacySurfacesRootSyncAndCanRetry(t *testing.T) {
81	81		t.Fatal(err)
82	82		}
83	83		injected := errors.New("root sync")
84		-	repo.hooks.syncDirectory = func(string) error { return injected }
	84	+	syncCalls := 0
	85	+	repo.hooks.syncDirectory = func(string) error {
	86	+	syncCalls++
	87	+	if syncCalls == 1 {
	88	+	return injected
	89	+	}
	90	+	return nil
	91	+	}
85	92		if err := repo.DeleteLegacy(context.Background(), "named"); !errors.Is(err, injected) {
86	93		t.Fatalf("DeleteLegacy error = %v, want root sync error", err)
87	94		}
88		-	repo.hooks.syncDirectory = nil
89	95		if err := repo.DeleteLegacy(context.Background(), "named"); err != nil {
90	96		t.Fatalf("retry DeleteLegacy: %v", err)
91	97		}
	98	+	if syncCalls != 2 {
	99	+	t.Fatalf("sync calls = %d, want retry to sync absent file", syncCalls)
	100	+	}
92	101		}

 		t.Fatal(err)
 	}
 	injected := errors.New("root sync")
-	repo.hooks.syncDirectory = func(string) error { return injected }
+	syncCalls := 0
+	repo.hooks.syncDirectory = func(string) error {
+		syncCalls++
+		if syncCalls == 1 {
+			return injected
+		}
+		return nil
+	}
 	if err := repo.DeleteLegacy(context.Background(), "named"); !errors.Is(err, injected) {
 		t.Fatalf("DeleteLegacy error = %v, want root sync error", err)
 	}
-	repo.hooks.syncDirectory = nil
 	if err := repo.DeleteLegacy(context.Background(), "named"); err != nil {
 		t.Fatalf("retry DeleteLegacy: %v", err)
 	}
+	if syncCalls != 2 {
+		t.Fatalf("sync calls = %d, want retry to sync absent file", syncCalls)
+	}
 }

M internal/adapters/snapshot/maintenance.go +79 -28


@@ -2,6 +2,7 @@ package snapshot
2	2
3	3		import (
4	4		"context"
	5	+	"crypto/sha256"
5	6		"errors"
6	7		"fmt"
7	8		"io"

@@ -140,8 +141,9 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
140	141		if err != nil {
141	142		return fmt.Errorf("read snapshot maintenance sessions: %w", safeFilesystemError(err))
142	143		}
143		-	for _, entry := range entries {
	144	+	for i, entry := range entries {
144	145		if budget == 0 {
	146	+	r.requeueMaintenanceEntries(sessions, "sessions", entries[i:])
145	147		return nil
146	148		}
147	149		if err := ctx.Err(); err != nil {

@@ -177,7 +179,10 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
177	179		return nil
178	180		}
179	181
180		-	type maintenanceCursor struct{ file *os.File }
	182	+	type maintenanceCursor struct {
	183	+	file *os.File
	184	+	pending []os.DirEntry
	185	+	}
181	186
182	187		type manifestMaintenance struct {
183	188		refs map[ports.SnapshotDigest]codec.ObjectRef

@@ -188,6 +193,7 @@ type sessionMaintenance struct {
188	193		token string
189	194		marked map[uint64]manifestMaintenance
190	195		uncertain bool
	196	+	conservative bool
191	197		markDone bool
192	198		manifestQueue []uint64
193	199		sweepQueue []string

@@ -196,17 +202,32 @@ type sessionMaintenance struct {
196	202
197	203		func (r *Repository) resetMaintenance() {
198	204		for _, cursor := range r.maintenanceCursors {
199		-	_ = cursor.file.Close()
	205	+	if cursor.file != nil {
	206	+	_ = cursor.file.Close()
	207	+	}
200	208		}
201	209		r.maintenanceCursors = nil
202	210		r.maintenanceSessions = nil
203	211		}
204	212
205		-	// readMaintenanceDir advances one named directory cursor. It closes and
206		-	// forgets the handle at EOF, so the following call begins a fresh cycle.
	213	+	func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
	214	+
	215	+	// readMaintenanceDir advances one named directory cursor. Callers must return
	216	+	// entries they could not process with requeueMaintenanceEntries before yielding
	217	+	// for budget exhaustion. This prevents a ReadDir batch from skipping work when
	218	+	// an earlier operation spends the remainder of the pass budget.
207	219		func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {
208		-	id := purpose + "\x00" + dir
	220	+	id := maintenanceCursorID(dir, purpose)
209	221		cursor := r.maintenanceCursors[id]
	222	+	if cursor != nil && len(cursor.pending) != 0 {
	223	+	entries := cursor.pending
	224	+	cursor.pending = nil
	225	+	return entries, cursor.file == nil, nil
	226	+	}
	227	+	if cursor != nil && cursor.file == nil {
	228	+	delete(r.maintenanceCursors, id)
	229	+	cursor = nil
	230	+	}
210	231		if cursor == nil {
211	232		file, err := os.Open(dir)
212	233		if err != nil {

@@ -234,6 +255,19 @@ func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os
234	255		return entries, true, nil
235	256		}
236	257
	258	+	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []os.DirEntry) {
	259	+	if len(entries) == 0 {
	260	+	return
	261	+	}
	262	+	id := maintenanceCursorID(dir, purpose)
	263	+	cursor := r.maintenanceCursors[id]
	264	+	if cursor == nil {
	265	+	cursor = &maintenanceCursor{}
	266	+	r.maintenanceCursors[id] = cursor
	267	+	}
	268	+	cursor.pending = append(entries, cursor.pending...)
	269	+	}
	270	+
237	271		func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
238	272
239	273		// readDirBatch is used only for a mutating quarantine walk, where every

@@ -324,8 +358,9 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
324	358		if err != nil && !errors.Is(err, os.ErrNotExist) {
325	359		return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
326	360		}
327		-	for _, shard := range shards {
	361	+	for i, shard := range shards {
328	362		if *budget == 0 {
	363	+	r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
329	364		return nil
330	365		}
331	366		if shard.IsDir() {

@@ -361,38 +396,39 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
361	396		}
362	397
363	398		func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, error) {
364		-	token, exists, err := r.maintenanceToken(key)
	399	+	token, conservative, err := r.maintenanceToken(key)
365	400		if err != nil {
366	401		return nil, err
367	402		}
368		-	if !exists {
369		-	r.clearSessionMaintenance(key)
370		-	return &sessionMaintenance{token: "missing", marked: make(map[uint64]manifestMaintenance)}, nil
371		-	}
372	403		state := r.maintenanceSessions[key]
373	404		if state == nil \|\| state.token != token {
374	405		r.clearSessionMaintenance(key)
375		-	state = &sessionMaintenance{token: token, marked: make(map[uint64]manifestMaintenance)}
	406	+	state = &sessionMaintenance{
	407	+	token: token,
	408	+	conservative: conservative,
	409	+	marked: make(map[uint64]manifestMaintenance),
	410	+	}
376	411		r.maintenanceSessions[key] = state
377	412		}
378	413		return state, nil
379	414		}
380	415
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.
	416	+	// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
	417	+	// itself stable maintenance state: we retain every classified reference until a
	418	+	// valid publication changes that state, rather than restarting the mark pass.
384	419		func (r *Repository) maintenanceToken(key string) (string, bool, error) {
385	420		data, exists, err := readOptionalBounded(r.headPath(key))
386		-	if errors.Is(err, os.ErrNotExist) {
387		-	return "", false, nil
388		-	}
389	421		if err != nil {
390	422		return "", false, err
391	423		}
392	424		if !exists {
393		-	return "", false, nil
	425	+	return "missing", true, nil
394	426		}
395		-	return string(data), true, nil
	427	+	sum := sha256.Sum256(data)
	428	+	if _, _, err := r.readHead(key); err != nil {
	429	+	return fmt.Sprintf("invalid:%x", sum), true, nil
	430	+	}
	431	+	return fmt.Sprintf("valid:%x", sum), false, nil
396	432		}
397	433
398	434		func (r *Repository) clearSessionMaintenance(key string) {

@@ -400,7 +436,9 @@ func (r *Repository) clearSessionMaintenance(key string) {
400	436		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
401	437		for id, cursor := range r.maintenanceCursors {
402	438		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {
403		-	_ = cursor.file.Close()
	439	+	if cursor.file != nil {
	440	+	_ = cursor.file.Close()
	441	+	}
404	442		delete(r.maintenanceCursors, id)
405	443		}
406	444		}

@@ -454,7 +492,9 @@ func (r Repository) markSession(ctx context.Context, key string, state session
454	492		}
455	493		if done {
456	494		state.markDone = true
457		-	state.manifestQueue = retainedManifestQueue(state.marked)
	495	+	if !state.conservative {
	496	+	state.manifestQueue = retainedManifestQueue(state.marked)
	497	+	}
458	498		}
459	499		return nil
460	500		}

@@ -502,7 +542,7 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st
502	542		}
503	543
504	544		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {
505		-	referenced := retainedReferences(state.marked)
	545	+	referenced := retainedReferences(state.marked, state.conservative)
506	546		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
507	547		if !state.sweepRootDone {
508	548		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)

@@ -535,7 +575,16 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
535	575		return nil
536	576		}
537	577
538		-	func retainedReferences(marked map[uint64]manifestMaintenance) map[ports.SnapshotDigest]struct{} {
	578	+	func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
	579	+	if conservative {
	580	+	references := make(map[ports.SnapshotDigest]struct{})
	581	+	for _, item := range marked {
	582	+	for digest := range item.refs {
	583	+	references[digest] = struct{}{}
	584	+	}
	585	+	}
	586	+	return references
	587	+	}
539	588		complete := make([]uint64, 0, len(marked))
540	589		for generation, item := range marked {
541	590		if item.complete {

@@ -575,9 +624,10 @@ func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[
575	624		if err != nil {
576	625		return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
577	626		}
578		-	for _, object := range entries {
	627	+	for i, object := range entries {
579	628		if *budget == 0 {
580		-	break
	629	+	r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
	630	+	return false, nil
581	631		}
582	632		if err := ctx.Err(); err != nil {
583	633		return false, err

@@ -615,8 +665,9 @@ func (r Repository) removeTemps(ctx context.Context, dir string, budget int, p
615	665		if err != nil {
616	666		return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
617	667		}
618		-	for _, entry := range entries {
	668	+	for i, entry := range entries {
619	669		if *budget == 0 {
	670	+	r.requeueMaintenanceEntries(dir, purpose, entries[i:])
620	671		return nil
621	672		}
622	673		if err := ctx.Err(); err != nil {


@@ -2,6 +2,7 @@ package snapshot
2			2
3		import (	3		import (
4		"context"	4		"context"
			5	+	"crypto/sha256"
5		"errors"	6		"errors"
6		"fmt"	7		"fmt"
7		"io"	8		"io"

@@ -140,8 +141,9 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
140		if err != nil {	141		if err != nil {
141		return fmt.Errorf("read snapshot maintenance sessions: %w", safeFilesystemError(err))	142		return fmt.Errorf("read snapshot maintenance sessions: %w", safeFilesystemError(err))
142		}	143		}
143	-	for _, entry := range entries {	144	+	for i, entry := range entries {
144		if budget == 0 {	145		if budget == 0 {
			146	+	r.requeueMaintenanceEntries(sessions, "sessions", entries[i:])
145		return nil	147		return nil
146		}	148		}
147		if err := ctx.Err(); err != nil {	149		if err := ctx.Err(); err != nil {

@@ -177,7 +179,10 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
177		return nil	179		return nil
178		}	180		}
179			181
180	-	type maintenanceCursor struct{ file *os.File }	182	+	type maintenanceCursor struct {
			183	+	file *os.File
			184	+	pending []os.DirEntry
			185	+	}
181			186
182		type manifestMaintenance struct {	187		type manifestMaintenance struct {
183		refs map[ports.SnapshotDigest]codec.ObjectRef	188		refs map[ports.SnapshotDigest]codec.ObjectRef

@@ -188,6 +193,7 @@ type sessionMaintenance struct {
188		token string	193		token string
189		marked map[uint64]manifestMaintenance	194		marked map[uint64]manifestMaintenance
190		uncertain bool	195		uncertain bool
			196	+	conservative bool
191		markDone bool	197		markDone bool
192		manifestQueue []uint64	198		manifestQueue []uint64
193		sweepQueue []string	199		sweepQueue []string

@@ -196,17 +202,32 @@ type sessionMaintenance struct {
196			202
197		func (r *Repository) resetMaintenance() {	203		func (r *Repository) resetMaintenance() {
198		for _, cursor := range r.maintenanceCursors {	204		for _, cursor := range r.maintenanceCursors {
199	-	_ = cursor.file.Close()	205	+	if cursor.file != nil {
			206	+	_ = cursor.file.Close()
			207	+	}
200		}	208		}
201		r.maintenanceCursors = nil	209		r.maintenanceCursors = nil
202		r.maintenanceSessions = nil	210		r.maintenanceSessions = nil
203		}	211		}
204			212
205	-	// readMaintenanceDir advances one named directory cursor. It closes and	213	+	func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
206	-	// forgets the handle at EOF, so the following call begins a fresh cycle.	214	+
			215	+	// readMaintenanceDir advances one named directory cursor. Callers must return
			216	+	// entries they could not process with requeueMaintenanceEntries before yielding
			217	+	// for budget exhaustion. This prevents a ReadDir batch from skipping work when
			218	+	// an earlier operation spends the remainder of the pass budget.
207		func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {	219		func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {
208	-	id := purpose + "\x00" + dir	220	+	id := maintenanceCursorID(dir, purpose)
209		cursor := r.maintenanceCursors[id]	221		cursor := r.maintenanceCursors[id]
			222	+	if cursor != nil && len(cursor.pending) != 0 {
			223	+	entries := cursor.pending
			224	+	cursor.pending = nil
			225	+	return entries, cursor.file == nil, nil
			226	+	}
			227	+	if cursor != nil && cursor.file == nil {
			228	+	delete(r.maintenanceCursors, id)
			229	+	cursor = nil
			230	+	}
210		if cursor == nil {	231		if cursor == nil {
211		file, err := os.Open(dir)	232		file, err := os.Open(dir)
212		if err != nil {	233		if err != nil {

@@ -234,6 +255,19 @@ func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os
234		return entries, true, nil	255		return entries, true, nil
235		}	256		}
236			257
			258	+	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []os.DirEntry) {
			259	+	if len(entries) == 0 {
			260	+	return
			261	+	}
			262	+	id := maintenanceCursorID(dir, purpose)
			263	+	cursor := r.maintenanceCursors[id]
			264	+	if cursor == nil {
			265	+	cursor = &maintenanceCursor{}
			266	+	r.maintenanceCursors[id] = cursor
			267	+	}
			268	+	cursor.pending = append(entries, cursor.pending...)
			269	+	}
			270	+
237		func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }	271		func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
238			272
239		// readDirBatch is used only for a mutating quarantine walk, where every	273		// readDirBatch is used only for a mutating quarantine walk, where every

@@ -324,8 +358,9 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
324		if err != nil && !errors.Is(err, os.ErrNotExist) {	358		if err != nil && !errors.Is(err, os.ErrNotExist) {
325		return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))	359		return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
326		}	360		}
327	-	for _, shard := range shards {	361	+	for i, shard := range shards {
328		if *budget == 0 {	362		if *budget == 0 {
			363	+	r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
329		return nil	364		return nil
330		}	365		}
331		if shard.IsDir() {	366		if shard.IsDir() {

@@ -361,38 +396,39 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
361		}	396		}
362			397
363		func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, error) {	398		func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, error) {
364	-	token, exists, err := r.maintenanceToken(key)	399	+	token, conservative, err := r.maintenanceToken(key)
365		if err != nil {	400		if err != nil {
366		return nil, err	401		return nil, err
367		}	402		}
368	-	if !exists {
369	-	r.clearSessionMaintenance(key)
370	-	return &sessionMaintenance{token: "missing", marked: make(map[uint64]manifestMaintenance)}, nil
371	-	}
372		state := r.maintenanceSessions[key]	403		state := r.maintenanceSessions[key]
373		if state == nil \|\| state.token != token {	404		if state == nil \|\| state.token != token {
374		r.clearSessionMaintenance(key)	405		r.clearSessionMaintenance(key)
375	-	state = &sessionMaintenance{token: token, marked: make(map[uint64]manifestMaintenance)}	406	+	state = &sessionMaintenance{
			407	+	token: token,
			408	+	conservative: conservative,
			409	+	marked: make(map[uint64]manifestMaintenance),
			410	+	}
376		r.maintenanceSessions[key] = state	411		r.maintenanceSessions[key] = state
377		}	412		}
378		return state, nil	413		return state, nil
379		}	414		}
380			415
381	-	// maintenanceToken is the publication boundary. Publish and Delete share the	416	+	// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
382	-	// per-session lock with maintenance, and a changed HEAD on a later call makes	417	+	// itself stable maintenance state: we retain every classified reference until a
383	-	// us discard partial marking/sweeping before deleting anything else.	418	+	// valid publication changes that state, rather than restarting the mark pass.
384		func (r *Repository) maintenanceToken(key string) (string, bool, error) {	419		func (r *Repository) maintenanceToken(key string) (string, bool, error) {
385		data, exists, err := readOptionalBounded(r.headPath(key))	420		data, exists, err := readOptionalBounded(r.headPath(key))
386	-	if errors.Is(err, os.ErrNotExist) {
387	-	return "", false, nil
388	-	}
389		if err != nil {	421		if err != nil {
390		return "", false, err	422		return "", false, err
391		}	423		}
392		if !exists {	424		if !exists {
393	-	return "", false, nil	425	+	return "missing", true, nil
394		}	426		}
395	-	return string(data), true, nil	427	+	sum := sha256.Sum256(data)
			428	+	if _, _, err := r.readHead(key); err != nil {
			429	+	return fmt.Sprintf("invalid:%x", sum), true, nil
			430	+	}
			431	+	return fmt.Sprintf("valid:%x", sum), false, nil
396		}	432		}
397			433
398		func (r *Repository) clearSessionMaintenance(key string) {	434		func (r *Repository) clearSessionMaintenance(key string) {

@@ -400,7 +436,9 @@ func (r *Repository) clearSessionMaintenance(key string) {
400		prefix := "\x00" + filepath.Clean(r.sessionPath(key))	436		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
401		for id, cursor := range r.maintenanceCursors {	437		for id, cursor := range r.maintenanceCursors {
402		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {	438		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {
403	-	_ = cursor.file.Close()	439	+	if cursor.file != nil {
			440	+	_ = cursor.file.Close()
			441	+	}
404		delete(r.maintenanceCursors, id)	442		delete(r.maintenanceCursors, id)
405		}	443		}
406		}	444		}

@@ -454,7 +492,9 @@ func (r Repository) markSession(ctx context.Context, key string, state session
454		}	492		}
455		if done {	493		if done {
456		state.markDone = true	494		state.markDone = true
457	-	state.manifestQueue = retainedManifestQueue(state.marked)	495	+	if !state.conservative {
			496	+	state.manifestQueue = retainedManifestQueue(state.marked)
			497	+	}
458		}	498		}
459		return nil	499		return nil
460		}	500		}

@@ -502,7 +542,7 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st
502		}	542		}
503			543
504		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {	544		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {
505	-	referenced := retainedReferences(state.marked)	545	+	referenced := retainedReferences(state.marked, state.conservative)
506		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)	546		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
507		if !state.sweepRootDone {	547		if !state.sweepRootDone {
508		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)	548		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)

@@ -535,7 +575,16 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
535		return nil	575		return nil
536		}	576		}
537			577
538	-	func retainedReferences(marked map[uint64]manifestMaintenance) map[ports.SnapshotDigest]struct{} {	578	+	func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
			579	+	if conservative {
			580	+	references := make(map[ports.SnapshotDigest]struct{})
			581	+	for _, item := range marked {
			582	+	for digest := range item.refs {
			583	+	references[digest] = struct{}{}
			584	+	}
			585	+	}
			586	+	return references
			587	+	}
539		complete := make([]uint64, 0, len(marked))	588		complete := make([]uint64, 0, len(marked))
540		for generation, item := range marked {	589		for generation, item := range marked {
541		if item.complete {	590		if item.complete {

@@ -575,9 +624,10 @@ func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[
575		if err != nil {	624		if err != nil {
576		return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))	625		return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
577		}	626		}
578	-	for _, object := range entries {	627	+	for i, object := range entries {
579		if *budget == 0 {	628		if *budget == 0 {
580	-	break	629	+	r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
			630	+	return false, nil
581		}	631		}
582		if err := ctx.Err(); err != nil {	632		if err := ctx.Err(); err != nil {
583		return false, err	633		return false, err

@@ -615,8 +665,9 @@ func (r Repository) removeTemps(ctx context.Context, dir string, budget int, p
615		if err != nil {	665		if err != nil {
616		return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))	666		return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
617		}	667		}
618	-	for _, entry := range entries {	668	+	for i, entry := range entries {
619		if *budget == 0 {	669		if *budget == 0 {
			670	+	r.requeueMaintenanceEntries(dir, purpose, entries[i:])
620		return nil	671		return nil
621		}	672		}
622		if err := ctx.Err(); err != nil {	673		if err := ctx.Err(); err != nil {

M internal/adapters/snapshot/maintenance_test.go +151 -0


@@ -379,3 +379,154 @@ func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
379	379		t.Fatalf("remaining stale entries = %d, want 1 after one batch", len(entries))
380	380		}
381	381		}
	382	+
	383	+	func TestRepositoryMaintainQueuesFetchedSessionsPastWorkBudget(t *testing.T) {
	384	+	repo := NewRepository(privateDir(t))
	385	+	if err := os.MkdirAll(filepath.Join(repo.dir, repositorySessionsDir), 0o700); err != nil {
	386	+	t.Fatal(err)
	387	+	}
	388	+	if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-consume-budget"), []byte("stale"), 0o600); err != nil {
	389	+	t.Fatal(err)
	390	+	}
	391	+
	392	+	removed := make(map[string]int)
	393	+	for i := 0; i < maintenanceBatch+1; i++ {
	394	+	path := filepath.Join(repo.dir, repositorySessionsDir, fmt.Sprintf(".deleting-%03d", i))
	395	+	if err := os.Mkdir(path, 0o700); err != nil {
	396	+	t.Fatal(err)
	397	+	}
	398	+	}
	399	+	repo.hooks.remove = func(path string) error {
	400	+	if strings.HasPrefix(filepath.Base(path), ".deleting-") {
	401	+	removed[path]++
	402	+	}
	403	+	return nil
	404	+	}
	405	+
	406	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
	407	+	f, err := os.Open(sessions)
	408	+	if err != nil {
	409	+	t.Fatal(err)
	410	+	}
	411	+	fetched, err := f.ReadDir(maintenanceBatch)
	412	+	if err != nil {
	413	+	t.Fatal(err)
	414	+	}
	415	+	unread, err := f.ReadDir(1)
	416	+	if err != nil {
	417	+	t.Fatal(err)
	418	+	}
	419	+	if err := f.Close(); err != nil {
	420	+	t.Fatal(err)
	421	+	}
	422	+	queued := fetched[len(fetched)-1].Name()
	423	+	if err := repo.Maintain(context.Background()); err != nil {
	424	+	t.Fatal(err)
	425	+	}
	426	+	if err := repo.Maintain(context.Background()); err != nil {
	427	+	t.Fatal(err)
	428	+	}
	429	+	if _, err := os.Lstat(filepath.Join(sessions, queued)); !errors.Is(err, os.ErrNotExist) {
	430	+	t.Fatalf("fetched but unprocessed session %q = %v, want removed on next call", queued, err)
	431	+	}
	432	+	if _, err := os.Lstat(filepath.Join(sessions, unread[0].Name())); err != nil {
	433	+	t.Fatalf("unread session %q = %v, want queued entry to run first", unread[0].Name(), err)
	434	+	}
	435	+
	436	+	for i := 0; i < 2; i++ {
	437	+	if err := repo.Maintain(context.Background()); err != nil {
	438	+	t.Fatal(err)
	439	+	}
	440	+	}
	441	+	entries, err := os.ReadDir(sessions)
	442	+	if err != nil {
	443	+	t.Fatal(err)
	444	+	}
	445	+	if len(entries) != 0 {
	446	+	t.Fatalf("queued sessions after resumed maintenance = %d, want none", len(entries))
	447	+	}
	448	+	if len(removed) != maintenanceBatch+1 {
	449	+	t.Fatalf("removed sessions = %d, want %d", len(removed), maintenanceBatch+1)
	450	+	}
	451	+	for path, calls := range removed {
	452	+	if calls != 1 {
	453	+	t.Fatalf("remove calls for %q = %d, want one", path, calls)
	454	+	}
	455	+	}
	456	+	}
	457	+
	458	+	func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testing.T) {
	459	+	for _, head := range []struct {
	460	+	name string
	461	+	set func(t testing.T, repo Repository, key string)
	462	+	}{
	463	+	{
	464	+	name: "missing",
	465	+	set: func(t testing.T, repo Repository, key string) {
	466	+	t.Helper()
	467	+	if err := os.Remove(repo.headPath(key)); err != nil {
	468	+	t.Fatal(err)
	469	+	}
	470	+	},
	471	+	},
	472	+	{
	473	+	name: "corrupt",
	474	+	set: func(t testing.T, repo Repository, key string) {
	475	+	t.Helper()
	476	+	if err := os.WriteFile(repo.headPath(key), []byte("corrupt"), 0o600); err != nil {
	477	+	t.Fatal(err)
	478	+	}
	479	+	},
	480	+	},
	481	+	} {
	482	+	t.Run(head.name, func(t *testing.T) {
	483	+	repo := NewRepository(privateDir(t))
	484	+	key := sessionKey("named")
	485	+	publications := make([]ports.SnapshotPublication, 0, maintenanceBatch+2)
	486	+	for generation := uint64(1); generation <= maintenanceBatch+2; generation++ {
	487	+	publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))
	488	+	if err := repo.Publish(context.Background(), publication); err != nil {
	489	+	t.Fatal(err)
	490	+	}
	491	+	publications = append(publications, publication)
	492	+	}
	493	+	head.set(t, repo, key)
	494	+	stale := sha256.Sum256([]byte("unpublished stale object"))
	495	+	stalePath := repo.objectPath(key, stale)
	496	+	if err := os.MkdirAll(filepath.Dir(stalePath), 0o700); err != nil {
	497	+	t.Fatal(err)
	498	+	}
	499	+	if err := os.WriteFile(stalePath, []byte("unpublished stale object"), 0o600); err != nil {
	500	+	t.Fatal(err)
	501	+	}
	502	+
	503	+	if err := repo.Maintain(context.Background()); err != nil {
	504	+	t.Fatal(err)
	505	+	}
	506	+	if _, err := os.Lstat(stalePath); err != nil {
	507	+	t.Fatalf("stale object swept before full manifest classification: %v", err)
	508	+	}
	509	+	for _, publication := range publications {
	510	+	if _, err := os.Lstat(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
	511	+	t.Fatalf("object for unpublished generation %d removed early: %v", publication.Generation, err)
	512	+	}
	513	+	}
	514	+
	515	+	for i := 0; i < 600; i++ {
	516	+	if err := repo.Maintain(context.Background()); err != nil {
	517	+	t.Fatal(err)
	518	+	}
	519	+	if _, err := os.Lstat(stalePath); errors.Is(err, os.ErrNotExist) {
	520	+	break
	521	+	} else if i == 599 {
	522	+	t.Fatal("stale object was not swept after complete classification")
	523	+	}
	524	+	}
	525	+	for _, publication := range publications {
	526	+	if _, err := os.Lstat(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
	527	+	t.Fatalf("object for unpublished generation %d removed: %v", publication.Generation, err)
	528	+	}
	529	+	}
	530	+	})
	531	+	}
	532	+	}


@@ -379,3 +379,154 @@ func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
379	t.Fatalf("remaining stale entries = %d, want 1 after one batch", len(entries))	379		t.Fatalf("remaining stale entries = %d, want 1 after one batch", len(entries))
380	}	380		}
381	}	381		}
		382	+
		383	+	func TestRepositoryMaintainQueuesFetchedSessionsPastWorkBudget(t *testing.T) {
		384	+	repo := NewRepository(privateDir(t))
		385	+	if err := os.MkdirAll(filepath.Join(repo.dir, repositorySessionsDir), 0o700); err != nil {
		386	+	t.Fatal(err)
		387	+	}
		388	+	if err := os.WriteFile(filepath.Join(repo.dir, ".tmp-consume-budget"), []byte("stale"), 0o600); err != nil {
		389	+	t.Fatal(err)
		390	+	}
		391	+
		392	+	removed := make(map[string]int)
		393	+	for i := 0; i < maintenanceBatch+1; i++ {
		394	+	path := filepath.Join(repo.dir, repositorySessionsDir, fmt.Sprintf(".deleting-%03d", i))
		395	+	if err := os.Mkdir(path, 0o700); err != nil {
		396	+	t.Fatal(err)
		397	+	}
		398	+	}
		399	+	repo.hooks.remove = func(path string) error {
		400	+	if strings.HasPrefix(filepath.Base(path), ".deleting-") {
		401	+	removed[path]++
		402	+	}
		403	+	return nil
		404	+	}
		405	+
		406	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
		407	+	f, err := os.Open(sessions)
		408	+	if err != nil {
		409	+	t.Fatal(err)
		410	+	}
		411	+	fetched, err := f.ReadDir(maintenanceBatch)
		412	+	if err != nil {
		413	+	t.Fatal(err)
		414	+	}
		415	+	unread, err := f.ReadDir(1)
		416	+	if err != nil {
		417	+	t.Fatal(err)
		418	+	}
		419	+	if err := f.Close(); err != nil {
		420	+	t.Fatal(err)
		421	+	}
		422	+	queued := fetched[len(fetched)-1].Name()
		423	+	if err := repo.Maintain(context.Background()); err != nil {
		424	+	t.Fatal(err)
		425	+	}
		426	+	if err := repo.Maintain(context.Background()); err != nil {
		427	+	t.Fatal(err)
		428	+	}
		429	+	if _, err := os.Lstat(filepath.Join(sessions, queued)); !errors.Is(err, os.ErrNotExist) {
		430	+	t.Fatalf("fetched but unprocessed session %q = %v, want removed on next call", queued, err)
		431	+	}
		432	+	if _, err := os.Lstat(filepath.Join(sessions, unread[0].Name())); err != nil {
		433	+	t.Fatalf("unread session %q = %v, want queued entry to run first", unread[0].Name(), err)
		434	+	}
		435	+
		436	+	for i := 0; i < 2; i++ {
		437	+	if err := repo.Maintain(context.Background()); err != nil {
		438	+	t.Fatal(err)
		439	+	}
		440	+	}
		441	+	entries, err := os.ReadDir(sessions)
		442	+	if err != nil {
		443	+	t.Fatal(err)
		444	+	}
		445	+	if len(entries) != 0 {
		446	+	t.Fatalf("queued sessions after resumed maintenance = %d, want none", len(entries))
		447	+	}
		448	+	if len(removed) != maintenanceBatch+1 {
		449	+	t.Fatalf("removed sessions = %d, want %d", len(removed), maintenanceBatch+1)
		450	+	}
		451	+	for path, calls := range removed {
		452	+	if calls != 1 {
		453	+	t.Fatalf("remove calls for %q = %d, want one", path, calls)
		454	+	}
		455	+	}
		456	+	}
		457	+
		458	+	func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testing.T) {
		459	+	for _, head := range []struct {
		460	+	name string
		461	+	set func(t testing.T, repo Repository, key string)
		462	+	}{
		463	+	{
		464	+	name: "missing",
		465	+	set: func(t testing.T, repo Repository, key string) {
		466	+	t.Helper()
		467	+	if err := os.Remove(repo.headPath(key)); err != nil {
		468	+	t.Fatal(err)
		469	+	}
		470	+	},
		471	+	},
		472	+	{
		473	+	name: "corrupt",
		474	+	set: func(t testing.T, repo Repository, key string) {
		475	+	t.Helper()
		476	+	if err := os.WriteFile(repo.headPath(key), []byte("corrupt"), 0o600); err != nil {
		477	+	t.Fatal(err)
		478	+	}
		479	+	},
		480	+	},
		481	+	} {
		482	+	t.Run(head.name, func(t *testing.T) {
		483	+	repo := NewRepository(privateDir(t))
		484	+	key := sessionKey("named")
		485	+	publications := make([]ports.SnapshotPublication, 0, maintenanceBatch+2)
		486	+	for generation := uint64(1); generation <= maintenanceBatch+2; generation++ {
		487	+	publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))
		488	+	if err := repo.Publish(context.Background(), publication); err != nil {
		489	+	t.Fatal(err)
		490	+	}
		491	+	publications = append(publications, publication)
		492	+	}
		493	+	head.set(t, repo, key)
		494	+	stale := sha256.Sum256([]byte("unpublished stale object"))
		495	+	stalePath := repo.objectPath(key, stale)
		496	+	if err := os.MkdirAll(filepath.Dir(stalePath), 0o700); err != nil {
		497	+	t.Fatal(err)
		498	+	}
		499	+	if err := os.WriteFile(stalePath, []byte("unpublished stale object"), 0o600); err != nil {
		500	+	t.Fatal(err)
		501	+	}
		502	+
		503	+	if err := repo.Maintain(context.Background()); err != nil {
		504	+	t.Fatal(err)
		505	+	}
		506	+	if _, err := os.Lstat(stalePath); err != nil {
		507	+	t.Fatalf("stale object swept before full manifest classification: %v", err)
		508	+	}
		509	+	for _, publication := range publications {
		510	+	if _, err := os.Lstat(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
		511	+	t.Fatalf("object for unpublished generation %d removed early: %v", publication.Generation, err)
		512	+	}
		513	+	}
		514	+
		515	+	for i := 0; i < 600; i++ {
		516	+	if err := repo.Maintain(context.Background()); err != nil {
		517	+	t.Fatal(err)
		518	+	}
		519	+	if _, err := os.Lstat(stalePath); errors.Is(err, os.ErrNotExist) {
		520	+	break
		521	+	} else if i == 599 {
		522	+	t.Fatal("stale object was not swept after complete classification")
		523	+	}
		524	+	}
		525	+	for _, publication := range publications {
		526	+	if _, err := os.Lstat(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
		527	+	t.Fatalf("object for unpublished generation %d removed: %v", publication.Generation, err)
		528	+	}
		529	+	}
		530	+	})
		531	+	}
		532	+	}

M internal/adapters/snapshot/repository.go +5 -0


@@ -32,6 +32,11 @@ type Repository struct {
32	32		maintenanceMu sync.Mutex
33	33		maintenanceCursors map[string]*maintenanceCursor
34	34		maintenanceSessions map[string]*sessionMaintenance
	35	+
	36	+	// pendingLegacySync records an unlink whose root-directory sync failed.
	37	+	// It is keyed by the deterministic legacy filename and shares a per-file
	38	+	// lock with DeleteLegacy so a retry cannot acknowledge a recreated file.
	39	+	pendingLegacySync sync.Map // map[string]struct{}
35	40		}
36	41
37	42		// repositoryHooks makes each persistence boundary fault-injectable. Hooks run


@@ -32,6 +32,11 @@ type Repository struct {
32	maintenanceMu sync.Mutex	32		maintenanceMu sync.Mutex
33	maintenanceCursors map[string]*maintenanceCursor	33		maintenanceCursors map[string]*maintenanceCursor
34	maintenanceSessions map[string]*sessionMaintenance	34		maintenanceSessions map[string]*sessionMaintenance
		35	+
		36	+	// pendingLegacySync records an unlink whose root-directory sync failed.
		37	+	// It is keyed by the deterministic legacy filename and shares a per-file
		38	+	// lock with DeleteLegacy so a retry cannot acknowledge a recreated file.
		39	+	pendingLegacySync sync.Map // map[string]struct{}
35	}	40		}
36		41
37	// repositoryHooks makes each persistence boundary fault-injectable. Hooks run	42		// repositoryHooks makes each persistence boundary fault-injectable. Hooks run