amadan

carlos / platform Public

dbdoor: restore to a TXID and plan a restore's exact size (Plan 5 Task 2)

RestoreToTarget restores to a named transaction (0: latest); RestoreTo
is RestoreToTarget at 0, unchanged for carlos db pull. PlanRestore asks
litestream for the restore plan and reads the last planned file's LTX
header for the restored size (Commit pages times PageSize). A target
older than the retained history is ErrHistoryGone.

Preflight rulings: the history test builds a snapshot-level file and
removes the L0 files under it (M3); RestorePlan carries Target, and a
restore pinned at plan.TXID reproduces plan.RestoredBytes after the
source moves on (m4, M2); the RestoreToTarget comment says what
litestream v0.5.14 does (m5); no placeholder lines (m6).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5f575f55c302f2e8897ac73cb420acba7cd5a959 parent 9c98d55
3 files changed, +325 −2
  • internal/dbdoor/plan.go +82 −0
  • internal/dbdoor/plan_test.go +221 −0
  • internal/dbdoor/restore.go +22 −2
diff --git a/internal/dbdoor/plan.go b/internal/dbdoor/plan.go
new file mode 100644
index 000000000..c5632bc31
--- /dev/null
+++ b/internal/dbdoor/plan.go
@@ -0,0 +1,82 @@
+package dbdoor
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
+)
+
+// RestorePlan is what a restore will read and produce, known before
+// anything is downloaded (seeded-instances spec §4 step 5, codex R2-19).
+type RestorePlan struct {
+ // TXID is the last planned file's MaxTXID: where the restore ends. For
+ // a target it is the target itself; for the latest copy it is the
+ // latest transaction when planned. RestoreToTarget at TXID reproduces
+ // exactly this plan's RestoredBytes, however far the source has moved
+ // on since, as long as that history is still kept.
+ TXID uint64
+ LTXBytes int64 // the planned files' compressed sizes, summed
+ RestoredBytes int64 // the restored database's size: Commit pages times PageSize
+ Files int
+ // Target is what was planned for: the requested target, or 0 (latest)
+ // when the request is at or past the latest transaction, which
+ // litestream would otherwise refuse as not available.
+ Target uint64
+}
+
+// PlanRestore plans a restore to target (0: latest) through c and reads the
+// header of the last planned file with a ranged open of its first
+// ltx.HeaderSize bytes. A target older than the retained history, or one
+// no retained chain of files ends at, is ErrHistoryGone; a replica with
+// nothing in it is ErrNothingReplicated, with or without a target.
+func PlanRestore(ctx context.Context, c *Client, target uint64) (RestorePlan, error) {
+ if target != 0 {
+ latest, err := PlanRestore(ctx, c, 0)
+ if err != nil {
+ return RestorePlan{}, err
+ }
+ if target >= latest.TXID {
+ return latest, nil // at or past the end: the latest copy
+ }
+ }
+ infos, err := litestream.CalcRestorePlan(ctx, c, ltx.TXID(target), time.Time{}, slog.Default())
+ switch {
+ case errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots):
+ if target != 0 {
+ return RestorePlan{}, ErrHistoryGone
+ }
+ return RestorePlan{}, ErrNothingReplicated
+ case err != nil:
+ return RestorePlan{}, err
+ case len(infos) == 0:
+ return RestorePlan{}, ErrNothingReplicated
+ }
+ p := RestorePlan{Target: target}
+ for _, f := range infos {
+ p.LTXBytes += f.Size
+ }
+ p.Files = len(infos)
+ last := infos[len(infos)-1]
+ p.TXID = uint64(last.MaxTXID)
+ rc, err := c.OpenLTXFile(ctx, last.Level, last.MinTXID, last.MaxTXID, 0, ltx.HeaderSize)
+ if err != nil {
+ return RestorePlan{}, fmt.Errorf("dbdoor: read the last planned file's header: %w", err)
+ }
+ defer rc.Close()
+ buf := make([]byte, ltx.HeaderSize)
+ if _, err := io.ReadFull(rc, buf); err != nil {
+ return RestorePlan{}, fmt.Errorf("dbdoor: read the last planned file's header: %w", err)
+ }
+ var h ltx.Header
+ if err := h.UnmarshalBinary(buf); err != nil {
+ return RestorePlan{}, fmt.Errorf("dbdoor: the last planned file's header: %w", err)
+ }
+ p.RestoredBytes = int64(h.Commit) * int64(h.PageSize)
+ return p, nil
+}
diff --git a/internal/dbdoor/plan_test.go b/internal/dbdoor/plan_test.go
new file mode 100644
index 000000000..6677fd6c4
--- /dev/null
+++ b/internal/dbdoor/plan_test.go
@@ -0,0 +1,221 @@
+package dbdoor
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "strings"
+ "testing"
+
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+)
+
+func TestPlanRestoreSizesTheRestoreExactly(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 4)
+ plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(r.Dir)
+ st, err := os.Stat(r.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if plan.RestoredBytes != st.Size() || plan.TXID != r.TXID || plan.LTXBytes <= 0 || plan.Files == 0 || plan.Target != 0 {
+ t.Fatalf("plan %+v, restored %d bytes at %d; want the plan's size and txid", plan, st.Size(), r.TXID)
+ }
+}
+
+func TestRestoreToATargetTXID(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 4)
+ files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
+ at := uint64(files[0].MaxTXID)
+ r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(r.Dir)
+ if r.TXID != at {
+ t.Fatalf("restored to %d, want the first file's end %d", r.TXID, at)
+ }
+ if rows := countRows(t, r.Path); rows != dbdoortest.Rows {
+ t.Fatalf("rows = %d, want one batch's %d", rows, dbdoortest.Rows)
+ }
+}
+
+// A target at or past the latest transaction plans the latest copy, and
+// says so with Target 0: litestream itself refuses a target past the end.
+func TestPlanRestoreTargetPastTheEndPlansTheLatest(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 3)
+ latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, target := range []uint64{latest.TXID, latest.TXID + 100} {
+ got, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), target)
+ if err != nil {
+ t.Fatalf("target %d: %v", target, err)
+ }
+ if got != latest {
+ t.Fatalf("target %d planned %+v, want the latest %+v", target, got, latest)
+ }
+ }
+}
+
+// A planned TXID pins the restore: after the source moves on, a restore at
+// plan.TXID still produces exactly the planned size, transaction and rows.
+// Task 5 restores at plan.TXID for that reason (preflight ruling M2).
+func TestRestorePinnedAtAPlannedTXIDKeepsThePlannedSize(t *testing.T) {
+ dir := t.TempDir()
+ db := dbdoortest.Replicate(t, dir, 2)
+ plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // The source grows by many pages and replicates again.
+ grow(t, db)
+ dbdoortest.Sync(t, db, dir)
+ later, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if later.TXID <= plan.TXID || later.RestoredBytes <= plan.RestoredBytes {
+ t.Fatalf("the source did not move on: planned %+v, later %+v", plan, later)
+ }
+
+ r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), plan.TXID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(r.Dir)
+ st, err := os.Stat(r.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Size() != plan.RestoredBytes || r.TXID != plan.TXID {
+ t.Fatalf("pinned restore: %d bytes at %d, want the plan's %d bytes at %d", st.Size(), r.TXID, plan.RestoredBytes, plan.TXID)
+ }
+ if rows := countRows(t, r.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("rows = %d, want the planned %d", rows, 2*dbdoortest.Rows)
+ }
+}
+
+// Retention keeps a snapshot and the L0 files after it. A target inside
+// the snapshot's span is history that is no longer kept, while the latest
+// copy (and a target at the snapshot's own end) still restores.
+func TestPlanRestoreTargetBeforeHistory(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 4)
+ files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0)
+ if len(files) < 4 {
+ t.Fatalf("want one L0 file per batch, got %d", len(files))
+ }
+ snap := snapshotThrough(t, src, files[:2])
+ for _, f := range files[:2] {
+ if err := os.Remove(litestream.LTXFilePath(src, 0, f.MinTXID, f.MaxTXID)); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0)
+ if err != nil {
+ t.Fatalf("the latest plan over a snapshot and later L0 files: %v", err)
+ }
+ if latest.TXID != uint64(files[len(files)-1].MaxTXID) || latest.Files != 1+len(files)-2 {
+ t.Fatalf("latest plan %+v, want the snapshot plus %d L0 files ending at %d", latest, len(files)-2, files[len(files)-1].MaxTXID)
+ }
+ r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), latest.TXID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(r.Dir)
+ if rows := countRows(t, r.Path); rows != 4*dbdoortest.Rows {
+ t.Fatalf("rows = %d, want every batch's %d", rows, 4*dbdoortest.Rows)
+ }
+
+ atSnap, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(snap.MaxTXID))
+ if err != nil || atSnap.TXID != uint64(snap.MaxTXID) || atSnap.Files != 1 || atSnap.Target != uint64(snap.MaxTXID) {
+ t.Fatalf("a target at the snapshot's end: plan %+v, err %v; want the snapshot alone", atSnap, err)
+ }
+
+ _, err = PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(files[0].MaxTXID))
+ if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("a target before the retained history: err = %v, want ErrHistoryGone", err)
+ }
+ _, err = RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), uint64(files[0].MaxTXID))
+ if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("a restore before the retained history: err = %v, want ErrHistoryGone", err)
+ }
+
+ if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 0); !errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("an empty replica: err = %v, want ErrNothingReplicated", err)
+ }
+ if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 5); !errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("an empty replica with a target: err = %v, want ErrNothingReplicated", err)
+ }
+}
+
+// snapshotThrough compacts the given L0 files, which must start at TXID 1,
+// into one snapshot-level file in the file replica at dir, the way
+// litestream's own compactor writes a level: ltx.Compactor over the source
+// files, written through the file client.
+func snapshotThrough(t *testing.T, dir string, files []*ltx.FileInfo) *ltx.FileInfo {
+ t.Helper()
+ ctx := context.Background()
+ rc := lsfile.NewReplicaClient(dir)
+ var rdrs []io.Reader
+ for _, f := range files {
+ r, err := rc.OpenLTXFile(ctx, 0, f.MinTXID, f.MaxTXID, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ rdrs = append(rdrs, r)
+ }
+ pr, pw := io.Pipe()
+ go func() {
+ c, err := ltx.NewCompactor(pw, rdrs)
+ if err != nil {
+ pw.CloseWithError(err)
+ return
+ }
+ c.HeaderFlags = ltx.HeaderFlagNoChecksum
+ pw.CloseWithError(c.Compact(ctx))
+ }()
+ info, err := rc.WriteLTXFile(ctx, litestream.SnapshotLevel, files[0].MinTXID, files[len(files)-1].MaxTXID, pr)
+ pr.CloseWithError(err)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return info
+}
+
+// grow adds enough rows to the source database at path to add many pages.
+func grow(t *testing.T, path string) {
+ t.Helper()
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ for i := 0; i < 50; i++ {
+ if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("grow-%d-%s", i, strings.Repeat("x", 2000))); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index 35a3ac947..caf2f7493 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -26,6 +26,9 @@ var (
// ErrCheckFailed: the copy restored, but lease.CheckDB refused it. It
// is never handed over.
ErrCheckFailed = errors.New("dbdoor: the restored copy failed its check")
+ // ErrHistoryGone: a restore target older than the replica's retained
+ // history (seeded-instances spec §4 step 4).
+ ErrHistoryGone = errors.New("dbdoor: history before the target is no longer kept")
)
// Restored is a private copy of a replica on this machine.
@@ -62,6 +65,19 @@ var checkDB = lease.CheckDBContext
// recovery. On any failure the directory is removed, so no copy of
// production data is left behind.
func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error) {
+ return RestoreToTarget(ctx, c, parent, 0)
+}
+
+// RestoreToTarget is RestoreTo at a target transaction: 0 is the latest.
+//
+// A nonzero target restores exactly to target or not at all. litestream
+// v0.5.14's CalcRestorePlan skips every file whose MaxTXID is past target
+// and refuses unless the chain it builds ends exactly at target, so the
+// copy never reaches past it and Restored.TXID is target. A target no
+// retained chain ends at (older than the retained history, inside a
+// compacted file, or past the latest transaction) is refused as
+// ErrHistoryGone. PlanRestore says which target to ask for.
+func RestoreToTarget(ctx context.Context, c *Client, parent string, target uint64) (*Restored, error) {
dir, err := os.MkdirTemp(parent, "carlos-db-")
if err != nil {
return nil, err
@@ -76,7 +92,7 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
var rec *recorder
for attempt := 0; ; attempt++ {
rec = &recorder{Client: c, dir: dir}
- err = restoreOnce(ctx, rec, out)
+ err = restoreOnce(ctx, rec, out, target)
if err == nil {
break
}
@@ -103,6 +119,9 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
}
switch {
case errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable):
+ if target != 0 {
+ return nil, ErrHistoryGone
+ }
return nil, ErrNothingReplicated
case vanished && !errors.Is(err, os.ErrNotExist):
return nil, fmt.Errorf("%w: %w", os.ErrNotExist, err)
@@ -129,9 +148,10 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
return &Restored{Path: out, Dir: dir, TXID: txid, At: at}, nil
}
-func restoreOnce(ctx context.Context, client litestream.ReplicaClient, out string) error {
+func restoreOnce(ctx context.Context, client litestream.ReplicaClient, out string, target uint64) error {
opt := litestream.NewRestoreOptions()
opt.OutputPath = out
+ opt.TXID = ltx.TXID(target)
// Follow stays false (spec §12 condition ii); IntegrityCheck is left at
// litestream's default and CheckDB runs after (condition iii).
return litestream.NewReplicaWithClient(nil, client).Restore(ctx, opt)