1package snapshot
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/binary"
7 "errors"
8 "os"
9 "path/filepath"
10 "testing"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
15 codec "github.com/bnema/vev/internal/usecase/snapshot"
16 "github.com/stretchr/testify/require"
17)
18
19func TestIncarnationNamespace(t *testing.T) {
20 repo := NewRepository(privateDir(t))
21 id := domain.IncarnationID{1}
22 pub := incarnationPublication(t, id, "old", 1, nil)
23 if err := repo.Publish(context.Background(), pub); err != nil {
24 t.Fatal(err)
25 }
26 parent := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
27 pub = incarnationPublication(t, id, "new", 2, &parent)
28 if err := repo.Publish(context.Background(), pub); err != nil {
29 t.Fatal(err)
30 }
31 if got := filepath.Base(repo.sessionPath(id)); got != id.String() {
32 t.Fatalf("namespace = %q", got)
33 }
34
35 wrong := pub
36 wrong.IncarnationID = domain.IncarnationID{2}
37 if err := repo.Publish(context.Background(), wrong); err == nil {
38 t.Fatal("mismatched publication incarnation accepted")
39 }
40}
41
42func TestDirectCheckpointLoad(t *testing.T) {
43 repo := NewRepository(privateDir(t))
44 id := domain.IncarnationID{1}
45 pub := incarnationPublication(t, id, "work", 1, nil)
46 if err := repo.Publish(context.Background(), pub); err != nil {
47 t.Fatal(err)
48 }
49 generations := filepath.Dir(repo.manifestPath(id, 1))
50 for i := 2; i <= 5001; i++ {
51 if err := os.WriteFile(filepath.Join(generations, generationFilename(uint64(i))), []byte("stale"), 0o600); err != nil {
52 t.Fatal(err)
53 }
54 }
55 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
56 got, err := repo.LoadCheckpoint(context.Background(), id, "work", ref)
57 if err != nil {
58 t.Fatal(err)
59 }
60 if got.Generation != 1 || got.IncarnationID != id {
61 t.Fatalf("generation = %#v", got)
62 }
63}
64
65func TestLoadCheckpointPreservesManifestVersionError(t *testing.T) {
66 for _, tt := range []struct {
67 name string
68 version uint16
69 }{
70 {name: "dense VT checkpoint", version: 3},
71 {name: "future", version: codec.ManifestVersion + 1},
72 } {
73 t.Run(tt.name, func(t *testing.T) {
74 repo := NewRepository(privateDir(t))
75 id := domain.IncarnationID{1}
76 pub := incarnationPublication(t, id, "work", 1, nil)
77 require.NoError(t, repo.Publish(t.Context(), pub))
78
79 manifest := append([]byte(nil), pub.Manifest...)
80 binary.BigEndian.PutUint16(manifest[4:6], tt.version)
81 require.NoError(t, os.WriteFile(repo.manifestPath(id, 1), manifest, 0o600))
82 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(manifest)}
83
84 _, err := repo.LoadCheckpoint(t.Context(), id, "work", ref)
85 require.ErrorIs(t, err, codec.ErrBadVersion)
86 })
87 }
88}
89
90func TestLoadCheckpointDoesNotClassifyCorruptionAsManifestVersion(t *testing.T) {
91 for _, tt := range []struct {
92 name string
93 mutate func(*testing.T, *Repository, ports.SnapshotPublication, *domain.CheckpointRef)
94 }{
95 {
96 name: "digest mismatch precedes version decode",
97 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, _ *domain.CheckpointRef) {
98 t.Helper()
99 manifest := append([]byte(nil), pub.Manifest...)
100 binary.BigEndian.PutUint16(manifest[4:6], codec.ManifestVersion-1)
101 require.NoError(t, os.WriteFile(repo.manifestPath(pub.IncarnationID, pub.Generation), manifest, 0o600))
102 },
103 },
104 {
105 name: "bad magic",
106 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, ref *domain.CheckpointRef) {
107 t.Helper()
108 manifest := append([]byte(nil), pub.Manifest...)
109 manifest[0] ^= 1
110 ref.ManifestDigest = sha256.Sum256(manifest)
111 require.NoError(t, os.WriteFile(repo.manifestPath(pub.IncarnationID, pub.Generation), manifest, 0o600))
112 },
113 },
114 {
115 name: "bad crc",
116 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, ref *domain.CheckpointRef) {
117 t.Helper()
118 manifest := append([]byte(nil), pub.Manifest...)
119 manifest[12] ^= 1
120 ref.ManifestDigest = sha256.Sum256(manifest)
121 require.NoError(t, os.WriteFile(repo.manifestPath(pub.IncarnationID, pub.Generation), manifest, 0o600))
122 },
123 },
124 {
125 name: "wrong identity",
126 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, ref *domain.CheckpointRef) {
127 t.Helper()
128 manifest, err := codec.UnmarshalManifest(pub.Manifest)
129 require.NoError(t, err)
130 manifest.IncarnationID = domain.IncarnationID{2}
131 encoded, err := codec.MarshalManifest(manifest)
132 require.NoError(t, err)
133 ref.ManifestDigest = sha256.Sum256(encoded)
134 require.NoError(t, os.WriteFile(repo.manifestPath(pub.IncarnationID, pub.Generation), encoded, 0o600))
135 },
136 },
137 {
138 name: "invalid object",
139 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, _ *domain.CheckpointRef) {
140 t.Helper()
141 require.NoError(t, os.WriteFile(repo.objectPath(pub.IncarnationID, pub.Objects[0].Digest), []byte("corrupt"), 0o600))
142 },
143 },
144 } {
145 t.Run(tt.name, func(t *testing.T) {
146 repo := NewRepository(privateDir(t))
147 id := domain.IncarnationID{1}
148 pub := incarnationPublication(t, id, "work", 1, nil)
149 require.NoError(t, repo.Publish(t.Context(), pub))
150 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
151 tt.mutate(t, repo, pub, &ref)
152
153 _, err := repo.LoadCheckpoint(t.Context(), id, "work", ref)
154 require.Error(t, err)
155 require.NotErrorIs(t, err, codec.ErrBadVersion)
156 })
157 }
158}
159
160func TestReconcileCheckpoint(t *testing.T) {
161 repo := NewRepository(privateDir(t))
162 id := domain.IncarnationID{1}
163 pub := incarnationPublication(t, id, "work", 1, nil)
164 if err := repo.Publish(context.Background(), pub); err != nil {
165 t.Fatal(err)
166 }
167 valid := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
168 if err := repo.ReconcileCheckpoint(context.Background(), id, valid); err != nil {
169 t.Fatal(err)
170 }
171 before, err := os.ReadFile(repo.headPath(id))
172 if err != nil {
173 t.Fatal(err)
174 }
175 invalid := valid
176 invalid.ManifestDigest[0]++
177 if err := repo.ReconcileCheckpoint(context.Background(), id, invalid); err == nil {
178 t.Fatal("invalid checkpoint repaired HEAD")
179 }
180 after, err := os.ReadFile(repo.headPath(id))
181 if err != nil {
182 t.Fatal(err)
183 }
184 if string(after) != string(before) {
185 t.Fatal("invalid repair changed HEAD")
186 }
187}
188
189func TestReconcileCheckpointRepairsOnlyDivergentHead(t *testing.T) {
190 tests := []struct {
191 name string
192 prepare func(t *testing.T, repo *Repository, id domain.IncarnationID)
193 wantWrite bool
194 }{
195 {name: "matching HEAD is left untouched", prepare: func(*testing.T, *Repository, domain.IncarnationID) {}},
196 {name: "missing HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
197 require.NoError(t, os.Remove(repo.headPath(id)))
198 }},
199 {name: "corrupt HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
200 require.NoError(t, os.WriteFile(repo.headPath(id), []byte("garbage"), 0o600))
201 }},
202 {name: "stale HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
203 require.NoError(t, os.WriteFile(repo.headPath(id), marshalHead(1, [32]byte{9}), 0o600))
204 }},
205 }
206 for _, tt := range tests {
207 t.Run(tt.name, func(t *testing.T) {
208 repo := NewRepository(privateDir(t))
209 id := domain.IncarnationID{1}
210 pub := incarnationPublication(t, id, "work", 1, nil)
211 require.NoError(t, repo.Publish(t.Context(), pub))
212 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
213 tt.prepare(t, repo, id)
214 before, _ := os.Stat(repo.headPath(id))
215
216 require.NoError(t, repo.ReconcileCheckpoint(t.Context(), id, ref))
217
218 generation, digest, err := repo.readHead(id)
219 require.NoError(t, err)
220 require.Equal(t, ref.Generation, generation)
221 require.Equal(t, ports.SnapshotDigest(ref.ManifestDigest), digest)
222 after, err := os.Stat(repo.headPath(id))
223 require.NoError(t, err)
224 require.Equal(t, tt.wantWrite, before == nil || !os.SameFile(before, after), "HEAD rewrite")
225 })
226 }
227}
228
229func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
230 for _, tt := range []struct {
231 name string
232 mutate func(*testing.T, *Repository, ports.SnapshotPublication)
233 }{
234 {
235 name: "missing object",
236 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
237 t.Helper()
238 require.NoError(t, os.Remove(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest)))
239 },
240 },
241 {
242 name: "corrupt object",
243 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
244 t.Helper()
245 require.NoError(t, os.WriteFile(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest), []byte("corrupt"), 0o600))
246 },
247 },
248 {
249 name: "corrupt manifest digest",
250 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
251 t.Helper()
252 head, err := os.ReadFile(repository.headPath(publication.IncarnationID))
253 require.NoError(t, err)
254 head[len(head)-1]++
255 require.NoError(t, os.WriteFile(repository.headPath(publication.IncarnationID), head, 0o600))
256 },
257 },
258 } {
259 t.Run(tt.name, func(t *testing.T) {
260 id := domain.IncarnationID{1}
261 publication := incarnationPublication(t, id, "work", 1, nil)
262 commitError := errors.New("catalogue commit failed")
263 catalogue := &forwardOrphanCatalogue{
264 record: domain.CatalogueRecord{Name: publication.Name, IncarnationID: id},
265 replaceError: commitError,
266 }
267 repository := NewRepository(privateDir(t))
268 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
269
270 _, err := coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
271 require.ErrorIs(t, err, commitError)
272 tt.mutate(t, repository, publication)
273 _, err = coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
274 require.Error(t, err)
275 require.Nil(t, catalogue.record.Committed, "an invalid forward orphan must not become authoritative")
276 require.Equal(t, 1, catalogue.replaceCalls, "retry must not advance the catalogue")
277 })
278 }
279}
280
281type forwardOrphanCatalogue struct {
282 record domain.CatalogueRecord
283 replaceError error
284 replaceCalls int
285}
286
287func (c *forwardOrphanCatalogue) Records() ([]domain.CatalogueRecord, error) {
288 return []domain.CatalogueRecord{c.record}, nil
289}
290
291func (c *forwardOrphanCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
292 return c.record, c.record.Name == name, nil
293}
294
295func (c *forwardOrphanCatalogue) Create(domain.CatalogueRecord) error {
296 return errors.New("unexpected create")
297}
298func (c *forwardOrphanCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
299 return errors.New("unexpected metadata update")
300}
301func (c *forwardOrphanCatalogue) Sync() error { return nil }
302func (c *forwardOrphanCatalogue) Rename(string, domain.CatalogueRecord) error {
303 return errors.New("unexpected rename")
304}
305func (c *forwardOrphanCatalogue) Delete(string) error { return errors.New("unexpected delete") }
306func (c *forwardOrphanCatalogue) Close() error { return nil }
307
308func (c *forwardOrphanCatalogue) Replace(name string, record domain.CatalogueRecord) error {
309 if name != c.record.Name || name != record.Name {
310 return errors.New("catalogue key mismatch")
311 }
312 c.replaceCalls++
313 if c.replaceError != nil {
314 err := c.replaceError
315 c.replaceError = nil
316 return err
317 }
318 c.record = record
319 return nil
320}
321
322func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {
323 t.Helper()
324 tail, err := codec.MarshalObject(codec.HistoryTail, canonicalHistoryBlob(t, "tail"))
325 if err != nil {
326 t.Fatal(err)
327 }
328 transcript, err := codec.MarshalObject(codec.RecoveryTranscript, canonicalHistoryBlob(t, "transcript"))
329 if err != nil {
330 t.Fatal(err)
331 }
332 manifest, err := codec.MarshalManifest(codec.Manifest{
333 Generation: generation, IncarnationID: id, ParentCheckpoint: parent, Name: name,
334 Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))}, Transcript: codec.ObjectRef{Kind: codec.RecoveryTranscript, Digest: transcript.Digest, Size: uint32(len(transcript.Data))}}}}},
335 })
336 if err != nil {
337 t.Fatal(err)
338 }
339 return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, ParentCheckpoint: parent, Manifest: manifest, Objects: []ports.SnapshotObject{tail, transcript}}
340}