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: "old", version: codec.ManifestVersion - 1},
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 TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
190 for _, tt := range []struct {
191 name string
192 mutate func(*testing.T, *Repository, ports.SnapshotPublication)
193 }{
194 {
195 name: "missing object",
196 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
197 t.Helper()
198 require.NoError(t, os.Remove(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest)))
199 },
200 },
201 {
202 name: "corrupt object",
203 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
204 t.Helper()
205 require.NoError(t, os.WriteFile(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest), []byte("corrupt"), 0o600))
206 },
207 },
208 {
209 name: "corrupt manifest digest",
210 mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
211 t.Helper()
212 head, err := os.ReadFile(repository.headPath(publication.IncarnationID))
213 require.NoError(t, err)
214 head[len(head)-1]++
215 require.NoError(t, os.WriteFile(repository.headPath(publication.IncarnationID), head, 0o600))
216 },
217 },
218 } {
219 t.Run(tt.name, func(t *testing.T) {
220 id := domain.IncarnationID{1}
221 publication := incarnationPublication(t, id, "work", 1, nil)
222 commitError := errors.New("catalogue commit failed")
223 catalogue := &forwardOrphanCatalogue{
224 record: domain.CatalogueRecord{Name: publication.Name, IncarnationID: id},
225 replaceError: commitError,
226 }
227 repository := NewRepository(privateDir(t))
228 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
229
230 _, err := coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
231 require.ErrorIs(t, err, commitError)
232 tt.mutate(t, repository, publication)
233 _, err = coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
234 require.Error(t, err)
235 require.Nil(t, catalogue.record.Committed, "an invalid forward orphan must not become authoritative")
236 require.Equal(t, 1, catalogue.replaceCalls, "retry must not advance the catalogue")
237 })
238 }
239}
240
241type forwardOrphanCatalogue struct {
242 record domain.CatalogueRecord
243 replaceError error
244 replaceCalls int
245}
246
247func (c *forwardOrphanCatalogue) Records() ([]domain.CatalogueRecord, error) {
248 return []domain.CatalogueRecord{c.record}, nil
249}
250
251func (c *forwardOrphanCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
252 return c.record, c.record.Name == name, nil
253}
254
255func (c *forwardOrphanCatalogue) Create(domain.CatalogueRecord) error {
256 return errors.New("unexpected create")
257}
258func (c *forwardOrphanCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
259 return errors.New("unexpected metadata update")
260}
261func (c *forwardOrphanCatalogue) Sync() error { return nil }
262func (c *forwardOrphanCatalogue) Rename(string, domain.CatalogueRecord) error {
263 return errors.New("unexpected rename")
264}
265func (c *forwardOrphanCatalogue) Delete(string) error { return errors.New("unexpected delete") }
266func (c *forwardOrphanCatalogue) Close() error { return nil }
267
268func (c *forwardOrphanCatalogue) Replace(name string, record domain.CatalogueRecord) error {
269 if name != c.record.Name || name != record.Name {
270 return errors.New("catalogue key mismatch")
271 }
272 c.replaceCalls++
273 if c.replaceError != nil {
274 err := c.replaceError
275 c.replaceError = nil
276 return err
277 }
278 c.record = record
279 return nil
280}
281
282func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {
283 t.Helper()
284 tail, err := codec.MarshalObject(codec.HistoryTail, canonicalHistoryBlob(t, "tail"))
285 if err != nil {
286 t.Fatal(err)
287 }
288 transcript, err := codec.MarshalObject(codec.RecoveryTranscript, canonicalHistoryBlob(t, "transcript"))
289 if err != nil {
290 t.Fatal(err)
291 }
292 manifest, err := codec.MarshalManifest(codec.Manifest{
293 Generation: generation, IncarnationID: id, ParentCheckpoint: parent, Name: name,
294 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))}}}}},
295 })
296 if err != nil {
297 t.Fatal(err)
298 }
299 return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, ParentCheckpoint: parent, Manifest: manifest, Objects: []ports.SnapshotObject{tail, transcript}}
300}