amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package moves
2
3import (
4 "bytes"
5 "context"
6 "crypto/sha256"
7 "database/sql"
8 "encoding/binary"
9 "encoding/hex"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "io"
14 "log/slog"
15 "os"
16 "path/filepath"
17 "sort"
18 "strings"
19 "testing"
20 "time"
21
22 "github.com/benbjohnson/litestream"
23 lsfile "github.com/benbjohnson/litestream/file"
24 "github.com/carlosframework/platform/internal/lease"
25 "github.com/carlosframework/platform/internal/lease/leasetest"
26 "github.com/superfly/ltx"
27 _ "modernc.org/sqlite"
28)
29
30// The tests here run real litestream v0.5.14 and ltx v0.5.1 over real
31// SQLite files (controller ruling: real data, never a fake provider as
32// proof). A source replica is produced the way internal/sim/litestreamfacts
33// produces one: litestream's file replica client over a database written
34// through modernc sqlite, every sync explicit.
35
36const (
37 testBucket = "data"
38 testHost = "h"
39 testOp = "op1"
40 testLane = "moves/h/op1/replica/"
41)
42
43func testGenPrefix() string { return GenerationPrefix("instances/h", testOp) }
44
45// openLS opens path under litestream against client with no background
46// monitors, as litestreamfacts does.
47func openLS(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB {
48 t.Helper()
49 db := litestream.NewDB(path)
50 db.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
51 db.SetMetaPath(meta)
52 db.MonitorInterval = time.Hour
53 db.Replica = litestream.NewReplicaWithClient(db, client)
54 db.Replica.SyncInterval = time.Hour
55 if err := db.Open(); err != nil {
56 t.Fatalf("open %s: %v", path, err)
57 }
58 return db
59}
60
61// writeRows inserts n rows, each its own transaction.
62func writeRows(t *testing.T, path string, n int) {
63 t.Helper()
64 d, err := sql.Open("sqlite", path)
65 if err != nil {
66 t.Fatal(err)
67 }
68 defer d.Close()
69 if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
70 t.Fatal(err)
71 }
72 for i := 0; i < n; i++ {
73 if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil {
74 t.Fatal(err)
75 }
76 }
77}
78
79func countRows(t *testing.T, path string) int {
80 t.Helper()
81 d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
82 if err != nil {
83 t.Fatal(err)
84 }
85 defer d.Close()
86 var n int
87 if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
88 t.Fatal(err)
89 }
90 return n
91}
92
93func pageCountPragma(t *testing.T, path string) int64 {
94 t.Helper()
95 d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
96 if err != nil {
97 t.Fatal(err)
98 }
99 defer d.Close()
100 var n int64
101 if err := d.QueryRow(`PRAGMA page_count`).Scan(&n); err != nil {
102 t.Fatal(err)
103 }
104 return n
105}
106
107func fileSHA256(t *testing.T, path string) string {
108 t.Helper()
109 b, err := os.ReadFile(path)
110 if err != nil {
111 t.Fatal(err)
112 }
113 s := sha256.Sum256(b)
114 return hex.EncodeToString(s[:])
115}
116
117// sourceReplica writes a database with rows rows in syncs separate
118// syncs, takes a level-9 snapshot when snapshot is set, and returns the
119// file replica client plus the newest level-0 position.
120func sourceReplica(t *testing.T, syncs int, snapshot bool) (litestream.ReplicaClient, uint64) {
121 t.Helper()
122 dir := t.TempDir()
123 path := filepath.Join(dir, "src.db")
124 writeRows(t, path, 1)
125 client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
126 db := openLS(t, path, filepath.Join(dir, "meta"), client)
127 ctx := context.Background()
128 for i := 0; i < syncs; i++ {
129 if err := db.SyncAndWait(ctx); err != nil {
130 t.Fatalf("sync: %v", err)
131 }
132 writeRows(t, path, 2)
133 }
134 if err := db.SyncAndWait(ctx); err != nil {
135 t.Fatalf("sync: %v", err)
136 }
137 if snapshot {
138 if _, err := db.Snapshot(ctx); err != nil {
139 t.Fatal(err)
140 }
141 }
142 if err := db.Close(ctx); err != nil {
143 t.Fatal(err)
144 }
145 var n uint64
146 for _, f := range levelFiles(t, client, 0) {
147 if uint64(f.MaxTXID) > n {
148 n = uint64(f.MaxTXID)
149 }
150 }
151 if n == 0 {
152 t.Fatal("source replica has no level-0 files")
153 }
154 return client, n
155}
156
157func levelFiles(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo {
158 t.Helper()
159 itr, err := client.LTXFiles(context.Background(), level, 0, false)
160 if err != nil {
161 t.Fatal(err)
162 }
163 infos, err := ltx.SliceFileIterator(itr)
164 if err != nil {
165 t.Fatal(err)
166 }
167 return infos
168}
169
170func readLTX(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) []byte {
171 t.Helper()
172 rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0)
173 if err != nil {
174 t.Fatal(err)
175 }
176 defer rc.Close()
177 b, err := io.ReadAll(rc)
178 if err != nil {
179 t.Fatal(err)
180 }
181 return b
182}
183
184// sealReplica copies every LTX file of client into the bucket under lane,
185// as the staging lane holds them, and returns the pinned version set.
186func sealReplica(t *testing.T, b *leasetest.VersionedBucket, client litestream.ReplicaClient, lane string) []SealedObject {
187 t.Helper()
188 var out []SealedObject
189 for level := 0; level <= litestream.SnapshotLevel; level++ {
190 for _, info := range levelFiles(t, client, level) {
191 body := readLTX(t, client, info)
192 key := lane + fmt.Sprintf("%04x/", level) + ltx.FormatFilename(info.MinTXID, info.MaxTXID)
193 v := b.Put(testBucket, key, body)
194 out = append(out, SealedObject{Key: key, VersionID: v, Size: int64(len(body))})
195 }
196 }
197 return out
198}
199
200func newBucket() *leasetest.VersionedBucket {
201 b := leasetest.NewVersionedBucket()
202 b.AddBucket(testBucket)
203 return b
204}
205
206func versionsOf(t *testing.T, b *leasetest.VersionedBucket, key string) []lease.ObjectVersion {
207 t.Helper()
208 all, err := b.ListVersions(context.Background(), testBucket, key)
209 if err != nil {
210 t.Fatal(err)
211 }
212 var out []lease.ObjectVersion
213 for _, v := range all {
214 if v.Key == key {
215 out = append(out, v)
216 }
217 }
218 return out
219}
220
221func decodeLTX(t *testing.T, data []byte) (ltx.Header, ltx.Trailer, int) {
222 t.Helper()
223 dec := ltx.NewDecoder(bytes.NewReader(data))
224 if err := dec.DecodeHeader(); err != nil {
225 t.Fatal(err)
226 }
227 page := make([]byte, dec.Header().PageSize)
228 pages := 0
229 for {
230 var ph ltx.PageHeader
231 if err := dec.DecodePage(&ph, page); err == io.EOF {
232 break
233 } else if err != nil {
234 t.Fatal(err)
235 }
236 pages++
237 }
238 if err := dec.Close(); err != nil {
239 t.Fatal(err)
240 }
241 if err := ltx.NewDecoder(bytes.NewReader(data)).Verify(); err != nil {
242 t.Fatalf("verify: %v", err)
243 }
244 return dec.Header(), dec.Trailer(), pages
245}
246
247// plannedFiles is how many files litestream's planner reads to reach n
248// over the pinned set.
249func plannedFiles(t *testing.T, b *leasetest.VersionedBucket, src []SealedObject, n uint64) int {
250 t.Helper()
251 c, err := NewSealedClient(b, testBucket, src)
252 if err != nil {
253 t.Fatal(err)
254 }
255 plan, err := litestream.CalcRestorePlan(context.Background(), c, ltx.TXID(n), time.Time{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
256 if err != nil {
257 t.Fatal(err)
258 }
259 return len(plan)
260}
261
262func sameManifest(a, b InstallManifest) bool {
263 a.BuiltAt, b.BuiltAt = time.Time{}, time.Time{}
264 return a == b
265}
266
267var testClock = time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)
268
269func inputs(b *leasetest.VersionedBucket, src []SealedObject, n uint64, work string) BuildInputs {
270 return inputsAt(b, src, n, work, testClock)
271}
272
273// inputsAt is inputs with the build's clock at now: a replay runs later
274// than the attempt it replays.
275func inputsAt(store lease.VersionedStore, src []SealedObject, n uint64, work string, now time.Time) BuildInputs {
276 return BuildInputs{
277 Store: store, Bucket: testBucket, Source: src, N: n,
278 GenerationPrefix: testGenPrefix(), Op: testOp, Host: testHost,
279 WorkDir: work,
280 Now: func() time.Time { return now },
281 }
282}
283
284// versionRewriter is a VersionedStore whose answers name a version id the
285// builder must refuse: what a bucket whose versioning was suspended after
286// the seal returns for a new write ("null"), or a provider that returns
287// none (""). putAs rewrites the id a fresh PutIfAbsent returns for a key;
288// listAs rewrites the current version a replay finds for a key.
289type versionRewriter struct {
290 lease.VersionedStore
291 putAs map[string]string
292 listAs map[string]string
293}
294
295func (r *versionRewriter) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
296 v, err := r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
297 if as, ok := r.putAs[key]; ok && err == nil {
298 return as, nil
299 }
300 return v, err
301}
302
303func (r *versionRewriter) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
304 vs, err := r.VersionedStore.ListVersions(ctx, bucket, prefix)
305 for i := range vs {
306 if as, ok := r.listAs[vs[i].Key]; ok && vs[i].IsLatest {
307 vs[i].VersionID = as
308 }
309 }
310 return vs, err
311}
312
313// restoreGeneration restores the built generation through litestream's own
314// Restore, reading exactly the objects the manifest names.
315func restoreGeneration(t *testing.T, b *leasetest.VersionedBucket, m InstallManifest, out string) {
316 t.Helper()
317 snapBody, err := b.GetVersion(context.Background(), testBucket, m.SnapshotKey, m.SnapshotVersion)
318 if err != nil {
319 t.Fatal(err)
320 }
321 anchorBody, err := b.GetVersion(context.Background(), testBucket, m.AnchorKey, m.AnchorVersion)
322 if err != nil {
323 t.Fatal(err)
324 }
325 gen, err := NewSealedClient(b, testBucket, []SealedObject{
326 {Key: m.SnapshotKey, VersionID: m.SnapshotVersion, Size: int64(len(snapBody))},
327 {Key: m.AnchorKey, VersionID: m.AnchorVersion, Size: int64(len(anchorBody))},
328 })
329 if err != nil {
330 t.Fatal(err)
331 }
332 r := litestream.NewReplicaWithClient(nil, gen)
333 opt := litestream.NewRestoreOptions()
334 opt.OutputPath = out
335 if err := r.Restore(context.Background(), opt); err != nil {
336 t.Fatalf("restore the generation: %v", err)
337 }
338}
339
340func TestGenerationPrefixAndKeys(t *testing.T) {
341 if got := GenerationPrefix("instances/h", "op1"); got != "instances/h/g-op1/" {
342 t.Fatalf("GenerationPrefix = %q", got)
343 }
344 if got := GenerationPrefix("instances/h/", "op1"); got != "instances/h/g-op1/" {
345 t.Fatalf("GenerationPrefix with a slash = %q", got)
346 }
347 if got := GenerationPrefix("instances/h/inc2", "op1"); got != "instances/h/inc2/g-op1/" {
348 t.Fatalf("GenerationPrefix with an incarnation = %q", got)
349 }
350 // The files twin (spec §7.6, merge round 6): every reader and writer of
351 // a generation's files names them by this rule.
352 for _, loc := range []string{"instances/h/inc2", "instances/h/inc2/"} {
353 if got := GenerationFilesPrefix(loc, "op1"); got != "instances/h/inc2/files/g-op1/" || !strings.HasPrefix(got, FilesGenerationBase(loc)) {
354 t.Fatalf("GenerationFilesPrefix(%q) = %q", loc, got)
355 }
356 }
357 p := "instances/h/g-op1/"
358 if got := SnapshotKey(p, 7); got != p+"0009/0000000000000001-0000000000000007.ltx" {
359 t.Fatalf("SnapshotKey = %q", got)
360 }
361 if got := AnchorKey(p, 7); got != p+"0000/0000000000000007-0000000000000007.ltx" {
362 t.Fatalf("AnchorKey = %q", got)
363 }
364 if got := BuildIntentKey(p); got != p+"build-intent.json" {
365 t.Fatalf("BuildIntentKey = %q", got)
366 }
367 if got := InstallManifestKey(p); got != p+"install.json" {
368 t.Fatalf("InstallManifestKey = %q", got)
369 }
370}
371
372func TestSealedClientServesOnlySealedVersions(t *testing.T) {
373 b := newBucket()
374 ctx := context.Background()
375 key := testLane + "0000/" + ltx.FormatFilename(1, 1)
376 // At least an LTX header's worth of bytes: the client refuses a
377 // smaller object, as litestream's restore would.
378 sealed := []byte(strings.Repeat("sealed bytes of 0000000000000001-0000000000000001 ", 3))
379 v1 := b.Put(testBucket, key, sealed)
380 b.Put(testBucket, key, []byte("a newer version that landed after the seal"))
381 // An unsealed sibling exists in the lane but is not in the pinned set.
382 b.Put(testBucket, testLane+"0000/"+ltx.FormatFilename(2, 2), []byte("unsealed"))
383
384 c, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: int64(len(sealed))}})
385 if err != nil {
386 t.Fatal(err)
387 }
388 if c.Type() != "sealed" {
389 t.Fatalf("Type = %q", c.Type())
390 }
391 if err := c.Init(ctx); err != nil {
392 t.Fatal(err)
393 }
394
395 rc, err := c.OpenLTXFile(ctx, 0, 1, 1, 0, 0)
396 if err != nil {
397 t.Fatal(err)
398 }
399 got, _ := io.ReadAll(rc)
400 rc.Close()
401 if !bytes.Equal(got, sealed) {
402 t.Fatalf("OpenLTXFile served %q, want the sealed version", got)
403 }
404 // A ranged open, as litestream's resumable reader asks for it.
405 rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 5)
406 if err != nil {
407 t.Fatal(err)
408 }
409 got, _ = io.ReadAll(rc)
410 rc.Close()
411 if string(got) != string(sealed[7:12]) {
412 t.Fatalf("ranged open = %q, want %q", got, sealed[7:12])
413 }
414 rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 0)
415 if err != nil {
416 t.Fatal(err)
417 }
418 got, _ = io.ReadAll(rc)
419 rc.Close()
420 if string(got) != string(sealed[7:]) {
421 t.Fatalf("open to the end = %q, want %q", got, sealed[7:])
422 }
423
424 if _, err := c.OpenLTXFile(ctx, 0, 2, 2, 0, 0); !errors.Is(err, os.ErrNotExist) {
425 t.Fatalf("an unsealed key opened: %v", err)
426 }
427 infos := levelFiles(t, c, 0)
428 if len(infos) != 1 || infos[0].MinTXID != 1 || infos[0].MaxTXID != 1 || infos[0].Size != int64(len(sealed)) || infos[0].Level != 0 {
429 t.Fatalf("LTXFiles(0) = %+v", infos)
430 }
431 if infos := levelFiles(t, c, 9); len(infos) != 0 {
432 t.Fatalf("LTXFiles(9) = %+v, want none", infos)
433 }
434 // seek skips files below it, as the file client does.
435 itr, err := c.LTXFiles(ctx, 0, 2, false)
436 if err != nil {
437 t.Fatal(err)
438 }
439 if itr.Next() {
440 t.Fatal("seek past the only file still yielded it")
441 }
442
443 if _, err := c.WriteLTXFile(ctx, 0, 3, 3, bytes.NewReader(nil)); !errors.Is(err, ErrSealedReadOnly) {
444 t.Fatalf("WriteLTXFile = %v", err)
445 }
446 if err := c.DeleteLTXFiles(ctx, infos); !errors.Is(err, ErrSealedReadOnly) {
447 t.Fatalf("DeleteLTXFiles = %v", err)
448 }
449 if err := c.DeleteAll(ctx); !errors.Is(err, ErrSealedReadOnly) {
450 t.Fatalf("DeleteAll = %v", err)
451 }
452 if n := len(versionsOf(t, b, key)); n != 2 {
453 t.Fatalf("the sealed key has %d versions after the refused writes, want 2", n)
454 }
455
456 // Each refusal below isolates one property: every other field of the
457 // fixture is valid (a usable version, an LTX key, a header-sized object).
458 okSize := int64(len(sealed))
459 // G27: a version id that pins nothing is refused before any read.
460 for _, bad := range []string{"", "null"} {
461 if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: bad, Size: okSize}}); err == nil {
462 t.Fatalf("version %q accepted", bad)
463 }
464 }
465 // A key that is not <level>/<min>-<max>.ltx has no position.
466 if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: testLane + "lock.json", VersionID: v1, Size: okSize}}); err == nil {
467 t.Fatal("a non-LTX key accepted")
468 }
469 // Two objects at one position would make the plan ambiguous.
470 if _, err := NewSealedClient(b, testBucket, []SealedObject{
471 {Key: key, VersionID: v1, Size: okSize},
472 {Key: "other/" + "0000/" + ltx.FormatFilename(1, 1), VersionID: v1, Size: okSize},
473 }); err == nil {
474 t.Fatal("two objects at one position accepted")
475 }
476 // A size below an LTX header is refused on its own.
477 if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize - 1}}); err == nil {
478 t.Fatal("an object smaller than an LTX header accepted")
479 }
480 if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize}}); err != nil {
481 t.Fatalf("a header-sized object refused: %v", err)
482 }
483}
484
485// Every artifact identity the build records, and every current version a
486// replay chooses, must pin an immutable object: "" and "null" fail the
487// build before they are read or recorded, and no install.json is written.
488func TestBuildRefusesUnusableArtifactVersions(t *testing.T) {
489 src, n := sourceReplica(t, 1, true)
490 prefix := testGenPrefix()
491 ctx := context.Background()
492 keys := map[string]string{
493 "intent": BuildIntentKey(prefix),
494 "snapshot": SnapshotKey(prefix, n),
495 "anchor": AnchorKey(prefix, n),
496 "manifest": InstallManifestKey(prefix),
497 }
498 for name, key := range keys {
499 for _, bad := range []string{"", "null"} {
500 t.Run("created "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
501 b := newBucket()
502 seal := sealReplica(t, b, src, testLane)
503 store := &versionRewriter{VersionedStore: b, putAs: map[string]string{key: bad}}
504 _, err := BuildGeneration(ctx, inputsAt(store, seal, n, filepath.Join(t.TempDir(), "build"), testClock))
505 if !errors.Is(err, ErrBuildUnpinned) {
506 t.Fatalf("err = %v, want ErrBuildUnpinned", err)
507 }
508 if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok && name != "manifest" {
509 t.Fatal("install.json was written")
510 }
511 })
512 }
513 }
514 // On replay: the attempt crashed after the snapshot PUT; the replay's
515 // listing names an unusable current version for the intent or the
516 // snapshot, as a bucket whose versioning was suspended would.
517 for _, name := range []string{"intent", "snapshot"} {
518 for _, bad := range []string{"", "null"} {
519 t.Run("replayed "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
520 b := newBucket()
521 seal := sealReplica(t, b, src, testLane)
522 work := filepath.Join(t.TempDir(), "build")
523 boom := errors.New("the box died")
524 b.FailOn = func(op, key string) error {
525 if op == "put" && key == keys["anchor"] {
526 return boom
527 }
528 return nil
529 }
530 if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) {
531 t.Fatalf("first run = %v, want the injected failure", err)
532 }
533 b.FailOn = nil
534 store := &versionRewriter{VersionedStore: b, listAs: map[string]string{keys[name]: bad}}
535 _, err := BuildGeneration(ctx, inputsAt(store, seal, n, work, testClock.Add(time.Hour)))
536 if !errors.Is(err, ErrBuildUnpinned) {
537 t.Fatalf("replay = %v, want ErrBuildUnpinned", err)
538 }
539 if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok {
540 t.Fatal("install.json was written on the replay")
541 }
542 if vs := versionsOf(t, b, keys["snapshot"]); len(vs) != 1 {
543 t.Fatalf("the snapshot has %d versions, want the attempt's one", len(vs))
544 }
545 })
546 }
547 }
548}
549
550func TestBuildGenerationReachesNAndWritesSnapshotAndAnchor(t *testing.T) {
551 src, n := sourceReplica(t, 3, true)
552 b := newBucket()
553 seal := sealReplica(t, b, src, testLane)
554 work := filepath.Join(t.TempDir(), "build")
555 ctx := context.Background()
556
557 m, err := BuildGeneration(ctx, inputs(b, seal, n, work))
558 if err != nil {
559 t.Fatalf("BuildGeneration: %v", err)
560 }
561 prefix := testGenPrefix()
562 if m.V != 1 || m.Op != testOp || m.Host != testHost || m.N != n {
563 t.Fatalf("manifest identity = %+v", m)
564 }
565 if m.SnapshotKey != SnapshotKey(prefix, n) || m.AnchorKey != AnchorKey(prefix, n) {
566 t.Fatalf("manifest keys = %q %q", m.SnapshotKey, m.AnchorKey)
567 }
568 if !lease.UsableVersionID(m.SnapshotVersion) || !lease.UsableVersionID(m.AnchorVersion) {
569 t.Fatalf("manifest versions = %q %q", m.SnapshotVersion, m.AnchorVersion)
570 }
571 if m.IntegrityCheck != "ok" {
572 t.Fatalf("IntegrityCheck = %q", m.IntegrityCheck)
573 }
574 dbPath := filepath.Join(work, testHost+".db")
575 if got := fileSHA256(t, dbPath); got != m.DBSHA256 {
576 t.Fatalf("DBSHA256 = %s, file is %s", m.DBSHA256, got)
577 }
578 if !m.BuiltAt.Equal(time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)) {
579 t.Fatalf("BuiltAt = %v", m.BuiltAt)
580 }
581
582 snapBody, err := b.GetVersion(ctx, testBucket, m.SnapshotKey, m.SnapshotVersion)
583 if err != nil {
584 t.Fatal(err)
585 }
586 sh, st, _ := decodeLTX(t, snapBody)
587 if sh.MinTXID != 1 || sh.MaxTXID != ltx.TXID(n) || sh.Flags != ltx.HeaderFlagNoChecksum || sh.WALOffset != 0 {
588 t.Fatalf("snapshot header = %+v", sh)
589 }
590 if uint64(st.FileChecksum) != m.SnapshotCRC64 {
591 t.Fatalf("SnapshotCRC64 = %x, trailer %x", m.SnapshotCRC64, uint64(st.FileChecksum))
592 }
593 anchorBody, err := b.GetVersion(ctx, testBucket, m.AnchorKey, m.AnchorVersion)
594 if err != nil {
595 t.Fatal(err)
596 }
597 ah, at, _ := decodeLTX(t, anchorBody)
598 if ah.MinTXID != ltx.TXID(n) || ah.MaxTXID != ltx.TXID(n) || ah.Flags != ltx.HeaderFlagNoChecksum || ah.WALOffset != litestream.WALHeaderSize {
599 t.Fatalf("anchor header = %+v", ah)
600 }
601 if ah.PageSize != sh.PageSize || ah.Commit != sh.Commit || ah.Timestamp != sh.Timestamp {
602 t.Fatalf("anchor and snapshot disagree: %+v vs %+v", ah, sh)
603 }
604 if uint64(at.FileChecksum) != m.AnchorCRC64 {
605 t.Fatalf("AnchorCRC64 = %x, trailer %x", m.AnchorCRC64, uint64(at.FileChecksum))
606 }
607 ps, pc, err := ReadDBHeader(dbPath)
608 if err != nil {
609 t.Fatal(err)
610 }
611 if sh.PageSize != ps || sh.Commit != pc {
612 t.Fatalf("snapshot header %d/%d, database %d/%d", sh.PageSize, sh.Commit, ps, pc)
613 }
614
615 // The intent was written and names the encoding inputs.
616 intentBody, ok := b.Current(testBucket, BuildIntentKey(prefix))
617 if !ok {
618 t.Fatal("no build intent")
619 }
620 var intent BuildIntent
621 if err := json.Unmarshal(intentBody, &intent); err != nil {
622 t.Fatal(err)
623 }
624 if intent.V != 1 || intent.Op != testOp || intent.Host != testHost || intent.N != n || intent.PageSize != ps || intent.Commit != pc || intent.Timestamp != sh.Timestamp {
625 t.Fatalf("intent = %+v", intent)
626 }
627
628 // The manifest in the bucket is the one returned.
629 manifestBody, ok := b.Current(testBucket, InstallManifestKey(prefix))
630 if !ok {
631 t.Fatal("no install manifest")
632 }
633 var stored InstallManifest
634 if err := json.Unmarshal(manifestBody, &stored); err != nil {
635 t.Fatal(err)
636 }
637 if !sameManifest(stored, m) || !stored.BuiltAt.Equal(m.BuiltAt) {
638 t.Fatalf("stored manifest %+v, returned %+v", stored, m)
639 }
640
641 // Nothing but the four objects was written to the generation.
642 gen, err := b.ListVersions(ctx, testBucket, prefix)
643 if err != nil {
644 t.Fatal(err)
645 }
646 var keys []string
647 for _, v := range gen {
648 keys = append(keys, v.Key)
649 }
650 sort.Strings(keys)
651 want := []string{AnchorKey(prefix, n), SnapshotKey(prefix, n), BuildIntentKey(prefix), InstallManifestKey(prefix)}
652 sort.Strings(want)
653 if strings.Join(keys, ",") != strings.Join(want, ",") {
654 t.Fatalf("generation holds %v, want %v", keys, want)
655 }
656
657 // Litestream's own restore of the generation yields the same database.
658 out := filepath.Join(t.TempDir(), "restored.db")
659 restoreGeneration(t, b, m, out)
660 if got := fileSHA256(t, out); got != m.DBSHA256 {
661 t.Fatalf("the generation restores to %s, the build's file is %s", got, m.DBSHA256)
662 }
663 if err := lease.CheckDB(out); err != nil {
664 t.Fatalf("CheckDB on the restored generation: %v", err)
665 }
666 if got, want := countRows(t, out), countRows(t, dbPath); got != want {
667 t.Fatalf("restored %d rows, built %d", got, want)
668 }
669}
670
671func TestInstallManifestKeys(t *testing.T) {
672 src, n := sourceReplica(t, 1, true)
673 b := newBucket()
674 seal := sealReplica(t, b, src, testLane)
675 if _, err := BuildGeneration(context.Background(), inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil {
676 t.Fatal(err)
677 }
678 body, ok := b.Current(testBucket, InstallManifestKey(testGenPrefix()))
679 if !ok {
680 t.Fatal("no install manifest")
681 }
682 var got map[string]any
683 if err := json.Unmarshal(body, &got); err != nil {
684 t.Fatal(err)
685 }
686 var keys []string
687 for k := range got {
688 keys = append(keys, k)
689 }
690 sort.Strings(keys)
691 want := []string{"anchor_crc64", "anchor_key", "anchor_version", "built_at", "db_sha256", "host", "integrity_check", "op", "snapshot_crc64", "snapshot_key", "snapshot_txid", "snapshot_version", "v"}
692 if strings.Join(keys, " ") != strings.Join(want, " ") {
693 t.Fatalf("install.json keys = %v, want %v", keys, want)
694 }
695}
696
697// Review Focus 3, witness wBuildCrashes: a build that dies after the
698// snapshot PUT, or during the restore, is replayed in the same WorkDir
699// and ends with one snapshot version and one manifest.
700func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) {
701 src, n := sourceReplica(t, 2, true)
702 prefix := testGenPrefix()
703 ctx := context.Background()
704
705 t.Run("after the snapshot PUT", func(t *testing.T) {
706 b := newBucket()
707 seal := sealReplica(t, b, src, testLane)
708 work := filepath.Join(t.TempDir(), "build")
709 boom := errors.New("the box died")
710 b.FailOn = func(op, key string) error {
711 if op == "put" && key == AnchorKey(prefix, n) {
712 return boom
713 }
714 return nil
715 }
716 if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) {
717 t.Fatalf("first run = %v, want the injected failure", err)
718 }
719 snaps := versionsOf(t, b, SnapshotKey(prefix, n))
720 if len(snaps) != 1 {
721 t.Fatalf("after the crash the snapshot has %d versions", len(snaps))
722 }
723 if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok {
724 t.Fatal("a manifest exists after a crash before the anchor")
725 }
726 if _, err := os.Stat(filepath.Join(work, testHost+".db")); err != nil {
727 t.Fatalf("the restored file should still be on disk: %v", err)
728 }
729 first, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), snaps[0].VersionID)
730
731 // The replay runs later: a build that encoded with its own clock
732 // instead of the intent's would write other bytes.
733 b.FailOn = nil
734 m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
735 if err != nil {
736 t.Fatalf("replay: %v", err)
737 }
738 if !m.BuiltAt.Equal(testClock.Add(time.Hour)) {
739 t.Fatalf("BuiltAt = %v, want the replay's clock", m.BuiltAt)
740 }
741 snaps = versionsOf(t, b, SnapshotKey(prefix, n))
742 if len(snaps) != 1 {
743 t.Fatalf("after the replay the snapshot has %d versions, want 1", len(snaps))
744 }
745 if m.SnapshotVersion != snaps[0].VersionID {
746 t.Fatalf("manifest names snapshot version %s, the bucket has %s", m.SnapshotVersion, snaps[0].VersionID)
747 }
748 again, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), m.SnapshotVersion)
749 if !bytes.Equal(first, again) {
750 t.Fatal("the replay's snapshot bytes differ from the first run's")
751 }
752 if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); !ok {
753 t.Fatal("no manifest after the replay")
754 }
755 if n := len(versionsOf(t, b, BuildIntentKey(prefix))); n != 1 {
756 t.Fatalf("intent has %d versions, want 1", n)
757 }
758 // A third run, later again, is a pure replay: the stored manifest,
759 // its BuiltAt included, and no new versions.
760 m2, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(2*time.Hour)))
761 if err != nil {
762 t.Fatalf("second replay: %v", err)
763 }
764 if !sameManifest(m2, m) || !m2.BuiltAt.Equal(m.BuiltAt) {
765 t.Fatalf("second replay returned %+v, first %+v", m2, m)
766 }
767 if n := len(versionsOf(t, b, AnchorKey(prefix, n))); n != 1 {
768 t.Fatalf("anchor has %d versions, want 1", n)
769 }
770 if n := len(versionsOf(t, b, InstallManifestKey(prefix))); n != 1 {
771 t.Fatalf("manifest has %d versions, want 1", n)
772 }
773 })
774
775 t.Run("during the restore", func(t *testing.T) {
776 b := newBucket()
777 seal := sealReplica(t, b, src, testLane)
778 work := filepath.Join(t.TempDir(), "build")
779 boom := errors.New("the network died")
780 // Every planned file is read once to verify it before the restore;
781 // the restore's own reads come after. Fail from then on.
782 planned := plannedFiles(t, b, seal, n)
783 gets := 0
784 b.FailOn = func(op, key string) error {
785 if op != "get" {
786 return nil
787 }
788 gets++
789 if gets > planned {
790 return boom
791 }
792 return nil
793 }
794 // Litestream's restore pipeline does not carry the read error's
795 // identity (its compactor closes a pipe and the decoder reports the
796 // EOF), so the assertion is that the restore failed and wrote
797 // nothing, not that boom survives the wrapping.
798 if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); err == nil {
799 t.Fatal("the restore succeeded with its reads failing")
800 } else if errors.Is(err, ErrBuildIncomplete) || errors.Is(err, ErrBuildDiverged) {
801 t.Fatalf("first run = %v, want a plain restore failure", err)
802 } else {
803 t.Logf("first run failed as injected: %v", err)
804 }
805 if gets <= planned {
806 t.Fatalf("the failure was not injected: %d gets, %d planned", gets, planned)
807 }
808 if vs, err := b.ListVersions(ctx, testBucket, prefix); err != nil || len(vs) != 0 {
809 t.Fatalf("a failed restore left %d objects in the generation", len(vs))
810 }
811 // A crash leaves whatever the writer had on disk: a partial output
812 // and litestream's temporary file.
813 if err := os.MkdirAll(work, 0o700); err != nil {
814 t.Fatal(err)
815 }
816 for _, p := range []string{testHost + ".db", testHost + ".db.tmp", testHost + ".db-wal", testHost + ".db-shm", testHost + ".db-txid"} {
817 if err := os.WriteFile(filepath.Join(work, p), []byte("partial"), 0o600); err != nil {
818 t.Fatal(err)
819 }
820 }
821 b.FailOn = nil
822 m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
823 if err != nil {
824 t.Fatalf("replay: %v", err)
825 }
826 if got := fileSHA256(t, filepath.Join(work, testHost+".db")); got != m.DBSHA256 {
827 t.Fatalf("the replay's file is %s, manifest %s", got, m.DBSHA256)
828 }
829 out := filepath.Join(t.TempDir(), "restored.db")
830 restoreGeneration(t, b, m, out)
831 if got := fileSHA256(t, out); got != m.DBSHA256 {
832 t.Fatalf("the generation restores to %s, want %s", got, m.DBSHA256)
833 }
834 })
835}
836
837func TestBuildRefusesAGapOrAShortRestore(t *testing.T) {
838 // Level 0 only: with no level-9 snapshot every level-0 file is on the
839 // restore path, so a missing one is a gap.
840 src, n := sourceReplica(t, 4, false)
841 b := newBucket()
842 seal := sealReplica(t, b, src, testLane)
843 if len(seal) < 3 {
844 t.Fatalf("need at least three level-0 files, have %d", len(seal))
845 }
846 prefix := testGenPrefix()
847 ctx := context.Background()
848
849 gapped := append(append([]SealedObject{}, seal[:1]...), seal[2:]...)
850 _, err := BuildGeneration(ctx, inputs(b, gapped, n, filepath.Join(t.TempDir(), "build")))
851 if !errors.Is(err, ErrBuildIncomplete) {
852 t.Fatalf("gap: %v, want ErrBuildIncomplete", err)
853 }
854 if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 {
855 t.Fatalf("a gap wrote %d objects into the generation", len(vs))
856 }
857
858 _, err = BuildGeneration(ctx, inputs(b, seal, n+5, filepath.Join(t.TempDir(), "build")))
859 if !errors.Is(err, ErrBuildIncomplete) {
860 t.Fatalf("short: %v, want ErrBuildIncomplete", err)
861 }
862 if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 {
863 t.Fatalf("a short restore wrote %d objects into the generation", len(vs))
864 }
865
866 // The whole set reaches n.
867 if _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil {
868 t.Fatalf("the complete set: %v", err)
869 }
870}
871
872func TestBuildRefusesADivergentIntent(t *testing.T) {
873 src, n := sourceReplica(t, 1, true)
874 b := newBucket()
875 seal := sealReplica(t, b, src, testLane)
876 prefix := testGenPrefix()
877 ctx := context.Background()
878 // An intent from another build of this generation with other inputs.
879 other, _ := json.Marshal(BuildIntent{V: 1, Op: testOp, Host: testHost, N: n, PageSize: 1024, Commit: 1, Timestamp: 5})
880 b.Put(testBucket, BuildIntentKey(prefix), other)
881 _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build")))
882 if !errors.Is(err, ErrBuildDiverged) {
883 t.Fatalf("divergent intent: %v, want ErrBuildDiverged", err)
884 }
885 if _, ok := b.Current(testBucket, SnapshotKey(prefix, n)); ok {
886 t.Fatal("a snapshot was written under a divergent intent")
887 }
888}
889
890func TestBuildRefusesADivergentSnapshot(t *testing.T) {
891 src, n := sourceReplica(t, 1, true)
892 b := newBucket()
893 seal := sealReplica(t, b, src, testLane)
894 prefix := testGenPrefix()
895 ctx := context.Background()
896 b.Put(testBucket, SnapshotKey(prefix, n), []byte("not this build's bytes"))
897 _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build")))
898 if !errors.Is(err, ErrBuildDiverged) {
899 t.Fatalf("divergent snapshot: %v, want ErrBuildDiverged", err)
900 }
901 if _, ok := b.Current(testBucket, AnchorKey(prefix, n)); ok {
902 t.Fatal("an anchor was written beside a divergent snapshot")
903 }
904}
905
906func TestReadDBHeader(t *testing.T) {
907 path := filepath.Join(t.TempDir(), "x.db")
908 writeRows(t, path, 3)
909 ps, pc, err := ReadDBHeader(path)
910 if err != nil {
911 t.Fatal(err)
912 }
913 st, _ := os.Stat(path)
914 if ps != 4096 || int64(pc)*int64(ps) != st.Size() {
915 t.Fatalf("header %d x %d, file %d bytes", pc, ps, st.Size())
916 }
917 if err := os.WriteFile(path, []byte("not a database at all, just text"), 0o600); err != nil {
918 t.Fatal(err)
919 }
920 if _, _, err := ReadDBHeader(path); err == nil {
921 t.Fatal("garbage accepted")
922 }
923}
924
925// TestEncodeFullSkipsTheLockPage: a database that spans SQLite's lock page
926// encodes without it, verifies, and restores through litestream to a file
927// whose page count is the original's.
928func TestEncodeFullSkipsTheLockPage(t *testing.T) {
929 if testing.Short() {
930 t.Skip("writes a 1 GiB database")
931 }
932 dir := t.TempDir()
933 path := filepath.Join(dir, "big.db")
934 writeRows(t, path, 1)
935 const pageSize = 4096
936 lock := ltx.LockPgno(pageSize)
937 pageCount := lock + 1
938 f, err := os.OpenFile(path, os.O_RDWR, 0)
939 if err != nil {
940 t.Fatal(err)
941 }
942 if err := f.Truncate(int64(pageCount) * pageSize); err != nil {
943 t.Fatal(err)
944 }
945 // The in-header size follows the file, as SQLite keeps it.
946 var hdr [4]byte
947 binary.BigEndian.PutUint32(hdr[:], pageCount)
948 if _, err := f.WriteAt(hdr[:], 28); err != nil {
949 t.Fatal(err)
950 }
951 ps, pc, err := ReadDBHeader(path)
952 if err != nil {
953 t.Fatal(err)
954 }
955 if ps != pageSize || pc != pageCount {
956 t.Fatalf("header %d x %d", pc, ps)
957 }
958 origPages := pageCountPragma(t, path)
959 if origPages != int64(pageCount) {
960 t.Fatalf("PRAGMA page_count = %d, want %d", origPages, pageCount)
961 }
962
963 data, crc, err := EncodeFull(f, ps, pc, 1, 1, 1234, 0)
964 f.Close()
965 if err != nil {
966 t.Fatalf("EncodeFull: %v", err)
967 }
968 h, tr, pages := decodeLTX(t, data)
969 if pages != int(pageCount)-1 {
970 t.Fatalf("encoded %d pages, want %d (every page but the lock page)", pages, pageCount-1)
971 }
972 if h.Commit != pageCount || uint64(tr.FileChecksum) != crc {
973 t.Fatalf("header commit %d crc %x, returned crc %x", h.Commit, uint64(tr.FileChecksum), crc)
974 }
975
976 b := newBucket()
977 key := SnapshotKey(testGenPrefix(), 1)
978 v := b.Put(testBucket, key, data)
979 gen, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v, Size: int64(len(data))}})
980 if err != nil {
981 t.Fatal(err)
982 }
983 out := filepath.Join(dir, "restored.db")
984 r := litestream.NewReplicaWithClient(nil, gen)
985 opt := litestream.NewRestoreOptions()
986 opt.OutputPath = out
987 if err := r.Restore(context.Background(), opt); err != nil {
988 t.Fatalf("restore: %v", err)
989 }
990 if got := pageCountPragma(t, out); got != origPages {
991 t.Fatalf("restored PRAGMA page_count = %d, original %d", got, origPages)
992 }
993 st, err := os.Stat(out)
994 if err != nil {
995 t.Fatal(err)
996 }
997 if st.Size() != int64(pageCount)*pageSize {
998 t.Fatalf("restored file is %d bytes, want %d", st.Size(), int64(pageCount)*pageSize)
999 }
1000}
1001