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 codec "github.com/bnema/vev/internal/snapshotcodec"
15 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
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 name: "well-formed object with wrong digest",
146 mutate: func(t *testing.T, repo *Repository, pub ports.SnapshotPublication, _ *domain.CheckpointRef) {
147 t.Helper()
148 object := pub.Objects[0]
149 kind, payload, err := codec.UnmarshalObject(object.Data)
150 require.NoError(t, err)
151 payload = append([]byte(nil), payload...)
152 payload[len(payload)-1] ^= 1
153 replacement, err := codec.MarshalObject(kind, payload)
154 require.NoError(t, err)
155 require.Len(t, replacement.Data, len(object.Data))
156 require.NoError(t, os.WriteFile(repo.objectPath(pub.IncarnationID, object.Digest), replacement.Data, 0o600))
157 },
158 },
159 } {
160 t.Run(tt.name, func(t *testing.T) {
161 repo := NewRepository(privateDir(t))
162 id := domain.IncarnationID{1}
163 pub := incarnationPublication(t, id, "work", 1, nil)
164 require.NoError(t, repo.Publish(t.Context(), pub))
165 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
166 tt.mutate(t, repo, pub, &ref)
167
168 _, err := repo.LoadCheckpoint(t.Context(), id, "work", ref)
169 require.Error(t, err)
170 require.NotErrorIs(t, err, codec.ErrBadVersion)
171 })
172 }
173}
174
175func TestReconcileCheckpoint(t *testing.T) {
176 repo := NewRepository(privateDir(t))
177 id := domain.IncarnationID{1}
178 pub := incarnationPublication(t, id, "work", 1, nil)
179 if err := repo.Publish(context.Background(), pub); err != nil {
180 t.Fatal(err)
181 }
182 valid := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
183 if err := repo.ReconcileCheckpoint(context.Background(), id, valid); err != nil {
184 t.Fatal(err)
185 }
186 before, err := os.ReadFile(repo.headPath(id))
187 if err != nil {
188 t.Fatal(err)
189 }
190 invalid := valid
191 invalid.ManifestDigest[0]++
192 if err := repo.ReconcileCheckpoint(context.Background(), id, invalid); err == nil {
193 t.Fatal("invalid checkpoint repaired HEAD")
194 }
195 after, err := os.ReadFile(repo.headPath(id))
196 if err != nil {
197 t.Fatal(err)
198 }
199 if string(after) != string(before) {
200 t.Fatal("invalid repair changed HEAD")
201 }
202}
203
204func TestReconcileCheckpointRepairsOnlyDivergentHead(t *testing.T) {
205 tests := []struct {
206 name string
207 prepare func(t *testing.T, repo *Repository, id domain.IncarnationID)
208 wantWrite bool
209 }{
210 {name: "matching HEAD is left untouched", prepare: func(*testing.T, *Repository, domain.IncarnationID) {}},
211 {name: "missing HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
212 require.NoError(t, os.Remove(repo.headPath(id)))
213 }},
214 {name: "corrupt HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
215 require.NoError(t, os.WriteFile(repo.headPath(id), []byte("garbage"), 0o600))
216 }},
217 {name: "stale HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
218 require.NoError(t, os.WriteFile(repo.headPath(id), marshalHead(1, [32]byte{9}), 0o600))
219 }},
220 }
221 for _, tt := range tests {
222 t.Run(tt.name, func(t *testing.T) {
223 repo := NewRepository(privateDir(t))
224 id := domain.IncarnationID{1}
225 pub := incarnationPublication(t, id, "work", 1, nil)
226 require.NoError(t, repo.Publish(t.Context(), pub))
227 ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
228 tt.prepare(t, repo, id)
229 before, _ := os.Stat(repo.headPath(id))
230
231 require.NoError(t, repo.ReconcileCheckpoint(t.Context(), id, ref))
232
233 generation, digest, err := repo.readHead(id)
234 require.NoError(t, err)
235 require.Equal(t, ref.Generation, generation)
236 require.Equal(t, ports.SnapshotDigest(ref.ManifestDigest), digest)
237 after, err := os.Stat(repo.headPath(id))
238 require.NoError(t, err)
239 require.Equal(t, tt.wantWrite, before == nil || !os.SameFile(before, after), "HEAD rewrite")
240 })
241 }
242}
243
244func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
245 for _, tt := range []struct {
246 name string
247 mutate func(*testing.T, *Repository, ports.SnapshotPublication)
248 }{
249 {
250 name: "missing object",
251 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
252 t.Helper()
253 require.NoError(t, os.Remove(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest)))
254 },
255 },
256 {
257 name: "corrupt object",
258 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
259 t.Helper()
260 require.NoError(t, os.WriteFile(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest), []byte("corrupt"), 0o600))
261 },
262 },
263 {
264 name: "corrupt manifest digest",
265 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
266 t.Helper()
267 head, err := os.ReadFile(repository.headPath(publication.IncarnationID))
268 require.NoError(t, err)
269 head[len(head)-1]++
270 require.NoError(t, os.WriteFile(repository.headPath(publication.IncarnationID), head, 0o600))
271 },
272 },
273 } {
274 t.Run(tt.name, func(t *testing.T) {
275 id := domain.IncarnationID{1}
276 publication := incarnationPublication(t, id, "work", 1, nil)
277 commitError := errors.New("catalogue commit failed")
278 catalogue := &forwardOrphanCatalogue{
279 record: domain.CatalogueRecord{Name: publication.Name, IncarnationID: id},
280 replaceError: commitError,
281 }
282 repository := NewRepository(privateDir(t))
283 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
284
285 _, err := coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
286 require.ErrorIs(t, err, commitError)
287 tt.mutate(t, repository, publication)
288 _, err = coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
289 require.Error(t, err)
290 require.Nil(t, catalogue.record.Committed, "an invalid forward orphan must not become authoritative")
291 require.Equal(t, 1, catalogue.replaceCalls, "retry must not advance the catalogue")
292 })
293 }
294}
295
296type forwardOrphanCatalogue struct {
297 record domain.CatalogueRecord
298 replaceError error
299 replaceCalls int
300}
301
302func (c *forwardOrphanCatalogue) Records() ([]domain.CatalogueRecord, error) {
303 return []domain.CatalogueRecord{c.record}, nil
304}
305
306func (c *forwardOrphanCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
307 return c.record, c.record.Name == name, nil
308}
309
310func (c *forwardOrphanCatalogue) Create(domain.CatalogueRecord) error {
311 return errors.New("unexpected create")
312}
313func (c *forwardOrphanCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
314 return errors.New("unexpected metadata update")
315}
316func (c *forwardOrphanCatalogue) Sync() error { return nil }
317func (c *forwardOrphanCatalogue) Rename(string, domain.CatalogueRecord) error {
318 return errors.New("unexpected rename")
319}
320func (c *forwardOrphanCatalogue) Delete(string) error { return errors.New("unexpected delete") }
321func (c *forwardOrphanCatalogue) Close() error { return nil }
322
323func (c *forwardOrphanCatalogue) Replace(name string, record domain.CatalogueRecord) error {
324 if name != c.record.Name || name != record.Name {
325 return errors.New("catalogue key mismatch")
326 }
327 c.replaceCalls++
328 if c.replaceError != nil {
329 err := c.replaceError
330 c.replaceError = nil
331 return err
332 }
333 c.record = record
334 return nil
335}
336
337func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {
338 t.Helper()
339 tail, err := codec.MarshalObject(codec.HistoryTail, canonicalHistoryBlob(t, "tail"))
340 if err != nil {
341 t.Fatal(err)
342 }
343 transcript, err := codec.MarshalObject(codec.RecoveryTranscript, canonicalHistoryBlob(t, "transcript"))
344 if err != nil {
345 t.Fatal(err)
346 }
347 manifest, err := codec.MarshalManifest(codec.Manifest{
348 Generation: generation, IncarnationID: id, ParentCheckpoint: parent, Name: name,
349 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))}}}}},
350 })
351 if err != nil {
352 t.Fatal(err)
353 }
354 return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, ParentCheckpoint: parent, Manifest: manifest, Objects: []ports.SnapshotObject{tail, transcript}}
355}