refactor(snapshot): bound maintenance cursors

commit: 11815fb3213db1e72125bafe8eadfe1bf73ccb6c
author: brice · 3mo ago · 2026-07-21 21:26 UTC
parent: 3565447
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +256 -78

M internal/adapters/snapshot/maintenance.go +114 -57


@@ -9,7 +9,9 @@ import (
9	9		"os"
10	10		"path/filepath"
11	11		"strings"
	12	+	"syscall"
12	13		"time"
	14	+	"unsafe"
13	15
14	16		"github.com/bnema/vev/internal/ports"
15	17		codec "github.com/bnema/vev/internal/usecase/snapshot"

@@ -152,8 +154,8 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
152	154		if err := ctx.Err(); err != nil {
153	155		return err
154	156		}
155		-	path := filepath.Join(sessions, entry.Name())
156		-	if isQuarantine(entry.Name()) {
	157	+	path := filepath.Join(sessions, entry.name)
	158	+	if isQuarantine(entry.name) {
157	159		changed, err := r.removeTreeBatch(ctx, path, &budget)
158	160		if err != nil && !errors.Is(err, os.ErrNotExist) {
159	161		return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))

@@ -168,12 +170,12 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
168	170		}
169	171		continue
170	172		}
171		-	if !entry.IsDir() \|\| !canonicalSessionKey(entry.Name()) {
	173	+	if !entry.isDir \|\| !canonicalSessionKey(entry.name) {
172	174		continue
173	175		}
174		-	lock := r.sessionLock(entry.Name())
	176	+	lock := r.sessionLock(entry.name)
175	177		lock.Lock()
176		-	err = r.maintainSession(ctx, entry.Name(), &budget)
	178	+	err = r.maintainSession(ctx, entry.name, &budget)
177	179		lock.Unlock()
178	180		if err != nil {
179	181		return err

@@ -182,9 +184,21 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
182	184		return nil
183	185		}
184	186
	187	+	// maintenanceCursor is deliberately reopenable metadata, not an open file.
	188	+	// On Linux, d_off is the seek cookie returned by getdents(2); seeking a newly
	189	+	// opened descriptor to it resumes the directory stream. Go's File.ReadDir
	190	+	// cannot be used here: its 8 KiB internal getdents buffer advances the kernel
	191	+	// offset beyond a short ReadDir result (see os/dir_unix.go), so saving SeekCur
	192	+	// after ReadDir would skip buffered entries.
185	193		type maintenanceCursor struct {
186		-	file *os.File
187		-	pending []os.DirEntry
	194	+	offset int64
	195	+	pending []maintenanceDirEntry
	196	+	done bool
	197	+	}
	198	+
	199	+	type maintenanceDirEntry struct {
	200	+	name string
	201	+	isDir bool
188	202		}
189	203
190	204		type manifestMaintenance struct {

@@ -205,68 +219,111 @@ type sessionMaintenance struct {
205	219		}
206	220
207	221		func (r *Repository) resetMaintenance() {
208		-	for _, cursor := range r.maintenanceCursors {
209		-	if cursor.file != nil {
210		-	_ = cursor.file.Close()
211		-	}
212		-	}
213	222		r.maintenanceCursors = nil
214	223		r.maintenanceSessions = nil
215	224		}
216	225
217	226		func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
218	227
219		-	// readMaintenanceDir advances one named directory cursor. Callers must return
220		-	// entries they could not process with requeueMaintenanceEntries before yielding
221		-	// for budget exhaustion. This prevents a ReadDir batch from skipping work when
222		-	// an earlier operation spends the remainder of the pass budget.
223		-	func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {
	228	+	// readMaintenanceDir advances one named directory cursor. Its continuation is
	229	+	// only a Linux seek cookie and at most one call's worth of pending names. The
	230	+	// descriptor used to read a call is always closed before this method returns.
	231	+	// Callers must return entries they could not process with
	232	+	// requeueMaintenanceEntries before yielding for budget exhaustion.
	233	+	func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
224	234		id := maintenanceCursorID(dir, purpose)
225	235		cursor := r.maintenanceCursors[id]
226	236		if cursor != nil && len(cursor.pending) != 0 {
227	237		entries := cursor.pending
228	238		cursor.pending = nil
229		-	return entries, cursor.file == nil, nil
230		-	}
231		-	if cursor != nil && cursor.file == nil {
232		-	delete(r.maintenanceCursors, id)
233		-	cursor = nil
	239	+	if cursor.done {
	240	+	delete(r.maintenanceCursors, id)
	241	+	}
	242	+	return entries, cursor.done, nil
234	243		}
235	244		if cursor == nil {
236		-	file, err := os.Open(dir)
237		-	if err != nil {
238		-	return nil, false, err
239		-	}
240		-	cursor = &maintenanceCursor{file: file}
	245	+	cursor = &maintenanceCursor{}
241	246		r.maintenanceCursors[id] = cursor
242	247		}
243		-	entries, err := cursor.file.ReadDir(n)
244		-	if err != nil && !errors.Is(err, io.EOF) {
	248	+
	249	+	entries, done, err := readMaintenanceDirent(dir, n, cursor)
	250	+	if err != nil {
245	251		delete(r.maintenanceCursors, id)
246		-	closeErr := cursor.file.Close()
247		-	if closeErr != nil {
248		-	return nil, false, closeErr
249		-	}
250	252		return nil, false, err
251	253		}
252		-	if !errors.Is(err, io.EOF) && len(entries) == n {
253		-	return entries, false, nil
	254	+	if done {
	255	+	delete(r.maintenanceCursors, id)
254	256		}
255		-	delete(r.maintenanceCursors, id)
256		-	if closeErr := cursor.file.Close(); closeErr != nil {
257		-	return nil, false, closeErr
	257	+	return entries, done, nil
	258	+	}
	259	+
	260	+	// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
	261	+	// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
	262	+	// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
	263	+	func readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) ([]maintenanceDirEntry, bool, error) {
	264	+	file, err := os.Open(dir)
	265	+	if err != nil {
	266	+	return nil, false, err
258	267		}
259		-	return entries, true, nil
	268	+	defer file.Close()
	269	+	fd := int(file.Fd())
	270	+	if cursor.offset != 0 {
	271	+	if _, err := syscall.Seek(fd, cursor.offset, io.SeekStart); err != nil {
	272	+	return nil, false, err
	273	+	}
	274	+	}
	275	+
	276	+	entries := make([]maintenanceDirEntry, 0, limit)
	277	+	buffer := make([]byte, 8192)
	278	+	for len(entries) < limit {
	279	+	count, err := syscall.ReadDirent(fd, buffer)
	280	+	if err != nil {
	281	+	return nil, false, err
	282	+	}
	283	+	if count == 0 {
	284	+	return entries, true, nil
	285	+	}
	286	+	for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
	287	+	if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
	288	+	return nil, false, syscall.EIO
	289	+	}
	290	+	record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
	291	+	reclen := int(record.Reclen)
	292	+	if reclen <= 0 \|\| reclen > len(data) {
	293	+	return nil, false, syscall.EIO
	294	+	}
	295	+	data = data[reclen:]
	296	+	// Advance past every raw record, including dot and disappeared entries.
	297	+	cursor.offset = record.Off
	298	+	nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
	299	+	if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
	300	+	nameBytes = nameBytes[:end]
	301	+	}
	302	+	name := string(nameBytes)
	303	+	if name == "." \|\| name == ".." \|\| name == "" {
	304	+	continue
	305	+	}
	306	+	info, err := os.Lstat(filepath.Join(dir, name))
	307	+	if errors.Is(err, os.ErrNotExist) {
	308	+	continue
	309	+	}
	310	+	if err != nil {
	311	+	return nil, false, err
	312	+	}
	313	+	entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
	314	+	}
	315	+	}
	316	+	return entries, false, nil
260	317		}
261	318
262		-	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []os.DirEntry) {
	319	+	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
263	320		if len(entries) == 0 {
264	321		return
265	322		}
266	323		id := maintenanceCursorID(dir, purpose)
267	324		cursor := r.maintenanceCursors[id]
268	325		if cursor == nil {
269		-	cursor = &maintenanceCursor{}
	326	+	cursor = &maintenanceCursor{done: true}
270	327		r.maintenanceCursors[id] = cursor
271	328		}
272	329		cursor.pending = append(entries, cursor.pending...)

@@ -367,8 +424,8 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
367	424		r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
368	425		return nil
369	426		}
370		-	if shard.IsDir() {
371		-	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget, "object-temps:"+key+":"+shard.Name()); err != nil {
	427	+	if shard.isDir {
	428	+	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
372	429		return err
373	430		}
374	431		}

@@ -445,11 +502,8 @@ func (r *Repository) maintenanceToken(key string) (string, bool, error) {
445	502		func (r *Repository) clearSessionMaintenance(key string) {
446	503		delete(r.maintenanceSessions, key)
447	504		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
448		-	for id, cursor := range r.maintenanceCursors {
	505	+	for id := range r.maintenanceCursors {
449	506		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {
450		-	if cursor.file != nil {
451		-	_ = cursor.file.Close()
452		-	}
453	507		delete(r.maintenanceCursors, id)
454	508		}
455	509		}

@@ -469,10 +523,10 @@ func (r Repository) markSession(ctx context.Context, key string, state session
469	523		if err := ctx.Err(); err != nil {
470	524		return err
471	525		}
472		-	if entry.IsDir() {
	526	+	if entry.isDir {
473	527		continue
474	528		}
475		-	generation, ok := parseGenerationFilename(entry.Name())
	529	+	generation, ok := parseGenerationFilename(entry.name)
476	530		if !ok {
477	531		continue
478	532		}

@@ -561,7 +615,10 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
561	615		}
562	616		referenced := retainedReferences(state.marked, state.conservative)
563	617		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
564		-	if !state.sweepRootDone {
	618	+	// Do not keep accumulating shard names while draining earlier work. This
	619	+	// bounds retained names to one directory batch even for a session with many
	620	+	// object shards.
	621	+	if !state.sweepRootDone && len(state.sweepQueue) == 0 {
565	622		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
566	623		if errors.Is(err, os.ErrNotExist) {
567	624		state.sweepRootDone = true

@@ -569,8 +626,8 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
569	626		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
570	627		} else {
571	628		for _, entry := range entries {
572		-	if entry.IsDir() {
573		-	state.sweepQueue = append(state.sweepQueue, entry.Name())
	629	+	if entry.isDir {
	630	+	state.sweepQueue = append(state.sweepQueue, entry.name)
574	631		}
575	632		}
576	633		state.sweepRootDone = done

@@ -649,14 +706,14 @@ func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[
649	706		if err := ctx.Err(); err != nil {
650	707		return false, err
651	708		}
652		-	digest, ok := parseObjectDigest(object.Name())
653		-	if !ok \|\| object.IsDir() {
	709	+	digest, ok := parseObjectDigest(object.name)
	710	+	if !ok \|\| object.isDir {
654	711		continue
655	712		}
656	713		if _, used := referenced[digest]; used {
657	714		continue
658	715		}
659		-	path := filepath.Join(dir, object.Name())
	716	+	path := filepath.Join(dir, object.name)
660	717		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
661	718		return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
662	719		}

@@ -690,10 +747,10 @@ func (r Repository) removeTemps(ctx context.Context, dir string, budget int, p
690	747		if err := ctx.Err(); err != nil {
691	748		return err
692	749		}
693		-	if entry.IsDir() \|\| !strings.HasPrefix(entry.Name(), ".tmp-") {
	750	+	if entry.isDir \|\| !strings.HasPrefix(entry.name, ".tmp-") {
694	751		continue
695	752		}
696		-	path := filepath.Join(dir, entry.Name())
	753	+	path := filepath.Join(dir, entry.name)
697	754		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
698	755		return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
699	756		}


@@ -9,7 +9,9 @@ import (
9		"os"	9		"os"
10		"path/filepath"	10		"path/filepath"
11		"strings"	11		"strings"
			12	+	"syscall"
12		"time"	13		"time"
			14	+	"unsafe"
13			15
14		"github.com/bnema/vev/internal/ports"	16		"github.com/bnema/vev/internal/ports"
15		codec "github.com/bnema/vev/internal/usecase/snapshot"	17		codec "github.com/bnema/vev/internal/usecase/snapshot"

@@ -152,8 +154,8 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
152		if err := ctx.Err(); err != nil {	154		if err := ctx.Err(); err != nil {
153		return err	155		return err
154		}	156		}
155	-	path := filepath.Join(sessions, entry.Name())	157	+	path := filepath.Join(sessions, entry.name)
156	-	if isQuarantine(entry.Name()) {	158	+	if isQuarantine(entry.name) {
157		changed, err := r.removeTreeBatch(ctx, path, &budget)	159		changed, err := r.removeTreeBatch(ctx, path, &budget)
158		if err != nil && !errors.Is(err, os.ErrNotExist) {	160		if err != nil && !errors.Is(err, os.ErrNotExist) {
159		return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))	161		return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))

@@ -168,12 +170,12 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
168		}	170		}
169		continue	171		continue
170		}	172		}
171	-	if !entry.IsDir() \|\| !canonicalSessionKey(entry.Name()) {	173	+	if !entry.isDir \|\| !canonicalSessionKey(entry.name) {
172		continue	174		continue
173		}	175		}
174	-	lock := r.sessionLock(entry.Name())	176	+	lock := r.sessionLock(entry.name)
175		lock.Lock()	177		lock.Lock()
176	-	err = r.maintainSession(ctx, entry.Name(), &budget)	178	+	err = r.maintainSession(ctx, entry.name, &budget)
177		lock.Unlock()	179		lock.Unlock()
178		if err != nil {	180		if err != nil {
179		return err	181		return err

@@ -182,9 +184,21 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
182		return nil	184		return nil
183		}	185		}
184			186
			187	+	// maintenanceCursor is deliberately reopenable metadata, not an open file.
			188	+	// On Linux, d_off is the seek cookie returned by getdents(2); seeking a newly
			189	+	// opened descriptor to it resumes the directory stream. Go's File.ReadDir
			190	+	// cannot be used here: its 8 KiB internal getdents buffer advances the kernel
			191	+	// offset beyond a short ReadDir result (see os/dir_unix.go), so saving SeekCur
			192	+	// after ReadDir would skip buffered entries.
185		type maintenanceCursor struct {	193		type maintenanceCursor struct {
186	-	file *os.File	194	+	offset int64
187	-	pending []os.DirEntry	195	+	pending []maintenanceDirEntry
			196	+	done bool
			197	+	}
			198	+
			199	+	type maintenanceDirEntry struct {
			200	+	name string
			201	+	isDir bool
188		}	202		}
189			203
190		type manifestMaintenance struct {	204		type manifestMaintenance struct {

@@ -205,68 +219,111 @@ type sessionMaintenance struct {
205		}	219		}
206			220
207		func (r *Repository) resetMaintenance() {	221		func (r *Repository) resetMaintenance() {
208	-	for _, cursor := range r.maintenanceCursors {
209	-	if cursor.file != nil {
210	-	_ = cursor.file.Close()
211	-	}
212	-	}
213		r.maintenanceCursors = nil	222		r.maintenanceCursors = nil
214		r.maintenanceSessions = nil	223		r.maintenanceSessions = nil
215		}	224		}
216			225
217		func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }	226		func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
218			227
219	-	// readMaintenanceDir advances one named directory cursor. Callers must return	228	+	// readMaintenanceDir advances one named directory cursor. Its continuation is
220	-	// entries they could not process with requeueMaintenanceEntries before yielding	229	+	// only a Linux seek cookie and at most one call's worth of pending names. The
221	-	// for budget exhaustion. This prevents a ReadDir batch from skipping work when	230	+	// descriptor used to read a call is always closed before this method returns.
222	-	// an earlier operation spends the remainder of the pass budget.	231	+	// Callers must return entries they could not process with
223	-	func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]os.DirEntry, bool, error) {	232	+	// requeueMaintenanceEntries before yielding for budget exhaustion.
			233	+	func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
224		id := maintenanceCursorID(dir, purpose)	234		id := maintenanceCursorID(dir, purpose)
225		cursor := r.maintenanceCursors[id]	235		cursor := r.maintenanceCursors[id]
226		if cursor != nil && len(cursor.pending) != 0 {	236		if cursor != nil && len(cursor.pending) != 0 {
227		entries := cursor.pending	237		entries := cursor.pending
228		cursor.pending = nil	238		cursor.pending = nil
229	-	return entries, cursor.file == nil, nil	239	+	if cursor.done {
230	-	}	240	+	delete(r.maintenanceCursors, id)
231	-	if cursor != nil && cursor.file == nil {	241	+	}
232	-	delete(r.maintenanceCursors, id)	242	+	return entries, cursor.done, nil
233	-	cursor = nil
234		}	243		}
235		if cursor == nil {	244		if cursor == nil {
236	-	file, err := os.Open(dir)	245	+	cursor = &maintenanceCursor{}
237	-	if err != nil {
238	-	return nil, false, err
239	-	}
240	-	cursor = &maintenanceCursor{file: file}
241		r.maintenanceCursors[id] = cursor	246		r.maintenanceCursors[id] = cursor
242		}	247		}
243	-	entries, err := cursor.file.ReadDir(n)	248	+
244	-	if err != nil && !errors.Is(err, io.EOF) {	249	+	entries, done, err := readMaintenanceDirent(dir, n, cursor)
			250	+	if err != nil {
245		delete(r.maintenanceCursors, id)	251		delete(r.maintenanceCursors, id)
246	-	closeErr := cursor.file.Close()
247	-	if closeErr != nil {
248	-	return nil, false, closeErr
249	-	}
250		return nil, false, err	252		return nil, false, err
251		}	253		}
252	-	if !errors.Is(err, io.EOF) && len(entries) == n {	254	+	if done {
253	-	return entries, false, nil	255	+	delete(r.maintenanceCursors, id)
254		}	256		}
255	-	delete(r.maintenanceCursors, id)	257	+	return entries, done, nil
256	-	if closeErr := cursor.file.Close(); closeErr != nil {	258	+	}
257	-	return nil, false, closeErr	259	+
			260	+	// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
			261	+	// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
			262	+	// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
			263	+	func readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) ([]maintenanceDirEntry, bool, error) {
			264	+	file, err := os.Open(dir)
			265	+	if err != nil {
			266	+	return nil, false, err
258		}	267		}
259	-	return entries, true, nil	268	+	defer file.Close()
			269	+	fd := int(file.Fd())
			270	+	if cursor.offset != 0 {
			271	+	if _, err := syscall.Seek(fd, cursor.offset, io.SeekStart); err != nil {
			272	+	return nil, false, err
			273	+	}
			274	+	}
			275	+
			276	+	entries := make([]maintenanceDirEntry, 0, limit)
			277	+	buffer := make([]byte, 8192)
			278	+	for len(entries) < limit {
			279	+	count, err := syscall.ReadDirent(fd, buffer)
			280	+	if err != nil {
			281	+	return nil, false, err
			282	+	}
			283	+	if count == 0 {
			284	+	return entries, true, nil
			285	+	}
			286	+	for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
			287	+	if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
			288	+	return nil, false, syscall.EIO
			289	+	}
			290	+	record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
			291	+	reclen := int(record.Reclen)
			292	+	if reclen <= 0 \|\| reclen > len(data) {
			293	+	return nil, false, syscall.EIO
			294	+	}
			295	+	data = data[reclen:]
			296	+	// Advance past every raw record, including dot and disappeared entries.
			297	+	cursor.offset = record.Off
			298	+	nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
			299	+	if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
			300	+	nameBytes = nameBytes[:end]
			301	+	}
			302	+	name := string(nameBytes)
			303	+	if name == "." \|\| name == ".." \|\| name == "" {
			304	+	continue
			305	+	}
			306	+	info, err := os.Lstat(filepath.Join(dir, name))
			307	+	if errors.Is(err, os.ErrNotExist) {
			308	+	continue
			309	+	}
			310	+	if err != nil {
			311	+	return nil, false, err
			312	+	}
			313	+	entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
			314	+	}
			315	+	}
			316	+	return entries, false, nil
260		}	317		}
261			318
262	-	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []os.DirEntry) {	319	+	func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
263		if len(entries) == 0 {	320		if len(entries) == 0 {
264		return	321		return
265		}	322		}
266		id := maintenanceCursorID(dir, purpose)	323		id := maintenanceCursorID(dir, purpose)
267		cursor := r.maintenanceCursors[id]	324		cursor := r.maintenanceCursors[id]
268		if cursor == nil {	325		if cursor == nil {
269	-	cursor = &maintenanceCursor{}	326	+	cursor = &maintenanceCursor{done: true}
270		r.maintenanceCursors[id] = cursor	327		r.maintenanceCursors[id] = cursor
271		}	328		}
272		cursor.pending = append(entries, cursor.pending...)	329		cursor.pending = append(entries, cursor.pending...)

@@ -367,8 +424,8 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
367		r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])	424		r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
368		return nil	425		return nil
369		}	426		}
370	-	if shard.IsDir() {	427	+	if shard.isDir {
371	-	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.Name()), budget, "object-temps:"+key+":"+shard.Name()); err != nil {	428	+	if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
372		return err	429		return err
373		}	430		}
374		}	431		}

@@ -445,11 +502,8 @@ func (r *Repository) maintenanceToken(key string) (string, bool, error) {
445		func (r *Repository) clearSessionMaintenance(key string) {	502		func (r *Repository) clearSessionMaintenance(key string) {
446		delete(r.maintenanceSessions, key)	503		delete(r.maintenanceSessions, key)
447		prefix := "\x00" + filepath.Clean(r.sessionPath(key))	504		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
448	-	for id, cursor := range r.maintenanceCursors {	505	+	for id := range r.maintenanceCursors {
449		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {	506		if strings.HasSuffix(id, prefix) \|\| strings.Contains(id, ":"+key+":") \|\| strings.Contains(id, ":"+key+"\x00") {
450	-	if cursor.file != nil {
451	-	_ = cursor.file.Close()
452	-	}
453		delete(r.maintenanceCursors, id)	507		delete(r.maintenanceCursors, id)
454		}	508		}
455		}	509		}

@@ -469,10 +523,10 @@ func (r Repository) markSession(ctx context.Context, key string, state session
469		if err := ctx.Err(); err != nil {	523		if err := ctx.Err(); err != nil {
470		return err	524		return err
471		}	525		}
472	-	if entry.IsDir() {	526	+	if entry.isDir {
473		continue	527		continue
474		}	528		}
475	-	generation, ok := parseGenerationFilename(entry.Name())	529	+	generation, ok := parseGenerationFilename(entry.name)
476		if !ok {	530		if !ok {
477		continue	531		continue
478		}	532		}

@@ -561,7 +615,10 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
561		}	615		}
562		referenced := retainedReferences(state.marked, state.conservative)	616		referenced := retainedReferences(state.marked, state.conservative)
563		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)	617		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
564	-	if !state.sweepRootDone {	618	+	// Do not keep accumulating shard names while draining earlier work. This
			619	+	// bounds retained names to one directory batch even for a session with many
			620	+	// object shards.
			621	+	if !state.sweepRootDone && len(state.sweepQueue) == 0 {
565		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)	622		entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
566		if errors.Is(err, os.ErrNotExist) {	623		if errors.Is(err, os.ErrNotExist) {
567		state.sweepRootDone = true	624		state.sweepRootDone = true

@@ -569,8 +626,8 @@ func (r Repository) sweepSession(ctx context.Context, key string, state sessio
569		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))	626		return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
570		} else {	627		} else {
571		for _, entry := range entries {	628		for _, entry := range entries {
572	-	if entry.IsDir() {	629	+	if entry.isDir {
573	-	state.sweepQueue = append(state.sweepQueue, entry.Name())	630	+	state.sweepQueue = append(state.sweepQueue, entry.name)
574		}	631		}
575		}	632		}
576		state.sweepRootDone = done	633		state.sweepRootDone = done

@@ -649,14 +706,14 @@ func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[
649		if err := ctx.Err(); err != nil {	706		if err := ctx.Err(); err != nil {
650		return false, err	707		return false, err
651		}	708		}
652	-	digest, ok := parseObjectDigest(object.Name())	709	+	digest, ok := parseObjectDigest(object.name)
653	-	if !ok \|\| object.IsDir() {	710	+	if !ok \|\| object.isDir {
654		continue	711		continue
655		}	712		}
656		if _, used := referenced[digest]; used {	713		if _, used := referenced[digest]; used {
657		continue	714		continue
658		}	715		}
659	-	path := filepath.Join(dir, object.Name())	716	+	path := filepath.Join(dir, object.name)
660		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {	717		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
661		return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))	718		return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
662		}	719		}

@@ -690,10 +747,10 @@ func (r Repository) removeTemps(ctx context.Context, dir string, budget int, p
690		if err := ctx.Err(); err != nil {	747		if err := ctx.Err(); err != nil {
691		return err	748		return err
692		}	749		}
693	-	if entry.IsDir() \|\| !strings.HasPrefix(entry.Name(), ".tmp-") {	750	+	if entry.isDir \|\| !strings.HasPrefix(entry.name, ".tmp-") {
694		continue	751		continue
695		}	752		}
696	-	path := filepath.Join(dir, entry.Name())	753	+	path := filepath.Join(dir, entry.name)
697		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {	754		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
698		return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))	755		return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
699		}	756		}

M internal/adapters/snapshot/maintenance_test.go +120 -17


@@ -77,13 +77,14 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) {
77	77		publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second")))
78	78		}()
79	79		<-entered
	80	+	deleteReached := make(chan struct{})
	81	+	deleteRelease := make(chan struct{})
	82	+	repo.hooks.beforeSessionLock = waitAtSessionLock(deleteReached, deleteRelease)
80	83		deleteDone := make(chan error, 1)
81	84		go func() { deleteDone <- repo.Delete(context.Background(), "named") }()
82		-	select {
83		-	case err := <-deleteDone:
84		-	t.Fatalf("Delete returned before Publish released its lock: %v", err)
85		-	default:
86		-	}
	85	+	<-deleteReached
	86	+	// Delete is parked at the session-lock boundary while Publish owns it.
	87	+	close(deleteRelease)
87	88		close(release)
88	89		if err := <-publishDone; err != nil {
89	90		t.Fatal(err)

@@ -152,14 +153,14 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
152	153		lock := repo.sessionLock(key)
153	154		lock.Lock()
154	155		ctx, cancel := context.WithCancel(context.Background())
155		-	started := make(chan struct{})
	156	+	reached := make(chan struct{})
	157	+	release := make(chan struct{})
	158	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
156	159		done := make(chan error, 1)
157		-	go func() {
158		-	close(started)
159		-	done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state")))
160		-	}()
161		-	<-started
	160	+	go func() { done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state"))) }()
	161	+	<-reached
162	162		cancel()
	163	+	close(release)
163	164		lock.Unlock()
164	165		if err := <-done; !errors.Is(err, context.Canceled) {
165	166		t.Fatalf("Publish error = %v, want canceled", err)

@@ -185,15 +186,19 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
185	186		lock := repo.sessionLock(key)
186	187		lock.Lock()
187	188		ctx, cancel := context.WithCancel(context.Background())
188		-	started := make(chan struct{})
	189	+	reached := make(chan struct{})
	190	+	release := make(chan struct{})
	191	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
189	192		done := make(chan error, 1)
190		-	go func() { close(started); done <- operation.run(ctx) }()
191		-	<-started
	193	+	go func() { done <- operation.run(ctx) }()
	194	+	<-reached
192	195		cancel()
	196	+	close(release)
193	197		lock.Unlock()
194	198		if err := <-done; !errors.Is(err, context.Canceled) {
195	199		t.Fatalf("%s error = %v, want canceled", operation.name, err)
196	200		}
	201	+	repo.hooks.beforeSessionLock = nil
197	202		if _, err := repo.Load(context.Background(), "named"); err != nil {
198	203		t.Fatalf("%s mutated session: %v", operation.name, err)
199	204		}

@@ -208,21 +213,32 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
208	213		lock := repo.sessionLock(sessionKey("named"))
209	214		lock.Lock()
210	215		ctx, cancel := context.WithCancel(context.Background())
211		-	started := make(chan struct{})
	216	+	reached := make(chan struct{})
	217	+	release := make(chan struct{})
	218	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
212	219		done := make(chan error, 1)
213		-	go func() { close(started); done <- repo.Maintain(ctx) }()
214		-	<-started
	220	+	go func() { done <- repo.Maintain(ctx) }()
	221	+	<-reached
215	222		cancel()
	223	+	close(release)
216	224		lock.Unlock()
217	225		if err := <-done; !errors.Is(err, context.Canceled) {
218	226		t.Fatalf("Maintain error = %v, want canceled", err)
219	227		}
	228	+	repo.hooks.beforeSessionLock = nil
220	229		if _, err := repo.Load(context.Background(), "named"); err != nil {
221	230		t.Fatalf("Maintain mutated session: %v", err)
222	231		}
223	232		})
224	233		}
225	234
	235	+	func waitAtSessionLock(reached chan<- struct{}, release <-chan struct{}) func(string) {
	236	+	return func(string) {
	237	+	close(reached)
	238	+	<-release
	239	+	}
	240	+	}
	241	+
226	242		func TestRepositoryDeleteRetriesPendingQuarantineSyncWithoutDeletingRecreatedSession(t *testing.T) {
227	243		repo := NewRepository(privateDir(t))
228	244		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("old"))); err != nil {

@@ -273,6 +289,93 @@ func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
273	289		}
274	290		}
275	291
	292	+	func TestRepositoryMaintainClosesContinuationDirectories(t *testing.T) {
	293	+	repo := NewRepository(privateDir(t))
	294	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
	295	+	for i := 0; i < maintenanceBatch+1; i++ {
	296	+	generations := filepath.Join(sessions, fmt.Sprintf("named-%03d", i), repositoryGenerations)
	297	+	if err := os.MkdirAll(generations, 0o700); err != nil {
	298	+	t.Fatal(err)
	299	+	}
	300	+	for j := 0; j < maintenanceBatch+1; j++ {
	301	+	if err := os.WriteFile(filepath.Join(generations, fmt.Sprintf("unclassified-%03d", j)), []byte("state"), 0o600); err != nil {
	302	+	t.Fatal(err)
	303	+	}
	304	+	}
	305	+	}
	306	+	before := openDescriptorCount(t)
	307	+	if err := repo.Maintain(context.Background()); err != nil {
	308	+	t.Fatal(err)
	309	+	}
	310	+	if got := openDescriptorCount(t); got > before+4 {
	311	+	t.Fatalf("open descriptors after bounded maintenance = %d, want at most %d", got, before+4)
	312	+	}
	313	+	}
	314	+
	315	+	func openDescriptorCount(t *testing.T) int {
	316	+	t.Helper()
	317	+	entries, err := os.ReadDir("/proc/self/fd")
	318	+	if err != nil {
	319	+	t.Fatal(err)
	320	+	}
	321	+	return len(entries)
	322	+	}
	323	+
	324	+	func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) {
	325	+	repo := NewRepository(privateDir(t))
	326	+	key := sessionKey("named")
	327	+	root := filepath.Join(repo.sessionPath(key), repositoryObjectsDir)
	328	+	for i := 0; i < maintenanceBatch*2; i++ {
	329	+	if err := os.MkdirAll(filepath.Join(root, fmt.Sprintf("%02x", i)), 0o700); err != nil {
	330	+	t.Fatal(err)
	331	+	}
	332	+	}
	333	+	repo.maintenanceCursors = make(map[string]*maintenanceCursor)
	334	+	state := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance), markDone: true}
	335	+	for i := 0; i < maintenanceBatch+2; i++ {
	336	+	budget := maintenanceBatch
	337	+	if err := repo.sweepSession(context.Background(), key, state, &budget); err != nil {
	338	+	t.Fatal(err)
	339	+	}
	340	+	if got := len(state.sweepQueue); got > maintenanceBatch {
	341	+	t.Fatalf("queued shard names = %d, want at most %d", got, maintenanceBatch)
	342	+	}
	343	+	}
	344	+	}
	345	+
	346	+	func TestRepositoryMaintainResumesCursorAfterDirectoryMutation(t *testing.T) {
	347	+	repo := NewRepository(privateDir(t))
	348	+	if err := os.Mkdir(repo.dir, 0o700); err != nil {
	349	+	t.Fatal(err)
	350	+	}
	351	+	for i := 0; i < maintenanceBatch+1; i++ {
	352	+	if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("live-%03d", i)), []byte("live"), 0o600); err != nil {
	353	+	t.Fatal(err)
	354	+	}
	355	+	}
	356	+	if err := repo.Maintain(context.Background()); err != nil {
	357	+	t.Fatal(err)
	358	+	}
	359	+	// Mutating entries ahead of the saved cookie must not reset traversal to
	360	+	// the prefix or lose a stale entry appended after that cookie.
	361	+	if err := os.Remove(filepath.Join(repo.dir, "live-000")); err != nil {
	362	+	t.Fatal(err)
	363	+	}
	364	+	stale := filepath.Join(repo.dir, ".tmp-appended")
	365	+	if err := os.WriteFile(stale, []byte("stale"), 0o600); err != nil {
	366	+	t.Fatal(err)
	367	+	}
	368	+	for i := 0; i < 4; i++ {
	369	+	if err := repo.Maintain(context.Background()); err != nil {
	370	+	t.Fatal(err)
	371	+	}
	372	+	if _, err := os.Lstat(stale); errors.Is(err, os.ErrNotExist) {
	373	+	return
	374	+	}
	375	+	}
	376	+	t.Fatal("stale entry appended after cursor mutation was not eventually removed")
	377	+	}
	378	+
276	379		func TestRepositoryMaintainDoesNotStarveLaterTemporaryEntries(t *testing.T) {
277	380		repo := NewRepository(privateDir(t))
278	381		if err := os.Mkdir(repo.dir, 0o700); err != nil {


@@ -77,13 +77,14 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) {
77		publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second")))	77		publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second")))
78		}()	78		}()
79		<-entered	79		<-entered
			80	+	deleteReached := make(chan struct{})
			81	+	deleteRelease := make(chan struct{})
			82	+	repo.hooks.beforeSessionLock = waitAtSessionLock(deleteReached, deleteRelease)
80		deleteDone := make(chan error, 1)	83		deleteDone := make(chan error, 1)
81		go func() { deleteDone <- repo.Delete(context.Background(), "named") }()	84		go func() { deleteDone <- repo.Delete(context.Background(), "named") }()
82	-	select {	85	+	<-deleteReached
83	-	case err := <-deleteDone:	86	+	// Delete is parked at the session-lock boundary while Publish owns it.
84	-	t.Fatalf("Delete returned before Publish released its lock: %v", err)	87	+	close(deleteRelease)
85	-	default:
86	-	}
87		close(release)	88		close(release)
88		if err := <-publishDone; err != nil {	89		if err := <-publishDone; err != nil {
89		t.Fatal(err)	90		t.Fatal(err)

@@ -152,14 +153,14 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
152		lock := repo.sessionLock(key)	153		lock := repo.sessionLock(key)
153		lock.Lock()	154		lock.Lock()
154		ctx, cancel := context.WithCancel(context.Background())	155		ctx, cancel := context.WithCancel(context.Background())
155	-	started := make(chan struct{})	156	+	reached := make(chan struct{})
			157	+	release := make(chan struct{})
			158	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
156		done := make(chan error, 1)	159		done := make(chan error, 1)
157	-	go func() {	160	+	go func() { done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state"))) }()
158	-	close(started)	161	+	<-reached
159	-	done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state")))
160	-	}()
161	-	<-started
162		cancel()	162		cancel()
			163	+	close(release)
163		lock.Unlock()	164		lock.Unlock()
164		if err := <-done; !errors.Is(err, context.Canceled) {	165		if err := <-done; !errors.Is(err, context.Canceled) {
165		t.Fatalf("Publish error = %v, want canceled", err)	166		t.Fatalf("Publish error = %v, want canceled", err)

@@ -185,15 +186,19 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
185		lock := repo.sessionLock(key)	186		lock := repo.sessionLock(key)
186		lock.Lock()	187		lock.Lock()
187		ctx, cancel := context.WithCancel(context.Background())	188		ctx, cancel := context.WithCancel(context.Background())
188	-	started := make(chan struct{})	189	+	reached := make(chan struct{})
			190	+	release := make(chan struct{})
			191	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
189		done := make(chan error, 1)	192		done := make(chan error, 1)
190	-	go func() { close(started); done <- operation.run(ctx) }()	193	+	go func() { done <- operation.run(ctx) }()
191	-	<-started	194	+	<-reached
192		cancel()	195		cancel()
			196	+	close(release)
193		lock.Unlock()	197		lock.Unlock()
194		if err := <-done; !errors.Is(err, context.Canceled) {	198		if err := <-done; !errors.Is(err, context.Canceled) {
195		t.Fatalf("%s error = %v, want canceled", operation.name, err)	199		t.Fatalf("%s error = %v, want canceled", operation.name, err)
196		}	200		}
			201	+	repo.hooks.beforeSessionLock = nil
197		if _, err := repo.Load(context.Background(), "named"); err != nil {	202		if _, err := repo.Load(context.Background(), "named"); err != nil {
198		t.Fatalf("%s mutated session: %v", operation.name, err)	203		t.Fatalf("%s mutated session: %v", operation.name, err)
199		}	204		}

@@ -208,21 +213,32 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
208		lock := repo.sessionLock(sessionKey("named"))	213		lock := repo.sessionLock(sessionKey("named"))
209		lock.Lock()	214		lock.Lock()
210		ctx, cancel := context.WithCancel(context.Background())	215		ctx, cancel := context.WithCancel(context.Background())
211	-	started := make(chan struct{})	216	+	reached := make(chan struct{})
			217	+	release := make(chan struct{})
			218	+	repo.hooks.beforeSessionLock = waitAtSessionLock(reached, release)
212		done := make(chan error, 1)	219		done := make(chan error, 1)
213	-	go func() { close(started); done <- repo.Maintain(ctx) }()	220	+	go func() { done <- repo.Maintain(ctx) }()
214	-	<-started	221	+	<-reached
215		cancel()	222		cancel()
			223	+	close(release)
216		lock.Unlock()	224		lock.Unlock()
217		if err := <-done; !errors.Is(err, context.Canceled) {	225		if err := <-done; !errors.Is(err, context.Canceled) {
218		t.Fatalf("Maintain error = %v, want canceled", err)	226		t.Fatalf("Maintain error = %v, want canceled", err)
219		}	227		}
			228	+	repo.hooks.beforeSessionLock = nil
220		if _, err := repo.Load(context.Background(), "named"); err != nil {	229		if _, err := repo.Load(context.Background(), "named"); err != nil {
221		t.Fatalf("Maintain mutated session: %v", err)	230		t.Fatalf("Maintain mutated session: %v", err)
222		}	231		}
223		})	232		})
224		}	233		}
225			234
			235	+	func waitAtSessionLock(reached chan<- struct{}, release <-chan struct{}) func(string) {
			236	+	return func(string) {
			237	+	close(reached)
			238	+	<-release
			239	+	}
			240	+	}
			241	+
226		func TestRepositoryDeleteRetriesPendingQuarantineSyncWithoutDeletingRecreatedSession(t *testing.T) {	242		func TestRepositoryDeleteRetriesPendingQuarantineSyncWithoutDeletingRecreatedSession(t *testing.T) {
227		repo := NewRepository(privateDir(t))	243		repo := NewRepository(privateDir(t))
228		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("old"))); err != nil {	244		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("old"))); err != nil {

@@ -273,6 +289,93 @@ func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
273		}	289		}
274		}	290		}
275			291
			292	+	func TestRepositoryMaintainClosesContinuationDirectories(t *testing.T) {
			293	+	repo := NewRepository(privateDir(t))
			294	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
			295	+	for i := 0; i < maintenanceBatch+1; i++ {
			296	+	generations := filepath.Join(sessions, fmt.Sprintf("named-%03d", i), repositoryGenerations)
			297	+	if err := os.MkdirAll(generations, 0o700); err != nil {
			298	+	t.Fatal(err)
			299	+	}
			300	+	for j := 0; j < maintenanceBatch+1; j++ {
			301	+	if err := os.WriteFile(filepath.Join(generations, fmt.Sprintf("unclassified-%03d", j)), []byte("state"), 0o600); err != nil {
			302	+	t.Fatal(err)
			303	+	}
			304	+	}
			305	+	}
			306	+	before := openDescriptorCount(t)
			307	+	if err := repo.Maintain(context.Background()); err != nil {
			308	+	t.Fatal(err)
			309	+	}
			310	+	if got := openDescriptorCount(t); got > before+4 {
			311	+	t.Fatalf("open descriptors after bounded maintenance = %d, want at most %d", got, before+4)
			312	+	}
			313	+	}
			314	+
			315	+	func openDescriptorCount(t *testing.T) int {
			316	+	t.Helper()
			317	+	entries, err := os.ReadDir("/proc/self/fd")
			318	+	if err != nil {
			319	+	t.Fatal(err)
			320	+	}
			321	+	return len(entries)
			322	+	}
			323	+
			324	+	func TestRepositoryMaintainBoundsQueuedShardNames(t *testing.T) {
			325	+	repo := NewRepository(privateDir(t))
			326	+	key := sessionKey("named")
			327	+	root := filepath.Join(repo.sessionPath(key), repositoryObjectsDir)
			328	+	for i := 0; i < maintenanceBatch*2; i++ {
			329	+	if err := os.MkdirAll(filepath.Join(root, fmt.Sprintf("%02x", i)), 0o700); err != nil {
			330	+	t.Fatal(err)
			331	+	}
			332	+	}
			333	+	repo.maintenanceCursors = make(map[string]*maintenanceCursor)
			334	+	state := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance), markDone: true}
			335	+	for i := 0; i < maintenanceBatch+2; i++ {
			336	+	budget := maintenanceBatch
			337	+	if err := repo.sweepSession(context.Background(), key, state, &budget); err != nil {
			338	+	t.Fatal(err)
			339	+	}
			340	+	if got := len(state.sweepQueue); got > maintenanceBatch {
			341	+	t.Fatalf("queued shard names = %d, want at most %d", got, maintenanceBatch)
			342	+	}
			343	+	}
			344	+	}
			345	+
			346	+	func TestRepositoryMaintainResumesCursorAfterDirectoryMutation(t *testing.T) {
			347	+	repo := NewRepository(privateDir(t))
			348	+	if err := os.Mkdir(repo.dir, 0o700); err != nil {
			349	+	t.Fatal(err)
			350	+	}
			351	+	for i := 0; i < maintenanceBatch+1; i++ {
			352	+	if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("live-%03d", i)), []byte("live"), 0o600); err != nil {
			353	+	t.Fatal(err)
			354	+	}
			355	+	}
			356	+	if err := repo.Maintain(context.Background()); err != nil {
			357	+	t.Fatal(err)
			358	+	}
			359	+	// Mutating entries ahead of the saved cookie must not reset traversal to
			360	+	// the prefix or lose a stale entry appended after that cookie.
			361	+	if err := os.Remove(filepath.Join(repo.dir, "live-000")); err != nil {
			362	+	t.Fatal(err)
			363	+	}
			364	+	stale := filepath.Join(repo.dir, ".tmp-appended")
			365	+	if err := os.WriteFile(stale, []byte("stale"), 0o600); err != nil {
			366	+	t.Fatal(err)
			367	+	}
			368	+	for i := 0; i < 4; i++ {
			369	+	if err := repo.Maintain(context.Background()); err != nil {
			370	+	t.Fatal(err)
			371	+	}
			372	+	if _, err := os.Lstat(stale); errors.Is(err, os.ErrNotExist) {
			373	+	return
			374	+	}
			375	+	}
			376	+	t.Fatal("stale entry appended after cursor mutation was not eventually removed")
			377	+	}
			378	+
276		func TestRepositoryMaintainDoesNotStarveLaterTemporaryEntries(t *testing.T) {	379		func TestRepositoryMaintainDoesNotStarveLaterTemporaryEntries(t *testing.T) {
277		repo := NewRepository(privateDir(t))	380		repo := NewRepository(privateDir(t))
278		if err := os.Mkdir(repo.dir, 0o700); err != nil {	381		if err := os.Mkdir(repo.dir, 0o700); err != nil {

M internal/adapters/snapshot/repository.go +22 -4


@@ -23,7 +23,7 @@ const (
23	23		// Store remains available separately for the one-way legacy bridge.
24	24		type Repository struct {
25	25		dir string
26		-	locks sync.Map // map[string]*sync.Mutex
	26	+	locks sync.Map // map[string]*sessionMutex
27	27		hooks repositoryHooks
28	28
29	29		// storageEpochs is guarded by each session's lock. It invalidates a

@@ -59,6 +59,7 @@ type repositoryHooks struct {
59	59		rename func(string) error
60	60		syncDirectory func(string) error
61	61		remove func(string) error
	62	+	beforeSessionLock func(string)
62	63		}
63	64
64	65		var _ ports.SnapshotRepository = (*Repository)(nil)

@@ -78,7 +79,24 @@ func (r *Repository) invalidateStorageEpoch(key string) {
78	79		// storageEpoch must be called with sessionLock(key) held.
79	80		func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
80	81
81		-	func (r Repository) sessionLock(key string) sync.Mutex {
82		-	lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{})
83		-	return lock.(*sync.Mutex)
	82	+	// sessionMutex provides a test-only boundary immediately before a contested
	83	+	// session mutex. Production hooks are nil, so normal locking is unchanged.
	84	+	type sessionMutex struct {
	85	+	repository *Repository
	86	+	key string
	87	+	mu sync.Mutex
	88	+	}
	89	+
	90	+	func (m *sessionMutex) Lock() {
	91	+	if hook := m.repository.hooks.beforeSessionLock; hook != nil {
	92	+	hook(m.key)
	93	+	}
	94	+	m.mu.Lock()
	95	+	}
	96	+
	97	+	func (m *sessionMutex) Unlock() { m.mu.Unlock() }
	98	+
	99	+	func (r Repository) sessionLock(key string) sessionMutex {
	100	+	lock, _ := r.locks.LoadOrStore(key, &sessionMutex{repository: r, key: key})
	101	+	return lock.(*sessionMutex)
84	102		}

 // Store remains available separately for the one-way legacy bridge.
 type Repository struct {
 	dir   string
-	locks sync.Map // map[string]*sync.Mutex
+	locks sync.Map // map[string]*sessionMutex
 	hooks repositoryHooks
 	// storageEpochs is guarded by each session's lock. It invalidates a
 	rename              func(string) error
 	syncDirectory       func(string) error
 	remove              func(string) error
+	beforeSessionLock   func(string)
 }
 var _ ports.SnapshotRepository = (*Repository)(nil)
 // storageEpoch must be called with sessionLock(key) held.
 func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
-func (r *Repository) sessionLock(key string) *sync.Mutex {
+// sessionMutex provides a test-only boundary immediately before a contested
-	lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{})
+// session mutex. Production hooks are nil, so normal locking is unchanged.
-	return lock.(*sync.Mutex)
+type sessionMutex struct {
+	repository *Repository
+	key        string
+	mu         sync.Mutex
+}
+func (m *sessionMutex) Lock() {
+	if hook := m.repository.hooks.beforeSessionLock; hook != nil {
+		hook(m.key)
+	}
+	m.mu.Lock()
+}
+func (m *sessionMutex) Unlock() { m.mu.Unlock() }
+func (r *Repository) sessionLock(key string) *sessionMutex {
+	lock, _ := r.locks.LoadOrStore(key, &sessionMutex{repository: r, key: key})
+	return lock.(*sessionMutex)
 }