7e93b8d fix(snapshot): bound repository session state · vev

M internal/adapters/snapshot/legacy.go +6 -9


@@ -122,9 +122,8 @@ func (r *Repository) DeleteLegacy(ctx context.Context, name string) error {
122	122		return err
123	123		}
124	124		filename := filenameForName(name)
125		-	lock := r.sessionLock("legacy-sync:" + filename)
126		-	lock.Lock()
127		-	defer lock.Unlock()
	125	+	lock := r.lockSession("legacy-sync:" + filename)
	126	+	defer r.unlockSession(lock)
128	127		if err := ctx.Err(); err != nil {
129	128		return err
130	129		}

@@ -252,9 +251,8 @@ func (r *Repository) Tombstone(ctx context.Context, name string) error {
252	251		return err
253	252		}
254	253		key := sessionKey(name)
255		-	lock := r.sessionLock(key)
256		-	lock.Lock()
257		-	defer lock.Unlock()
	254	+	lock := r.lockSession(key)
	255	+	defer r.unlockSession(lock)
258	256		if err := ctx.Err(); err != nil {
259	257		return err
260	258		}

@@ -280,9 +278,8 @@ func (r *Repository) DeleteTombstone(ctx context.Context, name string) error {
280	278		return err
281	279		}
282	280		key := sessionKey(name)
283		-	lock := r.sessionLock(key)
284		-	lock.Lock()
285		-	defer lock.Unlock()
	281	+	lock := r.lockSession(key)
	282	+	defer r.unlockSession(lock)
286	283		if err := ctx.Err(); err != nil {
287	284		return err
288	285		}

 		return err
 	}
 	filename := filenameForName(name)
-	lock := r.sessionLock("legacy-sync:" + filename)
+	lock := r.lockSession("legacy-sync:" + filename)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return err
 	}
 		return err
 	}
 	key := sessionKey(name)
-	lock := r.sessionLock(key)
+	lock := r.lockSession(key)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return err
 	}
 		return err
 	}
 	key := sessionKey(name)
-	lock := r.sessionLock(key)
+	lock := r.lockSession(key)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return err
 	}

M internal/adapters/snapshot/load.go +2 -3


@@ -65,9 +65,8 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener
65	65		if killed {
66	66		return ports.SnapshotGeneration{}, fmt.Errorf("snapshot session %q is killed", name)
67	67		}
68		-	lock := r.sessionLock(key)
69		-	lock.Lock()
70		-	defer lock.Unlock()
	68	+	lock := r.lockSession(key)
	69	+	defer r.unlockSession(lock)
71	70		if err := ctx.Err(); err != nil {
72	71		return ports.SnapshotGeneration{}, err
73	72		}

 	if killed {
 		return ports.SnapshotGeneration{}, fmt.Errorf("snapshot session %q is killed", name)
 	}
-	lock := r.sessionLock(key)
+	lock := r.lockSession(key)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return ports.SnapshotGeneration{}, err
 	}

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


@@ -42,6 +42,15 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
42	42		if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil \|\| budget == 0 {
43	43		return err
44	44		}
	45	+	// Resume an incomplete session before discovering another one. This keeps
	46	+	// retained marks (and their session lock/epoch references) bounded even
	47	+	// when a hostile repository contains arbitrarily many partial sessions.
	48	+	for key := range r.maintenanceSessions {
	49	+	lock := r.lockSession(key)
	50	+	err = r.maintainSession(ctx, key, &budget)
	51	+	r.unlockSession(lock)
	52	+	return err
	53	+	}
45	54		sessions := filepath.Join(r.dir, repositorySessionsDir)
46	55		entries, _, err := r.readMaintenanceDir(sessions, maintenanceBatch, "sessions")
47	56		if errors.Is(err, os.ErrNotExist) {

@@ -77,13 +86,15 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
77	86		if !entry.isDir \|\| !canonicalSessionKey(entry.name) {
78	87		continue
79	88		}
80		-	lock := r.sessionLock(entry.name)
81		-	lock.Lock()
	89	+	lock := r.lockSession(entry.name)
82	90		err = r.maintainSession(ctx, entry.name, &budget)
83		-	lock.Unlock()
	91	+	r.unlockSession(lock)
84	92		if err != nil {
85	93		return err
86	94		}
	95	+	if r.maintenanceSessions[entry.name] != nil {
	96	+	return nil
	97	+	}
87	98		}
88	99		return nil
89	100		}

 	if err := r.removeTemps(ctx, r.dir, &budget, "root-temps"); err != nil || budget == 0 {
 		return err
 	}
+	// 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.
+	for key := range r.maintenanceSessions {
+		lock := r.lockSession(key)
+		err = r.maintainSession(ctx, key, &budget)
+		r.unlockSession(lock)
+		return err
+	}
 	sessions := filepath.Join(r.dir, repositorySessionsDir)
 	entries, _, err := r.readMaintenanceDir(sessions, maintenanceBatch, "sessions")
 	if errors.Is(err, os.ErrNotExist) {
 		if !entry.isDir || !canonicalSessionKey(entry.name) {
 			continue
 		}
-		lock := r.sessionLock(entry.name)
+		lock := r.lockSession(entry.name)
-		lock.Lock()
 		err = r.maintainSession(ctx, entry.name, &budget)
-		lock.Unlock()
+		r.unlockSession(lock)
 		if err != nil {
 			return err
 		}
+		if r.maintenanceSessions[entry.name] != nil {
+			return nil
+		}
 	}
 	return nil
 }

M internal/adapters/snapshot/maintenance_cursor.go +3 -0


@@ -24,6 +24,9 @@ type maintenanceDirEntry struct {
24	24		}
25	25
26	26		func (r *Repository) resetMaintenance() {
	27	+	for key := range r.maintenanceSessions {
	28	+	r.clearSessionMaintenance(key)
	29	+	}
27	30		r.maintenanceCursors = nil
28	31		r.maintenanceSessions = nil
29	32		}


@@ -24,6 +24,9 @@ type maintenanceDirEntry struct {
24	}	24		}
25		25
26	func (r *Repository) resetMaintenance() {	26		func (r *Repository) resetMaintenance() {
		27	+	for key := range r.maintenanceSessions {
		28	+	r.clearSessionMaintenance(key)
		29	+	}
27	r.maintenanceCursors = nil	30		r.maintenanceCursors = nil
28	r.maintenanceSessions = nil	31		r.maintenanceSessions = nil
29	}	32		}

M internal/adapters/snapshot/maintenance_delete.go +18 -37


@@ -8,7 +8,6 @@ import (
8	8		"os"
9	9		"path/filepath"
10	10		"strings"
11		-	"time"
12	11		)
13	12
14	13		// Delete makes a session unavailable by durably moving it out of the canonical

@@ -19,9 +18,8 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
19	18		return err
20	19		}
21	20		key := sessionKey(name)
22		-	lock := r.sessionLock(key)
23		-	lock.Lock()
24		-	defer lock.Unlock()
	21	+	lock := r.lockSession(key)
	22	+	defer r.unlockSession(lock)
25	23		if err := ctx.Err(); err != nil {
26	24		return err
27	25		}

@@ -31,9 +29,9 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
31	29		// A prior rename may have succeeded while its parent sync failed. Complete
32	30		// that durability boundary before considering a canonical directory: this
33	31		// also leaves a newly recreated session untouched.
34		-	pending, err := pendingQuarantine(sessions, key)
	32	+	pending, err := r.pendingQuarantine(sessions, key)
35	33		if err != nil {
36		-	return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
	34	+	return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
37	35		}
38	36		if pending {
39	37		if err := ctx.Err(); err != nil {

@@ -49,15 +47,7 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
49	47		} else if err != nil {
50	48		return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
51	49		}
52		-	quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
53		-	for attempt := 0; ; attempt++ {
54		-	if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
55		-	break
56		-	} else if err != nil {
57		-	return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
58		-	}
59		-	quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
60		-	}
	50	+	quarantine := filepath.Join(sessions, deletingSessionName(key))
61	51		if err := ctx.Err(); err != nil {
62	52		return err
63	53		}

@@ -76,34 +66,25 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
76	66		return nil
77	67		}
78	68
79		-	func pendingQuarantine(dir, key string) (pending bool, err error) {
80		-	f, err := os.Open(dir)
	69	+	// deletingSessionName is a deterministic pending-delete record. Its presence
	70	+	// is checked directly rather than scanning an attacker-controlled sessions
	71	+	// directory, and it prevents a retry from deleting a session republished after
	72	+	// the initial rename reached disk but its directory sync failed.
	73	+	func deletingSessionName(key string) string { return ".deleting-" + key }
	74	+
	75	+	func (r *Repository) pendingQuarantine(dir, key string) (bool, error) {
	76	+	path := filepath.Join(dir, deletingSessionName(key))
	77	+	if hook := r.hooks.beforePendingQuarantineCheck; hook != nil {
	78	+	hook(path)
	79	+	}
	80	+	info, err := os.Lstat(path)
81	81		if errors.Is(err, os.ErrNotExist) {
82	82		return false, nil
83	83		}
84	84		if err != nil {
85	85		return false, err
86	86		}
87		-	defer func() {
88		-	if closeErr := f.Close(); closeErr != nil {
89		-	err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
90		-	}
91		-	}()
92		-	prefix := ".deleting-" + key + "-"
93		-	for {
94		-	entries, err := f.ReadDir(maintenanceBatch)
95		-	for _, entry := range entries {
96		-	if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
97		-	return true, nil
98		-	}
99		-	}
100		-	if errors.Is(err, io.EOF) {
101		-	return false, nil
102		-	}
103		-	if err != nil {
104		-	return false, err
105		-	}
106		-	}
	87	+	return info.IsDir(), nil
107	88		}
108	89
109	90		// Maintain reaps a bounded amount of stale state. Continuation handles are

 	"os"
 	"path/filepath"
 	"strings"
-	"time"
 )
 // Delete makes a session unavailable by durably moving it out of the canonical
 		return err
 	}
 	key := sessionKey(name)
-	lock := r.sessionLock(key)
+	lock := r.lockSession(key)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return err
 	}
 	// A prior rename may have succeeded while its parent sync failed. Complete
 	// that durability boundary before considering a canonical directory: this
 	// also leaves a newly recreated session untouched.
-	pending, err := pendingQuarantine(sessions, key)
+	pending, err := r.pendingQuarantine(sessions, key)
 	if err != nil {
-		return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
+		return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
 	}
 	if pending {
 		if err := ctx.Err(); err != nil {
 	} else if err != nil {
 		return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
 	}
-	quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
+	quarantine := filepath.Join(sessions, deletingSessionName(key))
-	for attempt := 0; ; attempt++ {
-		if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
-			break
-		} else if err != nil {
-			return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
-		}
-		quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
-	}
 	if err := ctx.Err(); err != nil {
 		return err
 	}
 	return nil
 }
-func pendingQuarantine(dir, key string) (pending bool, err error) {
+// deletingSessionName is a deterministic pending-delete record. Its presence
-	f, err := os.Open(dir)
+// is checked directly rather than scanning an attacker-controlled sessions
+// directory, and it prevents a retry from deleting a session republished after
+// the initial rename reached disk but its directory sync failed.
+func deletingSessionName(key string) string { return ".deleting-" + key }
+func (r *Repository) pendingQuarantine(dir, key string) (bool, error) {
+	path := filepath.Join(dir, deletingSessionName(key))
+	if hook := r.hooks.beforePendingQuarantineCheck; hook != nil {
+		hook(path)
+	}
+	info, err := os.Lstat(path)
 	if errors.Is(err, os.ErrNotExist) {
 		return false, nil
 	}
 	if err != nil {
 		return false, err
 	}
-	defer func() {
+	return info.IsDir(), nil
-		if closeErr := f.Close(); closeErr != nil {
-			err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
-		}
-	}()
-	prefix := ".deleting-" + key + "-"
-	for {
-		entries, err := f.ReadDir(maintenanceBatch)
-		for _, entry := range entries {
-			if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
-				return true, nil
-			}
-		}
-		if errors.Is(err, io.EOF) {
-			return false, nil
-		}
-		if err != nil {
-			return false, err
-		}
-	}
 }
 // Maintain reaps a bounded amount of stale state. Continuation handles are

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


@@ -19,6 +19,7 @@ type manifestMaintenance struct {
19	19		}
20	20
21	21		type sessionMaintenance struct {
	22	+	lock *sessionMutex
22	23		token string
23	24		epoch uint64
24	25		marked map[uint64]manifestMaintenance

@@ -115,6 +116,7 @@ func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, e
115	116		if state == nil \|\| state.token != token \|\| state.epoch != r.storageEpoch(key) {
116	117		r.clearSessionMaintenance(key)
117	118		state = &sessionMaintenance{
	119	+	lock: r.retainSessionState(key),
118	120		token: token,
119	121		epoch: r.storageEpoch(key),
120	122		conservative: conservative,

@@ -144,6 +146,9 @@ func (r *Repository) maintenanceToken(key string) (string, bool, error) {
144	146		}
145	147
146	148		func (r *Repository) clearSessionMaintenance(key string) {
	149	+	if state := r.maintenanceSessions[key]; state != nil && state.lock != nil {
	150	+	r.releaseSessionReference(state.lock)
	151	+	}
147	152		delete(r.maintenanceSessions, key)
148	153		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
149	154		for id := range r.maintenanceCursors {


@@ -19,6 +19,7 @@ type manifestMaintenance struct {
19	}	19		}
20		20
21	type sessionMaintenance struct {	21		type sessionMaintenance struct {
		22	+	lock *sessionMutex
22	token string	23		token string
23	epoch uint64	24		epoch uint64
24	marked map[uint64]manifestMaintenance	25		marked map[uint64]manifestMaintenance

@@ -115,6 +116,7 @@ func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, e
115	if state == nil \|\| state.token != token \|\| state.epoch != r.storageEpoch(key) {	116		if state == nil \|\| state.token != token \|\| state.epoch != r.storageEpoch(key) {
116	r.clearSessionMaintenance(key)	117		r.clearSessionMaintenance(key)
117	state = &sessionMaintenance{	118		state = &sessionMaintenance{
		119	+	lock: r.retainSessionState(key),
118	token: token,	120		token: token,
119	epoch: r.storageEpoch(key),	121		epoch: r.storageEpoch(key),
120	conservative: conservative,	122		conservative: conservative,

@@ -144,6 +146,9 @@ func (r *Repository) maintenanceToken(key string) (string, bool, error) {
144	}	146		}
145		147
146	func (r *Repository) clearSessionMaintenance(key string) {	148		func (r *Repository) clearSessionMaintenance(key string) {
		149	+	if state := r.maintenanceSessions[key]; state != nil && state.lock != nil {
		150	+	r.releaseSessionReference(state.lock)
		151	+	}
147	delete(r.maintenanceSessions, key)	152		delete(r.maintenanceSessions, key)
148	prefix := "\x00" + filepath.Clean(r.sessionPath(key))	153		prefix := "\x00" + filepath.Clean(r.sessionPath(key))
149	for id := range r.maintenanceCursors {	154		for id := range r.maintenanceCursors {

M internal/adapters/snapshot/maintenance_test.go +6 -9


@@ -224,8 +224,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
224	224		t.Run("publish", func(t *testing.T) {
225	225		repo := NewRepository(privateDir(t))
226	226		key := sessionKey("named")
227		-	lock := repo.sessionLock(key)
228		-	lock.Lock()
	227	+	lock := repo.lockSession(key)
229	228		ctx, cancel := context.WithCancel(context.Background())
230	229		reached := make(chan struct{})
231	230		release := make(chan struct{})

@@ -235,7 +234,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
235	234		<-reached
236	235		cancel()
237	236		close(release)
238		-	lock.Unlock()
	237	+	repo.unlockSession(lock)
239	238		if err := <-done; !errors.Is(err, context.Canceled) {
240	239		t.Fatalf("Publish error = %v, want canceled", err)
241	240		}

@@ -257,8 +256,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
257	256		{"delete", func(ctx context.Context) error { return repo.Delete(ctx, "named") }},
258	257		} {
259	258		t.Run(operation.name, func(t *testing.T) {
260		-	lock := repo.sessionLock(key)
261		-	lock.Lock()
	259	+	lock := repo.lockSession(key)
262	260		ctx, cancel := context.WithCancel(context.Background())
263	261		reached := make(chan struct{})
264	262		release := make(chan struct{})

@@ -268,7 +266,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
268	266		<-reached
269	267		cancel()
270	268		close(release)
271		-	lock.Unlock()
	269	+	repo.unlockSession(lock)
272	270		if err := <-done; !errors.Is(err, context.Canceled) {
273	271		t.Fatalf("%s error = %v, want canceled", operation.name, err)
274	272		}

@@ -284,8 +282,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
284	282		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
285	283		t.Fatal(err)
286	284		}
287		-	lock := repo.sessionLock(sessionKey("named"))
288		-	lock.Lock()
	285	+	lock := repo.lockSession(sessionKey("named"))
289	286		ctx, cancel := context.WithCancel(context.Background())
290	287		reached := make(chan struct{})
291	288		release := make(chan struct{})

@@ -295,7 +292,7 @@ func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.
295	292		<-reached
296	293		cancel()
297	294		close(release)
298		-	lock.Unlock()
	295	+	repo.unlockSession(lock)
299	296		if err := <-done; !errors.Is(err, context.Canceled) {
300	297		t.Fatalf("Maintain error = %v, want canceled", err)
301	298		}

 	t.Run("publish", func(t *testing.T) {
 		repo := NewRepository(privateDir(t))
 		key := sessionKey("named")
-		lock := repo.sessionLock(key)
+		lock := repo.lockSession(key)
-		lock.Lock()
 		ctx, cancel := context.WithCancel(context.Background())
 		reached := make(chan struct{})
 		release := make(chan struct{})
 		<-reached
 		cancel()
 		close(release)
-		lock.Unlock()
+		repo.unlockSession(lock)
 		if err := <-done; !errors.Is(err, context.Canceled) {
 			t.Fatalf("Publish error = %v, want canceled", err)
 		}
 			{"delete", func(ctx context.Context) error { return repo.Delete(ctx, "named") }},
 		} {
 			t.Run(operation.name, func(t *testing.T) {
-				lock := repo.sessionLock(key)
+				lock := repo.lockSession(key)
-				lock.Lock()
 				ctx, cancel := context.WithCancel(context.Background())
 				reached := make(chan struct{})
 				release := make(chan struct{})
 				<-reached
 				cancel()
 				close(release)
-				lock.Unlock()
+				repo.unlockSession(lock)
 				if err := <-done; !errors.Is(err, context.Canceled) {
 					t.Fatalf("%s error = %v, want canceled", operation.name, err)
 				}
 		if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
 			t.Fatal(err)
 		}
-		lock := repo.sessionLock(sessionKey("named"))
+		lock := repo.lockSession(sessionKey("named"))
-		lock.Lock()
 		ctx, cancel := context.WithCancel(context.Background())
 		reached := make(chan struct{})
 		release := make(chan struct{})
 		<-reached
 		cancel()
 		close(release)
-		lock.Unlock()
+		repo.unlockSession(lock)
 		if err := <-done; !errors.Is(err, context.Canceled) {
 			t.Fatalf("Maintain error = %v, want canceled", err)
 		}

M internal/adapters/snapshot/publish.go +2 -3


@@ -14,9 +14,8 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
14	14		return err
15	15		}
16	16		key := sessionKey(publication.Name)
17		-	lock := r.sessionLock(key)
18		-	lock.Lock()
19		-	defer lock.Unlock()
	17	+	lock := r.lockSession(key)
	18	+	defer r.unlockSession(lock)
20	19		if err := ctx.Err(); err != nil {
21	20		return err
22	21		}

 		return err
 	}
 	key := sessionKey(publication.Name)
-	lock := r.sessionLock(key)
+	lock := r.lockSession(key)
-	lock.Lock()
+	defer r.unlockSession(lock)
-	defer lock.Unlock()
 	if err := ctx.Err(); err != nil {
 		return err
 	}

M internal/adapters/snapshot/repository.go +97 -36


@@ -23,13 +23,15 @@ 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]*sessionMutex
27	26		hooks repositoryHooks
28	27
29		-	// storageEpochs is guarded by each session's lock. It invalidates a
30		-	// maintenance mark whenever publication or deletion changes that session's
31		-	// on-disk namespace.
32		-	storageEpochs map[string]uint64
	28	+	// sessionStateMu owns both maps. A caller retains a reference before it
	29	+	// waits on a session mutex, so an idle entry can never be removed while a
	30	+	// waiter still holds its old mutex. Epoch entries share that lifetime.
	31	+	sessionStateMu sync.Mutex
	32	+	locks map[string]*sessionMutex
	33	+	storageEpochs map[string]uint64
	34	+	nextEpoch uint64
33	35
34	36		// maintenanceMu owns bounded continuation metadata: seek cookies and
35	37		// pending directory entries. Directory descriptors are opened and closed

@@ -53,21 +55,22 @@ type repositoryHooks struct {
53	55		// Object hooks instrument the publication path in package tests. They keep
54	56		// the steady-state cost of retained history observable without exposing a
55	57		// production metrics surface.
56		-	beforeObjectRead func(string)
57		-	beforeObjectHash func([]byte)
58		-	beforeObjectCopy func([]byte)
59		-	beforeHeadWrite func(string) error
60		-	createTemp func(string) error
61		-	writeTemp func(string) error
62		-	syncFile func(string) error
63		-	closeFile func(string) error
64		-	installImmutable func(string) error
65		-	rename func(string) error
66		-	syncDirectory func(string) error
67		-	remove func(string) error
68		-	openMaintenanceDirectory func(string) (maintenanceDirectory, error)
69		-	openLegacyDirectory func(string) (legacyDirectory, error)
70		-	beforeSessionLock func(string)
	58	+	beforeObjectRead func(string)
	59	+	beforeObjectHash func([]byte)
	60	+	beforeObjectCopy func([]byte)
	61	+	beforeHeadWrite func(string) error
	62	+	beforePendingQuarantineCheck func(string)
	63	+	createTemp func(string) error
	64	+	writeTemp func(string) error
	65	+	syncFile func(string) error
	66	+	closeFile func(string) error
	67	+	installImmutable func(string) error
	68	+	rename func(string) error
	69	+	syncDirectory func(string) error
	70	+	remove func(string) error
	71	+	openMaintenanceDirectory func(string) (maintenanceDirectory, error)
	72	+	openLegacyDirectory func(string) (legacyDirectory, error)
	73	+	beforeSessionLock func(string)
71	74		}
72	75
73	76		var _ ports.SnapshotRepository = (*Repository)(nil)

@@ -76,35 +79,93 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
76	79		// NewRepository creates a repository rooted at dir. It does not create files
77	80		// until the first publication, so merely constructing it is side-effect free.
78	81		func NewRepository(dir string) *Repository {
79		-	return &Repository{dir: dir, storageEpochs: make(map[string]uint64)}
	82	+	return &Repository{
	83	+	dir: dir,
	84	+	locks: make(map[string]*sessionMutex),
	85	+	storageEpochs: make(map[string]uint64),
	86	+	}
80	87		}
81	88
82		-	// invalidateStorageEpoch must be called with sessionLock(key) held.
	89	+	// invalidateStorageEpoch must be called with lockSession(key) held. The
	90	+	// registry mutex makes accesses for different session keys race-free too.
83	91		func (r *Repository) invalidateStorageEpoch(key string) {
84		-	r.storageEpochs[key]++
	92	+	r.sessionStateMu.Lock()
	93	+	r.nextEpoch++
	94	+	r.storageEpochs[key] = r.nextEpoch
	95	+	r.sessionStateMu.Unlock()
85	96		}
86	97
87		-	// storageEpoch must be called with sessionLock(key) held.
88		-	func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
	98	+	// storageEpoch must be called with lockSession(key) held.
	99	+	func (r *Repository) storageEpoch(key string) uint64 {
	100	+	r.sessionStateMu.Lock()
	101	+	epoch := r.storageEpochs[key]
	102	+	r.sessionStateMu.Unlock()
	103	+	return epoch
	104	+	}
89	105
90		-	// sessionMutex provides a test-only boundary immediately before a contested
91		-	// session mutex. Production hooks are nil, so normal locking is unchanged.
	106	+	// sessionMutex is retained by each waiter and each resumable maintenance
	107	+	// state. That reference is what makes eviction safe: a new caller cannot lock
	108	+	// a replacement mutex until all users of the prior mutex have left it.
92	109		type sessionMutex struct {
93		-	repository *Repository
94	110		key string
95	111		mu sync.Mutex
	112	+	references int
96	113		}
97	114
98		-	func (m *sessionMutex) Lock() {
99		-	if hook := m.repository.hooks.beforeSessionLock; hook != nil {
100		-	hook(m.key)
	115	+	func (r Repository) lockSession(key string) sessionMutex {
	116	+	r.sessionStateMu.Lock()
	117	+	lock := r.locks[key]
	118	+	if lock == nil {
	119	+	lock = &sessionMutex{key: key}
	120	+	r.locks[key] = lock
	121	+	}
	122	+	lock.references++
	123	+	r.sessionStateMu.Unlock()
	124	+	if hook := r.hooks.beforeSessionLock; hook != nil {
	125	+	hook(key)
101	126		}
102		-	m.mu.Lock()
	127	+	lock.mu.Lock()
	128	+	return lock
103	129		}
104	130
105		-	func (m *sessionMutex) Unlock() { m.mu.Unlock() }
	131	+	func (r Repository) unlockSession(lock sessionMutex) {
	132	+	lock.mu.Unlock()
	133	+	r.releaseSessionReference(lock)
	134	+	}
	135	+
	136	+	func (r Repository) retainSessionState(key string) sessionMutex {
	137	+	r.sessionStateMu.Lock()
	138	+	lock := r.locks[key]
	139	+	// sessionMaintenanceState is called while the active session lock is held.
	140	+	if lock == nil {
	141	+	panic("snapshot: retain missing session lock")
	142	+	}
	143	+	lock.references++
	144	+	r.sessionStateMu.Unlock()
	145	+	return lock
	146	+	}
	147	+
	148	+	func (r Repository) releaseSessionReference(lock sessionMutex) {
	149	+	r.sessionStateMu.Lock()
	150	+	lock.references--
	151	+	if lock.references < 0 {
	152	+	r.sessionStateMu.Unlock()
	153	+	panic("snapshot: release of unretained session lock")
	154	+	}
	155	+	if lock.references == 0 {
	156	+	if r.locks[lock.key] != lock {
	157	+	r.sessionStateMu.Unlock()
	158	+	panic("snapshot: session lock registry mismatch")
	159	+	}
	160	+	delete(r.locks, lock.key)
	161	+	delete(r.storageEpochs, lock.key)
	162	+	}
	163	+	r.sessionStateMu.Unlock()
	164	+	}
106	165
107		-	func (r Repository) sessionLock(key string) sessionMutex {
108		-	lock, _ := r.locks.LoadOrStore(key, &sessionMutex{repository: r, key: key})
109		-	return lock.(*sessionMutex)
	166	+	// sessionStateCounts is test evidence that idle per-session state is evicted.
	167	+	func (r *Repository) sessionStateCounts() (locks, epochs int) {
	168	+	r.sessionStateMu.Lock()
	169	+	defer r.sessionStateMu.Unlock()
	170	+	return len(r.locks), len(r.storageEpochs)
110	171		}

 // Store remains available separately for the one-way legacy bridge.
 type Repository struct {
 	dir   string
-	locks sync.Map // map[string]*sessionMutex
 	hooks repositoryHooks
-	// storageEpochs is guarded by each session's lock. It invalidates a
+	// sessionStateMu owns both maps. A caller retains a reference before it
-	// maintenance mark whenever publication or deletion changes that session's
+	// waits on a session mutex, so an idle entry can never be removed while a
-	// on-disk namespace.
+	// waiter still holds its old mutex. Epoch entries share that lifetime.
-	storageEpochs map[string]uint64
+	sessionStateMu sync.Mutex
+	locks          map[string]*sessionMutex
+	storageEpochs  map[string]uint64
+	nextEpoch      uint64
 	// maintenanceMu owns bounded continuation metadata: seek cookies and
 	// pending directory entries. Directory descriptors are opened and closed
 	// Object hooks instrument the publication path in package tests. They keep
 	// the steady-state cost of retained history observable without exposing a
 	// production metrics surface.
-	beforeObjectRead         func(string)
+	beforeObjectRead             func(string)
-	beforeObjectHash         func([]byte)
+	beforeObjectHash             func([]byte)
-	beforeObjectCopy         func([]byte)
+	beforeObjectCopy             func([]byte)
-	beforeHeadWrite          func(string) error
+	beforeHeadWrite              func(string) error
-	createTemp               func(string) error
+	beforePendingQuarantineCheck func(string)
-	writeTemp                func(string) error
+	createTemp                   func(string) error
-	syncFile                 func(string) error
+	writeTemp                    func(string) error
-	closeFile                func(string) error
+	syncFile                     func(string) error
-	installImmutable         func(string) error
+	closeFile                    func(string) error
-	rename                   func(string) error
+	installImmutable             func(string) error
-	syncDirectory            func(string) error
+	rename                       func(string) error
-	remove                   func(string) error
+	syncDirectory                func(string) error
-	openMaintenanceDirectory func(string) (maintenanceDirectory, error)
+	remove                       func(string) error
-	openLegacyDirectory      func(string) (legacyDirectory, error)
+	openMaintenanceDirectory     func(string) (maintenanceDirectory, error)
-	beforeSessionLock        func(string)
+	openLegacyDirectory          func(string) (legacyDirectory, error)
+	beforeSessionLock            func(string)
 }
 var _ ports.SnapshotRepository = (*Repository)(nil)
 // NewRepository creates a repository rooted at dir. It does not create files
 // until the first publication, so merely constructing it is side-effect free.
 func NewRepository(dir string) *Repository {
-	return &Repository{dir: dir, storageEpochs: make(map[string]uint64)}
+	return &Repository{
+		dir:           dir,
+		locks:         make(map[string]*sessionMutex),
+		storageEpochs: make(map[string]uint64),
+	}
 }
-// invalidateStorageEpoch must be called with sessionLock(key) held.
+// invalidateStorageEpoch must be called with lockSession(key) held. The
+// registry mutex makes accesses for different session keys race-free too.
 func (r *Repository) invalidateStorageEpoch(key string) {
-	r.storageEpochs[key]++
+	r.sessionStateMu.Lock()
+	r.nextEpoch++
+	r.storageEpochs[key] = r.nextEpoch
+	r.sessionStateMu.Unlock()
 }
-// storageEpoch must be called with sessionLock(key) held.
+// storageEpoch must be called with lockSession(key) held.
-func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
+func (r *Repository) storageEpoch(key string) uint64 {
+	r.sessionStateMu.Lock()
+	epoch := r.storageEpochs[key]
+	r.sessionStateMu.Unlock()
+	return epoch
+}
-// sessionMutex provides a test-only boundary immediately before a contested
+// sessionMutex is retained by each waiter and each resumable maintenance
-// session mutex. Production hooks are nil, so normal locking is unchanged.
+// state. That reference is what makes eviction safe: a new caller cannot lock
+// a replacement mutex until all users of the prior mutex have left it.
 type sessionMutex struct {
-	repository *Repository
 	key        string
 	mu         sync.Mutex
+	references int
 }
-func (m *sessionMutex) Lock() {
+func (r *Repository) lockSession(key string) *sessionMutex {
-	if hook := m.repository.hooks.beforeSessionLock; hook != nil {
+	r.sessionStateMu.Lock()
-		hook(m.key)
+	lock := r.locks[key]
+	if lock == nil {
+		lock = &sessionMutex{key: key}
+		r.locks[key] = lock
+	}
+	lock.references++
+	r.sessionStateMu.Unlock()
+	if hook := r.hooks.beforeSessionLock; hook != nil {
+		hook(key)
 	}
-	m.mu.Lock()
+	lock.mu.Lock()
+	return lock
 }
-func (m *sessionMutex) Unlock() { m.mu.Unlock() }
+func (r *Repository) unlockSession(lock *sessionMutex) {
+	lock.mu.Unlock()
+	r.releaseSessionReference(lock)
+}
+func (r *Repository) retainSessionState(key string) *sessionMutex {
+	r.sessionStateMu.Lock()
+	lock := r.locks[key]
+	// sessionMaintenanceState is called while the active session lock is held.
+	if lock == nil {
+		panic("snapshot: retain missing session lock")
+	}
+	lock.references++
+	r.sessionStateMu.Unlock()
+	return lock
+}
+func (r *Repository) releaseSessionReference(lock *sessionMutex) {
+	r.sessionStateMu.Lock()
+	lock.references--
+	if lock.references < 0 {
+		r.sessionStateMu.Unlock()
+		panic("snapshot: release of unretained session lock")
+	}
+	if lock.references == 0 {
+		if r.locks[lock.key] != lock {
+			r.sessionStateMu.Unlock()
+			panic("snapshot: session lock registry mismatch")
+		}
+		delete(r.locks, lock.key)
+		delete(r.storageEpochs, lock.key)
+	}
+	r.sessionStateMu.Unlock()
+}
-func (r *Repository) sessionLock(key string) *sessionMutex {
+// sessionStateCounts is test evidence that idle per-session state is evicted.
-	lock, _ := r.locks.LoadOrStore(key, &sessionMutex{repository: r, key: key})
+func (r *Repository) sessionStateCounts() (locks, epochs int) {
-	return lock.(*sessionMutex)
+	r.sessionStateMu.Lock()
+	defer r.sessionStateMu.Unlock()
+	return len(r.locks), len(r.storageEpochs)
 }

A internal/adapters/snapshot/repository_phase6_test.go +107 -0

@@ -0,0 +1,107 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"os"
	7	+	"path/filepath"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	func TestRepositoryPublishDeleteManySessionsRaceFree(t *testing.T) {
	16	+	repo := NewRepository(privateDir(t))
	17	+	const sessions = 128
	18	+
	19	+	publications := make([]ports.SnapshotPublication, sessions)
	20	+	for i := range publications {
	21	+	name := fmt.Sprintf("named-%03d", i)
	22	+	publications[i] = repositoryPublication(t, name, 1, []byte(name))
	23	+	if err := repo.Publish(context.Background(), publications[i]); err != nil {
	24	+	t.Fatalf("initial Publish(%q): %v", name, err)
	25	+	}
	26	+	}
	27	+
	28	+	var wg sync.WaitGroup
	29	+	for _, publication := range publications {
	30	+	publication := publication
	31	+	wg.Add(2)
	32	+	go func() {
	33	+	defer wg.Done()
	34	+	if err := repo.Publish(context.Background(), publication); err != nil {
	35	+	t.Errorf("Publish(%q): %v", publication.Name, err)
	36	+	}
	37	+	}()
	38	+	go func() {
	39	+	defer wg.Done()
	40	+	if err := repo.Delete(context.Background(), publication.Name); err != nil {
	41	+	t.Errorf("Delete(%q): %v", publication.Name, err)
	42	+	}
	43	+	}()
	44	+	}
	45	+	wg.Wait()
	46	+
	47	+	locks, epochs := repo.sessionStateCounts()
	48	+	if locks != 0 \|\| epochs != 0 {
	49	+	t.Fatalf("retained session state = %d locks, %d epochs, want none", locks, epochs)
	50	+	}
	51	+	}
	52	+
	53	+	func TestRepositoryBoundsAndEvictsPartialMaintenanceSessionState(t *testing.T) {
	54	+	repo := NewRepository(privateDir(t))
	55	+	for _, key := range []string{"named-000", "named-001"} {
	56	+	generations := filepath.Join(repo.sessionPath(key), repositoryGenerations)
	57	+	for i := 0; i < maintenanceBatch+1; i++ {
	58	+	if err := os.MkdirAll(generations, 0o700); err != nil {
	59	+	t.Fatal(err)
	60	+	}
	61	+	if err := os.WriteFile(filepath.Join(generations, fmt.Sprintf("unclassified-%03d", i)), []byte("state"), 0o600); err != nil {
	62	+	t.Fatal(err)
	63	+	}
	64	+	}
	65	+	}
	66	+	if err := repo.Maintain(context.Background()); err != nil {
	67	+	t.Fatal(err)
	68	+	}
	69	+	locks, epochs := repo.sessionStateCounts()
	70	+	if locks != 1 \|\| epochs != 0 {
	71	+	t.Fatalf("partial maintenance state = %d locks, %d epochs, want one lock and no epochs", locks, epochs)
	72	+	}
	73	+
	74	+	ctx, cancel := context.WithCancel(context.Background())
	75	+	cancel()
	76	+	if err := repo.Maintain(ctx); err != context.Canceled {
	77	+	t.Fatalf("canceled Maintain error = %v, want context canceled", err)
	78	+	}
	79	+	locks, epochs = repo.sessionStateCounts()
	80	+	if locks != 0 \|\| epochs != 0 {
	81	+	t.Fatalf("session state after maintenance reset = %d locks, %d epochs, want none", locks, epochs)
	82	+	}
	83	+	}
	84	+
	85	+	func TestRepositoryDeleteUsesDeterministicPendingQuarantine(t *testing.T) {
	86	+	repo := NewRepository(privateDir(t))
	87	+	if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
	88	+	t.Fatal(err)
	89	+	}
	90	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
	91	+	for i := 0; i < maxDirectoryTraversalEntries*2; i++ {
	92	+	if err := os.Mkdir(filepath.Join(sessions, fmt.Sprintf("unrelated-%05d", i)), 0o700); err != nil {
	93	+	t.Fatal(err)
	94	+	}
	95	+	}
	96	+	var checks atomic.Int32
	97	+	repo.hooks.beforePendingQuarantineCheck = func(string) { checks.Add(1) }
	98	+	if err := repo.Delete(context.Background(), "named"); err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	if got := checks.Load(); got != 1 {
	102	+	t.Fatalf("pending quarantine checks = %d, want one deterministic lookup", got)
	103	+	}
	104	+	if _, err := os.Lstat(filepath.Join(sessions, deletingSessionName(sessionKey("named")))); err != nil {
	105	+	t.Fatalf("deterministic quarantine missing: %v", err)
	106	+	}
	107	+	}

@@ -0,0 +1,107 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"os"
	7	+	"path/filepath"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	func TestRepositoryPublishDeleteManySessionsRaceFree(t *testing.T) {
	16	+	repo := NewRepository(privateDir(t))
	17	+	const sessions = 128
	18	+
	19	+	publications := make([]ports.SnapshotPublication, sessions)
	20	+	for i := range publications {
	21	+	name := fmt.Sprintf("named-%03d", i)
	22	+	publications[i] = repositoryPublication(t, name, 1, []byte(name))
	23	+	if err := repo.Publish(context.Background(), publications[i]); err != nil {
	24	+	t.Fatalf("initial Publish(%q): %v", name, err)
	25	+	}
	26	+	}
	27	+
	28	+	var wg sync.WaitGroup
	29	+	for _, publication := range publications {
	30	+	publication := publication
	31	+	wg.Add(2)
	32	+	go func() {
	33	+	defer wg.Done()
	34	+	if err := repo.Publish(context.Background(), publication); err != nil {
	35	+	t.Errorf("Publish(%q): %v", publication.Name, err)
	36	+	}
	37	+	}()
	38	+	go func() {
	39	+	defer wg.Done()
	40	+	if err := repo.Delete(context.Background(), publication.Name); err != nil {
	41	+	t.Errorf("Delete(%q): %v", publication.Name, err)
	42	+	}
	43	+	}()
	44	+	}
	45	+	wg.Wait()
	46	+
	47	+	locks, epochs := repo.sessionStateCounts()
	48	+	if locks != 0 \|\| epochs != 0 {
	49	+	t.Fatalf("retained session state = %d locks, %d epochs, want none", locks, epochs)
	50	+	}
	51	+	}
	52	+
	53	+	func TestRepositoryBoundsAndEvictsPartialMaintenanceSessionState(t *testing.T) {
	54	+	repo := NewRepository(privateDir(t))
	55	+	for _, key := range []string{"named-000", "named-001"} {
	56	+	generations := filepath.Join(repo.sessionPath(key), repositoryGenerations)
	57	+	for i := 0; i < maintenanceBatch+1; i++ {
	58	+	if err := os.MkdirAll(generations, 0o700); err != nil {
	59	+	t.Fatal(err)
	60	+	}
	61	+	if err := os.WriteFile(filepath.Join(generations, fmt.Sprintf("unclassified-%03d", i)), []byte("state"), 0o600); err != nil {
	62	+	t.Fatal(err)
	63	+	}
	64	+	}
	65	+	}
	66	+	if err := repo.Maintain(context.Background()); err != nil {
	67	+	t.Fatal(err)
	68	+	}
	69	+	locks, epochs := repo.sessionStateCounts()
	70	+	if locks != 1 \|\| epochs != 0 {
	71	+	t.Fatalf("partial maintenance state = %d locks, %d epochs, want one lock and no epochs", locks, epochs)
	72	+	}
	73	+
	74	+	ctx, cancel := context.WithCancel(context.Background())
	75	+	cancel()
	76	+	if err := repo.Maintain(ctx); err != context.Canceled {
	77	+	t.Fatalf("canceled Maintain error = %v, want context canceled", err)
	78	+	}
	79	+	locks, epochs = repo.sessionStateCounts()
	80	+	if locks != 0 \|\| epochs != 0 {
	81	+	t.Fatalf("session state after maintenance reset = %d locks, %d epochs, want none", locks, epochs)
	82	+	}
	83	+	}
	84	+
	85	+	func TestRepositoryDeleteUsesDeterministicPendingQuarantine(t *testing.T) {
	86	+	repo := NewRepository(privateDir(t))
	87	+	if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
	88	+	t.Fatal(err)
	89	+	}
	90	+	sessions := filepath.Join(repo.dir, repositorySessionsDir)
	91	+	for i := 0; i < maxDirectoryTraversalEntries*2; i++ {
	92	+	if err := os.Mkdir(filepath.Join(sessions, fmt.Sprintf("unrelated-%05d", i)), 0o700); err != nil {
	93	+	t.Fatal(err)
	94	+	}
	95	+	}
	96	+	var checks atomic.Int32
	97	+	repo.hooks.beforePendingQuarantineCheck = func(string) { checks.Add(1) }
	98	+	if err := repo.Delete(context.Background(), "named"); err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	if got := checks.Load(); got != 1 {
	102	+	t.Fatalf("pending quarantine checks = %d, want one deterministic lookup", got)
	103	+	}
	104	+	if _, err := os.Lstat(filepath.Join(sessions, deletingSessionName(sessionKey("named")))); err != nil {
	105	+	t.Fatalf("deterministic quarantine missing: %v", err)
	106	+	}
	107	+	}