| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/ed25519" |
| 7 | "crypto/rand" |
| 8 | "database/sql" |
| 9 | "encoding/json" |
| 10 | "path/filepath" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | lsfile "github.com/benbjohnson/litestream/file" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/dbdoor" |
| 17 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 18 | "github.com/carlosframework/platform/internal/dbquery" |
| 19 | "github.com/carlosframework/platform/internal/dbreads" |
| 20 | "github.com/carlosframework/platform/internal/dbreads/dbreadstest" |
| 21 | "github.com/carlosframework/platform/internal/hostclaim" |
| 22 | "github.com/carlosframework/platform/internal/provision" |
| 23 | "github.com/carlosframework/platform/internal/release" |
| 24 | ) |
| 25 | |
| 26 | // The parity half of spec T6 that needs no console server (plan 1): the box |
| 27 | // path (dbreads.Agent through the real worker code, joined in process: not |
| 28 | // acceptance for the units) and the CLI's in-region path (dbdoor.RestoreAsOf |
| 29 | // over this package's own door code, replicaDoer and listReplicaLevel, then |
| 30 | // dbquery.RunContext) print the same bytes, as of the same transaction: the |
| 31 | // newest copy, a time between writes, and a time in the same second as a |
| 32 | // write. Plan 2's T6 runs the same comparison through the console server and |
| 33 | // the CLI binary. |
| 34 | func TestDBReadsParityWithTheInRegionDoor(t *testing.T) { |
| 35 | const host = "shop.example.com" |
| 36 | dir := t.TempDir() |
| 37 | src := filepath.Join(t.TempDir(), "shop.db") |
| 38 | db, err := sql.Open("sqlite", src) |
| 39 | if err != nil { |
| 40 | t.Fatal(err) |
| 41 | } |
| 42 | defer db.Close() |
| 43 | exec := func(q string, args ...any) { |
| 44 | t.Helper() |
| 45 | if _, err := db.Exec(q, args...); err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | } |
| 49 | nextSecond := func() time.Time { |
| 50 | time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| 51 | return time.Now() |
| 52 | } |
| 53 | exec(`PRAGMA journal_mode=WAL; CREATE TABLE orders (id INTEGER PRIMARY KEY, note TEXT, data BLOB, at DATETIME)`) |
| 54 | exec(`INSERT INTO orders (note, data, at) VALUES (?, ?, ?), (NULL, NULL, NULL), (?, ?, ?)`, |
| 55 | "a\ttab", []byte{0, 0xff}, time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC), |
| 56 | "two\nlines", []byte("x"), time.Date(2026, 10, 7, 14, 0, 1, 500, time.UTC)) |
| 57 | dbdoortest.Sync(t, src, dir) |
| 58 | between := nextSecond() |
| 59 | nextSecond() |
| 60 | exec(`INSERT INTO orders (note) VALUES ('later')`) |
| 61 | dbdoortest.Sync(t, src, dir) |
| 62 | sameSecond := time.Now() // in the second the last file was written |
| 63 | nextSecond() |
| 64 | |
| 65 | store := release.NewDirStore(t.TempDir()) |
| 66 | pub, priv, _ := ed25519.GenerateKey(rand.Reader) |
| 67 | yes := true |
| 68 | rec := provision.Record{Host: host, App: "shop", Account: "bab", Region: "uk", Hibernate: &yes, CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)} |
| 69 | body, _ := json.Marshal(rec) |
| 70 | if _, err := store.Put(context.Background(), provision.RecordKey("bab", "shop", host), body, ""); err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | inc, _ := hostclaim.Incarnation("bab", "shop", host, rec.CreatedAt) |
| 74 | box := dbreadstest.Start(t, dbreadstest.Options{Store: store, Region: "uk", Box: "uk-1", Pub: pub, HibernatePrefix: "instances", |
| 75 | OpenReplica: func(context.Context, string) (dbdoor.ReplicaReader, error) { return lsfile.NewReplicaClient(dir), nil }}) |
| 76 | const query = "SELECT id, note, data, at FROM orders ORDER BY id" |
| 77 | for name, at := range map[string]time.Time{"newest": {}, "between writes": between, "same second as a write": sameSecond} { |
| 78 | got, err := dbdoor.RestoreAsOf(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(dir))), t.TempDir(), at) |
| 79 | if err != nil { |
| 80 | t.Fatalf("%s: in-region: %v", name, err) |
| 81 | } |
| 82 | var want bytes.Buffer |
| 83 | if err := dbquery.RunContext(context.Background(), &want, got.Path, query, dbquery.Options{}); err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | reply, _ := dbreads.NewReplyKey() |
| 87 | id, _ := dbreads.MintID() |
| 88 | atRaw := "" |
| 89 | if !at.IsZero() { |
| 90 | atRaw = at.UTC().Format(time.RFC3339Nano) |
| 91 | } |
| 92 | req := dbreads.Request{ID: id, Account: "bab", App: "shop", Host: host, Incarnation: inc, Region: "uk", |
| 93 | Op: dbreads.OpQuery, SQL: query, At: atRaw, ReplyKey: dbreads.EncodeReplyKey(reply.PublicKey()), |
| 94 | By: "member@example.com", IssuedAt: time.Now().UTC().Format(time.RFC3339)} |
| 95 | dbreadstest.Submit(t, store, priv, req) |
| 96 | box.Serve(t, store, id, 30*time.Second) |
| 97 | rows, s, r := dbreadstest.Answer(t, store, id, reply, dbreads.StreamRows) |
| 98 | if r.State != dbreads.StateDone || !bytes.Equal(rows, want.Bytes()) || s.TXID != got.TXID || !s.At.Equal(got.At) { |
| 99 | t.Fatalf("%s: box %+v txid %d at %v rows %q; in-region txid %d at %v rows %q", |
| 100 | name, r, s.TXID, s.At, rows, got.TXID, got.At, want.Bytes()) |
| 101 | } |
| 102 | } |
| 103 | } |
| 104 | |
amadan