fix(snapshot): make fallback loading atomic and bounded

commit: 0edf864712f8b8c14aa169d83875b4a61a6fbf68
author: brice · 3mo ago · 2026-07-22 04:45 UTC
parent: f2d254a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +149 -53

M internal/adapters/snapshot/legacy.go +5 -1


@@ -362,7 +362,11 @@ func (r *Repository) tombstoned(name string) (bool, error) {
362	362		if name == "" {
363	363		return false, nil
364	364		}
365		-	data, err := r.readBounded(filepath.Join(r.dir, tombstoneFilename(sessionKey(name))))
	365	+	path := filepath.Join(r.dir, tombstoneFilename(sessionKey(name)))
	366	+	if hook := r.hooks.beforeTombstoneCheck; hook != nil {
	367	+	hook(path)
	368	+	}
	369	+	data, err := r.readBounded(path)
366	370		if errors.Is(err, os.ErrNotExist) {
367	371		return false, nil
368	372		}

 	if name == "" {
 		return false, nil
 	}
-	data, err := r.readBounded(filepath.Join(r.dir, tombstoneFilename(sessionKey(name))))
+	path := filepath.Join(r.dir, tombstoneFilename(sessionKey(name)))
+	if hook := r.hooks.beforeTombstoneCheck; hook != nil {
+		hook(path)
+	}
+	data, err := r.readBounded(path)
 	if errors.Is(err, os.ErrNotExist) {
 		return false, nil
 	}

M internal/adapters/snapshot/load.go +54 -52


@@ -23,7 +23,10 @@ const (
23	23		maxDirectoryTraversalEntries = 4096
24	24		)
25	25
26		-	var ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
	26	+	var (
	27	+	ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
	28	+	ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
	29	+	)
27	30
28	31		func (r *Repository) List(ctx context.Context) ([]string, error) {
29	32		if err := ctx.Err(); err != nil {

@@ -58,6 +61,11 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener
58	61		return ports.SnapshotGeneration{}, err
59	62		}
60	63		key := sessionKey(name)
	64	+	lock := r.lockSession(key)
	65	+	defer r.unlockSession(lock)
	66	+	if err := ctx.Err(); err != nil {
	67	+	return ports.SnapshotGeneration{}, err
	68	+	}
61	69		killed, err := r.tombstoned(name)
62	70		if err != nil {
63	71		return ports.SnapshotGeneration{}, fmt.Errorf("read killed session marker: %w", err)

@@ -65,20 +73,15 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener
65	73		if killed {
66	74		return ports.SnapshotGeneration{}, fmt.Errorf("snapshot session %q is killed", name)
67	75		}
68		-	lock := r.lockSession(key)
69		-	defer r.unlockSession(lock)
70		-	if err := ctx.Err(); err != nil {
71		-	return ports.SnapshotGeneration{}, err
72		-	}
73	76		return r.load(ctx, name, key)
74	77		}
75	78
76	79		func (r *Repository) load(ctx context.Context, name, key string) (ports.SnapshotGeneration, error) {
77	80		preferred, preferredDigest, err := r.readHead(key)
78		-	if err != nil && !errors.Is(err, os.ErrNotExist) {
79		-	preferred = 0
	81	+	skipped := errors.Is(err, ErrInvalidHEAD)
	82	+	if err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
	83	+	return ports.SnapshotGeneration{}, fmt.Errorf("read snapshot HEAD: %w", err)
80	84		}
81		-	skipped := false
82	85		if preferred != 0 {
83	86		got, err := r.loadGeneration(ctx, name, key, preferred)
84	87		if err == nil && sha256.Sum256(got.Manifest) == preferredDigest {

@@ -88,16 +91,11 @@ func (r *Repository) load(ctx context.Context, name, key string) (ports.Snapshot
88	91		}
89	92
90	93		budget := maxDirectoryTraversalEntries
91		-	before := uint64(0) // zero means no upper bound.
92		-	for {
93		-	generation, ok, err := r.nextGeneration(ctx, key, before, preferred, &budget)
94		-	if err != nil {
95		-	return ports.SnapshotGeneration{}, err
96		-	}
97		-	if !ok {
98		-	break
99		-	}
100		-	before = generation
	94	+	candidates, err := r.generationCandidates(ctx, key, preferred, &budget)
	95	+	if err != nil {
	96	+	return ports.SnapshotGeneration{}, err
	97	+	}
	98	+	for _, generation := range candidates {
101	99		got, err := r.loadGeneration(ctx, name, key, generation)
102	100		if err != nil {
103	101		skipped = true

@@ -116,33 +114,32 @@ func (r *Repository) currentGeneration(ctx context.Context, name, key string) (u
116	114		if err == nil && sha256.Sum256(got.Manifest) == digest {
117	115		return generation, got.Manifest, nil
118	116		}
	117	+	} else if !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
	118	+	return 0, nil, fmt.Errorf("read snapshot HEAD: %w", err)
119	119		}
120	120		budget := maxDirectoryTraversalEntries
121		-	before := uint64(0)
122		-	for {
123		-	generation, ok, err := r.nextGeneration(ctx, key, before, 0, &budget)
124		-	if err != nil \|\| !ok {
125		-	return 0, nil, err
126		-	}
127		-	before = generation
	121	+	candidates, err := r.generationCandidates(ctx, key, 0, &budget)
	122	+	if err != nil {
	123	+	return 0, nil, err
	124	+	}
	125	+	for _, generation := range candidates {
128	126		got, err := r.loadGeneration(ctx, name, key, generation)
129	127		if err == nil {
130	128		return generation, got.Manifest, nil
131	129		}
132	130		}
	131	+	return 0, nil, nil
133	132		}
134	133
135		-	// loadNewestGeneration validates candidates in descending generation order
136		-	// without materializing the generations directory. It is used by List, whose
137		-	// name is discovered from each manifest rather than supplied by a caller.
	134	+	// loadNewestGeneration validates a single bounded candidate batch in strict
	135	+	// descending generation order. It is used by List, whose name is discovered
	136	+	// from each manifest rather than supplied by a caller.
138	137		func (r Repository) loadNewestGeneration(ctx context.Context, key string, budget int) (ports.SnapshotGeneration, bool, error) {
139		-	before := uint64(0)
140		-	for {
141		-	generation, ok, err := r.nextGeneration(ctx, key, before, 0, budget)
142		-	if err != nil \|\| !ok {
143		-	return ports.SnapshotGeneration{}, false, err
144		-	}
145		-	before = generation
	138	+	candidates, err := r.generationCandidates(ctx, key, 0, budget)
	139	+	if err != nil {
	140	+	return ports.SnapshotGeneration{}, false, err
	141	+	}
	142	+	for _, generation := range candidates {
146	143		data, err := r.readBounded(r.manifestPath(key, generation))
147	144		if err != nil {
148	145		continue

@@ -163,33 +160,32 @@ func (r *Repository) loadNewestGeneration(ctx context.Context, key string, budge
163	160		return got, true, nil
164	161		}
165	162		}
	163	+	return ports.SnapshotGeneration{}, false, nil
166	164		}
167	165
168		-	// nextGeneration finds one candidate in a complete, batched directory cursor
169		-	// pass. Repeating it with the returned number as before preserves strict
170		-	// newest-to-oldest fallback while work remains bounded by budget.
171		-	func (r Repository) nextGeneration(ctx context.Context, key string, before, exclude uint64, budget int) (uint64, bool, error) {
172		-	var newest uint64
	166	+	// generationCandidates reads the generations directory once per load attempt.
	167	+	// The shared traversal budget bounds both retained names and work; sorting the
	168	+	// bounded batch preserves strict newest-to-oldest fallback without rescanning.
	169	+	func (r Repository) generationCandidates(ctx context.Context, key string, exclude uint64, budget int) ([]uint64, error) {
	170	+	candidates := make([]uint64, 0)
173	171		err := r.walkDirectory(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, func(entry os.DirEntry) error {
174	172		if entry.IsDir() {
175	173		return nil
176	174		}
177	175		generation, ok := parseGenerationFilename(entry.Name())
178		-	if !ok \|\| generation == exclude \|\| (before != 0 && generation >= before) {
179		-	return nil
180		-	}
181		-	if generation > newest {
182		-	newest = generation
	176	+	if ok && generation != exclude {
	177	+	candidates = append(candidates, generation)
183	178		}
184	179		return nil
185	180		})
186	181		if errors.Is(err, os.ErrNotExist) {
187		-	return 0, false, nil
	182	+	return candidates, nil
188	183		}
189	184		if err != nil {
190		-	return 0, false, err
	185	+	return nil, err
191	186		}
192		-	return newest, newest != 0, nil
	187	+	sort.Slice(candidates, func(i, j int) bool { return candidates[i] > candidates[j] })
	188	+	return candidates, nil
193	189		}
194	190
195	191		// walkDirectory is the common finite directory traversal primitive for reads.

@@ -266,18 +262,24 @@ func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {
266	262		return out
267	263		}
268	264		func (r *Repository) readHead(key string) (uint64, ports.SnapshotDigest, error) {
269		-	data, err := r.readBounded(r.headPath(key))
	265	+	path := r.headPath(key)
	266	+	if hook := r.hooks.beforeHeadRead; hook != nil {
	267	+	if err := hook(path); err != nil {
	268	+	return 0, ports.SnapshotDigest{}, err
	269	+	}
	270	+	}
	271	+	data, err := r.readBounded(path)
270	272		if err != nil {
271	273		return 0, ports.SnapshotDigest{}, err
272	274		}
273	275		if len(data) != 4+8+sha256.Size \|\| string(data[:4]) != "VEVH" {
274		-	return 0, ports.SnapshotDigest{}, fmt.Errorf("invalid HEAD")
	276	+	return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
275	277		}
276	278		generation := binary.BigEndian.Uint64(data[4:12])
277	279		var digest ports.SnapshotDigest
278	280		copy(digest[:], data[12:])
279	281		if generation == 0 {
280		-	return 0, ports.SnapshotDigest{}, fmt.Errorf("invalid HEAD")
	282	+	return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
281	283		}
282	284		return generation, digest, nil
283	285		}

 	maxDirectoryTraversalEntries = 4096
 )
-var ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
+var (
+	ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
+	ErrInvalidHEAD              = errors.New("invalid snapshot HEAD")
+)
 func (r *Repository) List(ctx context.Context) ([]string, error) {
 	if err := ctx.Err(); err != nil {
 		return ports.SnapshotGeneration{}, err
 	}
 	key := sessionKey(name)
+	lock := r.lockSession(key)
+	defer r.unlockSession(lock)
+	if err := ctx.Err(); err != nil {
+		return ports.SnapshotGeneration{}, err
+	}
 	killed, err := r.tombstoned(name)
 	if err != nil {
 		return ports.SnapshotGeneration{}, fmt.Errorf("read killed session marker: %w", err)
 	if killed {
 		return ports.SnapshotGeneration{}, fmt.Errorf("snapshot session %q is killed", name)
 	}
-	lock := r.lockSession(key)
-	defer r.unlockSession(lock)
-	if err := ctx.Err(); err != nil {
-		return ports.SnapshotGeneration{}, err
-	}
 	return r.load(ctx, name, key)
 }
 func (r *Repository) load(ctx context.Context, name, key string) (ports.SnapshotGeneration, error) {
 	preferred, preferredDigest, err := r.readHead(key)
-	if err != nil && !errors.Is(err, os.ErrNotExist) {
+	skipped := errors.Is(err, ErrInvalidHEAD)
-		preferred = 0
+	if err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
+		return ports.SnapshotGeneration{}, fmt.Errorf("read snapshot HEAD: %w", err)
 	}
-	skipped := false
 	if preferred != 0 {
 		got, err := r.loadGeneration(ctx, name, key, preferred)
 		if err == nil && sha256.Sum256(got.Manifest) == preferredDigest {
 	}
 	budget := maxDirectoryTraversalEntries
-	before := uint64(0) // zero means no upper bound.
+	candidates, err := r.generationCandidates(ctx, key, preferred, &budget)
-	for {
+	if err != nil {
-		generation, ok, err := r.nextGeneration(ctx, key, before, preferred, &budget)
+		return ports.SnapshotGeneration{}, err
-		if err != nil {
+	}
-			return ports.SnapshotGeneration{}, err
+	for _, generation := range candidates {
-		}
-		if !ok {
-			break
-		}
-		before = generation
 		got, err := r.loadGeneration(ctx, name, key, generation)
 		if err != nil {
 			skipped = true
 		if err == nil && sha256.Sum256(got.Manifest) == digest {
 			return generation, got.Manifest, nil
 		}
+	} else if !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
+		return 0, nil, fmt.Errorf("read snapshot HEAD: %w", err)
 	}
 	budget := maxDirectoryTraversalEntries
-	before := uint64(0)
+	candidates, err := r.generationCandidates(ctx, key, 0, &budget)
-	for {
+	if err != nil {
-		generation, ok, err := r.nextGeneration(ctx, key, before, 0, &budget)
+		return 0, nil, err
-		if err != nil || !ok {
+	}
-			return 0, nil, err
+	for _, generation := range candidates {
-		}
-		before = generation
 		got, err := r.loadGeneration(ctx, name, key, generation)
 		if err == nil {
 			return generation, got.Manifest, nil
 		}
 	}
+	return 0, nil, nil
 }
-// loadNewestGeneration validates candidates in descending generation order
+// loadNewestGeneration validates a single bounded candidate batch in strict
-// without materializing the generations directory. It is used by List, whose
+// descending generation order. It is used by List, whose name is discovered
-// name is discovered from each manifest rather than supplied by a caller.
+// from each manifest rather than supplied by a caller.
 func (r *Repository) loadNewestGeneration(ctx context.Context, key string, budget *int) (ports.SnapshotGeneration, bool, error) {
-	before := uint64(0)
+	candidates, err := r.generationCandidates(ctx, key, 0, budget)
-	for {
+	if err != nil {
-		generation, ok, err := r.nextGeneration(ctx, key, before, 0, budget)
+		return ports.SnapshotGeneration{}, false, err
-		if err != nil || !ok {
+	}
-			return ports.SnapshotGeneration{}, false, err
+	for _, generation := range candidates {
-		}
-		before = generation
 		data, err := r.readBounded(r.manifestPath(key, generation))
 		if err != nil {
 			continue
 			return got, true, nil
 		}
 	}
+	return ports.SnapshotGeneration{}, false, nil
 }
-// nextGeneration finds one candidate in a complete, batched directory cursor
+// generationCandidates reads the generations directory once per load attempt.
-// pass. Repeating it with the returned number as before preserves strict
+// The shared traversal budget bounds both retained names and work; sorting the
-// newest-to-oldest fallback while work remains bounded by budget.
+// bounded batch preserves strict newest-to-oldest fallback without rescanning.
-func (r *Repository) nextGeneration(ctx context.Context, key string, before, exclude uint64, budget *int) (uint64, bool, error) {
+func (r *Repository) generationCandidates(ctx context.Context, key string, exclude uint64, budget *int) ([]uint64, error) {
-	var newest uint64
+	candidates := make([]uint64, 0)
 	err := r.walkDirectory(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, func(entry os.DirEntry) error {
 		if entry.IsDir() {
 			return nil
 		}
 		generation, ok := parseGenerationFilename(entry.Name())
-		if !ok || generation == exclude || (before != 0 && generation >= before) {
+		if ok && generation != exclude {
-			return nil
+			candidates = append(candidates, generation)
-		}
-		if generation > newest {
-			newest = generation
 		}
 		return nil
 	})
 	if errors.Is(err, os.ErrNotExist) {
-		return 0, false, nil
+		return candidates, nil
 	}
 	if err != nil {
-		return 0, false, err
+		return nil, err
 	}
-	return newest, newest != 0, nil
+	sort.Slice(candidates, func(i, j int) bool { return candidates[i] > candidates[j] })
+	return candidates, nil
 }
 // walkDirectory is the common finite directory traversal primitive for reads.
 	return out
 }
 func (r *Repository) readHead(key string) (uint64, ports.SnapshotDigest, error) {
-	data, err := r.readBounded(r.headPath(key))
+	path := r.headPath(key)
+	if hook := r.hooks.beforeHeadRead; hook != nil {
+		if err := hook(path); err != nil {
+			return 0, ports.SnapshotDigest{}, err
+		}
+	}
+	data, err := r.readBounded(path)
 	if err != nil {
 		return 0, ports.SnapshotDigest{}, err
 	}
 	if len(data) != 4+8+sha256.Size || string(data[:4]) != "VEVH" {
-		return 0, ports.SnapshotDigest{}, fmt.Errorf("invalid HEAD")
+		return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
 	}
 	generation := binary.BigEndian.Uint64(data[4:12])
 	var digest ports.SnapshotDigest
 	copy(digest[:], data[12:])
 	if generation == 0 {
-		return 0, ports.SnapshotDigest{}, fmt.Errorf("invalid HEAD")
+		return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
 	}
 	return generation, digest, nil
 }

A internal/adapters/snapshot/load_major_test.go +88 -0

@@ -0,0 +1,88 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"os"
	6	+	"syscall"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+	)
	11	+
	12	+	func TestRepositoryLoadPropagatesOperationalHeadErrors(t *testing.T) {
	13	+	repo := NewRepository(privateDir(t))
	14	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	15	+
	16	+	repo.hooks.beforeHeadRead = func(string) error { return os.ErrPermission }
	17	+
	18	+	_, err := repo.Load(context.Background(), "named")
	19	+	require.ErrorIs(t, err, os.ErrPermission)
	20	+	}
	21	+
	22	+	func TestRepositoryCurrentGenerationPropagatesOperationalHeadErrors(t *testing.T) {
	23	+	repo := NewRepository(privateDir(t))
	24	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	25	+
	26	+	repo.hooks.beforeHeadRead = func(string) error { return syscall.EIO }
	27	+
	28	+	_, _, err := repo.currentGeneration(context.Background(), "named", sessionKey("named"))
	29	+	require.ErrorIs(t, err, syscall.EIO)
	30	+	}
	31	+
	32	+	func TestRepositoryLoadFallsBackFromInvalidHead(t *testing.T) {
	33	+	repo := NewRepository(privateDir(t))
	34	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	35	+	require.NoError(t, os.WriteFile(repo.headPath(sessionKey("named")), []byte("invalid"), 0o600))
	36	+
	37	+	got, err := repo.Load(context.Background(), "named")
	38	+	require.NoError(t, err)
	39	+	require.Equal(t, uint64(1), got.Generation)
	40	+	require.True(t, got.Fallback)
	41	+	}
	42	+
	43	+	func TestRepositoryLoadFallsBackAcrossMultipleCorruptCandidates(t *testing.T) {
	44	+	repo := NewRepository(privateDir(t))
	45	+	for generation := uint64(1); generation <= 4; generation++ {
	46	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})))
	47	+	}
	48	+	key := sessionKey("named")
	49	+	for _, generation := range []uint64{4, 3} {
	50	+	publication := repositoryPublication(t, "named", generation, []byte{byte(generation)})
	51	+	require.NoError(t, os.Remove(repo.objectPath(key, publication.Objects[0].Digest)))
	52	+	}
	53	+
	54	+	got, err := repo.Load(context.Background(), "named")
	55	+	require.NoError(t, err)
	56	+	require.Equal(t, uint64(2), got.Generation)
	57	+	require.True(t, got.Fallback)
	58	+	}
	59	+
	60	+	func TestRepositoryLoadHoldsSessionLockAcrossTombstoneCheck(t *testing.T) {
	61	+	repo := NewRepository(privateDir(t))
	62	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	63	+
	64	+	tombstoneCheckStarted := make(chan struct{})
	65	+	releaseTombstoneCheck := make(chan struct{})
	66	+	repo.hooks.beforeTombstoneCheck = func(string) {
	67	+	close(tombstoneCheckStarted)
	68	+	<-releaseTombstoneCheck
	69	+	}
	70	+	loadDone := make(chan error, 1)
	71	+	go func() {
	72	+	_, err := repo.Load(context.Background(), "named")
	73	+	loadDone <- err
	74	+	}()
	75	+	<-tombstoneCheckStarted
	76	+
	77	+	tombstoneDone := make(chan error, 1)
	78	+	go func() { tombstoneDone <- repo.Tombstone(context.Background(), "named") }()
	79	+	select {
	80	+	case err := <-tombstoneDone:
	81	+	t.Fatalf("Tombstone completed before Load released the session lock: %v", err)
	82	+	default:
	83	+	}
	84	+
	85	+	close(releaseTombstoneCheck)
	86	+	require.NoError(t, <-loadDone)
	87	+	require.NoError(t, <-tombstoneDone)
	88	+	}

@@ -0,0 +1,88 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"context"
	5	+	"os"
	6	+	"syscall"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+	)
	11	+
	12	+	func TestRepositoryLoadPropagatesOperationalHeadErrors(t *testing.T) {
	13	+	repo := NewRepository(privateDir(t))
	14	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	15	+
	16	+	repo.hooks.beforeHeadRead = func(string) error { return os.ErrPermission }
	17	+
	18	+	_, err := repo.Load(context.Background(), "named")
	19	+	require.ErrorIs(t, err, os.ErrPermission)
	20	+	}
	21	+
	22	+	func TestRepositoryCurrentGenerationPropagatesOperationalHeadErrors(t *testing.T) {
	23	+	repo := NewRepository(privateDir(t))
	24	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	25	+
	26	+	repo.hooks.beforeHeadRead = func(string) error { return syscall.EIO }
	27	+
	28	+	_, _, err := repo.currentGeneration(context.Background(), "named", sessionKey("named"))
	29	+	require.ErrorIs(t, err, syscall.EIO)
	30	+	}
	31	+
	32	+	func TestRepositoryLoadFallsBackFromInvalidHead(t *testing.T) {
	33	+	repo := NewRepository(privateDir(t))
	34	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	35	+	require.NoError(t, os.WriteFile(repo.headPath(sessionKey("named")), []byte("invalid"), 0o600))
	36	+
	37	+	got, err := repo.Load(context.Background(), "named")
	38	+	require.NoError(t, err)
	39	+	require.Equal(t, uint64(1), got.Generation)
	40	+	require.True(t, got.Fallback)
	41	+	}
	42	+
	43	+	func TestRepositoryLoadFallsBackAcrossMultipleCorruptCandidates(t *testing.T) {
	44	+	repo := NewRepository(privateDir(t))
	45	+	for generation := uint64(1); generation <= 4; generation++ {
	46	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})))
	47	+	}
	48	+	key := sessionKey("named")
	49	+	for _, generation := range []uint64{4, 3} {
	50	+	publication := repositoryPublication(t, "named", generation, []byte{byte(generation)})
	51	+	require.NoError(t, os.Remove(repo.objectPath(key, publication.Objects[0].Digest)))
	52	+	}
	53	+
	54	+	got, err := repo.Load(context.Background(), "named")
	55	+	require.NoError(t, err)
	56	+	require.Equal(t, uint64(2), got.Generation)
	57	+	require.True(t, got.Fallback)
	58	+	}
	59	+
	60	+	func TestRepositoryLoadHoldsSessionLockAcrossTombstoneCheck(t *testing.T) {
	61	+	repo := NewRepository(privateDir(t))
	62	+	require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))))
	63	+
	64	+	tombstoneCheckStarted := make(chan struct{})
	65	+	releaseTombstoneCheck := make(chan struct{})
	66	+	repo.hooks.beforeTombstoneCheck = func(string) {
	67	+	close(tombstoneCheckStarted)
	68	+	<-releaseTombstoneCheck
	69	+	}
	70	+	loadDone := make(chan error, 1)
	71	+	go func() {
	72	+	_, err := repo.Load(context.Background(), "named")
	73	+	loadDone <- err
	74	+	}()
	75	+	<-tombstoneCheckStarted
	76	+
	77	+	tombstoneDone := make(chan error, 1)
	78	+	go func() { tombstoneDone <- repo.Tombstone(context.Background(), "named") }()
	79	+	select {
	80	+	case err := <-tombstoneDone:
	81	+	t.Fatalf("Tombstone completed before Load released the session lock: %v", err)
	82	+	default:
	83	+	}
	84	+
	85	+	close(releaseTombstoneCheck)
	86	+	require.NoError(t, <-loadDone)
	87	+	require.NoError(t, <-tombstoneDone)
	88	+	}

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


@@ -59,7 +59,9 @@ type repositoryHooks struct {
59	59		beforeObjectRead func(string)
60	60		beforeObjectHash func([]byte)
61	61		beforeObjectCopy func([]byte)
	62	+	beforeHeadRead func(string) error
62	63		beforeHeadWrite func(string) error
	64	+	beforeTombstoneCheck func(string)
63	65		beforePendingQuarantineCheck func(string)
64	66		createTemp func(string) error
65	67		writeTemp func(string) error


@@ -59,7 +59,9 @@ type repositoryHooks struct {
59	beforeObjectRead func(string)	59		beforeObjectRead func(string)
60	beforeObjectHash func([]byte)	60		beforeObjectHash func([]byte)
61	beforeObjectCopy func([]byte)	61		beforeObjectCopy func([]byte)
		62	+	beforeHeadRead func(string) error
62	beforeHeadWrite func(string) error	63		beforeHeadWrite func(string) error
		64	+	beforeTombstoneCheck func(string)
63	beforePendingQuarantineCheck func(string)	65		beforePendingQuarantineCheck func(string)
64	createTemp func(string) error	66		createTemp func(string) error
65	writeTemp func(string) error	67		writeTemp func(string) error