fix(snapshot): invalidate stale gc marks

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

4 files · +127 -2

M internal/adapters/snapshot/maintenance.go +18 -1


@@ -70,6 +70,9 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
70	70		if err := ctx.Err(); err != nil {
71	71		return err
72	72		}
	73	+	// A renamed session can be replaced by a later publication. Invalidate any
	74	+	// mark made for the old namespace before that replacement becomes possible.
	75	+	r.invalidateStorageEpoch(key)
73	76		if err := r.rename(canonical, quarantine); err != nil {
74	77		return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
75	78		}

@@ -191,6 +194,7 @@ type manifestMaintenance struct {
191	194
192	195		type sessionMaintenance struct {
193	196		token string
	197	+	epoch uint64
194	198		marked map[uint64]manifestMaintenance
195	199		uncertain bool
196	200		conservative bool

@@ -380,6 +384,12 @@ func (r Repository) maintainSession(ctx context.Context, key string, budget in
380	384		return nil
381	385		}
382	386		}
	387	+	if state.epoch != r.storageEpoch(key) {
	388	+	// A publication may have left an unpublished manifest after this mark.
	389	+	// Discard all mark and sweep cursors; the next pass starts from it.
	390	+	r.clearSessionMaintenance(key)
	391	+	return nil
	392	+	}
383	393		if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil \|\| *budget == 0 {
384	394		return err
385	395		}

@@ -401,10 +411,11 @@ func (r Repository) sessionMaintenanceState(key string) (sessionMaintenance, e
401	411		return nil, err
402	412		}
403	413		state := r.maintenanceSessions[key]
404		-	if state == nil \|\| state.token != token {
	414	+	if state == nil \|\| state.token != token \|\| state.epoch != r.storageEpoch(key) {
405	415		r.clearSessionMaintenance(key)
406	416		state = &sessionMaintenance{
407	417		token: token,
	418	+	epoch: r.storageEpoch(key),
408	419		conservative: conservative,
409	420		marked: make(map[uint64]manifestMaintenance),
410	421		}

@@ -542,6 +553,12 @@ func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, st
542	553		}
543	554
544	555		func (r Repository) sweepSession(ctx context.Context, key string, state sessionMaintenance, budget *int) error {
	556	+	// Every bounded sweep batch is tied to the mark's storage epoch. Do not use
	557	+	// stale references after any publication or replacement.
	558	+	if state.epoch != r.storageEpoch(key) {
	559	+	r.clearSessionMaintenance(key)
	560	+	return nil
	561	+	}
545	562		referenced := retainedReferences(state.marked, state.conservative)
546	563		root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
547	564		if !state.sweepRootDone {

 	if err := ctx.Err(); err != nil {
 		return err
 	}
+	// A renamed session can be replaced by a later publication. Invalidate any
+	// mark made for the old namespace before that replacement becomes possible.
+	r.invalidateStorageEpoch(key)
 	if err := r.rename(canonical, quarantine); err != nil {
 		return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
 	}
 type sessionMaintenance struct {
 	token         string
+	epoch         uint64
 	marked        map[uint64]manifestMaintenance
 	uncertain     bool
 	conservative  bool
 			return nil
 		}
 	}
+	if state.epoch != r.storageEpoch(key) {
+		// A publication may have left an unpublished manifest after this mark.
+		// Discard all mark and sweep cursors; the next pass starts from it.
+		r.clearSessionMaintenance(key)
+		return nil
+	}
 	if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
 		return err
 	}
 		return nil, err
 	}
 	state := r.maintenanceSessions[key]
-	if state == nil || state.token != token {
+	if state == nil || state.token != token || state.epoch != r.storageEpoch(key) {
 		r.clearSessionMaintenance(key)
 		state = &sessionMaintenance{
 			token:        token,
+			epoch:        r.storageEpoch(key),
 			conservative: conservative,
 			marked:       make(map[uint64]manifestMaintenance),
 		}
 }
 func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
+	// Every bounded sweep batch is tied to the mark's storage epoch. Do not use
+	// stale references after any publication or replacement.
+	if state.epoch != r.storageEpoch(key) {
+		r.clearSessionMaintenance(key)
+		return nil
+	}
 	referenced := retainedReferences(state.marked, state.conservative)
 	root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
 	if !state.sweepRootDone {

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


@@ -530,3 +530,93 @@ func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testin
530	530		})
531	531		}
532	532		}
	533	+
	534	+	func TestRepositoryMaintainRestartsMarkAfterFailedPublication(t *testing.T) {
	535	+	repo := NewRepository(privateDir(t))
	536	+	key := sessionKey("named")
	537	+	first := publicationWithTailShard(t, "named", 1, "ff")
	538	+	stalePaths := make([]string, 0, maintenanceBatch+1)
	539	+	for i := 0; i < maintenanceBatch+1; i++ {
	540	+	digest := digestInShard(t, "00", i)
	541	+	path := repo.objectPath(key, digest)
	542	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
	543	+	t.Fatal(err)
	544	+	}
	545	+	if err := os.WriteFile(path, []byte("stale"), 0o600); err != nil {
	546	+	t.Fatal(err)
	547	+	}
	548	+	stalePaths = append(stalePaths, path)
	549	+	}
	550	+	if err := repo.Publish(context.Background(), first); err != nil {
	551	+	t.Fatal(err)
	552	+	}
	553	+
	554	+	// This completes marking but exhausts the sweep budget in the stale shard,
	555	+	// leaving the next sweep batch pending.
	556	+	if err := repo.Maintain(context.Background()); err != nil {
	557	+	t.Fatal(err)
	558	+	}
	559	+	staleRemaining := false
	560	+	for _, path := range stalePaths {
	561	+	if _, err := os.Lstat(path); err == nil {
	562	+	staleRemaining = true
	563	+	break
	564	+	}
	565	+	}
	566	+	if !staleRemaining {
	567	+	t.Fatal("sweep did not leave a pending stale object")
	568	+	}
	569	+
	570	+	second := publicationWithTailShard(t, "named", 2, "ff")
	571	+	injected := errors.New("before HEAD")
	572	+	repo.hooks.beforeHeadWrite = func(string) error { return injected }
	573	+	if err := repo.Publish(context.Background(), second); !errors.Is(err, injected) {
	574	+	t.Fatalf("Publish error = %v, want injected failure", err)
	575	+	}
	576	+	repo.hooks.beforeHeadWrite = nil
	577	+
	578	+	for i := 0; i < 20; i++ {
	579	+	if err := repo.Maintain(context.Background()); err != nil {
	580	+	t.Fatal(err)
	581	+	}
	582	+	for _, object := range second.Objects {
	583	+	if _, err := os.Lstat(repo.objectPath(key, object.Digest)); err != nil {
	584	+	t.Fatalf("object referenced by failed publication removed on pass %d: %v", i, err)
	585	+	}
	586	+	}
	587	+	staleRemaining = false
	588	+	for _, path := range stalePaths {
	589	+	if _, err := os.Lstat(path); err == nil {
	590	+	staleRemaining = true
	591	+	break
	592	+	}
	593	+	}
	594	+	}
	595	+	if staleRemaining {
	596	+	t.Fatal("stale object was not eventually swept")
	597	+	}
	598	+	}
	599	+
	600	+	func publicationWithTailShard(t *testing.T, name string, generation uint64, shard string) ports.SnapshotPublication {
	601	+	t.Helper()
	602	+	for i := 0; i < 4096; i++ {
	603	+	publication := repositoryPublication(t, name, generation, []byte(fmt.Sprintf("state-%d-%d", generation, i)))
	604	+	if fmt.Sprintf("%02x", publication.Objects[0].Digest[0]) == shard {
	605	+	return publication
	606	+	}
	607	+	}
	608	+	t.Fatalf("did not find publication tail in shard %q", shard)
	609	+	return ports.SnapshotPublication{}
	610	+	}
	611	+
	612	+	func digestInShard(t *testing.T, shard string, n int) ports.SnapshotDigest {
	613	+	t.Helper()
	614	+	for i := 0; i < 4096; i++ {
	615	+	digest := sha256.Sum256([]byte(fmt.Sprintf("stale-%d-%d", n, i)))
	616	+	if fmt.Sprintf("%02x", digest[0]) == shard {
	617	+	return digest
	618	+	}
	619	+	}
	620	+	t.Fatalf("did not find stale digest in shard %q", shard)
	621	+	return ports.SnapshotDigest{}
	622	+	}


@@ -530,3 +530,93 @@ func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testin
530	})	530		})
531	}	531		}
532	}	532		}
		533	+
		534	+	func TestRepositoryMaintainRestartsMarkAfterFailedPublication(t *testing.T) {
		535	+	repo := NewRepository(privateDir(t))
		536	+	key := sessionKey("named")
		537	+	first := publicationWithTailShard(t, "named", 1, "ff")
		538	+	stalePaths := make([]string, 0, maintenanceBatch+1)
		539	+	for i := 0; i < maintenanceBatch+1; i++ {
		540	+	digest := digestInShard(t, "00", i)
		541	+	path := repo.objectPath(key, digest)
		542	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		543	+	t.Fatal(err)
		544	+	}
		545	+	if err := os.WriteFile(path, []byte("stale"), 0o600); err != nil {
		546	+	t.Fatal(err)
		547	+	}
		548	+	stalePaths = append(stalePaths, path)
		549	+	}
		550	+	if err := repo.Publish(context.Background(), first); err != nil {
		551	+	t.Fatal(err)
		552	+	}
		553	+
		554	+	// This completes marking but exhausts the sweep budget in the stale shard,
		555	+	// leaving the next sweep batch pending.
		556	+	if err := repo.Maintain(context.Background()); err != nil {
		557	+	t.Fatal(err)
		558	+	}
		559	+	staleRemaining := false
		560	+	for _, path := range stalePaths {
		561	+	if _, err := os.Lstat(path); err == nil {
		562	+	staleRemaining = true
		563	+	break
		564	+	}
		565	+	}
		566	+	if !staleRemaining {
		567	+	t.Fatal("sweep did not leave a pending stale object")
		568	+	}
		569	+
		570	+	second := publicationWithTailShard(t, "named", 2, "ff")
		571	+	injected := errors.New("before HEAD")
		572	+	repo.hooks.beforeHeadWrite = func(string) error { return injected }
		573	+	if err := repo.Publish(context.Background(), second); !errors.Is(err, injected) {
		574	+	t.Fatalf("Publish error = %v, want injected failure", err)
		575	+	}
		576	+	repo.hooks.beforeHeadWrite = nil
		577	+
		578	+	for i := 0; i < 20; i++ {
		579	+	if err := repo.Maintain(context.Background()); err != nil {
		580	+	t.Fatal(err)
		581	+	}
		582	+	for _, object := range second.Objects {
		583	+	if _, err := os.Lstat(repo.objectPath(key, object.Digest)); err != nil {
		584	+	t.Fatalf("object referenced by failed publication removed on pass %d: %v", i, err)
		585	+	}
		586	+	}
		587	+	staleRemaining = false
		588	+	for _, path := range stalePaths {
		589	+	if _, err := os.Lstat(path); err == nil {
		590	+	staleRemaining = true
		591	+	break
		592	+	}
		593	+	}
		594	+	}
		595	+	if staleRemaining {
		596	+	t.Fatal("stale object was not eventually swept")
		597	+	}
		598	+	}
		599	+
		600	+	func publicationWithTailShard(t *testing.T, name string, generation uint64, shard string) ports.SnapshotPublication {
		601	+	t.Helper()
		602	+	for i := 0; i < 4096; i++ {
		603	+	publication := repositoryPublication(t, name, generation, []byte(fmt.Sprintf("state-%d-%d", generation, i)))
		604	+	if fmt.Sprintf("%02x", publication.Objects[0].Digest[0]) == shard {
		605	+	return publication
		606	+	}
		607	+	}
		608	+	t.Fatalf("did not find publication tail in shard %q", shard)
		609	+	return ports.SnapshotPublication{}
		610	+	}
		611	+
		612	+	func digestInShard(t *testing.T, shard string, n int) ports.SnapshotDigest {
		613	+	t.Helper()
		614	+	for i := 0; i < 4096; i++ {
		615	+	digest := sha256.Sum256([]byte(fmt.Sprintf("stale-%d-%d", n, i)))
		616	+	if fmt.Sprintf("%02x", digest[0]) == shard {
		617	+	return digest
		618	+	}
		619	+	}
		620	+	t.Fatalf("did not find stale digest in shard %q", shard)
		621	+	return ports.SnapshotDigest{}
		622	+	}

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


@@ -28,6 +28,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
28	28		if err := ctx.Err(); err != nil {
29	29		return err
30	30		}
	31	+	// A failed publication can still leave immutable blobs or a manifest behind.
	32	+	// Invalidate an in-progress GC mark before creating any publication storage.
	33	+	r.invalidateStorageEpoch(key)
31	34		if err := r.ensureSession(key); err != nil {
32	35		return err
33	36		}


@@ -28,6 +28,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
28	if err := ctx.Err(); err != nil {	28		if err := ctx.Err(); err != nil {
29	return err	29		return err
30	}	30		}
		31	+	// A failed publication can still leave immutable blobs or a manifest behind.
		32	+	// Invalidate an in-progress GC mark before creating any publication storage.
		33	+	r.invalidateStorageEpoch(key)
31	if err := r.ensureSession(key); err != nil {	34		if err := r.ensureSession(key); err != nil {
32	return err	35		return err
33	}	36		}

M internal/adapters/snapshot/repository.go +16 -1


@@ -26,6 +26,11 @@ type Repository struct {
26	26		locks sync.Map // map[string]*sync.Mutex
27	27		hooks repositoryHooks
28	28
	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
	33	+
29	34		// maintenanceMu owns continuation state for bounded maintenance. Keeping
30	35		// directory handles on the repository makes successive calls advance rather
31	36		// than repeatedly reading the first batch of a large directory.

@@ -61,7 +66,17 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
61	66
62	67		// NewRepository creates a repository rooted at dir. It does not create files
63	68		// until the first publication, so merely constructing it is side-effect free.
64		-	func NewRepository(dir string) *Repository { return &Repository{dir: dir} }
	69	+	func NewRepository(dir string) *Repository {
	70	+	return &Repository{dir: dir, storageEpochs: make(map[string]uint64)}
	71	+	}
	72	+
	73	+	// invalidateStorageEpoch must be called with sessionLock(key) held.
	74	+	func (r *Repository) invalidateStorageEpoch(key string) {
	75	+	r.storageEpochs[key]++
	76	+	}
	77	+
	78	+	// storageEpoch must be called with sessionLock(key) held.
	79	+	func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
65	80
66	81		func (r Repository) sessionLock(key string) sync.Mutex {
67	82		lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{})


@@ -26,6 +26,11 @@ type Repository struct {
26		locks sync.Map // map[string]*sync.Mutex	26		locks sync.Map // map[string]*sync.Mutex
27		hooks repositoryHooks	27		hooks repositoryHooks
28			28
			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
			33	+
29		// maintenanceMu owns continuation state for bounded maintenance. Keeping	34		// maintenanceMu owns continuation state for bounded maintenance. Keeping
30		// directory handles on the repository makes successive calls advance rather	35		// directory handles on the repository makes successive calls advance rather
31		// than repeatedly reading the first batch of a large directory.	36		// than repeatedly reading the first batch of a large directory.

@@ -61,7 +66,17 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
61			66
62		// NewRepository creates a repository rooted at dir. It does not create files	67		// NewRepository creates a repository rooted at dir. It does not create files
63		// until the first publication, so merely constructing it is side-effect free.	68		// until the first publication, so merely constructing it is side-effect free.
64	-	func NewRepository(dir string) *Repository { return &Repository{dir: dir} }	69	+	func NewRepository(dir string) *Repository {
			70	+	return &Repository{dir: dir, storageEpochs: make(map[string]uint64)}
			71	+	}
			72	+
			73	+	// invalidateStorageEpoch must be called with sessionLock(key) held.
			74	+	func (r *Repository) invalidateStorageEpoch(key string) {
			75	+	r.storageEpochs[key]++
			76	+	}
			77	+
			78	+	// storageEpoch must be called with sessionLock(key) held.
			79	+	func (r *Repository) storageEpoch(key string) uint64 { return r.storageEpochs[key] }
65			80
66		func (r Repository) sessionLock(key string) sync.Mutex {	81		func (r Repository) sessionLock(key string) sync.Mutex {
67		lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{})	82		lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{})