amadan
moves: build the generation from the sealed replica: planner restore, snapshot, anchor, intent and manifest (§5.1 steps 2 and 7, §7.6 building)
SealedClient is a litestream.ReplicaClient over exactly one pinned set of object versions (SealedObject: key, version id, size), refusing writes and any version id that pins nothing (G27). BuildGeneration runs litestream's own planner and restore through it into <WorkDir>/<host>.db, refuses a plan that does not end at N (ErrBuildIncomplete), verifies every planned file's trailer, checks the database, then writes create-only in order: the build intent, the snapshot [1, N], the anchor [N, N] at WALHeaderSize, and the InstallManifest from install_contract.go. A replay restores again, reads the intent it finds, encodes identical bytes and reuses equal objects; unequal ones are ErrBuildDiverged. EncodeFull skips SQLite's lock page while Commit keeps the full count. GenerationPrefix normalises the location through ReplicaSource (G24). Tests run real litestream v0.5.14 and ltx v0.5.1 over real SQLite files, including a restore of the built generation through litestream's Restore and a database spanning the lock page.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
5 files changed,
+1589
−0
approved_moves_copy_test.go+3 −0internal/moves/build.go+407 −0internal/moves/build_test.go+875 −0internal/moves/ltxbuild.go+109 −0internal/moves/sealedclient.go+195 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index 46e8b5c3a..0e480faf4 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -583,6 +583,7 @@ var movesScanFiles = []string{ |
| "internal/moves/archive/sync_other.go", |
| "internal/moves/archive/verify.go", |
| "internal/moves/budget.go", |
| + "internal/moves/build.go", |
| "internal/moves/check.go", |
| "internal/moves/envreport.go", |
| "internal/moves/evidence.go", |
| @@ -590,7 +591,9 @@ var movesScanFiles = []string{ |
| "internal/moves/ids.go", |
| "internal/moves/journal.go", |
| "internal/moves/lease.go", |
| + "internal/moves/ltxbuild.go", |
| "internal/moves/readiness.go", |
| + "internal/moves/sealedclient.go", |
| "internal/moves/source.go", |
| "internal/moves/staging.go", |
| "internal/moves/switch.go", |
diff --git a/internal/moves/build.go b/internal/moves/build.go| new file mode 100644 |
| index 000000000..d6dc9caa0 |
| --- /dev/null |
| +++ b/internal/moves/build.go |
| @@ -0,0 +1,407 @@ |
| +package moves |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log/slog" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "time" |
| + |
| + "github.com/benbjohnson/litestream" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/superfly/ltx" |
| +) |
| + |
| +// Errors a build refuses with. Both are internal: Task 10's landing decides |
| +// what an operator reads. |
| +var ( |
| + // ErrBuildIncomplete: the pinned set does not restore to exactly N (a |
| + // gap between files, or a set that stops short). Nothing was written. |
| + ErrBuildIncomplete = errors.New("moves: the sealed replica does not restore to the required position") |
| + // ErrBuildDiverged: an object already under one of the generation's |
| + // keys is not what this build would write (another intent, other |
| + // snapshot or anchor bytes, another manifest). The generation is new |
| + // and the build is deterministic, so this is never a replay. |
| + ErrBuildDiverged = errors.New("moves: the prepared generation holds an object this build did not make") |
| +) |
| + |
| +// buildContractVersion is the V of the intent and the manifest. |
| +const buildContractVersion = 1 |
| + |
| +// BuildIntent is written before any other PUT with the encoding inputs, |
| +// so a replay encodes identical bytes (§7.6 building, R7-4). Timestamp is |
| +// the LTX header timestamp in milliseconds since the epoch. |
| +type BuildIntent struct { |
| + V int `json:"v"` |
| + Op string `json:"op"` |
| + Host string `json:"host"` |
| + N uint64 `json:"snapshot_txid"` |
| + PageSize uint32 `json:"page_size"` |
| + Commit uint32 `json:"commit"` |
| + Timestamp int64 `json:"timestamp"` |
| +} |
| + |
| +// GenerationPrefix is `<location>/g-<op>/`: the prepared generation a move |
| +// (or a relocation) installs under a holder's location. The location is |
| +// normalised with TrimSuffix(location, "/") + "/" through ReplicaSource, |
| +// whose rule this shares with hostbox.GenerationPrefix and genselect's |
| +// prefix; the four change together (G24, I73). |
| +func GenerationPrefix(location, op string) string { |
| + prefix, _ := ReplicaSource(hostclaim.Holder{Location: location, Generation: op}) |
| + return prefix |
| +} |
| + |
| +// SnapshotKey is the generation's level-9 snapshot [1, n]. |
| +func SnapshotKey(prefix string, n uint64) string { |
| + return prefix + fmt.Sprintf("%04x/", litestream.SnapshotLevel) + ltx.FormatFilename(1, ltx.TXID(n)) |
| +} |
| + |
| +// AnchorKey is the generation's level-0 anchor [n, n]. |
| +func AnchorKey(prefix string, n uint64) string { |
| + return prefix + "0000/" + ltx.FormatFilename(ltx.TXID(n), ltx.TXID(n)) |
| +} |
| + |
| +// BuildIntentKey is where the intent lives in the generation. |
| +func BuildIntentKey(prefix string) string { return prefix + "build-intent.json" } |
| + |
| +// InstallManifestKey is where the completed install manifest lives. |
| +func InstallManifestKey(prefix string) string { return prefix + "install.json" } |
| + |
| +// BuildInputs are one generation build. Source is the pinned version set |
| +// (a move: the seal's replica/ objects; region HA's relocation: the |
| +// versions a genselect plan or a current listing pins, L16); the |
| +// SealedClient is built over it, so nothing here names a staging lane. |
| +// WorkDir is the build's scratch, `<JournalDir>/build/` (G34): the |
| +// restored database is left there as `<WorkDir>/<Host>.db` for the |
| +// landing's `restored` to rename into place on the same filesystem. |
| +// Verify is the database check; nil means lease.CheckDB. Now is the clock |
| +// for the intent's timestamp and the manifest's BuiltAt; nil means |
| +// time.Now. |
| +type BuildInputs struct { |
| + Store lease.VersionedStore |
| + Bucket string |
| + Source []SealedObject |
| + N uint64 |
| + GenerationPrefix string |
| + Op, Host string |
| + WorkDir string |
| + Now func() time.Time |
| + Verify func(dbPath string) error |
| +} |
| + |
| +// buildName refuses a host or op that is empty or could leave WorkDir: |
| +// the output path is built from the host and both name the generation. |
| +func buildName(kind, s string) error { |
| + if s == "" || s == "." || s == ".." || strings.ContainsAny(s, "/\\\x00") { |
| + return fmt.Errorf("moves: build has an unusable %s %q", kind, s) |
| + } |
| + return nil |
| +} |
| + |
| +func (in BuildInputs) validate() error { |
| + if in.Store == nil || in.Bucket == "" { |
| + return errors.New("moves: build needs a store and a bucket") |
| + } |
| + if in.N == 0 { |
| + return errors.New("moves: build needs a position") |
| + } |
| + if in.GenerationPrefix == "" || !strings.HasSuffix(in.GenerationPrefix, "/") { |
| + return fmt.Errorf("moves: build needs a generation prefix ending in /, not %q", in.GenerationPrefix) |
| + } |
| + if err := buildName("op", in.Op); err != nil { |
| + return err |
| + } |
| + if err := buildName("host", in.Host); err != nil { |
| + return err |
| + } |
| + if in.WorkDir == "" { |
| + return errors.New("moves: build needs a work directory") |
| + } |
| + if len(in.Source) == 0 { |
| + return fmt.Errorf("%w: the pinned set is empty", ErrBuildIncomplete) |
| + } |
| + return nil |
| +} |
| + |
| +// BuildGeneration builds the prepared generation from the pinned set |
| +// (§5.1 steps 2 and 7, §7.6 building): litestream's own planner and |
| +// restore over a SealedClient into `<WorkDir>/<Host>.db`, refused unless |
| +// the plan ends exactly at N and every planned file verifies; the database |
| +// check; then, in order, the intent, the snapshot [1, N], the anchor |
| +// [N, N] and the completed InstallManifest, each create-only. A replay at |
| +// any boundary restores again from the same versions (the scratch file is |
| +// removed first: litestream refuses an existing output and the file is |
| +// reproducible), reads the intent it finds and so encodes identical bytes, |
| +// and reuses an object it finds under its key only when the bytes are |
| +// equal. The build never deletes an object. |
| +func BuildGeneration(ctx context.Context, in BuildInputs) (InstallManifest, error) { |
| + if err := in.validate(); err != nil { |
| + return InstallManifest{}, err |
| + } |
| + now := in.Now |
| + if now == nil { |
| + now = time.Now |
| + } |
| + verify := in.Verify |
| + if verify == nil { |
| + verify = lease.CheckDB |
| + } |
| + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) |
| + |
| + client, err := NewSealedClient(in.Store, in.Bucket, in.Source) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + target := ltx.TXID(in.N) |
| + |
| + // The plan: litestream's own, rooted at the newest snapshot at or |
| + // below N and extended file by file. A gap under a TXID target is |
| + // ErrTxNotAvailable (the contiguity error is only raised for a |
| + // latest-state plan), as is a set that stops short. |
| + plan, err := litestream.CalcRestorePlan(ctx, client, target, time.Time{}, logger) |
| + if err != nil { |
| + if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) { |
| + return InstallManifest{}, fmt.Errorf("%w: planning to %s: %v", ErrBuildIncomplete, target, err) |
| + } |
| + return InstallManifest{}, fmt.Errorf("moves: plan the restore: %w", err) |
| + } |
| + if len(plan) == 0 || plan[len(plan)-1].MaxTXID != target { |
| + return InstallManifest{}, fmt.Errorf("%w: the plan ends at %s, not %s", ErrBuildIncomplete, plan[len(plan)-1].MaxTXID, target) |
| + } |
| + for _, info := range plan { |
| + if err := verifySealedFile(ctx, client, info); err != nil { |
| + return InstallManifest{}, err |
| + } |
| + } |
| + |
| + // A fresh scratch file. Litestream refuses an existing output path |
| + // and writes through `<path>.tmp`; a crash can leave either, plus the |
| + // sidecars the database check attaches. |
| + if err := os.MkdirAll(in.WorkDir, 0o700); err != nil { |
| + return InstallManifest{}, fmt.Errorf("moves: build scratch: %w", err) |
| + } |
| + dbPath := filepath.Join(in.WorkDir, in.Host+".db") |
| + for _, p := range []string{dbPath, dbPath + ".tmp", dbPath + "-wal", dbPath + "-shm", dbPath + "-txid"} { |
| + if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return InstallManifest{}, fmt.Errorf("moves: clear build scratch: %w", err) |
| + } |
| + } |
| + opt := litestream.NewRestoreOptions() |
| + opt.OutputPath = dbPath |
| + opt.TXID = target |
| + if err := litestream.NewReplicaWithClient(nil, client).Restore(ctx, opt); err != nil { |
| + if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) { |
| + return InstallManifest{}, fmt.Errorf("%w: restoring to %s: %v", ErrBuildIncomplete, target, err) |
| + } |
| + return InstallManifest{}, fmt.Errorf("moves: restore to %s: %w", target, err) |
| + } |
| + |
| + pageSize, pageCount, err := ReadDBHeader(dbPath) |
| + if err != nil { |
| + return InstallManifest{}, fmt.Errorf("moves: restored database: %w", err) |
| + } |
| + st, err := os.Stat(dbPath) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + if st.Size() != int64(pageSize)*int64(pageCount) { |
| + return InstallManifest{}, fmt.Errorf("moves: restored database is %d bytes, its header says %d pages of %d", st.Size(), pageCount, pageSize) |
| + } |
| + if err := verify(dbPath); err != nil { |
| + return InstallManifest{}, fmt.Errorf("moves: restored database check: %w", err) |
| + } |
| + // The check may attach a WAL and a shared-memory file; neither is part |
| + // of the build's result. |
| + for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| + _ = os.Remove(p) |
| + } |
| + |
| + // The intent, before any other PUT. On a replay the existing intent's |
| + // timestamp is the one that gives identical bytes. |
| + intent := BuildIntent{V: buildContractVersion, Op: in.Op, Host: in.Host, N: in.N, PageSize: pageSize, Commit: pageCount, Timestamp: now().UnixMilli()} |
| + intent, err = putOrReadIntent(ctx, in, intent) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + |
| + f, err := os.Open(dbPath) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + defer f.Close() |
| + |
| + snapshot, snapshotCRC, err := EncodeFull(f, pageSize, pageCount, 1, target, intent.Timestamp, 0) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + snapshotKey := SnapshotKey(in.GenerationPrefix, in.N) |
| + snapshotVersion, err := putOrReuse(ctx, in, snapshotKey, snapshot) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + snapshot = nil |
| + |
| + anchor, anchorCRC, err := EncodeFull(f, pageSize, pageCount, target, target, intent.Timestamp, litestream.WALHeaderSize) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + anchorKey := AnchorKey(in.GenerationPrefix, in.N) |
| + anchorVersion, err := putOrReuse(ctx, in, anchorKey, anchor) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + anchor = nil |
| + |
| + h := sha256.New() |
| + if _, err := io.Copy(h, io.NewSectionReader(f, 0, st.Size())); err != nil { |
| + return InstallManifest{}, fmt.Errorf("moves: hash the restored database: %w", err) |
| + } |
| + |
| + m := InstallManifest{ |
| + V: buildContractVersion, Op: in.Op, Host: in.Host, N: in.N, |
| + SnapshotKey: snapshotKey, SnapshotVersion: snapshotVersion, SnapshotCRC64: snapshotCRC, |
| + AnchorKey: anchorKey, AnchorVersion: anchorVersion, AnchorCRC64: anchorCRC, |
| + IntegrityCheck: "ok", |
| + DBSHA256: hex.EncodeToString(h.Sum(nil)), |
| + BuiltAt: now().UTC(), |
| + } |
| + return putOrReadManifest(ctx, in, m) |
| +} |
| + |
| +// verifySealedFile reads one planned file whole through the sealed client |
| +// and checks its CRC-64 trailer and that its header is the position the |
| +// seal named it by. |
| +func verifySealedFile(ctx context.Context, client *SealedClient, info *ltx.FileInfo) error { |
| + rc, err := client.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| + if err != nil { |
| + return fmt.Errorf("moves: open sealed %s: %w", ltx.FormatFilename(info.MinTXID, info.MaxTXID), err) |
| + } |
| + defer rc.Close() |
| + dec := ltx.NewDecoder(rc) |
| + if err := dec.Verify(); err != nil { |
| + return fmt.Errorf("moves: sealed %s at level %d does not verify: %w", ltx.FormatFilename(info.MinTXID, info.MaxTXID), info.Level, err) |
| + } |
| + if h := dec.Header(); h.MinTXID != info.MinTXID || h.MaxTXID != info.MaxTXID { |
| + return fmt.Errorf("moves: sealed %s at level %d carries [%s,%s]", ltx.FormatFilename(info.MinTXID, info.MaxTXID), info.Level, h.MinTXID, h.MaxTXID) |
| + } |
| + return nil |
| +} |
| + |
| +// currentVersion is the object version currently under key, found by an |
| +// exact-key listing (a store names versions, never a bare key). |
| +func currentVersion(ctx context.Context, in BuildInputs, key string) (lease.ObjectVersion, error) { |
| + vs, err := in.Store.ListVersions(ctx, in.Bucket, key) |
| + if err != nil { |
| + return lease.ObjectVersion{}, fmt.Errorf("moves: list %s: %w", key, err) |
| + } |
| + for _, v := range vs { |
| + if v.Key == key && v.IsLatest && !v.DeleteMarker { |
| + return v, nil |
| + } |
| + } |
| + return lease.ObjectVersion{}, fmt.Errorf("moves: %s exists but has no current version", key) |
| +} |
| + |
| +// putOrReuse writes body create-only under key and returns the version id. |
| +// When the key already holds an object, the current version's bytes must |
| +// equal body (identical by construction from the intent) and its id is the |
| +// answer; anything else is ErrBuildDiverged. |
| +func putOrReuse(ctx context.Context, in BuildInputs, key string, body []byte) (string, error) { |
| + v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body) |
| + if err == nil { |
| + return v, nil |
| + } |
| + if !errors.Is(err, lease.ErrExists) { |
| + return "", fmt.Errorf("moves: put %s: %w", key, err) |
| + } |
| + cur, err := currentVersion(ctx, in, key) |
| + if err != nil { |
| + return "", err |
| + } |
| + have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID) |
| + if err != nil { |
| + return "", fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err) |
| + } |
| + if !bytes.Equal(have, body) { |
| + return "", fmt.Errorf("%w: %s@%s", ErrBuildDiverged, key, cur.VersionID) |
| + } |
| + return cur.VersionID, nil |
| +} |
| + |
| +// putOrReadIntent writes the intent create-only, or reads the one already |
| +// there: it must name the same op, host, position, page size and commit, |
| +// and its timestamp is the one this build encodes with. |
| +func putOrReadIntent(ctx context.Context, in BuildInputs, want BuildIntent) (BuildIntent, error) { |
| + key := BuildIntentKey(in.GenerationPrefix) |
| + body, err := json.Marshal(want) |
| + if err != nil { |
| + return BuildIntent{}, err |
| + } |
| + if _, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil { |
| + return want, nil |
| + } else if !errors.Is(err, lease.ErrExists) { |
| + return BuildIntent{}, fmt.Errorf("moves: put %s: %w", key, err) |
| + } |
| + cur, err := currentVersion(ctx, in, key) |
| + if err != nil { |
| + return BuildIntent{}, err |
| + } |
| + have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID) |
| + if err != nil { |
| + return BuildIntent{}, fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err) |
| + } |
| + var got BuildIntent |
| + if err := json.Unmarshal(have, &got); err != nil { |
| + return BuildIntent{}, fmt.Errorf("%w: %s@%s is not an intent: %v", ErrBuildDiverged, key, cur.VersionID, err) |
| + } |
| + if got.V != want.V || got.Op != want.Op || got.Host != want.Host || got.N != want.N || got.PageSize != want.PageSize || got.Commit != want.Commit { |
| + return BuildIntent{}, fmt.Errorf("%w: %s@%s names other inputs", ErrBuildDiverged, key, cur.VersionID) |
| + } |
| + return got, nil |
| +} |
| + |
| +// putOrReadManifest writes the completed manifest create-only, or reads |
| +// the one already there, which must name the same position, keys, |
| +// versions, checksums and database digest; the stored one (its BuiltAt |
| +// included) is the answer. |
| +func putOrReadManifest(ctx context.Context, in BuildInputs, want InstallManifest) (InstallManifest, error) { |
| + key := InstallManifestKey(in.GenerationPrefix) |
| + body, err := json.Marshal(want) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + if _, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil { |
| + return want, nil |
| + } else if !errors.Is(err, lease.ErrExists) { |
| + return InstallManifest{}, fmt.Errorf("moves: put %s: %w", key, err) |
| + } |
| + cur, err := currentVersion(ctx, in, key) |
| + if err != nil { |
| + return InstallManifest{}, err |
| + } |
| + have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID) |
| + if err != nil { |
| + return InstallManifest{}, fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err) |
| + } |
| + var got InstallManifest |
| + if err := json.Unmarshal(have, &got); err != nil { |
| + return InstallManifest{}, fmt.Errorf("%w: %s@%s is not a manifest: %v", ErrBuildDiverged, key, cur.VersionID, err) |
| + } |
| + same := got.V == want.V && got.Op == want.Op && got.Host == want.Host && got.N == want.N && |
| + got.SnapshotKey == want.SnapshotKey && got.SnapshotVersion == want.SnapshotVersion && got.SnapshotCRC64 == want.SnapshotCRC64 && |
| + got.AnchorKey == want.AnchorKey && got.AnchorVersion == want.AnchorVersion && got.AnchorCRC64 == want.AnchorCRC64 && |
| + got.IntegrityCheck == want.IntegrityCheck && got.DBSHA256 == want.DBSHA256 |
| + if !same { |
| + return InstallManifest{}, fmt.Errorf("%w: %s@%s records another build", ErrBuildDiverged, key, cur.VersionID) |
| + } |
| + return got, nil |
| +} |
diff --git a/internal/moves/build_test.go b/internal/moves/build_test.go| new file mode 100644 |
| index 000000000..8958eb385 |
| --- /dev/null |
| +++ b/internal/moves/build_test.go |
| @@ -0,0 +1,875 @@ |
| +package moves |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/sha256" |
| + "database/sql" |
| + "encoding/binary" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log/slog" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/benbjohnson/litestream" |
| + lsfile "github.com/benbjohnson/litestream/file" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/lease/leasetest" |
| + "github.com/superfly/ltx" |
| + _ "modernc.org/sqlite" |
| +) |
| + |
| +// The tests here run real litestream v0.5.14 and ltx v0.5.1 over real |
| +// SQLite files (controller ruling: real data, never a fake provider as |
| +// proof). A source replica is produced the way internal/sim/litestreamfacts |
| +// produces one: litestream's file replica client over a database written |
| +// through modernc sqlite, every sync explicit. |
| + |
| +const ( |
| + testBucket = "data" |
| + testHost = "h" |
| + testOp = "op1" |
| + testLane = "moves/h/op1/replica/" |
| +) |
| + |
| +func testGenPrefix() string { return GenerationPrefix("instances/h", testOp) } |
| + |
| +// openLS opens path under litestream against client with no background |
| +// monitors, as litestreamfacts does. |
| +func openLS(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB { |
| + t.Helper() |
| + db := litestream.NewDB(path) |
| + db.Logger = slog.New(slog.NewTextHandler(io.Discard, nil)) |
| + db.SetMetaPath(meta) |
| + db.MonitorInterval = time.Hour |
| + db.Replica = litestream.NewReplicaWithClient(db, client) |
| + db.Replica.SyncInterval = time.Hour |
| + if err := db.Open(); err != nil { |
| + t.Fatalf("open %s: %v", path, err) |
| + } |
| + return db |
| +} |
| + |
| +// writeRows inserts n rows, each its own transaction. |
| +func writeRows(t *testing.T, path string, n int) { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i := 0; i < n; i++ { |
| + if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +func countRows(t *testing.T, path string) int { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + var n int |
| + if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| +} |
| + |
| +func pageCountPragma(t *testing.T, path string) int64 { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + var n int64 |
| + if err := d.QueryRow(`PRAGMA page_count`).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| +} |
| + |
| +func fileSHA256(t *testing.T, path string) string { |
| + t.Helper() |
| + b, err := os.ReadFile(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + s := sha256.Sum256(b) |
| + return hex.EncodeToString(s[:]) |
| +} |
| + |
| +// sourceReplica writes a database with rows rows in syncs separate |
| +// syncs, takes a level-9 snapshot when snapshot is set, and returns the |
| +// file replica client plus the newest level-0 position. |
| +func sourceReplica(t *testing.T, syncs int, snapshot bool) (litestream.ReplicaClient, uint64) { |
| + t.Helper() |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "src.db") |
| + writeRows(t, path, 1) |
| + client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| + db := openLS(t, path, filepath.Join(dir, "meta"), client) |
| + ctx := context.Background() |
| + for i := 0; i < syncs; i++ { |
| + if err := db.SyncAndWait(ctx); err != nil { |
| + t.Fatalf("sync: %v", err) |
| + } |
| + writeRows(t, path, 2) |
| + } |
| + if err := db.SyncAndWait(ctx); err != nil { |
| + t.Fatalf("sync: %v", err) |
| + } |
| + if snapshot { |
| + if _, err := db.Snapshot(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := db.Close(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + var n uint64 |
| + for _, f := range levelFiles(t, client, 0) { |
| + if uint64(f.MaxTXID) > n { |
| + n = uint64(f.MaxTXID) |
| + } |
| + } |
| + if n == 0 { |
| + t.Fatal("source replica has no level-0 files") |
| + } |
| + return client, n |
| +} |
| + |
| +func levelFiles(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo { |
| + t.Helper() |
| + itr, err := client.LTXFiles(context.Background(), level, 0, false) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + infos, err := ltx.SliceFileIterator(itr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return infos |
| +} |
| + |
| +func readLTX(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) []byte { |
| + t.Helper() |
| + rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rc.Close() |
| + b, err := io.ReadAll(rc) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| +} |
| + |
| +// sealReplica copies every LTX file of client into the bucket under lane, |
| +// as the staging lane holds them, and returns the pinned version set. |
| +func sealReplica(t *testing.T, b *leasetest.VersionedBucket, client litestream.ReplicaClient, lane string) []SealedObject { |
| + t.Helper() |
| + var out []SealedObject |
| + for level := 0; level <= litestream.SnapshotLevel; level++ { |
| + for _, info := range levelFiles(t, client, level) { |
| + body := readLTX(t, client, info) |
| + key := lane + fmt.Sprintf("%04x/", level) + ltx.FormatFilename(info.MinTXID, info.MaxTXID) |
| + v := b.Put(testBucket, key, body) |
| + out = append(out, SealedObject{Key: key, VersionID: v, Size: int64(len(body))}) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func newBucket() *leasetest.VersionedBucket { |
| + b := leasetest.NewVersionedBucket() |
| + b.AddBucket(testBucket) |
| + return b |
| +} |
| + |
| +func versionsOf(t *testing.T, b *leasetest.VersionedBucket, key string) []lease.ObjectVersion { |
| + t.Helper() |
| + all, err := b.ListVersions(context.Background(), testBucket, key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var out []lease.ObjectVersion |
| + for _, v := range all { |
| + if v.Key == key { |
| + out = append(out, v) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func decodeLTX(t *testing.T, data []byte) (ltx.Header, ltx.Trailer, int) { |
| + t.Helper() |
| + dec := ltx.NewDecoder(bytes.NewReader(data)) |
| + if err := dec.DecodeHeader(); err != nil { |
| + t.Fatal(err) |
| + } |
| + page := make([]byte, dec.Header().PageSize) |
| + pages := 0 |
| + for { |
| + var ph ltx.PageHeader |
| + if err := dec.DecodePage(&ph, page); err == io.EOF { |
| + break |
| + } else if err != nil { |
| + t.Fatal(err) |
| + } |
| + pages++ |
| + } |
| + if err := dec.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := ltx.NewDecoder(bytes.NewReader(data)).Verify(); err != nil { |
| + t.Fatalf("verify: %v", err) |
| + } |
| + return dec.Header(), dec.Trailer(), pages |
| +} |
| + |
| +// plannedFiles is how many files litestream's planner reads to reach n |
| +// over the pinned set. |
| +func plannedFiles(t *testing.T, b *leasetest.VersionedBucket, src []SealedObject, n uint64) int { |
| + t.Helper() |
| + c, err := NewSealedClient(b, testBucket, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + plan, err := litestream.CalcRestorePlan(context.Background(), c, ltx.TXID(n), time.Time{}, slog.New(slog.NewTextHandler(io.Discard, nil))) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return len(plan) |
| +} |
| + |
| +func sameManifest(a, b InstallManifest) bool { |
| + a.BuiltAt, b.BuiltAt = time.Time{}, time.Time{} |
| + return a == b |
| +} |
| + |
| +func inputs(b *leasetest.VersionedBucket, src []SealedObject, n uint64, work string) BuildInputs { |
| + return BuildInputs{ |
| + Store: b, Bucket: testBucket, Source: src, N: n, |
| + GenerationPrefix: testGenPrefix(), Op: testOp, Host: testHost, |
| + WorkDir: work, |
| + Now: func() time.Time { return time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) }, |
| + } |
| +} |
| + |
| +// restoreGeneration restores the built generation through litestream's own |
| +// Restore, reading exactly the objects the manifest names. |
| +func restoreGeneration(t *testing.T, b *leasetest.VersionedBucket, m InstallManifest, out string) { |
| + t.Helper() |
| + snapBody, err := b.GetVersion(context.Background(), testBucket, m.SnapshotKey, m.SnapshotVersion) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + anchorBody, err := b.GetVersion(context.Background(), testBucket, m.AnchorKey, m.AnchorVersion) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + gen, err := NewSealedClient(b, testBucket, []SealedObject{ |
| + {Key: m.SnapshotKey, VersionID: m.SnapshotVersion, Size: int64(len(snapBody))}, |
| + {Key: m.AnchorKey, VersionID: m.AnchorVersion, Size: int64(len(anchorBody))}, |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + r := litestream.NewReplicaWithClient(nil, gen) |
| + opt := litestream.NewRestoreOptions() |
| + opt.OutputPath = out |
| + if err := r.Restore(context.Background(), opt); err != nil { |
| + t.Fatalf("restore the generation: %v", err) |
| + } |
| +} |
| + |
| +func TestGenerationPrefixAndKeys(t *testing.T) { |
| + if got := GenerationPrefix("instances/h", "op1"); got != "instances/h/g-op1/" { |
| + t.Fatalf("GenerationPrefix = %q", got) |
| + } |
| + if got := GenerationPrefix("instances/h/", "op1"); got != "instances/h/g-op1/" { |
| + t.Fatalf("GenerationPrefix with a slash = %q", got) |
| + } |
| + if got := GenerationPrefix("instances/h/inc2", "op1"); got != "instances/h/inc2/g-op1/" { |
| + t.Fatalf("GenerationPrefix with an incarnation = %q", got) |
| + } |
| + p := "instances/h/g-op1/" |
| + if got := SnapshotKey(p, 7); got != p+"0009/0000000000000001-0000000000000007.ltx" { |
| + t.Fatalf("SnapshotKey = %q", got) |
| + } |
| + if got := AnchorKey(p, 7); got != p+"0000/0000000000000007-0000000000000007.ltx" { |
| + t.Fatalf("AnchorKey = %q", got) |
| + } |
| + if got := BuildIntentKey(p); got != p+"build-intent.json" { |
| + t.Fatalf("BuildIntentKey = %q", got) |
| + } |
| + if got := InstallManifestKey(p); got != p+"install.json" { |
| + t.Fatalf("InstallManifestKey = %q", got) |
| + } |
| +} |
| + |
| +func TestSealedClientServesOnlySealedVersions(t *testing.T) { |
| + b := newBucket() |
| + ctx := context.Background() |
| + key := testLane + "0000/" + ltx.FormatFilename(1, 1) |
| + // At least an LTX header's worth of bytes: the client refuses a |
| + // smaller object, as litestream's restore would. |
| + sealed := []byte(strings.Repeat("sealed bytes of 0000000000000001-0000000000000001 ", 3)) |
| + v1 := b.Put(testBucket, key, sealed) |
| + b.Put(testBucket, key, []byte("a newer version that landed after the seal")) |
| + // An unsealed sibling exists in the lane but is not in the pinned set. |
| + b.Put(testBucket, testLane+"0000/"+ltx.FormatFilename(2, 2), []byte("unsealed")) |
| + |
| + c, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: int64(len(sealed))}}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.Type() != "sealed" { |
| + t.Fatalf("Type = %q", c.Type()) |
| + } |
| + if err := c.Init(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + rc, err := c.OpenLTXFile(ctx, 0, 1, 1, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ := io.ReadAll(rc) |
| + rc.Close() |
| + if !bytes.Equal(got, sealed) { |
| + t.Fatalf("OpenLTXFile served %q, want the sealed version", got) |
| + } |
| + // A ranged open, as litestream's resumable reader asks for it. |
| + rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 5) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ = io.ReadAll(rc) |
| + rc.Close() |
| + if string(got) != string(sealed[7:12]) { |
| + t.Fatalf("ranged open = %q, want %q", got, sealed[7:12]) |
| + } |
| + rc, err = c.OpenLTXFile(ctx, 0, 1, 1, 7, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ = io.ReadAll(rc) |
| + rc.Close() |
| + if string(got) != string(sealed[7:]) { |
| + t.Fatalf("open to the end = %q, want %q", got, sealed[7:]) |
| + } |
| + |
| + if _, err := c.OpenLTXFile(ctx, 0, 2, 2, 0, 0); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("an unsealed key opened: %v", err) |
| + } |
| + infos := levelFiles(t, c, 0) |
| + if len(infos) != 1 || infos[0].MinTXID != 1 || infos[0].MaxTXID != 1 || infos[0].Size != int64(len(sealed)) || infos[0].Level != 0 { |
| + t.Fatalf("LTXFiles(0) = %+v", infos) |
| + } |
| + if infos := levelFiles(t, c, 9); len(infos) != 0 { |
| + t.Fatalf("LTXFiles(9) = %+v, want none", infos) |
| + } |
| + // seek skips files below it, as the file client does. |
| + itr, err := c.LTXFiles(ctx, 0, 2, false) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if itr.Next() { |
| + t.Fatal("seek past the only file still yielded it") |
| + } |
| + |
| + if _, err := c.WriteLTXFile(ctx, 0, 3, 3, bytes.NewReader(nil)); !errors.Is(err, ErrSealedReadOnly) { |
| + t.Fatalf("WriteLTXFile = %v", err) |
| + } |
| + if err := c.DeleteLTXFiles(ctx, infos); !errors.Is(err, ErrSealedReadOnly) { |
| + t.Fatalf("DeleteLTXFiles = %v", err) |
| + } |
| + if err := c.DeleteAll(ctx); !errors.Is(err, ErrSealedReadOnly) { |
| + t.Fatalf("DeleteAll = %v", err) |
| + } |
| + if n := len(versionsOf(t, b, key)); n != 2 { |
| + t.Fatalf("the sealed key has %d versions after the refused writes, want 2", n) |
| + } |
| + |
| + // G27: a version id that pins nothing is refused before any read. |
| + for _, bad := range []string{"", "null"} { |
| + if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: bad, Size: 10}}); err == nil { |
| + t.Fatalf("version %q accepted", bad) |
| + } |
| + } |
| + // A key that is not <level>/<min>-<max>.ltx has no position. |
| + if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: testLane + "lock.json", VersionID: v1, Size: 10}}); err == nil { |
| + t.Fatal("a non-LTX key accepted") |
| + } |
| + // Two objects at one position would make the plan ambiguous. |
| + if _, err := NewSealedClient(b, testBucket, []SealedObject{ |
| + {Key: key, VersionID: v1, Size: 10}, |
| + {Key: "other/" + "0000/" + ltx.FormatFilename(1, 1), VersionID: v1, Size: 10}, |
| + }); err == nil { |
| + t.Fatal("two objects at one position accepted") |
| + } |
| +} |
| + |
| +func TestBuildGenerationReachesNAndWritesSnapshotAndAnchor(t *testing.T) { |
| + src, n := sourceReplica(t, 3, true) |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + work := filepath.Join(t.TempDir(), "build") |
| + ctx := context.Background() |
| + |
| + m, err := BuildGeneration(ctx, inputs(b, seal, n, work)) |
| + if err != nil { |
| + t.Fatalf("BuildGeneration: %v", err) |
| + } |
| + prefix := testGenPrefix() |
| + if m.V != 1 || m.Op != testOp || m.Host != testHost || m.N != n { |
| + t.Fatalf("manifest identity = %+v", m) |
| + } |
| + if m.SnapshotKey != SnapshotKey(prefix, n) || m.AnchorKey != AnchorKey(prefix, n) { |
| + t.Fatalf("manifest keys = %q %q", m.SnapshotKey, m.AnchorKey) |
| + } |
| + if !lease.UsableVersionID(m.SnapshotVersion) || !lease.UsableVersionID(m.AnchorVersion) { |
| + t.Fatalf("manifest versions = %q %q", m.SnapshotVersion, m.AnchorVersion) |
| + } |
| + if m.IntegrityCheck != "ok" { |
| + t.Fatalf("IntegrityCheck = %q", m.IntegrityCheck) |
| + } |
| + dbPath := filepath.Join(work, testHost+".db") |
| + if got := fileSHA256(t, dbPath); got != m.DBSHA256 { |
| + t.Fatalf("DBSHA256 = %s, file is %s", m.DBSHA256, got) |
| + } |
| + if !m.BuiltAt.Equal(time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)) { |
| + t.Fatalf("BuiltAt = %v", m.BuiltAt) |
| + } |
| + |
| + snapBody, err := b.GetVersion(ctx, testBucket, m.SnapshotKey, m.SnapshotVersion) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + sh, st, _ := decodeLTX(t, snapBody) |
| + if sh.MinTXID != 1 || sh.MaxTXID != ltx.TXID(n) || sh.Flags != ltx.HeaderFlagNoChecksum || sh.WALOffset != 0 { |
| + t.Fatalf("snapshot header = %+v", sh) |
| + } |
| + if uint64(st.FileChecksum) != m.SnapshotCRC64 { |
| + t.Fatalf("SnapshotCRC64 = %x, trailer %x", m.SnapshotCRC64, uint64(st.FileChecksum)) |
| + } |
| + anchorBody, err := b.GetVersion(ctx, testBucket, m.AnchorKey, m.AnchorVersion) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ah, at, _ := decodeLTX(t, anchorBody) |
| + if ah.MinTXID != ltx.TXID(n) || ah.MaxTXID != ltx.TXID(n) || ah.Flags != ltx.HeaderFlagNoChecksum || ah.WALOffset != litestream.WALHeaderSize { |
| + t.Fatalf("anchor header = %+v", ah) |
| + } |
| + if ah.PageSize != sh.PageSize || ah.Commit != sh.Commit || ah.Timestamp != sh.Timestamp { |
| + t.Fatalf("anchor and snapshot disagree: %+v vs %+v", ah, sh) |
| + } |
| + if uint64(at.FileChecksum) != m.AnchorCRC64 { |
| + t.Fatalf("AnchorCRC64 = %x, trailer %x", m.AnchorCRC64, uint64(at.FileChecksum)) |
| + } |
| + ps, pc, err := ReadDBHeader(dbPath) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if sh.PageSize != ps || sh.Commit != pc { |
| + t.Fatalf("snapshot header %d/%d, database %d/%d", sh.PageSize, sh.Commit, ps, pc) |
| + } |
| + |
| + // The intent was written and names the encoding inputs. |
| + intentBody, ok := b.Current(testBucket, BuildIntentKey(prefix)) |
| + if !ok { |
| + t.Fatal("no build intent") |
| + } |
| + var intent BuildIntent |
| + if err := json.Unmarshal(intentBody, &intent); err != nil { |
| + t.Fatal(err) |
| + } |
| + if intent.V != 1 || intent.Op != testOp || intent.Host != testHost || intent.N != n || intent.PageSize != ps || intent.Commit != pc || intent.Timestamp != sh.Timestamp { |
| + t.Fatalf("intent = %+v", intent) |
| + } |
| + |
| + // The manifest in the bucket is the one returned. |
| + manifestBody, ok := b.Current(testBucket, InstallManifestKey(prefix)) |
| + if !ok { |
| + t.Fatal("no install manifest") |
| + } |
| + var stored InstallManifest |
| + if err := json.Unmarshal(manifestBody, &stored); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !sameManifest(stored, m) || !stored.BuiltAt.Equal(m.BuiltAt) { |
| + t.Fatalf("stored manifest %+v, returned %+v", stored, m) |
| + } |
| + |
| + // Nothing but the four objects was written to the generation. |
| + gen, err := b.ListVersions(ctx, testBucket, prefix) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var keys []string |
| + for _, v := range gen { |
| + keys = append(keys, v.Key) |
| + } |
| + sort.Strings(keys) |
| + want := []string{AnchorKey(prefix, n), SnapshotKey(prefix, n), BuildIntentKey(prefix), InstallManifestKey(prefix)} |
| + sort.Strings(want) |
| + if strings.Join(keys, ",") != strings.Join(want, ",") { |
| + t.Fatalf("generation holds %v, want %v", keys, want) |
| + } |
| + |
| + // Litestream's own restore of the generation yields the same database. |
| + out := filepath.Join(t.TempDir(), "restored.db") |
| + restoreGeneration(t, b, m, out) |
| + if got := fileSHA256(t, out); got != m.DBSHA256 { |
| + t.Fatalf("the generation restores to %s, the build's file is %s", got, m.DBSHA256) |
| + } |
| + if err := lease.CheckDB(out); err != nil { |
| + t.Fatalf("CheckDB on the restored generation: %v", err) |
| + } |
| + if got, want := countRows(t, out), countRows(t, dbPath); got != want { |
| + t.Fatalf("restored %d rows, built %d", got, want) |
| + } |
| +} |
| + |
| +func TestInstallManifestKeys(t *testing.T) { |
| + src, n := sourceReplica(t, 1, true) |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + if _, err := BuildGeneration(context.Background(), inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil { |
| + t.Fatal(err) |
| + } |
| + body, ok := b.Current(testBucket, InstallManifestKey(testGenPrefix())) |
| + if !ok { |
| + t.Fatal("no install manifest") |
| + } |
| + var got map[string]any |
| + if err := json.Unmarshal(body, &got); err != nil { |
| + t.Fatal(err) |
| + } |
| + var keys []string |
| + for k := range got { |
| + keys = append(keys, k) |
| + } |
| + sort.Strings(keys) |
| + 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"} |
| + if strings.Join(keys, " ") != strings.Join(want, " ") { |
| + t.Fatalf("install.json keys = %v, want %v", keys, want) |
| + } |
| +} |
| + |
| +// Review Focus 3, witness wBuildCrashes: a build that dies after the |
| +// snapshot PUT, or during the restore, is replayed in the same WorkDir |
| +// and ends with one snapshot version and one manifest. |
| +func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) { |
| + src, n := sourceReplica(t, 2, true) |
| + prefix := testGenPrefix() |
| + ctx := context.Background() |
| + |
| + t.Run("after the snapshot PUT", func(t *testing.T) { |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + work := filepath.Join(t.TempDir(), "build") |
| + boom := errors.New("the box died") |
| + b.FailOn = func(op, key string) error { |
| + if op == "put" && key == AnchorKey(prefix, n) { |
| + return boom |
| + } |
| + return nil |
| + } |
| + if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) { |
| + t.Fatalf("first run = %v, want the injected failure", err) |
| + } |
| + snaps := versionsOf(t, b, SnapshotKey(prefix, n)) |
| + if len(snaps) != 1 { |
| + t.Fatalf("after the crash the snapshot has %d versions", len(snaps)) |
| + } |
| + if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok { |
| + t.Fatal("a manifest exists after a crash before the anchor") |
| + } |
| + if _, err := os.Stat(filepath.Join(work, testHost+".db")); err != nil { |
| + t.Fatalf("the restored file should still be on disk: %v", err) |
| + } |
| + first, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), snaps[0].VersionID) |
| + |
| + b.FailOn = nil |
| + m, err := BuildGeneration(ctx, inputs(b, seal, n, work)) |
| + if err != nil { |
| + t.Fatalf("replay: %v", err) |
| + } |
| + snaps = versionsOf(t, b, SnapshotKey(prefix, n)) |
| + if len(snaps) != 1 { |
| + t.Fatalf("after the replay the snapshot has %d versions, want 1", len(snaps)) |
| + } |
| + if m.SnapshotVersion != snaps[0].VersionID { |
| + t.Fatalf("manifest names snapshot version %s, the bucket has %s", m.SnapshotVersion, snaps[0].VersionID) |
| + } |
| + again, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), m.SnapshotVersion) |
| + if !bytes.Equal(first, again) { |
| + t.Fatal("the replay's snapshot bytes differ from the first run's") |
| + } |
| + if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); !ok { |
| + t.Fatal("no manifest after the replay") |
| + } |
| + if n := len(versionsOf(t, b, BuildIntentKey(prefix))); n != 1 { |
| + t.Fatalf("intent has %d versions, want 1", n) |
| + } |
| + // A third run is a pure replay: same manifest, no new versions. |
| + m2, err := BuildGeneration(ctx, inputs(b, seal, n, work)) |
| + if err != nil { |
| + t.Fatalf("second replay: %v", err) |
| + } |
| + if !sameManifest(m2, m) || !m2.BuiltAt.Equal(m.BuiltAt) { |
| + t.Fatalf("second replay returned %+v, first %+v", m2, m) |
| + } |
| + if n := len(versionsOf(t, b, AnchorKey(prefix, n))); n != 1 { |
| + t.Fatalf("anchor has %d versions, want 1", n) |
| + } |
| + if n := len(versionsOf(t, b, InstallManifestKey(prefix))); n != 1 { |
| + t.Fatalf("manifest has %d versions, want 1", n) |
| + } |
| + }) |
| + |
| + t.Run("during the restore", func(t *testing.T) { |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + work := filepath.Join(t.TempDir(), "build") |
| + boom := errors.New("the network died") |
| + // Every planned file is read once to verify it before the restore; |
| + // the restore's own reads come after. Fail from then on. |
| + planned := plannedFiles(t, b, seal, n) |
| + gets := 0 |
| + b.FailOn = func(op, key string) error { |
| + if op != "get" { |
| + return nil |
| + } |
| + gets++ |
| + if gets > planned { |
| + return boom |
| + } |
| + return nil |
| + } |
| + // Litestream's restore pipeline does not carry the read error's |
| + // identity (its compactor closes a pipe and the decoder reports the |
| + // EOF), so the assertion is that the restore failed and wrote |
| + // nothing, not that boom survives the wrapping. |
| + if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); err == nil { |
| + t.Fatal("the restore succeeded with its reads failing") |
| + } else if errors.Is(err, ErrBuildIncomplete) || errors.Is(err, ErrBuildDiverged) { |
| + t.Fatalf("first run = %v, want a plain restore failure", err) |
| + } else { |
| + t.Logf("first run failed as injected: %v", err) |
| + } |
| + if gets <= planned { |
| + t.Fatalf("the failure was not injected: %d gets, %d planned", gets, planned) |
| + } |
| + if vs, err := b.ListVersions(ctx, testBucket, prefix); err != nil || len(vs) != 0 { |
| + t.Fatalf("a failed restore left %d objects in the generation", len(vs)) |
| + } |
| + // A crash leaves whatever the writer had on disk: a partial output |
| + // and litestream's temporary file. |
| + if err := os.MkdirAll(work, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, p := range []string{testHost + ".db", testHost + ".db.tmp", testHost + ".db-wal", testHost + ".db-shm", testHost + ".db-txid"} { |
| + if err := os.WriteFile(filepath.Join(work, p), []byte("partial"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + b.FailOn = nil |
| + m, err := BuildGeneration(ctx, inputs(b, seal, n, work)) |
| + if err != nil { |
| + t.Fatalf("replay: %v", err) |
| + } |
| + if got := fileSHA256(t, filepath.Join(work, testHost+".db")); got != m.DBSHA256 { |
| + t.Fatalf("the replay's file is %s, manifest %s", got, m.DBSHA256) |
| + } |
| + out := filepath.Join(t.TempDir(), "restored.db") |
| + restoreGeneration(t, b, m, out) |
| + if got := fileSHA256(t, out); got != m.DBSHA256 { |
| + t.Fatalf("the generation restores to %s, want %s", got, m.DBSHA256) |
| + } |
| + }) |
| +} |
| + |
| +func TestBuildRefusesAGapOrAShortRestore(t *testing.T) { |
| + // Level 0 only: with no level-9 snapshot every level-0 file is on the |
| + // restore path, so a missing one is a gap. |
| + src, n := sourceReplica(t, 4, false) |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + if len(seal) < 3 { |
| + t.Fatalf("need at least three level-0 files, have %d", len(seal)) |
| + } |
| + prefix := testGenPrefix() |
| + ctx := context.Background() |
| + |
| + gapped := append(append([]SealedObject{}, seal[:1]...), seal[2:]...) |
| + _, err := BuildGeneration(ctx, inputs(b, gapped, n, filepath.Join(t.TempDir(), "build"))) |
| + if !errors.Is(err, ErrBuildIncomplete) { |
| + t.Fatalf("gap: %v, want ErrBuildIncomplete", err) |
| + } |
| + if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 { |
| + t.Fatalf("a gap wrote %d objects into the generation", len(vs)) |
| + } |
| + |
| + _, err = BuildGeneration(ctx, inputs(b, seal, n+5, filepath.Join(t.TempDir(), "build"))) |
| + if !errors.Is(err, ErrBuildIncomplete) { |
| + t.Fatalf("short: %v, want ErrBuildIncomplete", err) |
| + } |
| + if vs, _ := b.ListVersions(ctx, testBucket, prefix); len(vs) != 0 { |
| + t.Fatalf("a short restore wrote %d objects into the generation", len(vs)) |
| + } |
| + |
| + // The whole set reaches n. |
| + if _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))); err != nil { |
| + t.Fatalf("the complete set: %v", err) |
| + } |
| +} |
| + |
| +func TestBuildRefusesADivergentIntent(t *testing.T) { |
| + src, n := sourceReplica(t, 1, true) |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + prefix := testGenPrefix() |
| + ctx := context.Background() |
| + // An intent from another build of this generation with other inputs. |
| + other, _ := json.Marshal(BuildIntent{V: 1, Op: testOp, Host: testHost, N: n, PageSize: 1024, Commit: 1, Timestamp: 5}) |
| + b.Put(testBucket, BuildIntentKey(prefix), other) |
| + _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))) |
| + if !errors.Is(err, ErrBuildDiverged) { |
| + t.Fatalf("divergent intent: %v, want ErrBuildDiverged", err) |
| + } |
| + if _, ok := b.Current(testBucket, SnapshotKey(prefix, n)); ok { |
| + t.Fatal("a snapshot was written under a divergent intent") |
| + } |
| +} |
| + |
| +func TestBuildRefusesADivergentSnapshot(t *testing.T) { |
| + src, n := sourceReplica(t, 1, true) |
| + b := newBucket() |
| + seal := sealReplica(t, b, src, testLane) |
| + prefix := testGenPrefix() |
| + ctx := context.Background() |
| + b.Put(testBucket, SnapshotKey(prefix, n), []byte("not this build's bytes")) |
| + _, err := BuildGeneration(ctx, inputs(b, seal, n, filepath.Join(t.TempDir(), "build"))) |
| + if !errors.Is(err, ErrBuildDiverged) { |
| + t.Fatalf("divergent snapshot: %v, want ErrBuildDiverged", err) |
| + } |
| + if _, ok := b.Current(testBucket, AnchorKey(prefix, n)); ok { |
| + t.Fatal("an anchor was written beside a divergent snapshot") |
| + } |
| +} |
| + |
| +func TestReadDBHeader(t *testing.T) { |
| + path := filepath.Join(t.TempDir(), "x.db") |
| + writeRows(t, path, 3) |
| + ps, pc, err := ReadDBHeader(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + st, _ := os.Stat(path) |
| + if ps != 4096 || int64(pc)*int64(ps) != st.Size() { |
| + t.Fatalf("header %d x %d, file %d bytes", pc, ps, st.Size()) |
| + } |
| + if err := os.WriteFile(path, []byte("not a database at all, just text"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := ReadDBHeader(path); err == nil { |
| + t.Fatal("garbage accepted") |
| + } |
| +} |
| + |
| +// TestEncodeFullSkipsTheLockPage: a database that spans SQLite's lock page |
| +// encodes without it, verifies, and restores through litestream to a file |
| +// whose page count is the original's. |
| +func TestEncodeFullSkipsTheLockPage(t *testing.T) { |
| + if testing.Short() { |
| + t.Skip("writes a 1 GiB database") |
| + } |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "big.db") |
| + writeRows(t, path, 1) |
| + const pageSize = 4096 |
| + lock := ltx.LockPgno(pageSize) |
| + pageCount := lock + 1 |
| + f, err := os.OpenFile(path, os.O_RDWR, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := f.Truncate(int64(pageCount) * pageSize); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The in-header size follows the file, as SQLite keeps it. |
| + var hdr [4]byte |
| + binary.BigEndian.PutUint32(hdr[:], pageCount) |
| + if _, err := f.WriteAt(hdr[:], 28); err != nil { |
| + t.Fatal(err) |
| + } |
| + ps, pc, err := ReadDBHeader(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if ps != pageSize || pc != pageCount { |
| + t.Fatalf("header %d x %d", pc, ps) |
| + } |
| + origPages := pageCountPragma(t, path) |
| + if origPages != int64(pageCount) { |
| + t.Fatalf("PRAGMA page_count = %d, want %d", origPages, pageCount) |
| + } |
| + |
| + data, crc, err := EncodeFull(f, ps, pc, 1, 1, 1234, 0) |
| + f.Close() |
| + if err != nil { |
| + t.Fatalf("EncodeFull: %v", err) |
| + } |
| + h, tr, pages := decodeLTX(t, data) |
| + if pages != int(pageCount)-1 { |
| + t.Fatalf("encoded %d pages, want %d (every page but the lock page)", pages, pageCount-1) |
| + } |
| + if h.Commit != pageCount || uint64(tr.FileChecksum) != crc { |
| + t.Fatalf("header commit %d crc %x, returned crc %x", h.Commit, uint64(tr.FileChecksum), crc) |
| + } |
| + |
| + b := newBucket() |
| + key := SnapshotKey(testGenPrefix(), 1) |
| + v := b.Put(testBucket, key, data) |
| + gen, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v, Size: int64(len(data))}}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + out := filepath.Join(dir, "restored.db") |
| + r := litestream.NewReplicaWithClient(nil, gen) |
| + opt := litestream.NewRestoreOptions() |
| + opt.OutputPath = out |
| + if err := r.Restore(context.Background(), opt); err != nil { |
| + t.Fatalf("restore: %v", err) |
| + } |
| + if got := pageCountPragma(t, out); got != origPages { |
| + t.Fatalf("restored PRAGMA page_count = %d, original %d", got, origPages) |
| + } |
| + st, err := os.Stat(out) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if st.Size() != int64(pageCount)*pageSize { |
| + t.Fatalf("restored file is %d bytes, want %d", st.Size(), int64(pageCount)*pageSize) |
| + } |
| +} |
diff --git a/internal/moves/ltxbuild.go b/internal/moves/ltxbuild.go| new file mode 100644 |
| index 000000000..af437922d |
| --- /dev/null |
| +++ b/internal/moves/ltxbuild.go |
| @@ -0,0 +1,109 @@ |
| +package moves |
| + |
| +import ( |
| + "bytes" |
| + "encoding/binary" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "os" |
| + |
| + "github.com/superfly/ltx" |
| +) |
| + |
| +// sqliteHeaderSize is the SQLite database header, the first 100 bytes of |
| +// page 1. |
| +const sqliteHeaderSize = 100 |
| + |
| +var sqliteMagic = []byte("SQLite format 3\x00") |
| + |
| +// ReadDBHeader reads the page size (header bytes 16-17; the value 1 means |
| +// 65536) and the in-header database size in pages (bytes 28-31) of the |
| +// SQLite database at path. The in-header size is what SQLite keeps current |
| +// in WAL mode and what litestream's restore writes; BuildGeneration checks |
| +// it against the file size before trusting it. |
| +func ReadDBHeader(path string) (pageSize uint32, pageCount uint32, err error) { |
| + f, err := os.Open(path) |
| + if err != nil { |
| + return 0, 0, err |
| + } |
| + defer f.Close() |
| + var hdr [sqliteHeaderSize]byte |
| + if _, err := io.ReadFull(f, hdr[:]); err != nil { |
| + return 0, 0, fmt.Errorf("read sqlite header of %s: %w", path, err) |
| + } |
| + if !bytes.Equal(hdr[:len(sqliteMagic)], sqliteMagic) { |
| + return 0, 0, fmt.Errorf("%s is not a SQLite database", path) |
| + } |
| + pageSize = uint32(binary.BigEndian.Uint16(hdr[16:18])) |
| + if pageSize == 1 { |
| + pageSize = 65536 |
| + } |
| + if !ltx.IsValidPageSize(pageSize) { |
| + return 0, 0, fmt.Errorf("%s has an invalid page size %d", path, pageSize) |
| + } |
| + pageCount = binary.BigEndian.Uint32(hdr[28:32]) |
| + if pageCount == 0 { |
| + return 0, 0, fmt.Errorf("%s has no in-header database size", path) |
| + } |
| + return pageSize, pageCount, nil |
| +} |
| + |
| +// EncodeFull encodes every page of db, pages 1 to pageCount in order, as |
| +// one LTX file [min, max] with the NoChecksum flag, and returns the bytes |
| +// and the trailer's CRC-64 file checksum. SQLite's lock page |
| +// (ltx.LockPgno(pageSize), the page holding PENDING_BYTE) is skipped: the |
| +// encoder refuses it, litestream never ships it, and the decoder writes it |
| +// back as zeros; Commit keeps the full page count. With min 1 the file is |
| +// a snapshot; with min == max == N and walOffset litestream.WALHeaderSize |
| +// it is the anchor litestream's open reads its position from (§5.1 step |
| +// 7). The header timestamp is an input, so the same inputs give the same |
| +// bytes, which is what lets a replayed PUT repeat its object. The bytes |
| +// are read back and verified before they are returned. |
| +func EncodeFull(db io.ReaderAt, pageSize, pageCount uint32, min, max ltx.TXID, timestamp int64, walOffset int64) (data []byte, crc64 uint64, err error) { |
| + if pageCount == 0 { |
| + return nil, 0, errors.New("moves: cannot encode an empty database") |
| + } |
| + var buf bytes.Buffer |
| + enc, err := ltx.NewEncoder(&buf) |
| + if err != nil { |
| + return nil, 0, err |
| + } |
| + if err := enc.EncodeHeader(ltx.Header{ |
| + Version: ltx.Version, |
| + Flags: ltx.HeaderFlagNoChecksum, |
| + PageSize: pageSize, |
| + Commit: pageCount, |
| + MinTXID: min, |
| + MaxTXID: max, |
| + Timestamp: timestamp, |
| + WALOffset: walOffset, |
| + }); err != nil { |
| + return nil, 0, fmt.Errorf("moves: encode header: %w", err) |
| + } |
| + lock := ltx.LockPgno(pageSize) |
| + page := make([]byte, pageSize) |
| + for pgno := uint32(1); pgno <= pageCount; pgno++ { |
| + if pgno == lock { |
| + continue |
| + } |
| + if _, err := db.ReadAt(page, int64(pgno-1)*int64(pageSize)); err != nil { |
| + return nil, 0, fmt.Errorf("moves: read page %d: %w", pgno, err) |
| + } |
| + if err := enc.EncodePage(ltx.PageHeader{Pgno: pgno}, page); err != nil { |
| + return nil, 0, fmt.Errorf("moves: encode page %d: %w", pgno, err) |
| + } |
| + } |
| + if err := enc.Close(); err != nil { |
| + return nil, 0, fmt.Errorf("moves: close encoder: %w", err) |
| + } |
| + data = buf.Bytes() |
| + dec := ltx.NewDecoder(bytes.NewReader(data)) |
| + if err := dec.Verify(); err != nil { |
| + return nil, 0, fmt.Errorf("moves: verify encoded %s: %w", ltx.FormatFilename(min, max), err) |
| + } |
| + if dec.Trailer().FileChecksum != enc.Trailer().FileChecksum { |
| + return nil, 0, fmt.Errorf("moves: encoded %s reads back with another checksum", ltx.FormatFilename(min, max)) |
| + } |
| + return data, uint64(dec.Trailer().FileChecksum), nil |
| +} |
diff --git a/internal/moves/sealedclient.go b/internal/moves/sealedclient.go| new file mode 100644 |
| index 000000000..e1d1ef2b2 |
| --- /dev/null |
| +++ b/internal/moves/sealedclient.go |
| @@ -0,0 +1,195 @@ |
| +package moves |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log/slog" |
| + "os" |
| + "strconv" |
| + "strings" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/superfly/ltx" |
| +) |
| + |
| +// SealedObject is one pinned object version of a generation build's |
| +// source: an LTX file named by its full key, the exact version id the seal |
| +// (or a genselect plan, or a current listing) pinned, and its size. Task 5's |
| +// Seal lists the staging lane's replica/ objects in this shape (plan:688); |
| +// region HA's relocation feeds the same shape from its own pins (L16). |
| +type SealedObject struct { |
| + Key string `json:"key"` |
| + VersionID string `json:"version_id"` |
| + Size int64 `json:"size"` |
| +} |
| + |
| +// ErrSealedReadOnly is every write a SealedClient refuses: the pinned set |
| +// is immutable after the flip, and the build never deletes (§7.6). |
| +var ErrSealedReadOnly = errors.New("moves: the sealed replica is read-only") |
| + |
| +// SealedClient is a litestream.ReplicaClient over exactly one pinned set of |
| +// object versions. Litestream's planner and restore read through it and |
| +// never see a version the seal did not name, whether a newer write to a |
| +// sealed key or an unsealed sibling under the same prefix. A key's position |
| +// (level, min TXID, max TXID) is its last two path elements, |
| +// `<level>/<min>-<max>.ltx`, so the client names no lane or generation. |
| +type SealedClient struct { |
| + store lease.VersionedStore |
| + bucket string |
| + byPos map[sealedPos]SealedObject |
| + infos []ltx.FileInfo |
| + logger *slog.Logger |
| +} |
| + |
| +type sealedPos struct { |
| + level int |
| + min, max ltx.TXID |
| +} |
| + |
| +var _ interface { |
| + Type() string |
| + Init(ctx context.Context) error |
| + LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) |
| + OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) |
| + WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) |
| + DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error |
| + DeleteAll(ctx context.Context) error |
| + SetLogger(logger *slog.Logger) |
| +} = (*SealedClient)(nil) |
| + |
| +// NewSealedClient pins source. It refuses, before any read, an object with |
| +// no usable version id (G27: "" is no version and "null" is what a |
| +// versioning-suspended bucket answers, so neither pins anything), a key |
| +// that is not an LTX file under a level directory, a size below an LTX |
| +// header (litestream's restore refuses such a file, and a size of zero |
| +// would make its premature-EOF detection blind), and two objects at one |
| +// position, which would make the plan ambiguous. |
| +func NewSealedClient(store lease.VersionedStore, bucket string, source []SealedObject) (*SealedClient, error) { |
| + if store == nil || bucket == "" { |
| + return nil, errors.New("moves: sealed client needs a store and a bucket") |
| + } |
| + c := &SealedClient{store: store, bucket: bucket, byPos: map[sealedPos]SealedObject{}, logger: slog.Default()} |
| + for _, o := range source { |
| + if !lease.UsableVersionID(o.VersionID) { |
| + return nil, fmt.Errorf("moves: sealed object %s has no usable version id %q", o.Key, o.VersionID) |
| + } |
| + pos, err := sealedPosition(o.Key) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if o.Size < ltx.HeaderSize { |
| + return nil, fmt.Errorf("moves: sealed object %s has size %d, below an LTX header", o.Key, o.Size) |
| + } |
| + if prev, dup := c.byPos[pos]; dup { |
| + return nil, fmt.Errorf("moves: sealed objects %s and %s share one position", prev.Key, o.Key) |
| + } |
| + c.byPos[pos] = o |
| + c.infos = append(c.infos, ltx.FileInfo{Level: pos.level, MinTXID: pos.min, MaxTXID: pos.max, Size: o.Size}) |
| + } |
| + return c, nil |
| +} |
| + |
| +// sealedPosition reads `<level>/<min>-<max>.ltx` off the end of key, with |
| +// the level directory's four lowercase hex digits and the canonical file |
| +// name, the same shape ReplicaSource's keep admits. |
| +func sealedPosition(key string) (sealedPos, error) { |
| + parts := strings.Split(key, "/") |
| + if len(parts) < 2 { |
| + return sealedPos{}, fmt.Errorf("moves: sealed object %s is not <level>/<min>-<max>.ltx", key) |
| + } |
| + dir, name := parts[len(parts)-2], parts[len(parts)-1] |
| + level, err := strconv.ParseUint(dir, 16, 16) |
| + if err != nil || len(dir) != 4 || fmt.Sprintf("%04x", level) != dir { |
| + return sealedPos{}, fmt.Errorf("moves: sealed object %s has no level directory", key) |
| + } |
| + min, max, err := ltx.ParseFilename(name) |
| + if err != nil || min == 0 || min > max || ltx.FormatFilename(min, max) != name { |
| + return sealedPos{}, fmt.Errorf("moves: sealed object %s is not an LTX file name", key) |
| + } |
| + return sealedPos{level: int(level), min: min, max: max}, nil |
| +} |
| + |
| +// Type names the client in litestream's logs. |
| +func (c *SealedClient) Type() string { return "sealed" } |
| + |
| +// Init has nothing to open: the store is already bound. |
| +func (c *SealedClient) Init(ctx context.Context) error { return nil } |
| + |
| +// SetLogger stores the logger litestream hands over. |
| +func (c *SealedClient) SetLogger(logger *slog.Logger) { |
| + if logger != nil { |
| + c.logger = logger |
| + } |
| +} |
| + |
| +// LTXFiles lists the pinned files at level from seek on, in TXID order. |
| +// useMetadata is ignored: a pinned set has no timestamps to fetch, and the |
| +// build restores by TXID, never by time. |
| +func (c *SealedClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { |
| + var out []*ltx.FileInfo |
| + for i := range c.infos { |
| + info := c.infos[i] |
| + if info.Level != level || info.MinTXID < seek { |
| + continue |
| + } |
| + out = append(out, &info) // a copy: the iterator sorts in place |
| + } |
| + return ltx.NewFileInfoSliceIterator(out), nil |
| +} |
| + |
| +// OpenLTXFile streams exactly the pinned version of the file at the |
| +// position, or os.ErrNotExist when the seal names none. offset and size |
| +// follow litestream's convention (size 0 reads to the end) and go through |
| +// the store's VersionRangeOpener when it has one, otherwise the whole |
| +// version is opened and the range is cut locally. |
| +func (c *SealedClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) { |
| + o, ok := c.byPos[sealedPos{level: level, min: minTXID, max: maxTXID}] |
| + if !ok { |
| + return nil, fmt.Errorf("moves: no sealed object at level %d %s: %w", level, ltx.FormatFilename(minTXID, maxTXID), os.ErrNotExist) |
| + } |
| + if offset < 0 || size < 0 { |
| + return nil, fmt.Errorf("moves: invalid range %d+%d for %s", offset, size, o.Key) |
| + } |
| + if offset == 0 && size == 0 { |
| + return c.store.OpenVersion(ctx, c.bucket, o.Key, o.VersionID) |
| + } |
| + if ranged, ok := c.store.(lease.VersionRangeOpener); ok { |
| + length := int64(-1) |
| + if size > 0 { |
| + length = size |
| + } |
| + return ranged.OpenVersionRange(ctx, c.bucket, o.Key, o.VersionID, offset, length) |
| + } |
| + rc, err := c.store.OpenVersion(ctx, c.bucket, o.Key, o.VersionID) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if offset > 0 { |
| + if _, err := io.CopyN(io.Discard, rc, offset); err != nil { |
| + rc.Close() |
| + return nil, fmt.Errorf("moves: skip %d bytes of %s: %w", offset, o.Key, err) |
| + } |
| + } |
| + if size > 0 { |
| + return struct { |
| + io.Reader |
| + io.Closer |
| + }{io.LimitReader(rc, size), rc}, nil |
| + } |
| + return rc, nil |
| +} |
| + |
| +// WriteLTXFile is refused: the pinned set is immutable. |
| +func (c *SealedClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) { |
| + return nil, ErrSealedReadOnly |
| +} |
| + |
| +// DeleteLTXFiles is refused: the build never deletes. |
| +func (c *SealedClient) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error { |
| + return ErrSealedReadOnly |
| +} |
| + |
| +// DeleteAll is refused: the build never deletes. |
| +func (c *SealedClient) DeleteAll(ctx context.Context) error { return ErrSealedReadOnly } |