1package snapshot
2
3import (
4 "context"
5 "crypto/sha256"
6 "os"
7 "path/filepath"
8 "slices"
9 "testing"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/stretchr/testify/require"
14)
15
16func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
17 t.Parallel()
18 repository := NewRepository(privateDir(t))
19 publication := repositoryPublication(t, "work", 1, []byte("first checkpoint"))
20 require.NoError(t, repository.Publish(context.Background(), publication))
21
22 // Repository publication completed, but the catalogue commit did not.
23 // GC knows the incarnation but has no committed checkpoint for it.
24 keep := map[domain.IncarnationID]domain.CheckpointRef{publication.IncarnationID: {}}
25 require.NoError(t, collectAll(context.Background(), repository, keep))
26 _, err := os.Stat(repository.headPath(publication.IncarnationID))
27 require.ErrorIs(t, err, os.ErrNotExist)
28 require.Empty(t, garbageCollectionGenerations(t, repository, publication.IncarnationID))
29
30 // Retrying starts from an empty repository, then the resulting checkpoint
31 // can be committed to the catalogue and restored.
32 require.NoError(t, repository.Publish(context.Background(), publication))
33 committed := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
34 keep[publication.IncarnationID] = committed
35 require.NoError(t, collectAll(context.Background(), repository, keep))
36 _, err = repository.LoadCheckpoint(context.Background(), publication.IncarnationID, publication.Name, committed)
37 require.NoError(t, err)
38}
39
40func TestCollectIncarnationGarbageKeepsHeadPublishable(t *testing.T) {
41 t.Parallel()
42 tests := []struct {
43 name string
44 prepare func(t *testing.T, repository *Repository, id domain.IncarnationID)
45 }{
46 // Generation 4 reached the repository but its catalogue commit failed.
47 {name: "forward orphan HEAD is rewound", prepare: func(*testing.T, *Repository, domain.IncarnationID) {}},
48 {name: "missing HEAD is restored", prepare: func(t *testing.T, repository *Repository, id domain.IncarnationID) {
49 require.NoError(t, os.Remove(repository.headPath(id)))
50 }},
51 // An interrupted earlier pass must never leave HEAD naming a removed
52 // manifest: HEAD is reconciled before generations are removed.
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)))
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 }},
69 }
70 for _, tt := range tests {
71 t.Run(tt.name, func(t *testing.T) {
72 t.Parallel()
73 repository := NewRepository(privateDir(t))
74 publications := publishGarbageCollectionGenerations(t, repository, "work", 4)
75 id := publications[0].IncarnationID
76 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
77 tt.prepare(t, repository, id)
78
79 require.NoError(t, repository.CollectIncarnationGarbage(t.Context(), id, &committed))
80
81 generation, digest, err := repository.readHead(id)
82 require.NoError(t, err)
83 require.Equal(t, committed.Generation, generation)
84 require.Equal(t, ports.SnapshotDigest(committed.ManifestDigest), digest)
85 // The live session retries generation 4 against its committed parent.
86 require.NoError(t, repository.Publish(t.Context(), publications[3]))
87 })
88 }
89}
90
91func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {
92 for _, tt := range []struct {
93 name string
94 mutate func(*testing.T, *Repository, domain.IncarnationID, domain.CheckpointRef) domain.CheckpointRef
95 }{
96 {
97 name: "committed digest mismatch",
98 mutate: func(_ *testing.T, _ *Repository, _ domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
99 committed.ManifestDigest[0]++
100 return committed
101 },
102 },
103 {
104 name: "committed manifest missing",
105 mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
106 t.Helper()
107 require.NoError(t, os.Remove(repository.manifestPath(id, committed.Generation)))
108 return committed
109 },
110 },
111 {
112 name: "committed manifest corrupt",
113 mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
114 t.Helper()
115 require.NoError(t, os.WriteFile(repository.manifestPath(id, committed.Generation), []byte("corrupt"), 0o600))
116 return committed
117 },
118 },
119 } {
120 t.Run(tt.name, func(t *testing.T) {
121 repository := NewRepository(privateDir(t))
122 publications := publishGarbageCollectionGenerations(t, repository, "work", 3)
123 id := publications[0].IncarnationID
124 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
125 committed = tt.mutate(t, repository, id, committed)
126 before := snapshotRepositoryFiles(t, repository.dir)
127
128 require.Error(t, collectAll(t.Context(), repository, map[domain.IncarnationID]domain.CheckpointRef{id: committed}))
129 require.Equal(t, before, snapshotRepositoryFiles(t, repository.dir), "malformed catalogue authority must fail closed")
130 })
131 }
132}
133
134func TestCollectGarbage(t *testing.T) {
135 t.Parallel()
136 tests := []struct {
137 name string
138 committed uint64
139 known bool
140 wantGenerations []uint64
141 }{
142 {name: "incarnation absent from catalogue is removed whole"},
143 {name: "committed checkpoint prunes forward orphan and keeps predecessor", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
144 }
145 for _, tt := range tests {
146 t.Run(tt.name, func(t *testing.T) {
147 t.Parallel()
148 repository := NewRepository(privateDir(t))
149 publications := publishGarbageCollectionGenerations(t, repository, "work", 4)
150 id := publications[0].IncarnationID
151 keep := make(map[domain.IncarnationID]domain.CheckpointRef)
152 if tt.known {
153 publication := publications[tt.committed-1]
154 keep[id] = domain.CheckpointRef{Generation: tt.committed, ManifestDigest: sha256.Sum256(publication.Manifest)}
155 }
156
157 require.NoError(t, collectAll(context.Background(), repository, keep))
158 if !tt.known {
159 _, err := os.Stat(repository.sessionPath(id))
160 require.ErrorIs(t, err, os.ErrNotExist)
161 return
162 }
163 require.Equal(t, tt.wantGenerations, garbageCollectionGenerations(t, repository, id))
164 retainedObjects := make(map[ports.SnapshotDigest]struct{})
165 for _, publication := range publications[tt.committed-2 : tt.committed] {
166 for _, object := range publication.Objects {
167 retainedObjects[object.Digest] = struct{}{}
168 }
169 }
170 for _, publication := range publications {
171 for _, object := range publication.Objects {
172 _, retained := retainedObjects[object.Digest]
173 _, err := os.Stat(repository.objectPath(id, object.Digest))
174 if retained {
175 require.NoError(t, err)
176 } else {
177 require.ErrorIs(t, err, os.ErrNotExist)
178 }
179 }
180 }
181 })
182 }
183}
184
185func publishGarbageCollectionGenerations(t *testing.T, repository *Repository, name string, count uint64) []ports.SnapshotPublication {
186 t.Helper()
187 publications := make([]ports.SnapshotPublication, 0, count)
188 for generation := uint64(1); generation <= count; generation++ {
189 publication := repositoryPublicationAfter(t, repository, name, generation, []byte{byte(generation)})
190 require.NoError(t, repository.Publish(context.Background(), publication))
191 publications = append(publications, publication)
192 }
193 return publications
194}
195
196func garbageCollectionGenerations(t *testing.T, repository *Repository, id domain.IncarnationID) []uint64 {
197 t.Helper()
198 entries, err := os.ReadDir(filepath.Join(repository.sessionPath(id), repositoryGenerations))
199 require.NoError(t, err)
200 generations := make([]uint64, 0, len(entries))
201 for _, entry := range entries {
202 if generation, ok := parseGenerationFilename(entry.Name()); ok {
203 generations = append(generations, generation)
204 }
205 }
206 slices.Sort(generations)
207 return generations
208}