detached ecf6fef

fix(snapshot): rewrite any HEAD that is not the committed checkpoint during GC

GC holds the mutation fence, so HEAD naming an older retained generation, a mismatched digest, or a malformed file is damage that blocks the next live publication. Rewrite it to the validated commit. Make the per-incarnation keep re-read test deterministic.
commit
ecf6fef00deb87b304c939c7430306d7badd7cad
author
brice · · 2026-10-02 08:03 UTC
parent
1401d80
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
refs
perf/daemon-startup
3 files · +40 -25
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance_gc.go +11 -11
@@ -88,7 +88,7 @@ func (r *Repository) pruneGenerationsLocked(ctx context.Context, id domain.Incar
8888 }
8989 // Reconcile HEAD before removing generations, so an interrupted pass never
9090 // leaves HEAD naming a removed manifest.
91- if err := r.reconcileHeadForRetention(id, kept, committed); err != nil {
91+ if err := r.reconcileHeadForRetention(id, committed); err != nil {
9292 return err
9393 }
9494
@@ -117,19 +117,19 @@ func (r *Repository) pruneGenerationsLocked(ctx context.Context, id domain.Incar
117117 return r.sweepUnreferencedObjects(ctx, id, references)
118118 }
119119
120-// reconcileHeadForRetention makes HEAD agree with the retained generations.
121-// With a committed checkpoint, a missing HEAD or one naming a discarded
122-// generation (such as a forward orphan) is rewound to the commit: GC runs while
123-// the daemon serves, and a live session's next publication must find its
124-// committed parent. Without one, no generation is retained and HEAD is removed
125-// so a crash-published HEAD cannot make a retry conflict with an orphan.
126-func (r *Repository) reconcileHeadForRetention(id domain.IncarnationID, kept map[uint64]struct{}, committed domain.CheckpointRef) error {
120+// reconcileHeadForRetention makes HEAD name the committed checkpoint. GC runs
121+// under the mutation fence, so no publication is in flight and any other HEAD
122+// (missing, malformed, a forward orphan, an older generation, or a digest
123+// mismatch) is damage: a live session's next publication must find its
124+// committed parent. Without a commit, no generation is retained and HEAD is
125+// removed so a crash-published HEAD cannot make a retry conflict with an orphan.
126+func (r *Repository) reconcileHeadForRetention(id domain.IncarnationID, committed domain.CheckpointRef) error {
127127 if committed.Generation > 0 {
128- generation, _, err := r.readHead(id)
129- if err != nil && !errors.Is(err, os.ErrNotExist) {
128+ generation, digest, err := r.readHead(id)
129+ if err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
130130 return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
131131 }
132- if _, retained := kept[generation]; err == nil && retained {
132+ if err == nil && generation == committed.Generation && digest == committed.ManifestDigest {
133133 return nil
134134 }
135135 // referencesForGenerations already validated the committed manifest digest.
@@ -88,7 +88,7 @@ func (r *Repository) pruneGenerationsLocked(ctx context.Context, id domain.Incar
88 }88 }
89 // Reconcile HEAD before removing generations, so an interrupted pass never89 // Reconcile HEAD before removing generations, so an interrupted pass never
90 // leaves HEAD naming a removed manifest.90 // leaves HEAD naming a removed manifest.
91- if err := r.reconcileHeadForRetention(id, kept, committed); err != nil {91+ if err := r.reconcileHeadForRetention(id, committed); err != nil {
92 return err92 return err
93 }93 }
94 94
@@ -117,19 +117,19 @@ func (r *Repository) pruneGenerationsLocked(ctx context.Context, id domain.Incar
117 return r.sweepUnreferencedObjects(ctx, id, references)117 return r.sweepUnreferencedObjects(ctx, id, references)
118 }118 }
119 119
120-// reconcileHeadForRetention makes HEAD agree with the retained generations.120+// reconcileHeadForRetention makes HEAD name the committed checkpoint. GC runs
121-// With a committed checkpoint, a missing HEAD or one naming a discarded121+// under the mutation fence, so no publication is in flight and any other HEAD
122-// generation (such as a forward orphan) is rewound to the commit: GC runs while122+// (missing, malformed, a forward orphan, an older generation, or a digest
123-// the daemon serves, and a live session's next publication must find its123+// mismatch) is damage: a live session's next publication must find its
124-// committed parent. Without one, no generation is retained and HEAD is removed124+// committed parent. Without a commit, no generation is retained and HEAD is
125-// so a crash-published HEAD cannot make a retry conflict with an orphan.125+// removed so a crash-published HEAD cannot make a retry conflict with an orphan.
126-func (r *Repository) reconcileHeadForRetention(id domain.IncarnationID, kept map[uint64]struct{}, committed domain.CheckpointRef) error {126+func (r *Repository) reconcileHeadForRetention(id domain.IncarnationID, committed domain.CheckpointRef) error {
127 if committed.Generation > 0 {127 if committed.Generation > 0 {
128- generation, _, err := r.readHead(id)128+ generation, digest, err := r.readHead(id)
129- if err != nil && !errors.Is(err, os.ErrNotExist) {129+ if err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, ErrInvalidHEAD) {
130 return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))130 return fmt.Errorf("read snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
131 }131 }
132- if _, retained := kept[generation]; err == nil && retained {132+ if err == nil && generation == committed.Generation && digest == committed.ManifestDigest {
133 return nil133 return nil
134 }134 }
135 // referencesForGenerations already validated the committed manifest digest.135 // referencesForGenerations already validated the committed manifest digest.
M internal/adapters/snapshot/maintenance_gc_test.go +13 -0
@@ -53,6 +53,19 @@ func TestCollectIncarnationGarbageKeepsHeadPublishable(t *testing.T) {
5353 {name: "HEAD naming a removed generation is rewound", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
5454 require.NoError(t, os.Remove(repository.manifestPath(id, 4)))
5555 }},
56+ // Generation 2 is retained as the committed parent, but HEAD must still
57+ // name the commit or the next publication is rejected.
58+ {name: "HEAD naming a retained older generation is moved forward", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
59+ manifest, err := os.ReadFile(repository.manifestPath(id, 2))
60+ require.NoError(t, err)
61+ require.NoError(t, os.WriteFile(repository.headPath(id), marshalHead(2, sha256.Sum256(manifest)), 0o600))
62+ }},
63+ {name: "HEAD with a mismatched digest is rewritten", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
64+ require.NoError(t, os.WriteFile(repository.headPath(id), marshalHead(3, ports.SnapshotDigest{1}), 0o600))
65+ }},
66+ {name: "malformed HEAD is restored", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
67+ require.NoError(t, os.WriteFile(repository.headPath(id), []byte("corrupt"), 0o600))
68+ }},
5669 }
5770 for _, tt := range tests {
5871 t.Run(tt.name, func(t *testing.T) {
@@ -53,6 +53,19 @@ func TestCollectIncarnationGarbageKeepsHeadPublishable(t *testing.T) {
53 {name: "HEAD naming a removed generation is rewound", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {53 {name: "HEAD naming a removed generation is rewound", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
54 require.NoError(t, os.Remove(repository.manifestPath(id, 4)))54 require.NoError(t, os.Remove(repository.manifestPath(id, 4)))
55 }},55 }},
56+ // Generation 2 is retained as the committed parent, but HEAD must still
57+ // name the commit or the next publication is rejected.
58+ {name: "HEAD naming a retained older generation is moved forward", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
59+ manifest, err := os.ReadFile(repository.manifestPath(id, 2))
60+ require.NoError(t, err)
61+ require.NoError(t, os.WriteFile(repository.headPath(id), marshalHead(2, sha256.Sum256(manifest)), 0o600))
62+ }},
63+ {name: "HEAD with a mismatched digest is rewritten", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
64+ require.NoError(t, os.WriteFile(repository.headPath(id), marshalHead(3, ports.SnapshotDigest{1}), 0o600))
65+ }},
66+ {name: "malformed HEAD is restored", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
67+ require.NoError(t, os.WriteFile(repository.headPath(id), []byte("corrupt"), 0o600))
68+ }},
56 }69 }
57 for _, tt := range tests {70 for _, tt := range tests {
58 t.Run(tt.name, func(t *testing.T) {71 t.Run(tt.name, func(t *testing.T) {
M internal/usecase/recovery/transactions_test.go +16 -14
@@ -4,6 +4,7 @@ import (
44 "bytes"
55 "context"
66 "errors"
7+ "slices"
78 "sync"
89 "testing"
910
@@ -141,6 +142,8 @@ func (r *garbageCollectionFenceRepository) SnapshotIncarnations(context.Context)
141142 for id := range seen {
142143 ids = append(ids, id)
143144 }
145+ // Deterministic order lets tests pause between known steps.
146+ slices.SortFunc(ids, func(a, b domain.IncarnationID) int { return bytes.Compare(a[:], b[:]) })
144147 return ids, nil
145148 }
146149
@@ -408,34 +411,32 @@ func TestGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *test
408411
409412 t.Run("a keep decision is re-read under the fence for every incarnation", func(t *testing.T) {
410413 t.Parallel()
411- // The pass lists incarnation 9 (an orphan) before "new" is created with
412- // that same ID; its step must observe the creation and keep it.
414+ // The pass lists incarnation 9 (an orphan) and pauses after step 1;
415+ // "new" is then created with ID 9, so step 9 must re-read the catalogue
416+ // and keep it. A keep decision read once per pass would remove it.
413417 reused := domain.IncarnationID{9}
414418 catalogue := newTransactionCatalogue()
415419 repository := &garbageCollectionFenceRepository{
416- collectionEntered: make(chan struct{}),
417- releaseCollection: make(chan struct{}),
418- incarnations: map[domain.IncarnationID]struct{}{{1}: {}, reused: {}},
420+ blocked: true,
421+ incarnations: map[domain.IncarnationID]struct{}{{1}: {}, reused: {}},
419422 }
420423 catalogue.onCreate = repository.addIncarnation
421424 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
425+ ctx := &betweenStepsContext{Context: context.Background(), repository: repository, reached: make(chan struct{}), release: make(chan struct{})}
422426
423427 collected := make(chan error, 1)
424428 go func() {
425- _, err := coordinator.CollectGarbage(context.Background())
429+ _, err := coordinator.CollectGarbage(ctx)
426430 collected <- err
427431 }()
428- <-repository.collectionEntered
432+ <-ctx.reached
433+ require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "step 1 ran first")
429434
430- created := make(chan error, 1)
431- go func() {
432- _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new", IncarnationID: reused})
433- created <- err
434- }()
435- close(repository.releaseCollection)
435+ _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new", IncarnationID: reused})
436+ require.NoError(t, err)
437+ close(ctx.release)
436438
437439 require.NoError(t, <-collected)
438- require.NoError(t, <-created)
439440 require.True(t, repository.hasIncarnation(reused), "a step after the creation must keep the new incarnation")
440441 require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "a true orphan is still removed")
441442 })
@@ -509,6 +510,7 @@ func TestDurableMutationWaitsForAtMostOneIncarnation(t *testing.T) {
509510 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
510511 work := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
511512 broken := degradedTransactionRecord()
513+ broken.IncarnationID = domain.IncarnationID{2}
512514 tests := []struct {
513515 name string
514516 mutate func(*Coordinator) error
@@ -4,6 +4,7 @@ import (
4 "bytes"4 "bytes"
5 "context"5 "context"
6 "errors"6 "errors"
7+ "slices"
7 "sync"8 "sync"
8 "testing"9 "testing"
9 10
@@ -141,6 +142,8 @@ func (r *garbageCollectionFenceRepository) SnapshotIncarnations(context.Context)
141 for id := range seen {142 for id := range seen {
142 ids = append(ids, id)143 ids = append(ids, id)
143 }144 }
145+ // Deterministic order lets tests pause between known steps.
146+ slices.SortFunc(ids, func(a, b domain.IncarnationID) int { return bytes.Compare(a[:], b[:]) })
144 return ids, nil147 return ids, nil
145 }148 }
146 149
@@ -408,34 +411,32 @@ func TestGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *test
408 411
409 t.Run("a keep decision is re-read under the fence for every incarnation", func(t *testing.T) {412 t.Run("a keep decision is re-read under the fence for every incarnation", func(t *testing.T) {
410 t.Parallel()413 t.Parallel()
411- // The pass lists incarnation 9 (an orphan) before "new" is created with414+ // The pass lists incarnation 9 (an orphan) and pauses after step 1;
412- // that same ID; its step must observe the creation and keep it.415+ // "new" is then created with ID 9, so step 9 must re-read the catalogue
416+ // and keep it. A keep decision read once per pass would remove it.
413 reused := domain.IncarnationID{9}417 reused := domain.IncarnationID{9}
414 catalogue := newTransactionCatalogue()418 catalogue := newTransactionCatalogue()
415 repository := &garbageCollectionFenceRepository{419 repository := &garbageCollectionFenceRepository{
416- collectionEntered: make(chan struct{}),420+ blocked: true,
417- releaseCollection: make(chan struct{}),421+ incarnations: map[domain.IncarnationID]struct{}{{1}: {}, reused: {}},
418- incarnations: map[domain.IncarnationID]struct{}{{1}: {}, reused: {}},
419 }422 }
420 catalogue.onCreate = repository.addIncarnation423 catalogue.onCreate = repository.addIncarnation
421 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))424 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
425+ ctx := &betweenStepsContext{Context: context.Background(), repository: repository, reached: make(chan struct{}), release: make(chan struct{})}
422 426
423 collected := make(chan error, 1)427 collected := make(chan error, 1)
424 go func() {428 go func() {
425- _, err := coordinator.CollectGarbage(context.Background())429+ _, err := coordinator.CollectGarbage(ctx)
426 collected <- err430 collected <- err
427 }()431 }()
428- <-repository.collectionEntered432+ <-ctx.reached
433+ require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "step 1 ran first")
429 434
430- created := make(chan error, 1)435+ _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new", IncarnationID: reused})
431- go func() {436+ require.NoError(t, err)
432- _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new", IncarnationID: reused})437+ close(ctx.release)
433- created <- err
434- }()
435- close(repository.releaseCollection)
436 438
437 require.NoError(t, <-collected)439 require.NoError(t, <-collected)
438- require.NoError(t, <-created)
439 require.True(t, repository.hasIncarnation(reused), "a step after the creation must keep the new incarnation")440 require.True(t, repository.hasIncarnation(reused), "a step after the creation must keep the new incarnation")
440 require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "a true orphan is still removed")441 require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "a true orphan is still removed")
441 })442 })
@@ -509,6 +510,7 @@ func TestDurableMutationWaitsForAtMostOneIncarnation(t *testing.T) {
509 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}510 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
510 work := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}511 work := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
511 broken := degradedTransactionRecord()512 broken := degradedTransactionRecord()
513+ broken.IncarnationID = domain.IncarnationID{2}
512 tests := []struct {514 tests := []struct {
513 name string515 name string
514 mutate func(*Coordinator) error516 mutate func(*Coordinator) error

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close