amadan

carlos / platform Public

dbdoor: restore into a 0700 directory, retrying the whole plan once

RestoreTo restores through the read-only client into a fresh 0700
directory, never sets Follow, and runs lease.CheckDB on the result
(spec 2026-09-02 §4, §12). A planned file that vanishes under the
restore gets one fresh Restore, since litestream's resumable reader
treats that as fatal. Nothing replicated yet is its own error. On any
failure the directory is removed, so no copy is left behind.

The result names the newest replicated transaction and its time, taken
from that file's LTX header rather than its upload time, for the
handover line. A console test now restores a real replica end to end
through the real door with this client.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4eb798c3072edf4ece37db03885c8f5ba5847596 parent bd9b6d2
3 files changed, +372 −0
  • internal/console/apidb_test.go +34 −0
  • internal/dbdoor/restore.go +183 −0
  • internal/dbdoor/restore_test.go +155 −0
diff --git a/internal/console/apidb_test.go b/internal/console/apidb_test.go
index 12e3d9625..7131e4704 100644
--- a/internal/console/apidb_test.go
+++ b/internal/console/apidb_test.go
@@ -2,6 +2,7 @@ package console
import (
"context"
+ "database/sql"
"encoding/json"
"errors"
"fmt"
@@ -545,3 +546,36 @@ func TestAPIDBHasItsOwnBudget(t *testing.T) {
}
}
}
+
+// End to end: the CLI's read-only client restores a real replica through
+// the real door, with a member's token, and gets every row.
+func TestAPIDBRestoresThroughTheDoor(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 3)
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ return nil, err
+ }
+ req.Header = h.Clone()
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ return http.DefaultClient.Do(req)
+ }
+ got, err := dbdoor.RestoreTo(context.Background(), dbdoor.NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var n int
+ if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ if n != 3*dbdoortest.Rows || got.TXID == 0 || got.At.IsZero() {
+ t.Fatalf("restored %d rows, txid %d at %v; want %d rows and an as-of", n, got.TXID, got.At, 3*dbdoortest.Rows)
+ }
+}
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
new file mode 100644
index 000000000..f96f79b57
--- /dev/null
+++ b/internal/dbdoor/restore.go
@@ -0,0 +1,183 @@
+package dbdoor
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "sync"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+var (
+ // ErrNothingReplicated: the replica holds no restorable state yet.
+ // litestream says so two ways (ErrNoSnapshots, ErrTxNotAvailable), the
+ // pair lease.Restore already treats as "nothing to restore".
+ ErrNothingReplicated = errors.New("dbdoor: nothing has been replicated yet")
+ // ErrCheckFailed: the copy restored, but lease.CheckDB refused it. It
+ // is never handed over.
+ ErrCheckFailed = errors.New("dbdoor: the restored copy failed its check")
+)
+
+// Restored is a private copy of a replica on this machine.
+type Restored struct {
+ // Path is the restored database. It lives inside Dir.
+ Path string
+ // Dir is the private 0700 directory holding Path and any -wal/-shm a
+ // read-only open makes. The caller removes it.
+ Dir string
+ // TXID is the newest replicated transaction in the copy, and At is
+ // that transaction's time from its LTX header: the time of the write,
+ // not of a later compaction (spec §8.1).
+ TXID uint64
+ At time.Time
+}
+
+// restoredName is the copy's file name inside its private directory.
+const restoredName = "replica.db"
+
+// checkDB is lease.CheckDB, the one check a disaster recovery leans on. A
+// variable only so a test can make it refuse.
+var checkDB = lease.CheckDB
+
+// RestoreTo restores the replica c reads into a new 0700 directory under
+// parent (os.TempDir() when empty) and proves it with lease.CheckDB.
+//
+// The three conditions spec §12 rests on hold here: c is not a
+// ReplicaClientV3, RestoreOptions.Follow is never set, and CheckDB runs on
+// the result rather than relying on litestream's own IntegrityCheck.
+//
+// A planned file can vanish under the restore when the writer's L0
+// retention fires, and litestream's ResumableReader treats that as fatal.
+// So the retry wraps the WHOLE Restore, once: a fresh plan is the only
+// 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) {
+ dir, err := os.MkdirTemp(parent, "carlos-db-")
+ if err != nil {
+ return nil, err
+ }
+ // MkdirTemp already creates 0700; say it anyway, because the privacy of
+ // everything restored rests on this one mode.
+ if err := os.Chmod(dir, 0o700); err != nil {
+ os.RemoveAll(dir)
+ return nil, err
+ }
+ out := filepath.Join(dir, restoredName)
+ var rec *recorder
+ for attempt := 0; ; attempt++ {
+ rec = &recorder{Client: c}
+ err = restoreOnce(ctx, rec, out)
+ if err == nil {
+ break
+ }
+ removeCopy(out)
+ vanished := errors.Is(err, os.ErrNotExist) || rec.sawVanish()
+ if attempt == 0 && vanished && ctx.Err() == nil {
+ continue
+ }
+ os.RemoveAll(dir)
+ switch {
+ case errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable):
+ return nil, ErrNothingReplicated
+ case vanished && !errors.Is(err, os.ErrNotExist):
+ return nil, fmt.Errorf("%w: %w", os.ErrNotExist, err)
+ }
+ return nil, err
+ }
+ if err := checkDB(out); err != nil {
+ os.RemoveAll(dir)
+ return nil, fmt.Errorf("%w: %v", ErrCheckFailed, err)
+ }
+ txid, at := rec.newest()
+ return &Restored{Path: out, Dir: dir, TXID: txid, At: at}, nil
+}
+
+func restoreOnce(ctx context.Context, client litestream.ReplicaClient, out string) error {
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = out
+ // 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)
+}
+
+func removeCopy(out string) {
+ for _, p := range []string{out, out + ".tmp", out + "-wal", out + "-shm"} {
+ _ = os.Remove(p)
+ }
+}
+
+// recorder sees every file a Restore opens, for two reasons: a file that
+// answered os.ErrNotExist is what makes the restore worth a second plan,
+// whatever litestream wrapped it in on the way out; and the header of the
+// newest file opened is the copy's "as of" (R1).
+type recorder struct {
+ *Client
+ mu sync.Mutex
+ vanished bool
+ maxTXID ltx.TXID
+ header *bytes.Buffer // first ltx.HeaderSize bytes of the maxTXID file
+}
+
+func (r *recorder) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ rc, err := r.Client.OpenLTXFile(ctx, level, minTXID, maxTXID, offset, size)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ r.mu.Lock()
+ r.vanished = true
+ r.mu.Unlock()
+ }
+ return nil, err
+ }
+ if offset != 0 {
+ return rc, nil
+ }
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if maxTXID < r.maxTXID {
+ return rc, nil
+ }
+ r.maxTXID = maxTXID
+ r.header = &bytes.Buffer{}
+ return &headerTee{ReadCloser: rc, buf: r.header}, nil
+}
+
+func (r *recorder) sawVanish() bool {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.vanished
+}
+
+// newest returns the copy's newest transaction and the time its LTX header
+// records. A header that cannot be read leaves At zero rather than guessed.
+func (r *recorder) newest() (uint64, time.Time) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ var h ltx.Header
+ if r.header == nil || r.header.Len() < ltx.HeaderSize || h.UnmarshalBinary(r.header.Bytes()[:ltx.HeaderSize]) != nil {
+ return uint64(r.maxTXID), time.Time{}
+ }
+ return uint64(r.maxTXID), time.UnixMilli(h.Timestamp).UTC()
+}
+
+// headerTee copies the first ltx.HeaderSize bytes read through it.
+type headerTee struct {
+ io.ReadCloser
+ buf *bytes.Buffer
+}
+
+func (t *headerTee) Read(p []byte) (int, error) {
+ n, err := t.ReadCloser.Read(p)
+ if need := ltx.HeaderSize - t.buf.Len(); need > 0 && n > 0 {
+ t.buf.Write(p[:min(n, need)])
+ }
+ return n, err
+}
diff --git a/internal/dbdoor/restore_test.go b/internal/dbdoor/restore_test.go
new file mode 100644
index 000000000..3d4dbeba5
--- /dev/null
+++ b/internal/dbdoor/restore_test.go
@@ -0,0 +1,155 @@
+package dbdoor
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+)
+
+// litestream creates <out>.tmp itself with os.Create, at the process umask,
+// so a 0600 file cannot be pre-created for it. The containing directory is
+// the only thing that can be private, and it must also confine the -wal
+// and -shm a read-only open still needs.
+func TestRestoreToUsesAPrivateDirectory(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 2)
+ parent := t.TempDir()
+ got, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, src).do), parent)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if filepath.Dir(got.Path) != got.Dir || filepath.Dir(got.Dir) != parent {
+ t.Fatalf("path %s, dir %s: want the copy inside a fresh directory under %s", got.Path, got.Dir, parent)
+ }
+ fi, err := os.Stat(got.Dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if perm := fi.Mode().Perm(); perm != 0o700 {
+ t.Fatalf("restore directory is %04o, want 0700", perm)
+ }
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
+ }
+}
+
+// R1: the handover names the newest replicated transaction and its time,
+// and the time is the LTX header's, the time of the write itself.
+func TestRestoreToReportsTheNewestTransaction(t *testing.T) {
+ src := t.TempDir()
+ before := time.Now().Add(-time.Second)
+ dbdoortest.Replicate(t, src, 3)
+ door := newTestDoor(t, src)
+ c := NewClient(door.do)
+ files := collect(t, c, 0, 0)
+ want := files[len(files)-1].MaxTXID
+ got, err := RestoreTo(context.Background(), c, t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if got.TXID != uint64(want) {
+ t.Errorf("TXID = %d, want %d", got.TXID, want)
+ }
+ if got.At.Before(before) || got.At.After(time.Now().Add(time.Second)) {
+ t.Errorf("At = %v, want the time of the last write (around now)", got.At)
+ }
+}
+
+// A planned file can vanish under the restore when the writer's L0
+// retention fires. ResumableReader treats ErrNotExist as FATAL and does not
+// retry, so the retry has to wrap the whole Restore: a fresh plan is the
+// only recovery.
+func TestRestoreToRetriesTheWholeRestoreOnce(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 2)
+ door := newTestDoor(t, src)
+ first := collect(t, NewClient(door.do), 0, 0)[0]
+ door.gone[Span(uint64(first.MinTXID), uint64(first.MaxTXID))] = true
+ door.lists = map[int]int{}
+ got, err := RestoreTo(context.Background(), NewClient(door.do), t.TempDir())
+ if err != nil {
+ t.Fatalf("RestoreTo with a file vanishing once = %v, want success on a fresh plan", err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if door.lists[9] != 2 {
+ t.Fatalf("snapshot level listed %d times, want 2 (one plan per Restore)", door.lists[9])
+ }
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
+ }
+}
+
+// Once, not forever: a file that keeps vanishing is a failure to report,
+// and the private directory goes with it.
+func TestRestoreToRetriesOnlyOnce(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 2)
+ door := newTestDoor(t, src)
+ first := collect(t, NewClient(door.do), 0, 0)[0]
+ span := Span(uint64(first.MinTXID), uint64(first.MaxTXID))
+ c := NewClient(func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ door.mu.Lock()
+ door.gone[span] = true
+ door.mu.Unlock()
+ return door.do(ctx, tail, h)
+ })
+ parent := t.TempDir()
+ if _, err := RestoreTo(context.Background(), c, parent); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("RestoreTo = %v, want the vanished file reported", err)
+ }
+ assertEmptyDir(t, parent)
+}
+
+// "Nothing replicated yet" is its own answer, for both of the errors
+// litestream gives it (spec §10), and leaves nothing behind.
+func TestRestoreToNamesAnEmptyReplica(t *testing.T) {
+ parent := t.TempDir()
+ _, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), parent)
+ if !errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("RestoreTo of an empty replica = %v, want ErrNothingReplicated", err)
+ }
+ assertEmptyDir(t, parent)
+}
+
+// A copy that fails lease.CheckDB is never handed over (spec §10). A
+// replica litestream wrote always holds its own _litestream_seq table and
+// a corrupt one fails inside Restore, so the check's refusal is injected:
+// what is under test is that RestoreTo honours it.
+func TestRestoreToRefusesACopyThatFailsTheCheck(t *testing.T) {
+ src := t.TempDir()
+ dbdoortest.Replicate(t, src, 1)
+ var checked string
+ defer func(old func(string) error) { checkDB = old }(checkDB)
+ checkDB = func(path string) error {
+ checked = path
+ return errors.New("integrity_check: page 2 is never used")
+ }
+ parent := t.TempDir()
+ _, err := RestoreTo(context.Background(), NewClient(newTestDoor(t, src).do), parent)
+ if !errors.Is(err, ErrCheckFailed) {
+ t.Fatalf("RestoreTo of a copy failing its check = %v, want ErrCheckFailed", err)
+ }
+ if filepath.Base(checked) != restoredName {
+ t.Fatalf("checked %q, want the restored copy", checked)
+ }
+ assertEmptyDir(t, parent)
+}
+
+func assertEmptyDir(t *testing.T, dir string) {
+ t.Helper()
+ ents, err := os.ReadDir(dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ents) != 0 {
+ t.Fatalf("%s holds %d entries after a failed restore, want none (no copy of production left behind)", dir, len(ents))
+ }
+}