| 1 | package dbdoor |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "net/http/httptest" |
| 11 | "os" |
| 12 | "reflect" |
| 13 | "strings" |
| 14 | "testing" |
| 15 | |
| 16 | "github.com/benbjohnson/litestream" |
| 17 | lsfile "github.com/benbjohnson/litestream/file" |
| 18 | "github.com/superfly/ltx" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 21 | ) |
| 22 | |
| 23 | func TestPlanRestoreSizesTheRestoreExactly(t *testing.T) { |
| 24 | src := t.TempDir() |
| 25 | dbdoortest.Replicate(t, src, 4) |
| 26 | plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0) |
| 27 | if err != nil { |
| 28 | t.Fatal(err) |
| 29 | } |
| 30 | r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), 0) |
| 31 | if err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | defer os.RemoveAll(r.Dir) |
| 35 | st, err := os.Stat(r.Path) |
| 36 | if err != nil { |
| 37 | t.Fatal(err) |
| 38 | } |
| 39 | if plan.RestoredBytes != st.Size() || plan.TXID != r.TXID || plan.LTXBytes <= 0 || plan.Files == 0 || plan.Target != 0 { |
| 40 | t.Fatalf("plan %+v, restored %d bytes at %d; want the plan's size and txid", plan, st.Size(), r.TXID) |
| 41 | } |
| 42 | // With no snapshot the plan is every L0 file, so LTXBytes is the |
| 43 | // listing's sizes summed. |
| 44 | var listed int64 |
| 45 | files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0) |
| 46 | for _, f := range files { |
| 47 | listed += f.Size |
| 48 | } |
| 49 | if plan.Files != len(files) || plan.LTXBytes != listed { |
| 50 | t.Fatalf("plan %+v, want %d files of %d bytes", plan, len(files), listed) |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | func TestRestoreToATargetTXID(t *testing.T) { |
| 55 | src := t.TempDir() |
| 56 | dbdoortest.Replicate(t, src, 4) |
| 57 | files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0) |
| 58 | at := uint64(files[0].MaxTXID) |
| 59 | r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), at) |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | defer os.RemoveAll(r.Dir) |
| 64 | if r.TXID != at { |
| 65 | t.Fatalf("restored to %d, want the first file's end %d", r.TXID, at) |
| 66 | } |
| 67 | if rows := countRows(t, r.Path); rows != dbdoortest.Rows { |
| 68 | t.Fatalf("rows = %d, want one batch's %d", rows, dbdoortest.Rows) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | // A target at or past the latest transaction plans the latest copy, and |
| 73 | // says so with Target 0: litestream itself refuses a target past the end. |
| 74 | func TestPlanRestoreTargetPastTheEndPlansTheLatest(t *testing.T) { |
| 75 | src := t.TempDir() |
| 76 | dbdoortest.Replicate(t, src, 3) |
| 77 | latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | for _, target := range []uint64{latest.TXID, latest.TXID + 100} { |
| 82 | got, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), target) |
| 83 | if err != nil { |
| 84 | t.Fatalf("target %d: %v", target, err) |
| 85 | } |
| 86 | if !reflect.DeepEqual(got, latest) { |
| 87 | t.Fatalf("target %d planned %+v, want the latest %+v", target, got, latest) |
| 88 | } |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | // A planned TXID pins the restore: after the source moves on, a restore at |
| 93 | // plan.TXID still produces exactly the planned size, transaction and rows. |
| 94 | // Task 5 restores at plan.TXID for that reason (preflight ruling M2). |
| 95 | func TestRestorePinnedAtAPlannedTXIDKeepsThePlannedSize(t *testing.T) { |
| 96 | dir := t.TempDir() |
| 97 | db := dbdoortest.Replicate(t, dir, 2) |
| 98 | plan, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0) |
| 99 | if err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | |
| 103 | // The source grows by many pages and replicates again. |
| 104 | grow(t, db) |
| 105 | dbdoortest.Sync(t, db, dir) |
| 106 | later, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0) |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | if later.TXID <= plan.TXID || later.RestoredBytes <= plan.RestoredBytes { |
| 111 | t.Fatalf("the source did not move on: planned %+v, later %+v", plan, later) |
| 112 | } |
| 113 | |
| 114 | r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), plan.TXID) |
| 115 | if err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | defer os.RemoveAll(r.Dir) |
| 119 | st, err := os.Stat(r.Path) |
| 120 | if err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | if st.Size() != plan.RestoredBytes || r.TXID != plan.TXID { |
| 124 | t.Fatalf("pinned restore: %d bytes at %d, want the plan's %d bytes at %d", st.Size(), r.TXID, plan.RestoredBytes, plan.TXID) |
| 125 | } |
| 126 | if rows := countRows(t, r.Path); rows != 2*dbdoortest.Rows { |
| 127 | t.Fatalf("rows = %d, want the planned %d", rows, 2*dbdoortest.Rows) |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | // Retention keeps a snapshot and the L0 files after it. A target inside |
| 132 | // the snapshot's span is history that is no longer kept, while the latest |
| 133 | // copy (and a target at the snapshot's own end) still restores. |
| 134 | func TestPlanRestoreTargetBeforeHistory(t *testing.T) { |
| 135 | src := t.TempDir() |
| 136 | dbdoortest.Replicate(t, src, 4) |
| 137 | files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0) |
| 138 | if len(files) < 4 { |
| 139 | t.Fatalf("want one L0 file per batch, got %d", len(files)) |
| 140 | } |
| 141 | snap := snapshotThrough(t, src, files[:2]) |
| 142 | for _, f := range files[:2] { |
| 143 | if err := os.Remove(litestream.LTXFilePath(src, 0, f.MinTXID, f.MaxTXID)); err != nil { |
| 144 | t.Fatal(err) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | latest, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), 0) |
| 149 | if err != nil { |
| 150 | t.Fatalf("the latest plan over a snapshot and later L0 files: %v", err) |
| 151 | } |
| 152 | if latest.TXID != uint64(files[len(files)-1].MaxTXID) || latest.Files != 1+len(files)-2 { |
| 153 | t.Fatalf("latest plan %+v, want the snapshot plus %d L0 files ending at %d", latest, len(files)-2, files[len(files)-1].MaxTXID) |
| 154 | } |
| 155 | r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), latest.TXID) |
| 156 | if err != nil { |
| 157 | t.Fatal(err) |
| 158 | } |
| 159 | defer os.RemoveAll(r.Dir) |
| 160 | if rows := countRows(t, r.Path); rows != 4*dbdoortest.Rows { |
| 161 | t.Fatalf("rows = %d, want every batch's %d", rows, 4*dbdoortest.Rows) |
| 162 | } |
| 163 | if st, err := os.Stat(r.Path); err != nil || st.Size() != latest.RestoredBytes { |
| 164 | t.Fatalf("restored over the snapshot: stat %v, err %v; want the planned %d bytes", st, err, latest.RestoredBytes) |
| 165 | } |
| 166 | |
| 167 | // A target at an L0 file's end after the snapshot: the snapshot plus |
| 168 | // that one file. |
| 169 | mid := uint64(files[2].MaxTXID) |
| 170 | atMid, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), mid) |
| 171 | if err != nil || atMid.TXID != mid || atMid.Files != 2 || atMid.Target != mid { |
| 172 | t.Fatalf("a target after the snapshot: plan %+v, err %v; want the snapshot and one L0 file ending at %d", atMid, err, mid) |
| 173 | } |
| 174 | rm, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), atMid.TXID) |
| 175 | if err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | defer os.RemoveAll(rm.Dir) |
| 179 | if st, err := os.Stat(rm.Path); err != nil || st.Size() != atMid.RestoredBytes || countRows(t, rm.Path) != 3*dbdoortest.Rows { |
| 180 | t.Fatalf("restored at %d: stat %v, err %v; want %d bytes and three batches", mid, st, err, atMid.RestoredBytes) |
| 181 | } |
| 182 | |
| 183 | atSnap, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(snap.MaxTXID)) |
| 184 | if err != nil || atSnap.TXID != uint64(snap.MaxTXID) || atSnap.Files != 1 || atSnap.Target != uint64(snap.MaxTXID) { |
| 185 | t.Fatalf("a target at the snapshot's end: plan %+v, err %v; want the snapshot alone", atSnap, err) |
| 186 | } |
| 187 | |
| 188 | _, err = PlanRestore(context.Background(), NewClient(newTestDoor(t, src).do), uint64(files[0].MaxTXID)) |
| 189 | if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) { |
| 190 | t.Fatalf("a target before the retained history: err = %v, want ErrHistoryGone", err) |
| 191 | } |
| 192 | _, err = RestoreToTarget(context.Background(), NewClient(newTestDoor(t, src).do), t.TempDir(), uint64(files[0].MaxTXID)) |
| 193 | if !errors.Is(err, ErrHistoryGone) || errors.Is(err, ErrNothingReplicated) { |
| 194 | t.Fatalf("a restore before the retained history: err = %v, want ErrHistoryGone", err) |
| 195 | } |
| 196 | |
| 197 | if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 0); !errors.Is(err, ErrNothingReplicated) { |
| 198 | t.Fatalf("an empty replica: err = %v, want ErrNothingReplicated", err) |
| 199 | } |
| 200 | if _, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), 5); !errors.Is(err, ErrNothingReplicated) { |
| 201 | t.Fatalf("an empty replica with a target: err = %v, want ErrNothingReplicated", err) |
| 202 | } |
| 203 | } |
| 204 | |
| 205 | // snapshotThrough compacts the given L0 files, which must start at TXID 1, |
| 206 | // into one snapshot-level file in the file replica at dir, the way |
| 207 | // litestream's own compactor writes a level: ltx.Compactor over the source |
| 208 | // files, written through the file client. |
| 209 | func snapshotThrough(t *testing.T, dir string, files []*ltx.FileInfo) *ltx.FileInfo { |
| 210 | t.Helper() |
| 211 | ctx := context.Background() |
| 212 | rc := lsfile.NewReplicaClient(dir) |
| 213 | var rdrs []io.Reader |
| 214 | for _, f := range files { |
| 215 | r, err := rc.OpenLTXFile(ctx, 0, f.MinTXID, f.MaxTXID, 0, 0) |
| 216 | if err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | defer r.Close() |
| 220 | rdrs = append(rdrs, r) |
| 221 | } |
| 222 | pr, pw := io.Pipe() |
| 223 | go func() { |
| 224 | c, err := ltx.NewCompactor(pw, rdrs) |
| 225 | if err != nil { |
| 226 | pw.CloseWithError(err) |
| 227 | return |
| 228 | } |
| 229 | c.HeaderFlags = ltx.HeaderFlagNoChecksum |
| 230 | pw.CloseWithError(c.Compact(ctx)) |
| 231 | }() |
| 232 | info, err := rc.WriteLTXFile(ctx, litestream.SnapshotLevel, files[0].MinTXID, files[len(files)-1].MaxTXID, pr) |
| 233 | pr.CloseWithError(err) |
| 234 | if err != nil { |
| 235 | t.Fatal(err) |
| 236 | } |
| 237 | return info |
| 238 | } |
| 239 | |
| 240 | // grow adds enough rows to the source database at path to add many pages. |
| 241 | func grow(t *testing.T, path string) { |
| 242 | t.Helper() |
| 243 | db, err := sql.Open("sqlite", path) |
| 244 | if err != nil { |
| 245 | t.Fatal(err) |
| 246 | } |
| 247 | defer db.Close() |
| 248 | for i := 0; i < 50; i++ { |
| 249 | if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("grow-%d-%s", i, strings.Repeat("x", 2000))); err != nil { |
| 250 | t.Fatal(err) |
| 251 | } |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | // The header PlanRestore sizes the restore from comes from the replica, |
| 256 | // so it is untrusted. A page size SQLite cannot have, a zero commit, or |
| 257 | // TXIDs that are not the listed file's are refused rather than sized; an |
| 258 | // unchecked Commit times PageSize can wrap negative and slip under any |
| 259 | // size cap. |
| 260 | func TestPlanRestoreRefusesAHeaderThatDoesNotDescribeItsFile(t *testing.T) { |
| 261 | src := t.TempDir() |
| 262 | dbdoortest.Replicate(t, src, 2) |
| 263 | files := collect(t, NewClient(newTestDoor(t, src).do), 0, 0) |
| 264 | last := files[len(files)-1] |
| 265 | rc, err := lsfile.NewReplicaClient(src).OpenLTXFile(context.Background(), 0, last.MinTXID, last.MaxTXID, 0, ltx.HeaderSize) |
| 266 | if err != nil { |
| 267 | t.Fatal(err) |
| 268 | } |
| 269 | honest := make([]byte, ltx.HeaderSize) |
| 270 | _, err = io.ReadFull(rc, honest) |
| 271 | rc.Close() |
| 272 | if err != nil { |
| 273 | t.Fatal(err) |
| 274 | } |
| 275 | |
| 276 | for _, tc := range []struct { |
| 277 | name string |
| 278 | edit func(h *ltx.Header) |
| 279 | }{ |
| 280 | {"a page size that wraps the size", func(h *ltx.Header) { h.PageSize, h.Commit = 0xFFFFFFFF, 0xFFFFFFFF }}, |
| 281 | {"a page size that is not a power of two", func(h *ltx.Header) { h.PageSize = 4000 }}, |
| 282 | {"a zero page size", func(h *ltx.Header) { h.PageSize = 0 }}, |
| 283 | {"a zero commit", func(h *ltx.Header) { h.Commit = 0 }}, |
| 284 | {"another file's max txid", func(h *ltx.Header) { h.MaxTXID++ }}, |
| 285 | {"another file's min txid", func(h *ltx.Header) { h.MinTXID++ }}, |
| 286 | } { |
| 287 | t.Run(tc.name, func(t *testing.T) { |
| 288 | var h ltx.Header |
| 289 | if err := h.UnmarshalBinary(honest); err != nil { |
| 290 | t.Fatal(err) |
| 291 | } |
| 292 | tc.edit(&h) |
| 293 | crafted, _ := h.MarshalBinary() |
| 294 | door := newTestDoor(t, src) |
| 295 | headerTail := GetTail(0, uint64(last.MinTXID), uint64(last.MaxTXID)) |
| 296 | do := func(ctx context.Context, tail string, hd http.Header) (*http.Response, error) { |
| 297 | if tail == headerTail && hd.Get("Range") == RangeHeader(0, ltx.HeaderSize) { |
| 298 | rec := httptest.NewRecorder() |
| 299 | rec.WriteHeader(http.StatusPartialContent) |
| 300 | rec.Write(crafted) |
| 301 | return rec.Result(), nil |
| 302 | } |
| 303 | return door.do(ctx, tail, hd) |
| 304 | } |
| 305 | plan, err := PlanRestore(context.Background(), NewClient(do), 0) |
| 306 | if err == nil || errors.Is(err, ErrNothingReplicated) || errors.Is(err, ErrHistoryGone) { |
| 307 | t.Fatalf("plan %+v, err = %v; want the header refused as its own error", plan, err) |
| 308 | } |
| 309 | if !reflect.DeepEqual(plan, RestorePlan{}) { |
| 310 | t.Fatalf("a refused header still planned %+v", plan) |
| 311 | } |
| 312 | }) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // A database that shrinks (rows deleted, then VACUUM) is sized by its last |
| 317 | // file's commit, not by the largest the history ever held. |
| 318 | func TestPlanRestoreFollowsAShrinkingDatabase(t *testing.T) { |
| 319 | dir := t.TempDir() |
| 320 | db := dbdoortest.Replicate(t, dir, 1) |
| 321 | grow(t, db) |
| 322 | dbdoortest.Sync(t, db, dir) |
| 323 | big, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0) |
| 324 | if err != nil { |
| 325 | t.Fatal(err) |
| 326 | } |
| 327 | sdb, err := sql.Open("sqlite", db) |
| 328 | if err != nil { |
| 329 | t.Fatal(err) |
| 330 | } |
| 331 | if _, err := sdb.Exec(`DELETE FROM t WHERE body LIKE 'grow-%'; VACUUM`); err != nil { |
| 332 | sdb.Close() |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | sdb.Close() |
| 336 | dbdoortest.Sync(t, db, dir) |
| 337 | |
| 338 | small, err := PlanRestore(context.Background(), NewClient(newTestDoor(t, dir).do), 0) |
| 339 | if err != nil { |
| 340 | t.Fatal(err) |
| 341 | } |
| 342 | if small.TXID <= big.TXID || small.RestoredBytes >= big.RestoredBytes { |
| 343 | t.Fatalf("after the shrink: planned %+v, before %+v; want a later, smaller plan", small, big) |
| 344 | } |
| 345 | r, err := RestoreToTarget(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), small.TXID) |
| 346 | if err != nil { |
| 347 | t.Fatal(err) |
| 348 | } |
| 349 | defer os.RemoveAll(r.Dir) |
| 350 | st, err := os.Stat(r.Path) |
| 351 | if err != nil { |
| 352 | t.Fatal(err) |
| 353 | } |
| 354 | if st.Size() != small.RestoredBytes { |
| 355 | t.Fatalf("restored %d bytes, planned %d", st.Size(), small.RestoredBytes) |
| 356 | } |
| 357 | } |
| 358 | |
amadan