amadan
db-pull-at: carlos db pull/query --at reads a database as of a past time (#262)
Squashed from db-pull-at into main.
6 files changed,
+503
−10
db.go+70 −6db_at_test.go+75 −0db_e2e_test.go+35 −1docs/site/cli.md+8 −0internal/dbdoor/restore.go+107 −3internal/dbdoor/restore_asof_test.go+208 −0
CI cancelled — superseded by a newer push — run details
diff --git a/db.go b/db.go| index 8efc220a4..596afd821 100644 |
| --- a/db.go |
| +++ b/db.go |
| @@ -45,6 +45,11 @@ const dbUsage = `usage: carlos db <verb> --app <a> [--instance <host>] [flags] |
| query --sql <text> restore a copy and run one read-only query against it; |
| rows go to stdout, tab-separated, with a header row |
| +Add --at <time> to either verb to read the database as it was at that |
| +time, for example --at 2026-10-07T14:00:00Z. The copy holds every change |
| +replicated before then, so it can be a little older than the time you give. |
| +How far back you can go depends on how much history the platform keeps. |
| + |
| Both read the platform's replica of the database, never the live one, and |
| nothing you do to the copy reaches the instance. --instance names which of |
| the app's instances to read; it may be left out when the app has one. The |
| @@ -76,6 +81,9 @@ type dbTarget struct { |
| acct apiAcct |
| app string |
| host string |
| + // at is --at: the copy holds what was replicated before it. Zero is |
| + // the newest copy. |
| + at time.Time |
| } |
| // dbFlags registers the flags both verbs share and returns the resolver. |
| @@ -86,7 +94,14 @@ func dbFlags(fs *flag.FlagSet, verb string) func() dbTarget { |
| instance := fs.String("instance", "", "the instance whose database to read (may be left out when the app has one)") |
| acctRaw, acctIDRaw, _ := accountFlags(fs) |
| consoleFlag := consoleOption(fs, "console to read through (default: the only logged-in one)") |
| + atRaw := fs.String("at", "", "read the database as it was at this time (RFC 3339, for example 2026-10-07T14:00:00Z)") |
| return func() dbTarget { |
| + // Before any network: a time that can't be read is the person's |
| + // to fix, and costs nothing to refuse. |
| + at, err := parseDBAt(*atRaw, time.Now()) |
| + if err != nil { |
| + log.Fatalf("carlos db %s: %v", verb, err) |
| + } |
| sess := requireDBAPISession(*consoleFlag) |
| printUsing(os.Stderr, sess) |
| acct := resolvedAccount(sess, acctRaw, acctIDRaw) |
| @@ -95,10 +110,36 @@ func dbFlags(fs *flag.FlagSet, verb string) func() dbTarget { |
| if err != nil { |
| log.Fatalf("carlos db %s: %v", verb, err) |
| } |
| - return dbTarget{sess: sess, acct: acct, app: name, host: host} |
| + return dbTarget{sess: sess, acct: acct, app: name, host: host, at: at} |
| } |
| } |
| +// parseDBAt reads --at. Empty is the zero time, the newest copy. A time |
| +// must carry its zone: a bare clock time would be read in whichever zone |
| +// this shell is in, and the copy would silently be hours off. A time after |
| +// now is refused rather than quietly answered with the newest copy, so |
| +// nobody holds a copy labelled with a moment that hasn't happened. |
| +func parseDBAt(raw string, now time.Time) (time.Time, error) { |
| + raw = strings.TrimSpace(raw) |
| + if raw == "" { |
| + return time.Time{}, nil |
| + } |
| + at, err := time.Parse(time.RFC3339, raw) |
| + if err != nil { |
| + return time.Time{}, fmt.Errorf("--at %q is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC", raw) |
| + } |
| + if at.After(now) { |
| + return time.Time{}, fmt.Errorf("--at %s is in the future. Leave --at out to read the newest copy", raw) |
| + } |
| + // 0001-01-01T00:00:00Z parses to Go's zero time, which means "no |
| + // --at". It is before any history, so it is asked as the earliest |
| + // time instead, and gets the before-history answer. |
| + if at.Before(dbdoor.EarliestAsOf) { |
| + at = dbdoor.EarliestAsOf |
| + } |
| + return at, nil |
| +} |
| + |
| // requireDBAPISession resolves signed-in mode, refusing direct bucket mode |
| // by name: the door is the console's, and an operator holding the bucket |
| // already has a box. |
| @@ -146,7 +187,7 @@ func resolveDBInstance(ctx context.Context, sess *apiSession, acct apiAcct, app, |
| // dbRestore restores the target's replica into a private directory under |
| // the system temp dir. The caller removes Restored.Dir. |
| func dbRestore(ctx context.Context, t dbTarget) (*dbdoor.Restored, error) { |
| - r, err := dbdoor.RestoreTo(ctx, t.sess.dbClient(t.acct, t.app, t.host), "") |
| + r, err := dbdoor.RestoreAsOf(ctx, t.sess.dbClient(t.acct, t.app, t.host), "", t.at) |
| if err != nil { |
| return nil, dbRestoreError(t.host, err) |
| } |
| @@ -158,7 +199,12 @@ func dbRestore(ctx context.Context, t dbTarget) (*dbdoor.Restored, error) { |
| // named here so none of them reads as a transient "no replica found". |
| func dbRestoreError(host string, err error) error { |
| var door *dbdoor.DoorError |
| + var before *dbdoor.BeforeHistoryError |
| switch { |
| + case errors.As(err, &before): |
| + return fmt.Errorf("%s's replica does not go back that far. The oldest point it keeps is %s", host, dbTime(before.Oldest)) |
| + case errors.Is(err, dbdoor.ErrNoTimes): |
| + return fmt.Errorf("%s's replica can't be read as of a time yet. Leave --at out to read the newest copy", host) |
| case errors.As(err, &door) && door.Status == 404: |
| return fmt.Errorf("no instance %s that you can read", host) |
| case errors.As(err, &door): |
| @@ -181,7 +227,26 @@ func dbAsOf(r *dbdoor.Restored) string { |
| if r.At.IsZero() { |
| return fmt.Sprintf("transaction %d", r.TXID) |
| } |
| - return fmt.Sprintf("transaction %d, written %s", r.TXID, r.At.UTC().Format("2006-01-02 15:04:05 UTC")) |
| + return fmt.Sprintf("transaction %d, written %s", r.TXID, dbTime(r.At)) |
| +} |
| + |
| +func dbTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05 UTC") } |
| + |
| +// dbPullHolds is pull's line saying what the copy holds; with --at it also |
| +// names the requested time, since the copy can be older than it. |
| +func dbPullHolds(r *dbdoor.Restored, at time.Time) string { |
| + if at.IsZero() { |
| + return fmt.Sprintf("It holds every change up to %s.\n", dbAsOf(r)) |
| + } |
| + return fmt.Sprintf("It holds every change up to %s, the last one replicated before %s.\n", dbAsOf(r), dbTime(at)) |
| +} |
| + |
| +// dbQueryStatus is query's stderr line before the rows. |
| +func dbQueryStatus(host string, r *dbdoor.Restored, at time.Time) string { |
| + if at.IsZero() { |
| + return fmt.Sprintf("Querying a copy of %s's database, up to %s. Nothing you run here reaches %s.\n", host, dbAsOf(r), host) |
| + } |
| + return fmt.Sprintf("Querying a copy of %s's database as it was before %s, up to %s. Nothing you run here reaches %s.\n", host, dbTime(at), dbAsOf(r), host) |
| } |
| func runDBPull(args []string) { |
| @@ -219,7 +284,7 @@ func dbPull(ctx context.Context, t dbTarget, out string) error { |
| return err |
| } |
| fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out) |
| - fmt.Fprintf(stdout, "It holds every change up to %s.\n", dbAsOf(r)) |
| + fmt.Fprint(stdout, dbPullHolds(r, t.at)) |
| fmt.Fprintf(stdout, "It is a copy. Nothing you do to it reaches %s.\n", t.host) |
| return nil |
| } |
| @@ -312,8 +377,7 @@ func dbQuery(ctx context.Context, t dbTarget, sqlText string) error { |
| return err |
| } |
| defer os.RemoveAll(r.Dir) |
| - fmt.Fprintf(os.Stderr, "Querying a copy of %s's database, up to %s. Nothing you run here reaches %s.\n", |
| - t.host, dbAsOf(r), t.host) |
| + fmt.Fprint(os.Stderr, dbQueryStatus(t.host, r, t.at)) |
| return runQueryContext(ctx, stdout, r.Path, sqlText) |
| } |
diff --git a/db_at_test.go b/db_at_test.go| new file mode 100644 |
| index 000000000..78e64ee43 |
| --- /dev/null |
| +++ b/db_at_test.go |
| @@ -0,0 +1,75 @@ |
| +package main |
| + |
| +import ( |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/dbdoor" |
| +) |
| + |
| +func TestParseDBAt(t *testing.T) { |
| + now := time.Date(2026, 10, 8, 12, 0, 0, 0, time.UTC) |
| + for _, ok := range []string{"2026-10-07T14:00:00Z", "2026-10-07T15:00:00+01:00", "2026-10-07T14:00:00.5Z"} { |
| + got, err := parseDBAt(ok, now) |
| + if err != nil { |
| + t.Errorf("parseDBAt(%q) = %v, want a time", ok, err) |
| + } else if !got.Before(now) { |
| + t.Errorf("parseDBAt(%q) = %v, want before %v", ok, got, now) |
| + } |
| + } |
| + // An explicit time that is Go's zero time must not read as "no --at" |
| + // (found in Astra's review). |
| + for _, early := range []string{"0001-01-01T00:00:00Z", "0001-01-01T01:00:00+01:00"} { |
| + if got, err := parseDBAt(early, now); err != nil || got.IsZero() { |
| + t.Errorf("parseDBAt(%q) = %v, %v, want a non-zero time before any history", early, got, err) |
| + } |
| + } |
| + if got, _ := parseDBAt("", now); !got.IsZero() { |
| + t.Errorf("parseDBAt(\"\") = %v, want zero (the newest copy)", got) |
| + } |
| + bad := map[string]string{ |
| + "2026-10-07 14:00": `--at "2026-10-07 14:00" is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC`, |
| + "2026-10-07T14:00:00": `--at "2026-10-07T14:00:00" is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC`, |
| + "yesterday": `--at "yesterday" is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC`, |
| + "2026-10-09T00:00:00Z": `--at 2026-10-09T00:00:00Z is in the future. Leave --at out to read the newest copy`, |
| + } |
| + for in, want := range bad { |
| + if _, err := parseDBAt(in, now); err == nil || err.Error() != want { |
| + t.Errorf("parseDBAt(%q) = %v, want %q", in, err, want) |
| + } |
| + } |
| +} |
| + |
| +func TestDBRestoreErrorsForAnAsOfRead(t *testing.T) { |
| + oldest := time.Date(2026, 10, 1, 9, 12, 44, 0, time.UTC) |
| + cases := map[error]string{ |
| + &dbdoor.BeforeHistoryError{Oldest: oldest}: "a.go.tito.io's replica does not go back that far. The oldest point it keeps is 2026-10-01 09:12:44 UTC", |
| + dbdoor.ErrNoTimes: "a.go.tito.io's replica can't be read as of a time yet. Leave --at out to read the newest copy", |
| + } |
| + for err, want := range cases { |
| + if got := dbRestoreError("a.go.tito.io", err).Error(); got != want { |
| + t.Errorf("%v -> %q, want %q", err, got, want) |
| + } |
| + } |
| +} |
| + |
| +func TestDBAsOfLines(t *testing.T) { |
| + r := &dbdoor.Restored{TXID: 42, At: time.Date(2026, 10, 7, 13, 58, 2, 0, time.UTC)} |
| + at := time.Date(2026, 10, 7, 15, 0, 0, 0, time.FixedZone("", 3600)) |
| + if got, want := dbPullHolds(r, at), "It holds every change up to transaction 42, written 2026-10-07 13:58:02 UTC, the last one replicated before 2026-10-07 14:00:00 UTC.\n"; got != want { |
| + t.Errorf("pull with --at:\n got %q\nwant %q", got, want) |
| + } |
| + if got, want := dbPullHolds(r, time.Time{}), "It holds every change up to transaction 42, written 2026-10-07 13:58:02 UTC.\n"; got != want { |
| + t.Errorf("pull without --at:\n got %q\nwant %q", got, want) |
| + } |
| + if got, want := dbQueryStatus("a.go.tito.io", r, at), "Querying a copy of a.go.tito.io's database as it was before 2026-10-07 14:00:00 UTC, up to transaction 42, written 2026-10-07 13:58:02 UTC. Nothing you run here reaches a.go.tito.io.\n"; got != want { |
| + t.Errorf("query with --at:\n got %q\nwant %q", got, want) |
| + } |
| + if got, want := dbQueryStatus("a.go.tito.io", r, time.Time{}), "Querying a copy of a.go.tito.io's database, up to transaction 42, written 2026-10-07 13:58:02 UTC. Nothing you run here reaches a.go.tito.io.\n"; got != want { |
| + t.Errorf("query without --at:\n got %q\nwant %q", got, want) |
| + } |
| + if !strings.Contains(dbUsage, "Add --at <time> to either verb") { |
| + t.Error("dbUsage does not explain --at") |
| + } |
| +} |
diff --git a/db_e2e_test.go b/db_e2e_test.go| index 29fad228b..2c8d08b41 100644 |
| --- a/db_e2e_test.go |
| +++ b/db_e2e_test.go |
| @@ -9,8 +9,10 @@ package main |
| import ( |
| "bytes" |
| "context" |
| + "database/sql" |
| "encoding/json" |
| "encoding/xml" |
| + "fmt" |
| "io/fs" |
| "net/http" |
| "net/http/httptest" |
| @@ -123,7 +125,26 @@ func TestE2EDBThroughConsole(t *testing.T) { |
| tmp := t.TempDir() |
| const host = "acme.go.tito.io" |
| replicaRoot := filepath.Join(tmp, "data") |
| - dbdoortest.Replicate(t, filepath.Join(replicaRoot, "instances", host), 3) |
| + // Two batches, then a third in a later whole second, so --at can |
| + // name a time between them in the fake S3's second-resolution listing. |
| + replicaDir := filepath.Join(replicaRoot, "instances", host) |
| + src := dbdoortest.Replicate(t, replicaDir, 2) |
| + time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| + between := time.Now() |
| + time.Sleep(time.Until(between.Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| + { |
| + 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-2-%d", i)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + db.Close() |
| + dbdoortest.Sync(t, src, replicaDir) |
| + } |
| // lock.json sits under the same prefix in production. Nothing may fetch it. |
| if err := os.WriteFile(filepath.Join(replicaRoot, "instances", host, "lock.json"), []byte(`{"owner":"box"}`), 0o600); err != nil { |
| t.Fatal(err) |
| @@ -220,6 +241,19 @@ func TestE2EDBThroughConsole(t *testing.T) { |
| t.Fatalf("pulled copy = %q %v", rows.String(), err) |
| } |
| + // pull --at: the copy as of a time between the second and third |
| + // batches holds the first two, and says which time it was asked for. |
| + outAt := filepath.Join(t.TempDir(), "pulled-at.db") |
| + buf.Reset() |
| + runDB([]string{"pull", "--console", ts.URL, "--app", "titogo", "--account", "ops", "--instance", host, "--out", outAt, "--at", between.UTC().Format(time.RFC3339Nano)}) |
| + if want := "the last one replicated before " + dbTime(between); !strings.Contains(buf.String(), want) { |
| + t.Errorf("pull --at said %q, want it to mention %q", buf.String(), want) |
| + } |
| + rows.Reset() |
| + if err := runQuery(&rows, outAt, "SELECT count(*) AS n FROM t"); err != nil || rows.String() != "n\n"+strconv.Itoa(2*dbdoortest.Rows)+"\n" { |
| + t.Fatalf("copy as of %v = %q %v, want %d rows", between, rows.String(), err, 2*dbdoortest.Rows) |
| + } |
| + |
| // query: rows on stdout, nothing else. |
| buf.Reset() |
| runDB([]string{"query", "--console", ts.URL, "--app", "titogo", "--account", "ops", "--instance", host, "--sql", "SELECT body FROM t ORDER BY id LIMIT 2"}) |
diff --git a/docs/site/cli.md b/docs/site/cli.md| index 44dc7d488..489572e6f 100644 |
| --- a/docs/site/cli.md |
| +++ b/docs/site/cli.md |
| @@ -459,6 +459,14 @@ Both say how recent the copy is: the last transaction it holds, and when |
| that transaction was written. A sleeping instance's copy is as recent as its |
| last sync before it went to sleep. A busy instance can be ahead of its copy. |
| +To read the database as it was at an earlier time, add `--at` with a time |
| +in RFC 3339 form, such as `--at 2026-10-07T14:00:00Z`. The copy holds every |
| +change replicated before that time. It can be a little older than the time |
| +you give, never newer, and the command says which transaction it stopped |
| +at. How far back you can go depends on how much history the platform keeps |
| +for the instance. Ask for a time earlier than that and the command names |
| +the oldest time you can use instead. |
| + |
| The copy is read-only. A query that tries to change it fails with SQLite's |
| own error, `attempt to write a readonly database`. That is on purpose. |
| Changing a copy would change nothing on the instance, and a command that |
diff --git a/internal/dbdoor/restore.go b/internal/dbdoor/restore.go| index 35a3ac947..2c78ffb37 100644 |
| --- a/internal/dbdoor/restore.go |
| +++ b/internal/dbdoor/restore.go |
| @@ -26,8 +26,34 @@ 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") |
| ) |
| +// EarliestAsOf is the earliest time RestoreAsOf asks litestream for. Any |
| +// earlier request is moved to it: it is before every replica's history, |
| +// and unlike Go's zero time it cannot mean "the newest copy". |
| +var EarliestAsOf = time.Unix(0, 0).UTC() |
| + |
| +// BeforeHistoryError is ErrBeforeHistory with the oldest point the replica |
| +// keeps: the earliest time an as-of restore can ask for, the first whole |
| +// second after the oldest file a restore could start from was listed. |
| +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 +88,26 @@ 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. at is rounded down to its second first (restoreOnce says |
| +// why). 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) { |
| + // A requested time that rounds down to Go's zero time would read as |
| + // "the newest copy" to litestream. Anything that early is before every |
| + // replica's history, so it is asked as the Unix epoch, which still is. |
| + if !at.IsZero() && at.Truncate(time.Second).Before(EarliestAsOf) { |
| + at = EarliestAsOf |
| + } |
| dir, err := os.MkdirTemp(parent, "carlos-db-") |
| if err != nil { |
| return nil, err |
| @@ -75,8 +121,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 +148,17 @@ 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.Truncate(time.Second)) { |
| + // The first whole second after the oldest file: the |
| + // earliest --at that, rounded down, still includes it. |
| + return nil, &BeforeHistoryError{Oldest: oldest.Truncate(time.Second).Add(time.Second)} |
| + } |
| + return nil, err |
| + } |
| return nil, ErrNothingReplicated |
| case vanished && !errors.Is(err, os.ErrNotExist): |
| return nil, fmt.Errorf("%w: %w", os.ErrNotExist, err) |
| @@ -129,9 +185,14 @@ 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 |
| + // S3 lists LastModified to the whole second, rounded down: a file |
| + // uploaded at 14:00:00.7 lists as 14:00:00, before a T of 14:00:00.5, |
| + // though it holds writes after T. Rounding T down to its second keeps |
| + // it out: a listed time before a whole second means an upload before it. |
| + opt.Timestamp = at.Truncate(time.Second) |
| // 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 +214,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..3143a4d1f |
| --- /dev/null |
| +++ b/internal/dbdoor/restore_asof_test.go |
| @@ -0,0 +1,208 @@ |
| +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 returns a time in a whole second of its own: after every file |
| +// written so far, and before the next one. RestoreAsOf rounds its time down |
| +// to the second, as S3's listing does. |
| +func pause() time.Time { |
| + time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| + at := time.Now() |
| + time.Sleep(time.Until(at.Truncate(time.Second).Add(time.Second + 10*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) |
| + |
| + // The point it names is one a person can ask for. |
| + got, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), bh.Oldest) |
| + if err != nil { |
| + t.Fatalf("RestoreAsOf at the named oldest point %v = %v, want a copy", bh.Oldest, err) |
| + } |
| + os.RemoveAll(got.Dir) |
| +} |
| + |
| +// 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) |
| + untimed := relist(newTestDoor(t, dir), func(f *LTXFile) { f.CreatedAt = time.Time{} }) |
| + 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) |
| +} |
| + |
| +// relist serves door's listings with each file passed through edit. |
| +func relist(door *testDoor, edit func(*LTXFile)) Doer { |
| + return 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 { |
| + edit(&list.Files[i]) |
| + } |
| + rec := httptest.NewRecorder() |
| + json.NewEncoder(rec).Encode(list) |
| + return rec.Result(), nil |
| + } |
| +} |
| + |
| +// S3 lists LastModified to the whole second, rounded down, so a file |
| +// uploaded just after T in the same second lists as before T. The copy must |
| +// still leave it out. |
| +func TestRestoreAsOfWithSecondResolutionTimes(t *testing.T) { |
| + dir := t.TempDir() |
| + src := dbdoortest.Replicate(t, dir, 2) |
| + // Start just after a whole second, so batch three lands in the same |
| + // second as T but after it. |
| + time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 100*time.Millisecond))) |
| + at := time.Now() |
| + time.Sleep(50 * time.Millisecond) |
| + addBatch(t, src, dir, 2) |
| + if done := time.Now(); done.Truncate(time.Second) != at.Truncate(time.Second) { |
| + t.Skipf("batch three took until %v, past T's second; the case needs it inside", done) |
| + } |
| + c := NewClient(relist(newTestDoor(t, dir), func(f *LTXFile) { f.CreatedAt = f.CreatedAt.Truncate(time.Second) })) |
| + 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: a write after T is in it", at, rows, 2*dbdoortest.Rows) |
| + } |
| +} |
| + |
| +// Go's zero time means "the newest copy" to RestoreAsOf and to litestream. |
| +// A requested time at or near it (year 1, or a fraction of a second into |
| +// it that rounds down to zero) must still be a time before the history, |
| +// never the newest copy. Found in Astra's review. The zero instant itself, |
| +// in any zone, is "the newest copy" by contract; parseDBAt keeps a typed |
| +// one from reaching here. |
| +func TestRestoreAsOfNeverTreatsAnEarlyTimeAsTheNewest(t *testing.T) { |
| + dir := t.TempDir() |
| + dbdoortest.Replicate(t, dir, 2) |
| + for _, at := range []time.Time{ |
| + time.Time{}.Add(500 * time.Millisecond), |
| + time.Time{}.Add(time.Hour), |
| + } { |
| + _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), at) |
| + if !errors.Is(err, ErrBeforeHistory) { |
| + t.Errorf("RestoreAsOf(%v) = %v, want ErrBeforeHistory", at, err) |
| + } |
| + } |
| +} |