amadan

carlos / platform Public

dbdoor: RestoreAsOf restores a replica as of a point in time (#262)

Sets litestream's Timestamp option. The door lists S3 LastModified, which
is never earlier than a file's writes, so a copy is never newer than the
requested time. Refuses a listing with no times (claim mode) rather than
letting litestream read zero as 'before T' and hand back the newest copy.
A time before the kept history is ErrBeforeHistory naming the oldest point.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Human

Prompt

Yes, go ahead.

From this commit's own message. The prompt ledger has no entries for it — which is what a squashed, rebased or amended commit looks like, and what every commit looks like in a repo without a ledger.

Summary

Adds dbdoor.RestoreAsOf; RestoreTo is RestoreAsOf with a zero time.
Agent notes
Mutation-checked: dropping the no-times guard restores the newest copy silently; dropping opt.Timestamp restores all rows.
claude 1f8d9b18ef1152eaefb4b6286510533a031b2077 parent 4a51a6d
2 files changed, +239 −3
  • internal/dbdoor/restore.go +89 −3
  • internal/dbdoor/restore_asof_test.go +150 −0

CI passed — run details

diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go
index 35a3ac947..8eee6edb2 100644
--- a/internal/dbdoor/restore.go
+++ b/internal/dbdoor/restore.go
@@ -26,8 +26,29 @@ 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")
+ // ErrBeforeHistory: an as-of restore asked for a time earlier than
+ // anything the replica keeps. RestoreAsOf returns it as a
+ // *BeforeHistoryError naming the oldest point there is.
+ ErrBeforeHistory = errors.New("dbdoor: the time is earlier than the history the replica keeps")
+ // ErrNoTimes: an as-of restore met a listing with no file times (the
+ // claim-mode door sends none). litestream reads a zero time as "before
+ // T" and would restore the newest copy, so this refuses instead.
+ ErrNoTimes = errors.New("dbdoor: the replica's listing carries no times, so it cannot be read as of a time")
)
+// BeforeHistoryError is ErrBeforeHistory with the oldest point the replica
+// keeps: the earliest time a file it could start a restore from was
+// written. Oldest is zero only when no listing named one.
+type BeforeHistoryError struct {
+ Oldest time.Time
+}
+
+func (e *BeforeHistoryError) Error() string {
+ return fmt.Sprintf("%v (the oldest point kept is %s)", ErrBeforeHistory, e.Oldest.UTC().Format(time.RFC3339))
+}
+
+func (e *BeforeHistoryError) Is(target error) bool { return target == ErrBeforeHistory }
+
// Restored is a private copy of a replica on this machine.
type Restored struct {
// Path is the restored database. It lives inside Dir.
@@ -62,6 +83,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 RestoreAsOf(ctx, c, parent, time.Time{})
+}
+
+// RestoreAsOf is RestoreTo stopped at a point in time: the copy holds every
+// file replicated before at and nothing after it. A zero at is the newest
+// copy, which is RestoreTo.
+//
+// The plan is litestream's own (Timestamp set, TXID not). It keeps a file
+// only if its listed time is before at, and the door lists S3's
+// LastModified, the upload time, which is never earlier than the writes a
+// file holds. So the copy can be older than at, never newer. Restored.At
+// says what it actually holds.
+func RestoreAsOf(ctx context.Context, c *Client, parent string, at time.Time) (*Restored, error) {
dir, err := os.MkdirTemp(parent, "carlos-db-")
if err != nil {
return nil, err
@@ -75,8 +109,8 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
out := filepath.Join(dir, restoredName)
var rec *recorder
for attempt := 0; ; attempt++ {
- rec = &recorder{Client: c, dir: dir}
- err = restoreOnce(ctx, rec, out)
+ rec = &recorder{Client: c, dir: dir, at: at}
+ err = restoreOnce(ctx, rec, out, at)
if err == nil {
break
}
@@ -102,7 +136,15 @@ func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error)
return nil, cerr
}
switch {
+ case errors.Is(err, ErrNoTimes):
+ return nil, ErrNoTimes
case errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable):
+ if oldest := rec.oldestStart(); !at.IsZero() && !oldest.IsZero() {
+ if !oldest.Before(at) {
+ return nil, &BeforeHistoryError{Oldest: oldest}
+ }
+ return nil, err
+ }
return nil, ErrNothingReplicated
case vanished && !errors.Is(err, os.ErrNotExist):
return nil, fmt.Errorf("%w: %w", os.ErrNotExist, err)
@@ -129,9 +171,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, at time.Time) error {
opt := litestream.NewRestoreOptions()
opt.OutputPath = out
+ opt.Timestamp = at
// 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)
@@ -153,10 +196,53 @@ type recorder struct {
mu sync.Mutex
vanished bool
checkFailed bool
+ at time.Time // the as-of time; zero for the newest copy
+ oldest time.Time // earliest time of a file a restore can start from
maxTXID ltx.TXID
header *bytes.Buffer // first ltx.HeaderSize bytes of the maxTXID file
}
+// LTXFiles passes the listing through, and for an as-of restore checks it
+// first: every file must carry a time, and the earliest file a restore can
+// start from (any snapshot, or a file beginning at the first transaction)
+// is remembered so a time before it can be named.
+func (r *recorder) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
+ itr, err := r.Client.LTXFiles(ctx, level, seek, useMetadata)
+ if err != nil || r.at.IsZero() {
+ return itr, err
+ }
+ var infos []*ltx.FileInfo
+ for itr.Next() {
+ infos = append(infos, itr.Item())
+ }
+ err = itr.Err()
+ if cerr := itr.Close(); err == nil {
+ err = cerr
+ }
+ if err != nil {
+ return nil, err
+ }
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ for _, f := range infos {
+ if f.CreatedAt.IsZero() {
+ return nil, ErrNoTimes
+ }
+ if level == litestream.SnapshotLevel || f.MinTXID == 1 {
+ if r.oldest.IsZero() || f.CreatedAt.Before(r.oldest) {
+ r.oldest = f.CreatedAt
+ }
+ }
+ }
+ return ltx.NewFileInfoSliceIterator(infos), nil
+}
+
+func (r *recorder) oldestStart() time.Time {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.oldest
+}
+
func (r *recorder) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
// The first open goes through download's loop too, so a full door's
// 429 waits out its Retry-After here rather than failing back to
diff --git a/internal/dbdoor/restore_asof_test.go b/internal/dbdoor/restore_asof_test.go
new file mode 100644
index 000000000..22bbd2f37
--- /dev/null
+++ b/internal/dbdoor/restore_asof_test.go
@@ -0,0 +1,150 @@
+package dbdoor
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+)
+
+// addBatch inserts one more batch into src and replicates it into dir.
+func addBatch(t *testing.T, src, dir string, batch int) {
+ t.Helper()
+ db, err := sql.Open("sqlite", src)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < dbdoortest.Rows; i++ {
+ if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d-%d", batch, i)); err != nil {
+ db.Close()
+ t.Fatal(err)
+ }
+ }
+ db.Close()
+ dbdoortest.Sync(t, src, dir)
+}
+
+// pause makes sure the next file's time is strictly after the last one's,
+// whatever the filesystem's timestamp resolution.
+func pause() time.Time {
+ time.Sleep(20 * time.Millisecond)
+ at := time.Now()
+ time.Sleep(20 * time.Millisecond)
+ return at
+}
+
+// A copy as of T holds what was replicated before T and nothing after it.
+func TestRestoreAsOfStopsAtTheTime(t *testing.T) {
+ dir := t.TempDir()
+ src := dbdoortest.Replicate(t, dir, 2)
+ at := pause()
+ addBatch(t, src, dir, 2)
+ c := NewClient(newTestDoor(t, dir).do)
+ latest := collect(t, c, 0, 0)
+
+ got, err := RestoreAsOf(context.Background(), c, t.TempDir(), at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("copy as of %v holds %d rows, want %d (the third batch came after)", at, rows, 2*dbdoortest.Rows)
+ }
+ if got.TXID >= uint64(latest[len(latest)-1].MaxTXID) {
+ t.Fatalf("TXID = %d, want one before the newest (%d)", got.TXID, latest[len(latest)-1].MaxTXID)
+ }
+ if !got.At.Before(at) {
+ t.Fatalf("At = %v, want a time before the requested %v", got.At, at)
+ }
+}
+
+// A time after the newest write is simply the newest copy.
+func TestRestoreAsOfAfterTheNewestWrite(t *testing.T) {
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 2)
+ got, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), pause())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows {
+ t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows)
+ }
+}
+
+// A time before anything the replica keeps is its own answer, naming the
+// oldest point it does keep, and not "nothing replicated".
+func TestRestoreAsOfBeforeTheHistory(t *testing.T) {
+ at := time.Now().Add(-time.Hour)
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 2)
+ parent := t.TempDir()
+ _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), parent, at)
+ if !errors.Is(err, ErrBeforeHistory) {
+ t.Fatalf("RestoreAsOf before the history = %v, want ErrBeforeHistory", err)
+ }
+ if errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("RestoreAsOf before the history = %v, must not read as nothing replicated", err)
+ }
+ var bh *BeforeHistoryError
+ if !errors.As(err, &bh) || bh.Oldest.IsZero() || !bh.Oldest.After(at) {
+ t.Fatalf("RestoreAsOf before the history = %#v, want the oldest kept point, after %v", err, at)
+ }
+ assertEmptyDir(t, parent)
+}
+
+// An empty replica is still "nothing replicated", with or without a time.
+func TestRestoreAsOfEmptyReplica(t *testing.T) {
+ parent := t.TempDir()
+ _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), parent, time.Now())
+ if !errors.Is(err, ErrNothingReplicated) {
+ t.Fatalf("RestoreAsOf of an empty replica = %v, want ErrNothingReplicated", err)
+ }
+ assertEmptyDir(t, parent)
+}
+
+// A listing with no times (the claim-mode door sends none) must refuse:
+// litestream reads a zero time as "before T" and would hand back the
+// newest copy labelled as the requested one.
+func TestRestoreAsOfRefusesAListingWithoutTimes(t *testing.T) {
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 2)
+ door := newTestDoor(t, dir)
+ untimed := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ res, err := door.do(ctx, tail, h)
+ if err != nil || res.StatusCode != http.StatusOK || h.Get("Range") != "" {
+ return res, err
+ }
+ var list ListResponse
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
+ return nil, err
+ }
+ for i := range list.Files {
+ list.Files[i].CreatedAt = time.Time{}
+ }
+ rec := httptest.NewRecorder()
+ json.NewEncoder(rec).Encode(list)
+ return rec.Result(), nil
+ }
+ parent := t.TempDir()
+ _, err := RestoreAsOf(context.Background(), NewClient(untimed), parent, time.Now())
+ if !errors.Is(err, ErrNoTimes) {
+ t.Fatalf("RestoreAsOf over an untimed listing = %v, want ErrNoTimes", err)
+ }
+ assertEmptyDir(t, parent)
+
+ // Without a time, the same listing restores as before.
+ got, err := RestoreTo(context.Background(), NewClient(untimed), t.TempDir())
+ if err != nil {
+ t.Fatalf("RestoreTo over an untimed listing = %v, want the newest copy", err)
+ }
+ os.RemoveAll(got.Dir)
+}