26baad6 fix(snapshot): bound maintenance cursor state · vev

M internal/adapters/snapshot/maintenance.go +25 -3


@@ -39,9 +39,24 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
39	39		}
40	40
41	41		budget := maintenanceBatch
42		-	if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil \|\| budget == 0 {
	42	+	if _, err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil \|\| budget == 0 {
43	43		return err
44	44		}
	45	+	if r.maintenanceQuarantine != nil {
	46	+	changed, done, err := r.maintainQuarantine(ctx, &budget)
	47	+	if err != nil {
	48	+	return err
	49	+	}
	50	+	if changed {
	51	+	if err := r.syncDirectory(filepath.Join(r.dir, repositorySessionsDir)); err != nil {
	52	+	return fmt.Errorf("sync snapshot maintenance directory: %w", safeFilesystemError(err))
	53	+	}
	54	+	}
	55	+	if done {
	56	+	r.maintenanceQuarantine = nil
	57	+	}
	58	+	return nil
	59	+	}
45	60		// Resume an incomplete session before discovering another one. This keeps
46	61		// retained marks (and their session lock/epoch references) bounded even
47	62		// when a hostile repository contains arbitrarily many partial sessions.

@@ -69,7 +84,8 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
69	84		}
70	85		path := filepath.Join(sessions, entry.name)
71	86		if isQuarantine(entry.name) {
72		-	changed, err := r.removeTreeBatch(ctx, path, &budget)
	87	+	r.maintenanceQuarantine = &quarantineMaintenance{root: path, current: path}
	88	+	changed, done, err := r.maintainQuarantine(ctx, &budget)
73	89		if err != nil && !errors.Is(err, os.ErrNotExist) {
74	90		return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
75	91		}

@@ -81,7 +97,13 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
81	97		return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
82	98		}
83	99		}
84		-	continue
	100	+	if done {
	101	+	r.maintenanceQuarantine = nil
	102	+	}
	103	+	if i+1 < len(entries) {
	104	+	r.requeueMaintenanceEntries(sessions, "sessions", entries[i+1:])
	105	+	}
	106	+	return nil
85	107		}
86	108		if !entry.isDir \|\| !canonicalSessionKey(entry.name) {
87	109		continue

 	}
 	budget := maintenanceBatch
-	if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil || budget == 0 {
+	if _, err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil || budget == 0 {
 		return err
 	}
+	if r.maintenanceQuarantine != nil {
+		changed, done, err := r.maintainQuarantine(ctx, &budget)
+		if err != nil {
+			return err
+		}
+		if changed {
+			if err := r.syncDirectory(filepath.Join(r.dir, repositorySessionsDir)); err != nil {
+				return fmt.Errorf("sync snapshot maintenance directory: %w", safeFilesystemError(err))
+			}
+		}
+		if done {
+			r.maintenanceQuarantine = nil
+		}
+		return nil
+	}
 	// Resume an incomplete session before discovering another one. This keeps
 	// retained marks (and their session lock/epoch references) bounded even
 	// when a hostile repository contains arbitrarily many partial sessions.
 		}
 		path := filepath.Join(sessions, entry.name)
 		if isQuarantine(entry.name) {
-			changed, err := r.removeTreeBatch(ctx, path, &budget)
+			r.maintenanceQuarantine = &quarantineMaintenance{root: path, current: path}
+			changed, done, err := r.maintainQuarantine(ctx, &budget)
 			if err != nil && !errors.Is(err, os.ErrNotExist) {
 				return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
 			}
 					return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
 				}
 			}
-			continue
+			if done {
+				r.maintenanceQuarantine = nil
+			}
+			if i+1 < len(entries) {
+				r.requeueMaintenanceEntries(sessions, "sessions", entries[i+1:])
+			}
+			return nil
 		}
 		if !entry.isDir || !canonicalSessionKey(entry.name) {
 			continue

M internal/adapters/snapshot/maintenance_cursor.go +22 -11


@@ -12,6 +12,13 @@ import (
12	12		"unsafe"
13	13		)
14	14
	15	+	// maxMaintenanceCursors bounds all retained directory seek cookies, including
	16	+	// cursors for attacker-controlled object shards. Maintenance processes only
	17	+	// one shard at a time, so normal operation stays well below this ceiling.
	18	+	const maxMaintenanceCursors = 8
	19	+
	20	+	var errMaintenanceCursorLimit = errors.New("snapshot maintenance cursor limit")
	21	+
15	22		type maintenanceCursor struct {
16	23		offset int64
17	24		pending []maintenanceDirEntry

@@ -29,6 +36,7 @@ func (r *Repository) resetMaintenance() {
29	36		}
30	37		r.maintenanceCursors = nil
31	38		r.maintenanceSessions = nil
	39	+	r.maintenanceQuarantine = nil
32	40		}
33	41
34	42		func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }

@@ -50,6 +58,9 @@ func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]ma
50	58		return entries, cursor.done, nil
51	59		}
52	60		if cursor == nil {
	61	+	if len(r.maintenanceCursors) >= maxMaintenanceCursors {
	62	+	return nil, false, errMaintenanceCursorLimit
	63	+	}
53	64		cursor = &maintenanceCursor{}
54	65		r.maintenanceCursors[id] = cursor
55	66		}

@@ -185,41 +196,41 @@ func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []ma
185	196		cursor.pending = append(entries, cursor.pending...)
186	197		}
187	198
188		-	func (r Repository) removeTemps(ctx context.Context, dir string, budget int, purpose string) error {
	199	+	func (r Repository) removeTemps(ctx context.Context, dir string, budget int, purpose string) (bool, error) {
189	200		if *budget == 0 {
190		-	return nil
	201	+	return false, nil
191	202		}
192		-	entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
	203	+	entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
193	204		if errors.Is(err, os.ErrNotExist) {
194		-	return nil
	205	+	return true, nil
195	206		}
196	207		if err != nil {
197		-	return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
	208	+	return false, fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
198	209		}
199	210		for i, entry := range entries {
200	211		if *budget == 0 {
201	212		r.requeueMaintenanceEntries(dir, purpose, entries[i:])
202		-	return nil
	213	+	return false, nil
203	214		}
204	215		if err := ctx.Err(); err != nil {
205		-	return err
	216	+	return false, err
206	217		}
207	218		if entry.isDir \|\| !strings.HasPrefix(entry.name, ".tmp-") {
208	219		continue
209	220		}
210	221		path := filepath.Join(dir, entry.name)
211	222		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
212		-	return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
	223	+	return false, fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
213	224		}
214	225		if err := ctx.Err(); err != nil {
215		-	return err
	226	+	return false, err
216	227		}
217	228		if err := r.syncDirectory(dir); err != nil {
218		-	return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
	229	+	return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
219	230		}
220	231		*budget--
221	232		}
222		-	return nil
	233	+	return done, nil
223	234		}
224	235
225	236		// maintenancePath avoids exposing the repository root while retaining a

 	"unsafe"
 )
+// maxMaintenanceCursors bounds all retained directory seek cookies, including
+// cursors for attacker-controlled object shards. Maintenance processes only
+// one shard at a time, so normal operation stays well below this ceiling.
+const maxMaintenanceCursors = 8
+var errMaintenanceCursorLimit = errors.New("snapshot maintenance cursor limit")
 type maintenanceCursor struct {
 	offset  int64
 	pending []maintenanceDirEntry
 	}
 	r.maintenanceCursors = nil
 	r.maintenanceSessions = nil
+	r.maintenanceQuarantine = nil
 }
 func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
 		return entries, cursor.done, nil
 	}
 	if cursor == nil {
+		if len(r.maintenanceCursors) >= maxMaintenanceCursors {
+			return nil, false, errMaintenanceCursorLimit
+		}
 		cursor = &maintenanceCursor{}
 		r.maintenanceCursors[id] = cursor
 	}
 	cursor.pending = append(entries, cursor.pending...)
 }
-func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
+func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) (bool, error) {
 	if *budget == 0 {
-		return nil
+		return false, nil
 	}
-	entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
+	entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
 	if errors.Is(err, os.ErrNotExist) {
-		return nil
+		return true, nil
 	}
 	if err != nil {
-		return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
+		return false, fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
 	}
 	for i, entry := range entries {
 		if *budget == 0 {
 			r.requeueMaintenanceEntries(dir, purpose, entries[i:])
-			return nil
+			return false, nil
 		}
 		if err := ctx.Err(); err != nil {
-			return err
+			return false, err
 		}
 		if entry.isDir || !strings.HasPrefix(entry.name, ".tmp-") {
 			continue
 		}
 		path := filepath.Join(dir, entry.name)
 		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
-			return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
+			return false, fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
 		}
 		if err := ctx.Err(); err != nil {
-			return err
+			return false, err
 		}
 		if err := r.syncDirectory(dir); err != nil {
-			return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
+			return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
 		}
 		*budget--
 	}
-	return nil
+	return done, nil
 }
 // maintenancePath avoids exposing the repository root while retaining a

M internal/adapters/snapshot/maintenance_delete.go +88 -57


@@ -93,73 +93,104 @@ func (r *Repository) pendingQuarantine(dir, key string) (bool, error) {
93	93		// repository (and the next fresh cycle) always begins at a safe boundary.
94	94		func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
95	95
96		-	// readDirBatch is used only for a mutating quarantine walk, where every
97		-	// successful step removes the entry it selected. General maintenance scans use
98		-	// repository-owned cursors above.
99		-	func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
100		-	f, err := os.Open(dir)
101		-	if err != nil {
102		-	return nil, err
103		-	}
104		-	entries, readErr := f.ReadDir(n)
105		-	closeErr := f.Close()
106		-	if readErr != nil && !errors.Is(readErr, io.EOF) {
107		-	return nil, readErr
108		-	}
109		-	if closeErr != nil {
110		-	return nil, closeErr
111		-	}
112		-	return entries, nil
113		-	}
114		-
115		-	func (r Repository) removeTreeBatch(ctx context.Context, path string, budget int) (bool, error) {
116		-	changed := false
117		-	for *budget > 0 {
118		-	did, err := r.removeTreeStep(ctx, path, budget)
119		-	if err != nil \|\| !did {
120		-	return changed, err
121		-	}
122		-	changed = true
123		-	}
124		-	return changed, nil
	96	+	// quarantineMaintenance retains one root-to-current path, never an unbounded
	97	+	// stack of directory handles or names. Parent paths are recovered after a
	98	+	// removal, while descent resumes from current on the next budgeted step.
	99	+	type quarantineMaintenance struct {
	100	+	root string
	101	+	current string
125	102		}
126	103
127		-	func (r Repository) removeTreeStep(ctx context.Context, path string, budget int) (bool, error) {
	104	+	// consumeQuarantineWork charges every filesystem operation that can be driven
	105	+	// by quarantine contents. This makes deep and wide trees subject to exactly
	106	+	// the same maintenanceBatch ceiling as removal work.
	107	+	func (r Repository) consumeQuarantineWork(budget int, operation string) bool {
128	108		if *budget == 0 {
129		-	return false, nil
	109	+	return false
130	110		}
131		-	info, err := os.Lstat(path)
132		-	if errors.Is(err, os.ErrNotExist) {
133		-	return false, nil
	111	+	*budget--
	112	+	if hook := r.hooks.beforeMaintenanceWork; hook != nil {
	113	+	hook(operation)
134	114		}
135		-	if err != nil {
136		-	return false, err
	115	+	return true
	116	+	}
	117	+
	118	+	// maintainQuarantine makes one bounded depth-first traversal step at a time.
	119	+	// It deliberately reopens a directory to select its first child: every
	120	+	// successful child removal changes that prefix, so no directory cursor is
	121	+	// retained at each hostile nesting level.
	122	+	func (r Repository) maintainQuarantine(ctx context.Context, budget int) (changed, done bool, err error) {
	123	+	state := r.maintenanceQuarantine
	124	+	if state == nil {
	125	+	return false, true, nil
137	126		}
138		-	if !info.IsDir() {
	127	+	for *budget > 0 {
139	128		if err := ctx.Err(); err != nil {
140		-	return false, err
	129	+	return changed, false, err
141	130		}
142		-	if err := r.remove(path); err != nil {
143		-	return false, err
	131	+	if !r.consumeQuarantineWork(budget, "stat") {
	132	+	return changed, false, nil
144	133		}
145		-	*budget--
146		-	return true, nil
147		-	}
148		-	// A quarantine tree contains no live entries. Reopening from its start is
149		-	// safe here because each successful step removes that first descendant.
150		-	entries, err := readDirBatch(path, 1)
151		-	if err != nil {
152		-	return false, err
153		-	}
154		-	if len(entries) == 0 {
155		-	if err := ctx.Err(); err != nil {
156		-	return false, err
	134	+	info, err := os.Lstat(state.current)
	135	+	if errors.Is(err, os.ErrNotExist) {
	136	+	if state.current == state.root {
	137	+	return changed, true, nil
	138	+	}
	139	+	state.current = filepath.Dir(state.current)
	140	+	continue
	141	+	}
	142	+	if err != nil {
	143	+	return changed, false, err
	144	+	}
	145	+	if !info.IsDir() {
	146	+	if !r.consumeQuarantineWork(budget, "remove") {
	147	+	return changed, false, nil
	148	+	}
	149	+	if err := r.remove(state.current); err != nil && !errors.Is(err, os.ErrNotExist) {
	150	+	return changed, false, err
	151	+	}
	152	+	changed = true
	153	+	state.current = filepath.Dir(state.current)
	154	+	continue
	155	+	}
	156	+
	157	+	if !r.consumeQuarantineWork(budget, "open") {
	158	+	return changed, false, nil
	159	+	}
	160	+	dir, err := os.Open(state.current)
	161	+	if err != nil {
	162	+	return changed, false, err
	163	+	}
	164	+	if !r.consumeQuarantineWork(budget, "read") {
	165	+	_ = dir.Close()
	166	+	return changed, false, nil
	167	+	}
	168	+	entries, readErr := dir.ReadDir(1)
	169	+	closeErr := dir.Close()
	170	+	if readErr != nil && !errors.Is(readErr, io.EOF) {
	171	+	return changed, false, readErr
	172	+	}
	173	+	if closeErr != nil {
	174	+	return changed, false, closeErr
	175	+	}
	176	+	if len(entries) == 0 {
	177	+	if !r.consumeQuarantineWork(budget, "remove") {
	178	+	return changed, false, nil
	179	+	}
	180	+	if err := r.remove(state.current); err != nil && !errors.Is(err, os.ErrNotExist) {
	181	+	return changed, false, err
	182	+	}
	183	+	changed = true
	184	+	if state.current == state.root {
	185	+	return changed, true, nil
	186	+	}
	187	+	state.current = filepath.Dir(state.current)
	188	+	continue
157	189		}
158		-	if err := r.remove(path); err != nil {
159		-	return false, err
	190	+	if !r.consumeQuarantineWork(budget, "descend") {
	191	+	return changed, false, nil
160	192		}
161		-	*budget--
162		-	return true, nil
	193	+	state.current = filepath.Join(state.current, entries[0].Name())
163	194		}
164		-	return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
	195	+	return changed, false, nil
165	196		}

 // repository (and the next fresh cycle) always begins at a safe boundary.
 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
-// readDirBatch is used only for a mutating quarantine walk, where every
+// quarantineMaintenance retains one root-to-current path, never an unbounded
-// successful step removes the entry it selected. General maintenance scans use
+// stack of directory handles or names. Parent paths are recovered after a
-// repository-owned cursors above.
+// removal, while descent resumes from current on the next budgeted step.
-func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
+type quarantineMaintenance struct {
-	f, err := os.Open(dir)
+	root    string
-	if err != nil {
+	current string
-		return nil, err
-	}
-	entries, readErr := f.ReadDir(n)
-	closeErr := f.Close()
-	if readErr != nil && !errors.Is(readErr, io.EOF) {
-		return nil, readErr
-	}
-	if closeErr != nil {
-		return nil, closeErr
-	}
-	return entries, nil
-}
-func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
-	changed := false
-	for *budget > 0 {
-		did, err := r.removeTreeStep(ctx, path, budget)
-		if err != nil || !did {
-			return changed, err
-		}
-		changed = true
-	}
-	return changed, nil
 }
-func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
+// consumeQuarantineWork charges every filesystem operation that can be driven
+// by quarantine contents. This makes deep and wide trees subject to exactly
+// the same maintenanceBatch ceiling as removal work.
+func (r *Repository) consumeQuarantineWork(budget *int, operation string) bool {
 	if *budget == 0 {
-		return false, nil
+		return false
 	}
-	info, err := os.Lstat(path)
+	*budget--
-	if errors.Is(err, os.ErrNotExist) {
+	if hook := r.hooks.beforeMaintenanceWork; hook != nil {
-		return false, nil
+		hook(operation)
 	}
-	if err != nil {
+	return true
-		return false, err
+}
+// maintainQuarantine makes one bounded depth-first traversal step at a time.
+// It deliberately reopens a directory to select its first child: every
+// successful child removal changes that prefix, so no directory cursor is
+// retained at each hostile nesting level.
+func (r *Repository) maintainQuarantine(ctx context.Context, budget *int) (changed, done bool, err error) {
+	state := r.maintenanceQuarantine
+	if state == nil {
+		return false, true, nil
 	}
-	if !info.IsDir() {
+	for *budget > 0 {
 		if err := ctx.Err(); err != nil {
-			return false, err
+			return changed, false, err
 		}
-		if err := r.remove(path); err != nil {
+		if !r.consumeQuarantineWork(budget, "stat") {
-			return false, err
+			return changed, false, nil
 		}
-		*budget--
+		info, err := os.Lstat(state.current)
-		return true, nil
+		if errors.Is(err, os.ErrNotExist) {
-	}
+			if state.current == state.root {
-	// A quarantine tree contains no live entries. Reopening from its start is
+				return changed, true, nil
-	// safe here because each successful step removes that first descendant.
+			}
-	entries, err := readDirBatch(path, 1)
+			state.current = filepath.Dir(state.current)
-	if err != nil {
+			continue
-		return false, err
+		}
-	}
+		if err != nil {
-	if len(entries) == 0 {
+			return changed, false, err
-		if err := ctx.Err(); err != nil {
+		}
-			return false, err
+		if !info.IsDir() {
+			if !r.consumeQuarantineWork(budget, "remove") {
+				return changed, false, nil
+			}
+			if err := r.remove(state.current); err != nil && !errors.Is(err, os.ErrNotExist) {
+				return changed, false, err
+			}
+			changed = true
+			state.current = filepath.Dir(state.current)
+			continue
+		}
+		if !r.consumeQuarantineWork(budget, "open") {
+			return changed, false, nil
+		}
+		dir, err := os.Open(state.current)
+		if err != nil {
+			return changed, false, err
+		}
+		if !r.consumeQuarantineWork(budget, "read") {
+			_ = dir.Close()
+			return changed, false, nil
+		}
+		entries, readErr := dir.ReadDir(1)
+		closeErr := dir.Close()
+		if readErr != nil && !errors.Is(readErr, io.EOF) {
+			return changed, false, readErr
+		}
+		if closeErr != nil {
+			return changed, false, closeErr
+		}
+		if len(entries) == 0 {
+			if !r.consumeQuarantineWork(budget, "remove") {
+				return changed, false, nil
+			}
+			if err := r.remove(state.current); err != nil && !errors.Is(err, os.ErrNotExist) {
+				return changed, false, err
+			}
+			changed = true
+			if state.current == state.root {
+				return changed, true, nil
+			}
+			state.current = filepath.Dir(state.current)
+			continue
 		}
-		if err := r.remove(path); err != nil {
+		if !r.consumeQuarantineWork(budget, "descend") {
-			return false, err
+			return changed, false, nil
 		}
-		*budget--
+		state.current = filepath.Join(state.current, entries[0].Name())
-		return true, nil
 	}
-	return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
+	return changed, false, nil
 }

M internal/adapters/snapshot/maintenance_gc.go +74 -51


@@ -19,17 +19,19 @@ type manifestMaintenance struct {
19	19		}
20	20
21	21		type sessionMaintenance struct {
22		-	lock *sessionMutex
23		-	token string
24		-	epoch uint64
25		-	marked map[uint64]manifestMaintenance
26		-	referenceCount int
27		-	uncertain bool
28		-	conservative bool
29		-	markDone bool
30		-	manifestQueue []uint64
31		-	sweepQueue []string
32		-	sweepRootDone bool
	22	+	lock *sessionMutex
	23	+	token string
	24	+	epoch uint64
	25	+	marked map[uint64]manifestMaintenance
	26	+	referenceCount int
	27	+	uncertain bool
	28	+	conservative bool
	29	+	markDone bool
	30	+	manifestQueue []uint64
	31	+	objectTempShard string
	32	+	objectTempsDone bool
	33	+	sweepShard string
	34	+	sweepRootDone bool
33	35		}
34	36
35	37		// canRetainManifest reserves bounded state before retaining a manifest's

@@ -54,29 +56,15 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
54	56		if err != nil {
55	57		return err
56	58		}
57		-	if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil \|\| *budget == 0 {
	59	+	if _, err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil \|\| *budget == 0 {
58	60		return err
59	61		}
60		-
61		-	objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
62		-	shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
63		-	if err != nil && !errors.Is(err, os.ErrNotExist) {
64		-	return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
65		-	}
66		-	for i, shard := range shards {
67		-	if *budget == 0 {
68		-	r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
69		-	return nil
70		-	}
71		-	if shard.isDir {
72		-	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
73		-	return err
74		-	}
	62	+	if !state.objectTempsDone {
	63	+	done, err := r.removeObjectTemps(ctx, key, state, budget)
	64	+	if err != nil \|\| !done \|\| *budget == 0 {
	65	+	return err
75	66		}
76	67		}
77		-	if *budget == 0 {
78		-	return nil
79		-	}
80	68		if !state.markDone {
81	69		if err := r.markSession(ctx, key, state); err != nil {
82	70		return err

@@ -280,6 +268,43 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st
280	268		return nil
281	269		}
282	270
	271	+	// removeObjectTemps processes a single shard to completion before advancing
	272	+	// the shard-root cursor. Thus a hostile set of shards can retain at most the
	273	+	// root cursor and the one active shard cursor, rather than one cursor per
	274	+	// discovered shard.
	275	+	func (r Repository) removeObjectTemps(ctx context.Context, key string, state sessionMaintenance, budget *int) (bool, error) {
	276	+	root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
	277	+	for *budget > 0 {
	278	+	if state.objectTempShard != "" {
	279	+	done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard)
	280	+	if err != nil \|\| !done {
	281	+	return false, err
	282	+	}
	283	+	// This releases the active shard cursor before the next root entry
	284	+	// is discovered. A completed small shard may therefore advance in
	285	+	// this call without accumulating continuation state.
	286	+	state.objectTempShard = ""
	287	+	continue
	288	+	}
	289	+	if state.objectTempsDone {
	290	+	return true, nil
	291	+	}
	292	+	entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key)
	293	+	if errors.Is(err, os.ErrNotExist) {
	294	+	state.objectTempsDone = true
	295	+	return true, nil
	296	+	}
	297	+	if err != nil {
	298	+	return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
	299	+	}
	300	+	state.objectTempsDone = done
	301	+	if len(entries) != 0 && entries[0].isDir {
	302	+	state.objectTempShard = entries[0].name
	303	+	}
	304	+	}
	305	+	return state.objectTempsDone && state.objectTempShard == "", nil
	306	+	}
	307	+
283	308		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {
284	309		// Every bounded sweep batch is tied to the mark's storage epoch. Do not use
285	310		// stale references after any publication or replacement.

@@ -289,35 +314,33 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
289	314		}
290	315		referenced := retainedReferences(state.marked, state.conservative)
291	316		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
292		-	// Do not keep accumulating shard names while draining earlier work. This
293		-	// bounds retained names to one directory batch even for a session with many
294		-	// object shards.
295		-	if !state.sweepRootDone && len(state.sweepQueue) == 0 {
296		-	entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
	317	+	if state.sweepShard != "" && *budget > 0 {
	318	+	shard := state.sweepShard
	319	+	done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
	320	+	if err != nil {
	321	+	return err
	322	+	}
	323	+	if !done {
	324	+	return nil
	325	+	}
	326	+	state.sweepShard = ""
	327	+	// Release this shard's cursor before discovering another shard.
	328	+	return nil
	329	+	}
	330	+	if !state.sweepRootDone {
	331	+	entries, done, err := r.readMaintenanceDir(root, 1, "sweep-shards:"+key)
297	332		if errors.Is(err, os.ErrNotExist) {
298	333		state.sweepRootDone = true
299	334		} else if err != nil {
300	335		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
301	336		} else {
302		-	for _, entry := range entries {
303		-	if entry.isDir {
304		-	state.sweepQueue = append(state.sweepQueue, entry.name)
305		-	}
306		-	}
307	337		state.sweepRootDone = done
	338	+	if len(entries) != 0 && entries[0].isDir {
	339	+	state.sweepShard = entries[0].name
	340	+	}
308	341		}
309	342		}
310		-	if len(state.sweepQueue) != 0 && *budget > 0 {
311		-	shard := state.sweepQueue[0]
312		-	done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
313		-	if err != nil {
314		-	return err
315		-	}
316		-	if done {
317		-	state.sweepQueue = state.sweepQueue[1:]
318		-	}
319		-	}
320		-	if state.sweepRootDone && len(state.sweepQueue) == 0 {
	343	+	if state.sweepRootDone && state.sweepShard == "" {
321	344		r.clearSessionMaintenance(key)
322	345		}
323	346		return nil


@@ -19,17 +19,19 @@ type manifestMaintenance struct {
19		}	19		}
20			20
21		type sessionMaintenance struct {	21		type sessionMaintenance struct {
22	-	lock *sessionMutex	22	+	lock *sessionMutex
23	-	token string	23	+	token string
24	-	epoch uint64	24	+	epoch uint64
25	-	marked map[uint64]manifestMaintenance	25	+	marked map[uint64]manifestMaintenance
26	-	referenceCount int	26	+	referenceCount int
27	-	uncertain bool	27	+	uncertain bool
28	-	conservative bool	28	+	conservative bool
29	-	markDone bool	29	+	markDone bool
30	-	manifestQueue []uint64	30	+	manifestQueue []uint64
31	-	sweepQueue []string	31	+	objectTempShard string
32	-	sweepRootDone bool	32	+	objectTempsDone bool
			33	+	sweepShard string
			34	+	sweepRootDone bool
33		}	35		}
34			36
35		// canRetainManifest reserves bounded state before retaining a manifest's	37		// canRetainManifest reserves bounded state before retaining a manifest's

@@ -54,29 +56,15 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
54		if err != nil {	56		if err != nil {
55		return err	57		return err
56		}	58		}
57	-	if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil \|\| *budget == 0 {	59	+	if _, err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil \|\| *budget == 0 {
58		return err	60		return err
59		}	61		}
60	-		62	+	if !state.objectTempsDone {
61	-	objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)	63	+	done, err := r.removeObjectTemps(ctx, key, state, budget)
62	-	shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)	64	+	if err != nil \|\| !done \|\| *budget == 0 {
63	-	if err != nil && !errors.Is(err, os.ErrNotExist) {	65	+	return err
64	-	return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
65	-	}
66	-	for i, shard := range shards {
67	-	if *budget == 0 {
68	-	r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
69	-	return nil
70	-	}
71	-	if shard.isDir {
72	-	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
73	-	return err
74	-	}
75		}	66		}
76		}	67		}
77	-	if *budget == 0 {
78	-	return nil
79	-	}
80		if !state.markDone {	68		if !state.markDone {
81		if err := r.markSession(ctx, key, state); err != nil {	69		if err := r.markSession(ctx, key, state); err != nil {
82		return err	70		return err

@@ -280,6 +268,43 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st
280		return nil	268		return nil
281		}	269		}
282			270
			271	+	// removeObjectTemps processes a single shard to completion before advancing
			272	+	// the shard-root cursor. Thus a hostile set of shards can retain at most the
			273	+	// root cursor and the one active shard cursor, rather than one cursor per
			274	+	// discovered shard.
			275	+	func (r Repository) removeObjectTemps(ctx context.Context, key string, state sessionMaintenance, budget *int) (bool, error) {
			276	+	root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
			277	+	for *budget > 0 {
			278	+	if state.objectTempShard != "" {
			279	+	done, err := r.removeTemps(ctx, filepath.Join(root, state.objectTempShard), budget, "object-temps:"+key+":"+state.objectTempShard)
			280	+	if err != nil \|\| !done {
			281	+	return false, err
			282	+	}
			283	+	// This releases the active shard cursor before the next root entry
			284	+	// is discovered. A completed small shard may therefore advance in
			285	+	// this call without accumulating continuation state.
			286	+	state.objectTempShard = ""
			287	+	continue
			288	+	}
			289	+	if state.objectTempsDone {
			290	+	return true, nil
			291	+	}
			292	+	entries, done, err := r.readMaintenanceDir(root, 1, "object-temps-shards:"+key)
			293	+	if errors.Is(err, os.ErrNotExist) {
			294	+	state.objectTempsDone = true
			295	+	return true, nil
			296	+	}
			297	+	if err != nil {
			298	+	return false, fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
			299	+	}
			300	+	state.objectTempsDone = done
			301	+	if len(entries) != 0 && entries[0].isDir {
			302	+	state.objectTempShard = entries[0].name
			303	+	}
			304	+	}
			305	+	return state.objectTempsDone && state.objectTempShard == "", nil
			306	+	}
			307	+
283		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {	308		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {
284		// Every bounded sweep batch is tied to the mark's storage epoch. Do not use	309		// Every bounded sweep batch is tied to the mark's storage epoch. Do not use
285		// stale references after any publication or replacement.	310		// stale references after any publication or replacement.

@@ -289,35 +314,33 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
289		}	314		}
290		referenced := retainedReferences(state.marked, state.conservative)	315		referenced := retainedReferences(state.marked, state.conservative)
291		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)	316		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
292	-	// Do not keep accumulating shard names while draining earlier work. This	317	+	if state.sweepShard != "" && *budget > 0 {
293	-	// bounds retained names to one directory batch even for a session with many	318	+	shard := state.sweepShard
294	-	// object shards.	319	+	done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
295	-	if !state.sweepRootDone && len(state.sweepQueue) == 0 {	320	+	if err != nil {
296	-	entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)	321	+	return err
			322	+	}
			323	+	if !done {
			324	+	return nil
			325	+	}
			326	+	state.sweepShard = ""
			327	+	// Release this shard's cursor before discovering another shard.
			328	+	return nil
			329	+	}
			330	+	if !state.sweepRootDone {
			331	+	entries, done, err := r.readMaintenanceDir(root, 1, "sweep-shards:"+key)
297		if errors.Is(err, os.ErrNotExist) {	332		if errors.Is(err, os.ErrNotExist) {
298		state.sweepRootDone = true	333		state.sweepRootDone = true
299		} else if err != nil {	334		} else if err != nil {
300		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))	335		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
301		} else {	336		} else {
302	-	for _, entry := range entries {
303	-	if entry.isDir {
304	-	state.sweepQueue = append(state.sweepQueue, entry.name)
305	-	}
306	-	}
307		state.sweepRootDone = done	337		state.sweepRootDone = done
			338	+	if len(entries) != 0 && entries[0].isDir {
			339	+	state.sweepShard = entries[0].name
			340	+	}
308		}	341		}
309		}	342		}
310	-	if len(state.sweepQueue) != 0 && *budget > 0 {	343	+	if state.sweepRootDone && state.sweepShard == "" {
311	-	shard := state.sweepQueue[0]
312	-	done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
313	-	if err != nil {
314	-	return err
315	-	}
316	-	if done {
317	-	state.sweepQueue = state.sweepQueue[1:]
318	-	}
319	-	}
320	-	if state.sweepRootDone && len(state.sweepQueue) == 0 {
321		r.clearSessionMaintenance(key)	344		r.clearSessionMaintenance(key)
322		}	345		}
323		return nil	346		return nil

A internal/adapters/snapshot/maintenance_hardening_test.go +94 -0

@@ -0,0 +1,94 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"os"
	8	+	"path/filepath"
	9	+	"testing"
	10	+	)
	11	+
	12	+	func TestRepositoryMaintainBoundsCursorsAcrossHostileObjectShards(t *testing.T) {
	13	+	repo := NewRepository(privateDir(t))
	14	+	key := sessionKey("named")
	15	+	objects := filepath.Join(repo.sessionPath(key), repositoryObjectsDir)
	16	+	const shards = maintenanceBatch * 2
	17	+	for i := 0; i < shards; i++ {
	18	+	shard := filepath.Join(objects, fmt.Sprintf("%03d", i))
	19	+	if err := os.MkdirAll(shard, 0o700); err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	if err := os.WriteFile(filepath.Join(shard, ".tmp-stale"), []byte("stale"), 0o600); err != nil {
	23	+	t.Fatal(err)
	24	+	}
	25	+	// Force a resumable cursor in every attacker-controlled shard.
	26	+	for j := 0; j < maintenanceBatch+1; j++ {
	27	+	if err := os.WriteFile(filepath.Join(shard, fmt.Sprintf("live-%03d", j)), []byte("live"), 0o600); err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	}
	31	+	}
	32	+
	33	+	for pass := 0; pass < shards*4; pass++ {
	34	+	if err := repo.Maintain(context.Background()); err != nil {
	35	+	t.Fatal(err)
	36	+	}
	37	+	if got := len(repo.maintenanceCursors); got > maxMaintenanceCursors {
	38	+	t.Fatalf("retained maintenance cursors = %d, want at most %d", got, maxMaintenanceCursors)
	39	+	}
	40	+	}
	41	+	for i := 0; i < shards; i++ {
	42	+	_, err := os.Lstat(filepath.Join(objects, fmt.Sprintf("%03d", i), ".tmp-stale"))
	43	+	if !errors.Is(err, os.ErrNotExist) {
	44	+	t.Fatalf("stale temporary object in shard %d = %v, want removed", i, err)
	45	+	}
	46	+	}
	47	+	}
	48	+
	49	+	func TestRepositoryMaintainBoundsAndResumesDeepWideQuarantine(t *testing.T) {
	50	+	repo := NewRepository(privateDir(t))
	51	+	quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-hostile")
	52	+	deep := quarantine
	53	+	for i := 0; i < maintenanceBatch*2; i++ {
	54	+	deep = filepath.Join(deep, "d")
	55	+	}
	56	+	leaf := filepath.Join(deep, "leaf")
	57	+	if err := os.MkdirAll(filepath.Dir(leaf), 0o700); err != nil {
	58	+	t.Fatal(err)
	59	+	}
	60	+	if err := os.WriteFile(leaf, []byte("stale"), 0o600); err != nil {
	61	+	t.Fatal(err)
	62	+	}
	63	+	for i := 0; i < maintenanceBatch*2; i++ {
	64	+	if err := os.WriteFile(filepath.Join(quarantine, fmt.Sprintf("wide-%03d", i)), []byte("stale"), 0o600); err != nil {
	65	+	t.Fatal(err)
	66	+	}
	67	+	}
	68	+
	69	+	work := 0
	70	+	repo.hooks.beforeMaintenanceWork = func(string) { work++ }
	71	+	if err := repo.Maintain(context.Background()); err != nil {
	72	+	t.Fatal(err)
	73	+	}
	74	+	if work == 0 \|\| work > maintenanceBatch {
	75	+	t.Fatalf("quarantine work in one maintenance call = %d, want 1..%d", work, maintenanceBatch)
	76	+	}
	77	+	if _, err := os.Lstat(leaf); err != nil {
	78	+	t.Fatalf("deep quarantine leaf removed despite bounded traversal: %v", err)
	79	+	}
	80	+
	81	+	for pass := 0; pass < 1000; pass++ {
	82	+	work = 0
	83	+	if err := repo.Maintain(context.Background()); err != nil {
	84	+	t.Fatal(err)
	85	+	}
	86	+	if work > maintenanceBatch {
	87	+	t.Fatalf("quarantine work in pass %d = %d, want at most %d", pass, work, maintenanceBatch)
	88	+	}
	89	+	if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
	90	+	return
	91	+	}
	92	+	}
	93	+	t.Fatal("deep and wide quarantine was not eventually removed")
	94	+	}

@@ -0,0 +1,94 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"os"
	8	+	"path/filepath"
	9	+	"testing"
	10	+	)
	11	+
	12	+	func TestRepositoryMaintainBoundsCursorsAcrossHostileObjectShards(t *testing.T) {
	13	+	repo := NewRepository(privateDir(t))
	14	+	key := sessionKey("named")
	15	+	objects := filepath.Join(repo.sessionPath(key), repositoryObjectsDir)
	16	+	const shards = maintenanceBatch * 2
	17	+	for i := 0; i < shards; i++ {
	18	+	shard := filepath.Join(objects, fmt.Sprintf("%03d", i))
	19	+	if err := os.MkdirAll(shard, 0o700); err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	if err := os.WriteFile(filepath.Join(shard, ".tmp-stale"), []byte("stale"), 0o600); err != nil {
	23	+	t.Fatal(err)
	24	+	}
	25	+	// Force a resumable cursor in every attacker-controlled shard.
	26	+	for j := 0; j < maintenanceBatch+1; j++ {
	27	+	if err := os.WriteFile(filepath.Join(shard, fmt.Sprintf("live-%03d", j)), []byte("live"), 0o600); err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	}
	31	+	}
	32	+
	33	+	for pass := 0; pass < shards*4; pass++ {
	34	+	if err := repo.Maintain(context.Background()); err != nil {
	35	+	t.Fatal(err)
	36	+	}
	37	+	if got := len(repo.maintenanceCursors); got > maxMaintenanceCursors {
	38	+	t.Fatalf("retained maintenance cursors = %d, want at most %d", got, maxMaintenanceCursors)
	39	+	}
	40	+	}
	41	+	for i := 0; i < shards; i++ {
	42	+	_, err := os.Lstat(filepath.Join(objects, fmt.Sprintf("%03d", i), ".tmp-stale"))
	43	+	if !errors.Is(err, os.ErrNotExist) {
	44	+	t.Fatalf("stale temporary object in shard %d = %v, want removed", i, err)
	45	+	}
	46	+	}
	47	+	}
	48	+
	49	+	func TestRepositoryMaintainBoundsAndResumesDeepWideQuarantine(t *testing.T) {
	50	+	repo := NewRepository(privateDir(t))
	51	+	quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-hostile")
	52	+	deep := quarantine
	53	+	for i := 0; i < maintenanceBatch*2; i++ {
	54	+	deep = filepath.Join(deep, "d")
	55	+	}
	56	+	leaf := filepath.Join(deep, "leaf")
	57	+	if err := os.MkdirAll(filepath.Dir(leaf), 0o700); err != nil {
	58	+	t.Fatal(err)
	59	+	}
	60	+	if err := os.WriteFile(leaf, []byte("stale"), 0o600); err != nil {
	61	+	t.Fatal(err)
	62	+	}
	63	+	for i := 0; i < maintenanceBatch*2; i++ {
	64	+	if err := os.WriteFile(filepath.Join(quarantine, fmt.Sprintf("wide-%03d", i)), []byte("stale"), 0o600); err != nil {
	65	+	t.Fatal(err)
	66	+	}
	67	+	}
	68	+
	69	+	work := 0
	70	+	repo.hooks.beforeMaintenanceWork = func(string) { work++ }
	71	+	if err := repo.Maintain(context.Background()); err != nil {
	72	+	t.Fatal(err)
	73	+	}
	74	+	if work == 0 \|\| work > maintenanceBatch {
	75	+	t.Fatalf("quarantine work in one maintenance call = %d, want 1..%d", work, maintenanceBatch)
	76	+	}
	77	+	if _, err := os.Lstat(leaf); err != nil {
	78	+	t.Fatalf("deep quarantine leaf removed despite bounded traversal: %v", err)
	79	+	}
	80	+
	81	+	for pass := 0; pass < 1000; pass++ {
	82	+	work = 0
	83	+	if err := repo.Maintain(context.Background()); err != nil {
	84	+	t.Fatal(err)
	85	+	}
	86	+	if work > maintenanceBatch {
	87	+	t.Fatalf("quarantine work in pass %d = %d, want at most %d", pass, work, maintenanceBatch)
	88	+	}
	89	+	if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
	90	+	return
	91	+	}
	92	+	}
	93	+	t.Fatal("deep and wide quarantine was not eventually removed")
	94	+	}

M internal/adapters/snapshot/maintenance_test.go +23 -12


@@ -350,10 +350,13 @@ func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
350	350		if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
351	351		t.Fatalf("quarantine after one bounded call = %v, want remaining tree", err)
352	352		}
353		-	for i := 0; i < 4; i++ {
	353	+	for i := 0; i < 12; i++ {
354	354		if err := repo.Maintain(context.Background()); err != nil {
355	355		t.Fatal(err)
356	356		}
	357	+	if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
	358	+	return
	359	+	}
357	360		}
358	361		if _, err := os.Lstat(quarantine); !errors.Is(err, os.ErrNotExist) {
359	362		t.Fatalf("quarantine after resumed maintenance = %v, want removed", err)

@@ -408,8 +411,8 @@ func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) {
408	411		if err := repo.sweepSession(context.Background(), key, state, &budget); err != nil {
409	412		t.Fatal(err)
410	413		}
411		-	if got := len(state.sweepQueue); got > maintenanceBatch {
412		-	t.Fatalf("queued shard names = %d, want at most %d", got, maintenanceBatch)
	414	+	if state.sweepShard != "" && len(repo.maintenanceCursors) > maxMaintenanceCursors {
	415	+	t.Fatalf("retained maintenance cursors = %d, want at most %d", len(repo.maintenanceCursors), maxMaintenanceCursors)
413	416		}
414	417		}
415	418		}

@@ -594,23 +597,31 @@ func TestRepositoryMaintainQueuesFetchedSessionsPastWorkBudget(t *testing.T) {
594	597		t.Fatal(err)
595	598		}
596	599		queued := fetched[len(fetched)-1].Name()
597		-	if err := repo.Maintain(context.Background()); err != nil {
598		-	t.Fatal(err)
599		-	}
600		-	if err := repo.Maintain(context.Background()); err != nil {
601		-	t.Fatal(err)
602		-	}
603		-	if _, err := os.Lstat(filepath.Join(sessions, queued)); !errors.Is(err, os.ErrNotExist) {
604		-	t.Fatalf("fetched but unprocessed session %q = %v, want removed on next call", queued, err)
	600	+	for i := 0; i < maintenanceBatch+4; i++ {
	601	+	if err := repo.Maintain(context.Background()); err != nil {
	602	+	t.Fatal(err)
	603	+	}
	604	+	if _, err := os.Lstat(filepath.Join(sessions, queued)); errors.Is(err, os.ErrNotExist) {
	605	+	break
	606	+	} else if i == maintenanceBatch+3 {
	607	+	t.Fatalf("fetched but unprocessed session %q = %v, want eventually removed", queued, err)
	608	+	}
605	609		}
606	610		if _, err := os.Lstat(filepath.Join(sessions, unread[0].Name())); err != nil {
607	611		t.Fatalf("unread session %q = %v, want queued entry to run first", unread[0].Name(), err)
608	612		}
609	613
610		-	for i := 0; i < 2; i++ {
	614	+	for i := 0; i < maintenanceBatch+4; i++ {
611	615		if err := repo.Maintain(context.Background()); err != nil {
612	616		t.Fatal(err)
613	617		}
	618	+	entries, err := os.ReadDir(sessions)
	619	+	if err != nil {
	620	+	t.Fatal(err)
	621	+	}
	622	+	if len(entries) == 0 {
	623	+	break
	624	+	}
614	625		}
615	626		entries, err := os.ReadDir(sessions)
616	627		if err != nil {

 	if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
 		t.Fatalf("quarantine after one bounded call = %v, want remaining tree", err)
 	}
-	for i := 0; i < 4; i++ {
+	for i := 0; i < 12; i++ {
 		if err := repo.Maintain(context.Background()); err != nil {
 			t.Fatal(err)
 		}
+		if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
+			return
+		}
 	}
 	if _, err := os.Lstat(quarantine); !errors.Is(err, os.ErrNotExist) {
 		t.Fatalf("quarantine after resumed maintenance = %v, want removed", err)
 		if err := repo.sweepSession(context.Background(), key, state, &budget); err != nil {
 			t.Fatal(err)
 		}
-		if got := len(state.sweepQueue); got > maintenanceBatch {
+		if state.sweepShard != "" && len(repo.maintenanceCursors) > maxMaintenanceCursors {
-			t.Fatalf("queued shard names = %d, want at most %d", got, maintenanceBatch)
+			t.Fatalf("retained maintenance cursors = %d, want at most %d", len(repo.maintenanceCursors), maxMaintenanceCursors)
 		}
 	}
 }
 		t.Fatal(err)
 	}
 	queued := fetched[len(fetched)-1].Name()
-	if err := repo.Maintain(context.Background()); err != nil {
+	for i := 0; i < maintenanceBatch+4; i++ {
-		t.Fatal(err)
+		if err := repo.Maintain(context.Background()); err != nil {
-	}
+			t.Fatal(err)
-	if err := repo.Maintain(context.Background()); err != nil {
+		}
-		t.Fatal(err)
+		if _, err := os.Lstat(filepath.Join(sessions, queued)); errors.Is(err, os.ErrNotExist) {
-	}
+			break
-	if _, err := os.Lstat(filepath.Join(sessions, queued)); !errors.Is(err, os.ErrNotExist) {
+		} else if i == maintenanceBatch+3 {
-		t.Fatalf("fetched but unprocessed session %q = %v, want removed on next call", queued, err)
+			t.Fatalf("fetched but unprocessed session %q = %v, want eventually removed", queued, err)
+		}
 	}
 	if _, err := os.Lstat(filepath.Join(sessions, unread[0].Name())); err != nil {
 		t.Fatalf("unread session %q = %v, want queued entry to run first", unread[0].Name(), err)
 	}
-	for i := 0; i < 2; i++ {
+	for i := 0; i < maintenanceBatch+4; i++ {
 		if err := repo.Maintain(context.Background()); err != nil {
 			t.Fatal(err)
 		}
+		entries, err := os.ReadDir(sessions)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(entries) == 0 {
+			break
+		}
 	}
 	entries, err := os.ReadDir(sessions)
 	if err != nil {

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


@@ -36,9 +36,10 @@ type Repository struct {
36	36		// maintenanceMu owns bounded continuation metadata: seek cookies and
37	37		// pending directory entries. Directory descriptors are opened and closed
38	38		// for each maintenance call.
39		-	maintenanceMu sync.Mutex
40		-	maintenanceCursors map[string]*maintenanceCursor
41		-	maintenanceSessions map[string]*sessionMaintenance
	39	+	maintenanceMu sync.Mutex
	40	+	maintenanceCursors map[string]*maintenanceCursor
	41	+	maintenanceSessions map[string]*sessionMaintenance
	42	+	maintenanceQuarantine *quarantineMaintenance
42	43
43	44		// pendingLegacySync records an unlink whose root-directory sync failed.
44	45		// It is keyed by the deterministic legacy filename and shares a per-file

@@ -69,8 +70,11 @@ type repositoryHooks struct {
69	70		syncDirectory func(string) error
70	71		remove func(string) error
71	72		openMaintenanceDirectory func(string) (maintenanceDirectory, error)
72		-	openLegacyDirectory func(string) (legacyDirectory, error)
73		-	beforeSessionLock func(string)
	73	+	// beforeMaintenanceWork observes each budgeted quarantine filesystem step.
	74	+	// It is test-only instrumentation for hostile traversal bounds.
	75	+	beforeMaintenanceWork func(string)
	76	+	openLegacyDirectory func(string) (legacyDirectory, error)
	77	+	beforeSessionLock func(string)
74	78		}
75	79
76	80		var _ ports.SnapshotRepository = (*Repository)(nil)

 	// maintenanceMu owns bounded continuation metadata: seek cookies and
 	// pending directory entries. Directory descriptors are opened and closed
 	// for each maintenance call.
-	maintenanceMu       sync.Mutex
+	maintenanceMu         sync.Mutex
-	maintenanceCursors  map[string]*maintenanceCursor
+	maintenanceCursors    map[string]*maintenanceCursor
-	maintenanceSessions map[string]*sessionMaintenance
+	maintenanceSessions   map[string]*sessionMaintenance
+	maintenanceQuarantine *quarantineMaintenance
 	// pendingLegacySync records an unlink whose root-directory sync failed.
 	// It is keyed by the deterministic legacy filename and shares a per-file
 	syncDirectory                func(string) error
 	remove                       func(string) error
 	openMaintenanceDirectory     func(string) (maintenanceDirectory, error)
-	openLegacyDirectory          func(string) (legacyDirectory, error)
+	// beforeMaintenanceWork observes each budgeted quarantine filesystem step.
-	beforeSessionLock            func(string)
+	// It is test-only instrumentation for hostile traversal bounds.
+	beforeMaintenanceWork func(string)
+	openLegacyDirectory   func(string) (legacyDirectory, error)
+	beforeSessionLock     func(string)
 }
 var _ ports.SnapshotRepository = (*Repository)(nil)