| 1 | // Package litestreamfacts establishes, by observation of the pinned |
| 2 | // libraries (litestream v0.5.14, superfly/ltx v0.5.1), the facts the |
| 3 | // move-any-app spec's restore contract rests on (moves spec rev 7.7 §5.1; |
| 4 | // joint rounds 5 and 6, R5-1 to R5-3, R6-2, R6-5). It is test-only and |
| 5 | // drives litestream as internal/lease does, with the file replica client |
| 6 | // the lease tests use. Each test asserts the exact fact the spec relies on |
| 7 | // and fails on any alternative behaviour, so a library upgrade that |
| 8 | // changes a fact fails here first (R6-2): a passing run is the contract, |
| 9 | // not a record of one observation. TestGate_AnchoredLineageThroughStore |
| 10 | // takes one anchored lineage through the production Store configuration |
| 11 | // (internal/lease's Held.Replicate, intervals shortened) with repeated |
| 12 | // writes, compaction, snapshot rotation, time-based level-0 expiry, disk |
| 13 | // loss and repeated cold restores, asserting exact positions and contents. |
| 14 | // |
| 15 | // What the file client cannot show (R6 and R7 reviews): S3 lists LTX |
| 16 | // files in filename order and ages them for retention by LastModified, |
| 17 | // the file client by the encoded LTX timestamp kept as the file's mtime; |
| 18 | // S3 keeps versions and delete markers, the file client replaces and |
| 19 | // deletes outright; S3 requests can complete late, and an upload |
| 20 | // outstanding across a writer replacement is the R7-1 hazard the model |
| 21 | // (internal/sim/hostlife) encodes. The repository has no S3 fake that |
| 22 | // litestream's client can talk to (the lease tests' fakeMirror is a |
| 23 | // MirrorStore seam), so those remain the model's and §15 step 8's. |
| 24 | // TestFact_R7_1_UnconditionalPut reads the pinned S3 client's source for |
| 25 | // the two facts the hazard rests on: an unconditional PutObject, and a key |
| 26 | // of prefix, level and TXID range only. |
| 27 | // |
| 28 | // The observed facts are recorded in ~/.cache/nearest-region/litestream-facts.md. |
| 29 | package litestreamfacts |
| 30 | |
| 31 | import ( |
| 32 | "bytes" |
| 33 | "context" |
| 34 | "database/sql" |
| 35 | "fmt" |
| 36 | "go/build" |
| 37 | "io" |
| 38 | "os" |
| 39 | "path/filepath" |
| 40 | "strings" |
| 41 | "testing" |
| 42 | "time" |
| 43 | |
| 44 | "github.com/benbjohnson/litestream" |
| 45 | lsfile "github.com/benbjohnson/litestream/file" |
| 46 | "github.com/superfly/ltx" |
| 47 | _ "modernc.org/sqlite" |
| 48 | ) |
| 49 | |
| 50 | // openDB opens path under litestream against client, with its own metadata |
| 51 | // directory, no background monitors (every sync is explicit), and returns |
| 52 | // the DB. The caller closes it. |
| 53 | func openDB(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB { |
| 54 | t.Helper() |
| 55 | db := litestream.NewDB(path) |
| 56 | db.SetMetaPath(meta) |
| 57 | db.MonitorInterval = time.Hour |
| 58 | db.Replica = litestream.NewReplicaWithClient(db, client) |
| 59 | db.Replica.SyncInterval = time.Hour |
| 60 | if err := db.Open(); err != nil { |
| 61 | t.Fatalf("open %s: %v", path, err) |
| 62 | } |
| 63 | return db |
| 64 | } |
| 65 | |
| 66 | func closeDB(t *testing.T, db *litestream.DB) { |
| 67 | t.Helper() |
| 68 | if err := db.Close(context.Background()); err != nil { |
| 69 | t.Fatalf("close: %v", err) |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | // write inserts n rows, each in its own transaction, so each is a |
| 74 | // transaction litestream can see. |
| 75 | func write(t *testing.T, path string, n int) { |
| 76 | t.Helper() |
| 77 | d, err := sql.Open("sqlite", path) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | defer d.Close() |
| 82 | if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | for i := 0; i < n; i++ { |
| 86 | if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | // rows reads the table's contents, id -> body, so a restore can be |
| 93 | // compared row by row, updates and deletes included (R7-5). |
| 94 | func rows(t *testing.T, path string) map[int]string { |
| 95 | t.Helper() |
| 96 | d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | defer d.Close() |
| 101 | rs, err := d.Query(`SELECT id, body FROM t ORDER BY id`) |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | defer rs.Close() |
| 106 | out := map[int]string{} |
| 107 | for rs.Next() { |
| 108 | var id int |
| 109 | var body string |
| 110 | if err := rs.Scan(&id, &body); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | out[id] = body |
| 114 | } |
| 115 | return out |
| 116 | } |
| 117 | |
| 118 | // mustRows fails unless the database at path holds exactly want. |
| 119 | func mustRows(t *testing.T, path string, want map[int]string) { |
| 120 | t.Helper() |
| 121 | got := rows(t, path) |
| 122 | if len(got) != len(want) { |
| 123 | t.Fatalf("%s: %d rows, want %d", filepath.Base(path), len(got), len(want)) |
| 124 | } |
| 125 | for id, body := range want { |
| 126 | if got[id] != body { |
| 127 | t.Fatalf("%s: row %d is %q, want %q", filepath.Base(path), id, got[id], body) |
| 128 | } |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | // exec runs one statement in its own transaction. |
| 133 | func exec(t *testing.T, path, stmt string, args ...any) { |
| 134 | t.Helper() |
| 135 | d, err := sql.Open("sqlite", path) |
| 136 | if err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | defer d.Close() |
| 140 | if _, err := d.Exec(stmt, args...); err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | // waitFor polls cond until it holds or the deadline passes (R7-5: a |
| 146 | // completion condition, not a sleep). |
| 147 | func waitFor(t *testing.T, what string, d time.Duration, cond func() bool) { |
| 148 | t.Helper() |
| 149 | deadline := time.Now().Add(d) |
| 150 | for !cond() { |
| 151 | if time.Now().After(deadline) { |
| 152 | t.Fatalf("timed out waiting for %s", what) |
| 153 | } |
| 154 | time.Sleep(20 * time.Millisecond) |
| 155 | } |
| 156 | } |
| 157 | |
| 158 | func count(t *testing.T, path string) int { |
| 159 | t.Helper() |
| 160 | d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | defer d.Close() |
| 165 | var n int |
| 166 | if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | return n |
| 170 | } |
| 171 | |
| 172 | func sync(t *testing.T, db *litestream.DB) { |
| 173 | t.Helper() |
| 174 | if err := db.SyncAndWait(context.Background()); err != nil { |
| 175 | t.Fatalf("sync: %v", err) |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | // files lists one level of a replica, in litestream's order. |
| 180 | func files(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo { |
| 181 | t.Helper() |
| 182 | itr, err := client.LTXFiles(context.Background(), level, 0, false) |
| 183 | if err != nil { |
| 184 | t.Fatal(err) |
| 185 | } |
| 186 | infos, err := ltx.SliceFileIterator(itr) |
| 187 | if err != nil { |
| 188 | t.Fatal(err) |
| 189 | } |
| 190 | return infos |
| 191 | } |
| 192 | |
| 193 | func ranges(infos []*ltx.FileInfo) string { |
| 194 | var out []string |
| 195 | for _, f := range infos { |
| 196 | out = append(out, fmt.Sprintf("[%d,%d]", f.MinTXID, f.MaxTXID)) |
| 197 | } |
| 198 | if len(out) == 0 { |
| 199 | return "none" |
| 200 | } |
| 201 | return strings.Join(out, " ") |
| 202 | } |
| 203 | |
| 204 | // copyFile copies one LTX file between replica clients, as the install's |
| 205 | // CopyObject would. |
| 206 | func copyFile(t *testing.T, from, to litestream.ReplicaClient, info *ltx.FileInfo) { |
| 207 | t.Helper() |
| 208 | ctx := context.Background() |
| 209 | rc, err := from.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| 210 | if err != nil { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | defer rc.Close() |
| 214 | if _, err := to.WriteLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, rc); err != nil { |
| 215 | t.Fatal(err) |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // restore restores client's newest state to out, litestream's own planner. |
| 220 | func restore(t *testing.T, client litestream.ReplicaClient, out string) error { |
| 221 | t.Helper() |
| 222 | r := litestream.NewReplicaWithClient(nil, client) |
| 223 | opt := litestream.NewRestoreOptions() |
| 224 | opt.OutputPath = out |
| 225 | return r.Restore(context.Background(), opt) |
| 226 | } |
| 227 | |
| 228 | // decode reads an LTX file's header and trailer and verifies it. |
| 229 | func decode(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) (ltx.Header, ltx.Trailer) { |
| 230 | t.Helper() |
| 231 | rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| 232 | if err != nil { |
| 233 | t.Fatal(err) |
| 234 | } |
| 235 | defer rc.Close() |
| 236 | b, err := io.ReadAll(rc) |
| 237 | if err != nil { |
| 238 | t.Fatal(err) |
| 239 | } |
| 240 | dec := ltx.NewDecoder(bytes.NewReader(b)) |
| 241 | if err := dec.Verify(); err != nil { |
| 242 | t.Fatalf("verify %s: %v", ltx.FormatFilename(info.MinTXID, info.MaxTXID), err) |
| 243 | } |
| 244 | return dec.Header(), dec.Trailer() |
| 245 | } |
| 246 | |
| 247 | // encodeAnchor re-encodes a snapshot [1,N] as a level-0 file [N,N] with |
| 248 | // every page: the file litestream's position initialization fetches, so |
| 249 | // that replication continues from N (R5-1, the remedy under test). The |
| 250 | // header timestamp is an input (R6-5): with the same inputs the bytes are |
| 251 | // identical, which is what lets a replayed PUT repeat the same object. |
| 252 | func encodeAnchor(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, ts int64) []byte { |
| 253 | t.Helper() |
| 254 | ctx := context.Background() |
| 255 | rc, err := from.OpenLTXFile(ctx, snap.Level, snap.MinTXID, snap.MaxTXID, 0, 0) |
| 256 | if err != nil { |
| 257 | t.Fatal(err) |
| 258 | } |
| 259 | defer rc.Close() |
| 260 | dec := ltx.NewDecoder(rc) |
| 261 | if err := dec.DecodeHeader(); err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | hdr := dec.Header() |
| 265 | var buf bytes.Buffer |
| 266 | enc, err := ltx.NewEncoder(&buf) |
| 267 | if err != nil { |
| 268 | t.Fatal(err) |
| 269 | } |
| 270 | // WALOffset at the WAL header: litestream's verify reads the position |
| 271 | // from the newest local level-0 file and, finding the WAL does not |
| 272 | // match it, takes a fresh snapshot at the next TXID; an offset below |
| 273 | // the header size is refused outright ("prev WAL offset is less than |
| 274 | // the header size", observed) |
| 275 | if err := enc.EncodeHeader(ltx.Header{ |
| 276 | Version: ltx.Version, |
| 277 | Flags: ltx.HeaderFlagNoChecksum, |
| 278 | PageSize: hdr.PageSize, |
| 279 | Commit: hdr.Commit, |
| 280 | MinTXID: snap.MaxTXID, |
| 281 | MaxTXID: snap.MaxTXID, |
| 282 | Timestamp: ts, |
| 283 | WALOffset: litestream.WALHeaderSize, |
| 284 | }); err != nil { |
| 285 | t.Fatal(err) |
| 286 | } |
| 287 | page := make([]byte, hdr.PageSize) |
| 288 | for { |
| 289 | var ph ltx.PageHeader |
| 290 | if err := dec.DecodePage(&ph, page); err == io.EOF { |
| 291 | break |
| 292 | } else if err != nil { |
| 293 | t.Fatal(err) |
| 294 | } |
| 295 | if err := enc.EncodePage(ph, page); err != nil { |
| 296 | t.Fatal(err) |
| 297 | } |
| 298 | } |
| 299 | if err := dec.Close(); err != nil { |
| 300 | t.Fatal(err) |
| 301 | } |
| 302 | if err := enc.Close(); err != nil { |
| 303 | t.Fatal(err) |
| 304 | } |
| 305 | return buf.Bytes() |
| 306 | } |
| 307 | |
| 308 | // anchorL0 writes the anchor for snap into to. |
| 309 | func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, to litestream.ReplicaClient) { |
| 310 | t.Helper() |
| 311 | b := encodeAnchor(t, from, snap, time.Now().UnixMilli()) |
| 312 | if _, err := to.WriteLTXFile(context.Background(), 0, snap.MaxTXID, snap.MaxTXID, bytes.NewReader(b)); err != nil { |
| 313 | t.Fatal(err) |
| 314 | } |
| 315 | } |
| 316 | |
| 317 | // mustRanges fails unless the level's current files are exactly want. |
| 318 | func mustRanges(t *testing.T, client litestream.ReplicaClient, level int, want string) { |
| 319 | t.Helper() |
| 320 | if got := ranges(files(t, client, level)); got != want { |
| 321 | t.Fatalf("level %d: have %s, want exactly %s", level, got, want) |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | // mustCount fails unless the database at path holds exactly want rows. |
| 326 | func mustCount(t *testing.T, path string, want int) { |
| 327 | t.Helper() |
| 328 | if n := count(t, path); n != want { |
| 329 | t.Fatalf("%s: %d rows, want exactly %d", filepath.Base(path), n, want) |
| 330 | } |
| 331 | } |
| 332 | |
| 333 | // source builds a replicated source database with n rows and a level-9 |
| 334 | // snapshot of it, and returns the client and the snapshot's info. |
| 335 | func source(t *testing.T, n int) (litestream.ReplicaClient, *ltx.FileInfo) { |
| 336 | t.Helper() |
| 337 | dir := t.TempDir() |
| 338 | path := filepath.Join(dir, "src.db") |
| 339 | write(t, path, 1) |
| 340 | client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| 341 | db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| 342 | sync(t, db) |
| 343 | write(t, path, n-1) |
| 344 | sync(t, db) |
| 345 | snap, err := db.Snapshot(context.Background()) |
| 346 | if err != nil { |
| 347 | t.Fatal(err) |
| 348 | } |
| 349 | closeDB(t, db) |
| 350 | t.Logf("source: L0 %s; snapshot [%d,%d]", ranges(files(t, client, 0)), snap.MinTXID, snap.MaxTXID) |
| 351 | return client, snap |
| 352 | } |
| 353 | |
| 354 | // wipe removes the local database, its WAL and SHM, and litestream's |
| 355 | // metadata: a lost disk. |
| 356 | func wipe(t *testing.T, path, meta string) { |
| 357 | t.Helper() |
| 358 | for _, p := range []string{path, path + "-wal", path + "-shm", path + "-journal"} { |
| 359 | os.Remove(p) |
| 360 | } |
| 361 | if err := os.RemoveAll(meta); err != nil { |
| 362 | t.Fatal(err) |
| 363 | } |
| 364 | } |
| 365 | |
| 366 | // TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne: install only the level-9 |
| 367 | // snapshot [1,N], restore it, start replication with fresh metadata and no |
| 368 | // remote level 0. The next upload carries TXID 1, not N+1, and a cold |
| 369 | // restore afterwards loses the new writes. |
| 370 | func TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne(t *testing.T) { |
| 371 | src, snap := source(t, 5) |
| 372 | dir := t.TempDir() |
| 373 | gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| 374 | copyFile(t, src, gen, snap) |
| 375 | out := filepath.Join(dir, "out.db") |
| 376 | if err := restore(t, gen, out); err != nil { |
| 377 | t.Fatal(err) |
| 378 | } |
| 379 | if n := count(t, out); n != 5 { |
| 380 | t.Fatalf("restored %d rows, want 5", n) |
| 381 | } |
| 382 | meta := filepath.Join(dir, "meta") |
| 383 | db := openDB(t, out, meta, gen) |
| 384 | write(t, out, 2) |
| 385 | sync(t, db) |
| 386 | closeDB(t, db) |
| 387 | l0 := files(t, gen, 0) |
| 388 | t.Logf("FACT R5-1a: after a snapshot-only install (snapshot [1,%d]) the first upload is level 0 %s", snap.MaxTXID, ranges(l0)) |
| 389 | if len(l0) == 0 || l0[0].MinTXID != 1 { |
| 390 | t.Fatalf("expected replication to restart at TXID 1, got %s", ranges(l0)) |
| 391 | } |
| 392 | wipe(t, out, meta) |
| 393 | out2 := filepath.Join(dir, "out2.db") |
| 394 | if err := restore(t, gen, out2); err != nil { |
| 395 | t.Fatal(err) |
| 396 | } |
| 397 | n := count(t, out2) |
| 398 | t.Logf("FACT R5-1b: a cold restore after that holds %d rows of 7: the planner roots at the snapshot and skips level-0 files below its TXID", n) |
| 399 | if n != 5 { |
| 400 | t.Fatalf("expected the two new rows to be lost (5 rows), got %d", n) |
| 401 | } |
| 402 | } |
| 403 | |
| 404 | // TestFact_R5_1_AnchorL0ContinuesFromN: the remedy. The install also writes |
| 405 | // a level-0 file [N,N] with every page. Litestream's open fetches the |
| 406 | // newest remote level-0 file as its local position, so the next upload |
| 407 | // carries N+1 and a cold restore has the new writes. |
| 408 | func TestFact_R5_1_AnchorL0ContinuesFromN(t *testing.T) { |
| 409 | src, snap := source(t, 5) |
| 410 | dir := t.TempDir() |
| 411 | gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| 412 | copyFile(t, src, gen, snap) |
| 413 | anchorL0(t, src, snap, gen) |
| 414 | out := filepath.Join(dir, "out.db") |
| 415 | if err := restore(t, gen, out); err != nil { |
| 416 | t.Fatal(err) |
| 417 | } |
| 418 | if n := count(t, out); n != 5 { |
| 419 | t.Fatalf("restored %d rows, want 5", n) |
| 420 | } |
| 421 | meta := filepath.Join(dir, "meta") |
| 422 | db := openDB(t, out, meta, gen) |
| 423 | write(t, out, 2) |
| 424 | sync(t, db) |
| 425 | closeDB(t, db) |
| 426 | l0 := files(t, gen, 0) |
| 427 | t.Logf("FACT R5-1c: with a level-0 anchor [%d,%d] beside the snapshot, level 0 after two writes and one sync is %s", snap.MaxTXID, snap.MaxTXID, ranges(l0)) |
| 428 | // exactly: the anchor, then the next TXID N+1 and nothing else (R6-2) |
| 429 | n := snap.MaxTXID |
| 430 | mustRanges(t, gen, 0, fmt.Sprintf("[%d,%d] [%d,%d]", n, n, n+1, n+1)) |
| 431 | wipe(t, out, meta) |
| 432 | out2 := filepath.Join(dir, "out2.db") |
| 433 | if err := restore(t, gen, out2); err != nil { |
| 434 | t.Fatal(err) |
| 435 | } |
| 436 | t.Logf("FACT R5-1d: a cold restore after that holds %d rows of 7", count(t, out2)) |
| 437 | mustCount(t, out2, 7) |
| 438 | } |
| 439 | |
| 440 | // TestFact_R5_1_TXIDSidecarIsFollowOnly: a -txid sidecar beside the |
| 441 | // restored database does not set the replication position of an |
| 442 | // ordinary open. |
| 443 | func TestFact_R5_1_TXIDSidecarIsFollowOnly(t *testing.T) { |
| 444 | src, snap := source(t, 5) |
| 445 | dir := t.TempDir() |
| 446 | gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| 447 | copyFile(t, src, gen, snap) |
| 448 | out := filepath.Join(dir, "out.db") |
| 449 | if err := restore(t, gen, out); err != nil { |
| 450 | t.Fatal(err) |
| 451 | } |
| 452 | if err := litestream.WriteTXIDFile(out, snap.MaxTXID); err != nil { |
| 453 | t.Fatal(err) |
| 454 | } |
| 455 | db := openDB(t, out, filepath.Join(dir, "meta"), gen) |
| 456 | write(t, out, 1) |
| 457 | sync(t, db) |
| 458 | closeDB(t, db) |
| 459 | l0 := files(t, gen, 0) |
| 460 | t.Logf("FACT R5-1e: with a -txid sidecar of %d, the first upload is level 0 %s", snap.MaxTXID, ranges(l0)) |
| 461 | if len(l0) == 0 || l0[0].MinTXID != 1 { |
| 462 | t.Fatalf("the sidecar unexpectedly set the position: %s", ranges(l0)) |
| 463 | } |
| 464 | } |
| 465 | |
| 466 | // TestFact_R5_1_L0ExpiryAndAFreshBox: a fresh box's position after |
| 467 | // litestream's own level-0 expiry. TXID-based retention keeps the newest |
| 468 | // level-0 file, and so does time-based retention (L0Retention, the store's |
| 469 | // monitor); a fresh box restored from the replica therefore continues at |
| 470 | // N+1 and a cold restore afterwards holds every row. Each of those is |
| 471 | // asserted exactly (R6-2): a library that removed the newest level-0 file, |
| 472 | // restarted numbering or lost the new writes fails here. |
| 473 | func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) { |
| 474 | dir := t.TempDir() |
| 475 | path := filepath.Join(dir, "src.db") |
| 476 | write(t, path, 1) |
| 477 | client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| 478 | db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| 479 | for i := 0; i < 3; i++ { |
| 480 | write(t, path, 2) |
| 481 | sync(t, db) |
| 482 | } |
| 483 | if _, err := db.Compact(context.Background(), 1); err != nil { |
| 484 | t.Fatal(err) |
| 485 | } |
| 486 | snap, err := db.Snapshot(context.Background()) |
| 487 | if err != nil { |
| 488 | t.Fatal(err) |
| 489 | } |
| 490 | // TXID-based retention at the snapshot level |
| 491 | if err := db.EnforceRetentionByTXID(context.Background(), 0, snap.MaxTXID+1); err != nil { |
| 492 | t.Fatal(err) |
| 493 | } |
| 494 | n := snap.MaxTXID // the newest level-0 file's TXID, the position |
| 495 | newest := fmt.Sprintf("[%d,%d]", n, n) |
| 496 | byTXID := ranges(files(t, client, 0)) |
| 497 | if byTXID != newest { |
| 498 | t.Fatalf("TXID-based retention at %d left level 0 %s, want exactly the newest file %s retained", n+1, byTXID, newest) |
| 499 | } |
| 500 | // time-based level-0 expiry, as the store's L0 retention monitor runs it |
| 501 | db.L0Retention = time.Millisecond |
| 502 | time.Sleep(20 * time.Millisecond) |
| 503 | if err := db.EnforceL0RetentionByTime(context.Background()); err != nil { |
| 504 | t.Fatal(err) |
| 505 | } |
| 506 | closeDB(t, db) |
| 507 | t.Logf("FACT R5-1f: after TXID-based level-0 retention at %d the level-0 listing is %s; after time-based L0 expiry it is %s (L9 %s; L1 %s)", n+1, byTXID, ranges(files(t, client, 0)), ranges(files(t, client, 9)), ranges(files(t, client, 1))) |
| 508 | mustRanges(t, client, 0, newest) // the newest level-0 file survives time-based expiry too, by identity |
| 509 | out := filepath.Join(dir, "out.db") |
| 510 | if err := restore(t, client, out); err != nil { |
| 511 | t.Fatal(err) |
| 512 | } |
| 513 | mustCount(t, out, 7) |
| 514 | meta := filepath.Join(dir, "meta2") |
| 515 | db2 := openDB(t, out, meta, client) |
| 516 | write(t, out, 2) |
| 517 | sync(t, db2) |
| 518 | closeDB(t, db2) |
| 519 | t.Logf("FACT R5-1g: on a fresh box restored at TXID %d, the first upload is level 0 %s", n, ranges(files(t, client, 0))) |
| 520 | mustRanges(t, client, 0, fmt.Sprintf("%s [%d,%d]", newest, n+1, n+1)) // exactly N+1: a restart at 1 fails here |
| 521 | wipe(t, out, meta) |
| 522 | out2 := filepath.Join(dir, "out2.db") |
| 523 | if err := restore(t, client, out2); err != nil { |
| 524 | t.Fatal(err) |
| 525 | } |
| 526 | t.Logf("FACT R5-1h: a cold restore after that holds %d rows of 9", count(t, out2)) |
| 527 | mustCount(t, out2, 9) // every row: a run that kept 7 of 9 fails here |
| 528 | } |
| 529 | |
| 530 | // TestFact_R5_2_Checksums: what the pinned writers put in LTX headers and |
| 531 | // trailers: the level-0 writer, the snapshot writer and the compactor. |
| 532 | func TestFact_R5_2_Checksums(t *testing.T) { |
| 533 | dir := t.TempDir() |
| 534 | path := filepath.Join(dir, "src.db") |
| 535 | write(t, path, 1) |
| 536 | client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| 537 | db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| 538 | for i := 0; i < 3; i++ { |
| 539 | write(t, path, 2) |
| 540 | sync(t, db) |
| 541 | } |
| 542 | snap, err := db.Snapshot(context.Background()) |
| 543 | if err != nil { |
| 544 | t.Fatal(err) |
| 545 | } |
| 546 | l1, err := db.Compact(context.Background(), 1) |
| 547 | if err != nil { |
| 548 | t.Fatal(err) |
| 549 | } |
| 550 | closeDB(t, db) |
| 551 | check := func(name string, info *ltx.FileInfo) { |
| 552 | hdr, tr := decode(t, client, info) |
| 553 | t.Logf("FACT R5-2 %s [%d,%d]: HeaderFlagNoChecksum=%v PreApplyChecksum=%#x PostApplyChecksum=%#x FileChecksum!=0=%v (Verify passed: CRC-64 of the file)", |
| 554 | name, info.MinTXID, info.MaxTXID, hdr.NoChecksum(), uint64(hdr.PreApplyChecksum), uint64(tr.PostApplyChecksum), tr.FileChecksum != 0) |
| 555 | if !hdr.NoChecksum() || tr.PostApplyChecksum != 0 || tr.FileChecksum == 0 { |
| 556 | t.Fatalf("%s: unexpected checksum fields", name) |
| 557 | } |
| 558 | } |
| 559 | l0 := files(t, client, 0) |
| 560 | check("level 0", l0[len(l0)-1]) |
| 561 | check("snapshot", snap) |
| 562 | check("level 1 (compactor)", l1) |
| 563 | } |
| 564 | |
| 565 | // TestFact_R5_3_SnapshotRetention: newer snapshots appear, and snapshot |
| 566 | // retention expires older snapshots and the lower-level ranges below the |
| 567 | // oldest remaining one. Afterwards no current file continues from the |
| 568 | // original snapshot's TXID, so a restore rooted there finds a gap, while |
| 569 | // litestream's own restore from the newest snapshot succeeds. |
| 570 | func TestFact_R5_3_SnapshotRetention(t *testing.T) { |
| 571 | dir := t.TempDir() |
| 572 | path := filepath.Join(dir, "src.db") |
| 573 | write(t, path, 1) |
| 574 | client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| 575 | db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| 576 | sync(t, db) |
| 577 | snap1, err := db.Snapshot(context.Background()) |
| 578 | if err != nil { |
| 579 | t.Fatal(err) |
| 580 | } |
| 581 | // keep the original snapshot's bytes: on S3 it would be a pinned, |
| 582 | // noncurrent version; the file client deletes outright |
| 583 | rc, err := client.OpenLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, 0, 0) |
| 584 | if err != nil { |
| 585 | t.Fatal(err) |
| 586 | } |
| 587 | snap1Bytes, _ := io.ReadAll(rc) |
| 588 | rc.Close() |
| 589 | // two more rotations: writes, a level-1 compaction, a snapshot, as the |
| 590 | // store's monitors would over days |
| 591 | var snap2 *ltx.FileInfo |
| 592 | for round := 0; round < 2; round++ { |
| 593 | for i := 0; i < 3; i++ { |
| 594 | write(t, path, 2) |
| 595 | sync(t, db) |
| 596 | } |
| 597 | if _, err := db.Compact(context.Background(), 1); err != nil { |
| 598 | t.Fatal(err) |
| 599 | } |
| 600 | time.Sleep(20 * time.Millisecond) |
| 601 | if snap2, err = db.Snapshot(context.Background()); err != nil { |
| 602 | t.Fatal(err) |
| 603 | } |
| 604 | } |
| 605 | before := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| 606 | // snapshot retention keeps the newest snapshot; minSnapshotTXID is the |
| 607 | // MaxTXID of the snapshot before the oldest kept one, and the store |
| 608 | // then expires levels 1 and up below it (level 0 expires by time, |
| 609 | // L0Retention, 5 minutes by default: here it is enforced by TXID too) |
| 610 | minSnap, err := db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second)) |
| 611 | if err != nil { |
| 612 | t.Fatal(err) |
| 613 | } |
| 614 | if err := db.EnforceRetentionByTXID(context.Background(), 1, minSnap); err != nil { |
| 615 | t.Fatal(err) |
| 616 | } |
| 617 | if err := db.EnforceRetentionByTXID(context.Background(), 0, minSnap); err != nil { |
| 618 | t.Fatal(err) |
| 619 | } |
| 620 | after := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| 621 | t.Logf("FACT R5-3a: before retention: %s", before) |
| 622 | t.Logf("FACT R5-3b: after snapshot retention (minSnapshotTXID %d): %s", minSnap, after) |
| 623 | if len(files(t, client, 9)) != 1 || files(t, client, 9)[0].MaxTXID != snap2.MaxTXID { |
| 624 | t.Fatalf("expected only the newest snapshot to remain") |
| 625 | } |
| 626 | continuation := func() bool { |
| 627 | // any current file with MinTXID <= snap1.MaxTXID+1 and MaxTXID > snap1.MaxTXID |
| 628 | for _, lvl := range []int{0, 1, 2, 3} { |
| 629 | for _, f := range files(t, client, lvl) { |
| 630 | if f.MinTXID <= snap1.MaxTXID+1 && f.MaxTXID > snap1.MaxTXID { |
| 631 | return true |
| 632 | } |
| 633 | } |
| 634 | } |
| 635 | return false |
| 636 | } |
| 637 | t.Logf("FACT R5-3c: after one retention pass a continuation from the original snapshot [1,%d] remains: %v (the level-1 file that starts at 1 ends at minSnapshotTXID and deletion is strictly below)", snap1.MaxTXID, continuation()) |
| 638 | if !continuation() { |
| 639 | t.Fatalf("expected the level-1 file starting at 1 to survive the first pass (it ends at minSnapshotTXID %d)", minSnap) |
| 640 | } |
| 641 | // one more rotation: the level-1 file starting at 1 now ends below the |
| 642 | // new minSnapshotTXID and is expired |
| 643 | for i := 0; i < 3; i++ { |
| 644 | write(t, path, 2) |
| 645 | sync(t, db) |
| 646 | } |
| 647 | if _, err := db.Compact(context.Background(), 1); err != nil { |
| 648 | t.Fatal(err) |
| 649 | } |
| 650 | time.Sleep(20 * time.Millisecond) |
| 651 | if snap2, err = db.Snapshot(context.Background()); err != nil { |
| 652 | t.Fatal(err) |
| 653 | } |
| 654 | if minSnap, err = db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second)); err != nil { |
| 655 | t.Fatal(err) |
| 656 | } |
| 657 | for _, lvl := range []int{1, 0} { |
| 658 | if err := db.EnforceRetentionByTXID(context.Background(), lvl, minSnap); err != nil { |
| 659 | t.Fatal(err) |
| 660 | } |
| 661 | } |
| 662 | closeDB(t, db) |
| 663 | after2 := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| 664 | cont := continuation() |
| 665 | t.Logf("FACT R5-3f: after a second rotation and retention (minSnapshotTXID %d): %s; a continuation from the original snapshot remains: %v", minSnap, after2, cont) |
| 666 | if cont { |
| 667 | t.Fatalf("expected every continuation from the original snapshot [1,%d] to be expired after the second rotation", snap1.MaxTXID) |
| 668 | } |
| 669 | mustRanges(t, client, 9, fmt.Sprintf("[1,%d]", snap2.MaxTXID)) // only the newest snapshot |
| 670 | // restore rooted at the original snapshot: the original's bytes plus the |
| 671 | // current lower levels, without the newer snapshot |
| 672 | rooted := lsfile.NewReplicaClient(filepath.Join(dir, "rooted")) |
| 673 | if _, err := rooted.WriteLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, bytes.NewReader(snap1Bytes)); err != nil { |
| 674 | t.Fatal(err) |
| 675 | } |
| 676 | for _, lvl := range []int{0, 1, 2, 3} { |
| 677 | for _, f := range files(t, client, lvl) { |
| 678 | copyFile(t, client, rooted, f) |
| 679 | } |
| 680 | } |
| 681 | errRooted := restore(t, rooted, filepath.Join(dir, "rooted.db")) |
| 682 | t.Logf("FACT R5-3d: restore rooted at the original snapshot after retention: %v", errRooted) |
| 683 | errNewest := restore(t, client, filepath.Join(dir, "newest.db")) |
| 684 | if errNewest != nil { |
| 685 | t.Fatal(errNewest) |
| 686 | } |
| 687 | t.Logf("FACT R5-3e: restore from the newest snapshot: ok (rows %d)", count(t, filepath.Join(dir, "newest.db"))) |
| 688 | mustCount(t, filepath.Join(dir, "newest.db"), 19) // 1 row, then 3 rounds of 3 syncs of 2 rows |
| 689 | // the rooted restore fails closed, with litestream's own error (R6-2) |
| 690 | want := fmt.Sprintf("non-contiguous ltx files: have up to %016x but next file starts at %016x", uint64(snap1.MaxTXID), 5) |
| 691 | if errRooted == nil || !strings.Contains(errRooted.Error(), want) { |
| 692 | t.Fatalf("expected the rooted restore to fail closed with %q, got %v", want, errRooted) |
| 693 | } |
| 694 | } |
| 695 | |
| 696 | // TestFact_R6_5_AnchorReplayBytes: the anchor's bytes depend on the header |
| 697 | // timestamp. Encoded twice with the same inputs the bytes are identical; |
| 698 | // with another timestamp they differ, so a builder that replays the anchor |
| 699 | // PUT (a restarted install) must persist the timestamp it used, or reuse |
| 700 | // the object it already wrote (moves spec §5.1 step 2, §7.6 building). |
| 701 | func TestFact_R6_5_AnchorReplayBytes(t *testing.T) { |
| 702 | src, snap := source(t, 5) |
| 703 | ts := time.Now().UnixMilli() |
| 704 | a := encodeAnchor(t, src, snap, ts) |
| 705 | b := encodeAnchor(t, src, snap, ts) |
| 706 | c := encodeAnchor(t, src, snap, ts+1) |
| 707 | t.Logf("FACT R6-5: anchor [%d,%d], %d bytes; same timestamp: identical=%v; timestamp+1 ms: identical=%v", snap.MaxTXID, snap.MaxTXID, len(a), bytes.Equal(a, b), bytes.Equal(a, c)) |
| 708 | if !bytes.Equal(a, b) { |
| 709 | t.Fatalf("the same inputs produced different anchor bytes") |
| 710 | } |
| 711 | if bytes.Equal(a, c) { |
| 712 | t.Fatalf("a different timestamp produced identical bytes: the premise of persisting it is gone, review the contract") |
| 713 | } |
| 714 | // the replayed object is accepted by the decoder and equals the first |
| 715 | dec := ltx.NewDecoder(bytes.NewReader(b)) |
| 716 | if err := dec.Verify(); err != nil { |
| 717 | t.Fatal(err) |
| 718 | } |
| 719 | } |
| 720 | |
| 721 | // TestGate_AnchoredLineageThroughStore: one anchored lineage through the |
| 722 | // production Store configuration (internal/lease Held.Replicate: a DB with |
| 723 | // its own metadata directory, a replica client, litestream.NewStore with |
| 724 | // the compaction ladder, the L0 retention monitor), with every interval |
| 725 | // shortened from hours to fractions of a second so the monitors run inside |
| 726 | // the test: repeated writes with distinct bodies, updates and deletes, |
| 727 | // level-1 to level-3 compaction, snapshot rotation and retention, |
| 728 | // time-based level-0 expiry, then disk loss and a cold restore, three |
| 729 | // times over. Exact positions (every sync advances the position by one, |
| 730 | // and a reopened box continues at exactly the newest remote level-0 TXID |
| 731 | // plus one) and exact row contents are asserted; each maintenance event |
| 732 | // is awaited as a condition, not assumed after a sleep; the last round |
| 733 | // forces a restore that consumes compacted continuation files (R7-5). |
| 734 | func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| 735 | src, snap := source(t, 5) |
| 736 | dir := t.TempDir() |
| 737 | gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| 738 | copyFile(t, src, gen, snap) |
| 739 | anchorL0(t, src, snap, gen) |
| 740 | path := filepath.Join(dir, "db") |
| 741 | if err := restore(t, gen, path); err != nil { |
| 742 | t.Fatal(err) |
| 743 | } |
| 744 | want := rows(t, path) // the source's five rows, distinct bodies |
| 745 | if len(want) != 5 { |
| 746 | t.Fatalf("restored %d rows, want 5", len(want)) |
| 747 | } |
| 748 | pos := int(snap.MaxTXID) |
| 749 | next := 6 // the next row id to insert |
| 750 | // internal/lease's ladder is L1 1h, L2 6h, L3 24h, the snapshot interval |
| 751 | // 24h, L0 retention 5 min checked hourly, the DB monitor 10 s: here each |
| 752 | // is shortened by the same shape |
| 753 | levels := litestream.CompactionLevels{ |
| 754 | {Level: 0, Interval: 0}, |
| 755 | {Level: 1, Interval: 150 * time.Millisecond}, |
| 756 | {Level: 2, Interval: 400 * time.Millisecond}, |
| 757 | {Level: 3, Interval: 900 * time.Millisecond}, |
| 758 | } |
| 759 | ctx := context.Background() |
| 760 | maxOf := func(level int) int { |
| 761 | fs := files(t, gen, level) |
| 762 | if len(fs) == 0 { |
| 763 | return 0 |
| 764 | } |
| 765 | return int(fs[len(fs)-1].MaxTXID) |
| 766 | } |
| 767 | for round := 0; round < 3; round++ { |
| 768 | meta := filepath.Join(dir, fmt.Sprintf("meta%d", round)) |
| 769 | db := litestream.NewDB(path) |
| 770 | db.SetMetaPath(meta) |
| 771 | db.MonitorInterval = 50 * time.Millisecond |
| 772 | db.Replica = litestream.NewReplicaWithClient(db, gen) |
| 773 | db.Replica.SyncInterval = 50 * time.Millisecond |
| 774 | store := litestream.NewStore([]*litestream.DB{db}, levels) |
| 775 | store.SnapshotInterval = 300 * time.Millisecond |
| 776 | store.SnapshotRetention = 400 * time.Millisecond |
| 777 | store.SetL0Retention(200 * time.Millisecond) // NewStore copied the default into the DB; SetL0Retention propagates |
| 778 | store.L0RetentionCheckInterval = 50 * time.Millisecond |
| 779 | if err := store.Open(ctx); err != nil { |
| 780 | t.Fatalf("round %d: open store: %v", round, err) |
| 781 | } |
| 782 | snapsBefore := maxOf(9) |
| 783 | l0Before := len(files(t, gen, 0)) |
| 784 | // eight transactions: inserts of distinct bodies, an update and a |
| 785 | // delete among them, each advancing the position by exactly one |
| 786 | for i := 0; i < 8; i++ { |
| 787 | switch { |
| 788 | case i == 3: |
| 789 | id := next - 2 |
| 790 | exec(t, path, `UPDATE t SET body = ? WHERE id = ?`, fmt.Sprintf("r%d-updated-in-round-%d", id, round), id) |
| 791 | want[id] = fmt.Sprintf("r%d-updated-in-round-%d", id, round) |
| 792 | case i == 6: |
| 793 | id := next - 4 |
| 794 | exec(t, path, `DELETE FROM t WHERE id = ?`, id) |
| 795 | delete(want, id) |
| 796 | default: |
| 797 | exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-round-%d", next, round)) |
| 798 | want[next] = fmt.Sprintf("r%d-round-%d", next, round) |
| 799 | next++ |
| 800 | } |
| 801 | sync(t, db) |
| 802 | pos++ |
| 803 | p, err := db.Pos() |
| 804 | if err != nil { |
| 805 | t.Fatal(err) |
| 806 | } |
| 807 | if int(p.TXID) != pos { |
| 808 | t.Fatalf("round %d write %d: position %d, want exactly %d (a reopened box continues at the newest remote level-0 TXID plus one)", round, i, p.TXID, pos) |
| 809 | } |
| 810 | time.Sleep(120 * time.Millisecond) // the monitors run between writes |
| 811 | } |
| 812 | // the maintenance events, each as a condition: a snapshot rotated |
| 813 | // past the previous newest, level 1, 2 and 3 files exist, the older |
| 814 | // snapshot was removed by retention, and level-0 expiry removed |
| 815 | // compacted files while keeping the newest |
| 816 | waitFor(t, "a snapshot rotation", 5*time.Second, func() bool { return maxOf(9) > snapsBefore }) |
| 817 | for _, lvl := range []int{1, 2, 3} { |
| 818 | waitFor(t, fmt.Sprintf("a level-%d compaction", lvl), 5*time.Second, func() bool { return len(files(t, gen, lvl)) > 0 }) |
| 819 | } |
| 820 | // Retention is awaited at the state that lasts: one snapshot, at the |
| 821 | // round's final position. One snapshot alone can be a moment in |
| 822 | // passing. The monitor snapshots every interval and keeps the previous |
| 823 | // snapshot while it is younger than the retention, so a tick that runs |
| 824 | // 100 ms late removes the previous one and leaves a single snapshot |
| 825 | // that is behind the position; the next tick adds the final snapshot |
| 826 | // beside it, and a round that had stopped waiting closed the store |
| 827 | // with two (#266). Once the only snapshot is at the final position a |
| 828 | // later tick can only rewrite that same file. |
| 829 | waitFor(t, "snapshot retention", 5*time.Second, func() bool { |
| 830 | l9 := files(t, gen, 9) |
| 831 | return len(l9) == 1 && int(l9[0].MaxTXID) == pos |
| 832 | }) |
| 833 | waitFor(t, "level-0 expiry", 5*time.Second, func() bool { return len(files(t, gen, 0)) < l0Before+8 }) |
| 834 | if err := store.Close(ctx); err != nil { |
| 835 | t.Fatal(err) |
| 836 | } |
| 837 | l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9) |
| 838 | t.Logf("GATE round %d: position %d, rows %d; L0 %s (%d files, %d before); L1 %s; L2 %s; L3 %s; L9 %s", round, pos, len(want), ranges(l0), len(l0), l0Before, ranges(l1), ranges(files(t, gen, 2)), ranges(files(t, gen, 3)), ranges(l9)) |
| 839 | if len(l0) == 0 || int(l0[len(l0)-1].MaxTXID) != pos { |
| 840 | t.Fatalf("round %d: newest remote level-0 file is %s, want MaxTXID exactly %d", round, ranges(l0), pos) |
| 841 | } |
| 842 | if len(l9) != 1 || int(l9[0].MaxTXID) <= snapsBefore { |
| 843 | t.Fatalf("round %d: snapshot retention did not leave exactly the newest snapshot: L9 %s", round, ranges(l9)) |
| 844 | } |
| 845 | // disk loss, cold restore: exact contents, row by row |
| 846 | wipe(t, path, meta) |
| 847 | if err := restore(t, gen, path); err != nil { |
| 848 | t.Fatalf("round %d: cold restore: %v", round, err) |
| 849 | } |
| 850 | mustRows(t, path, want) |
| 851 | } |
| 852 | // a restore that must consume compacted continuation files: with the |
| 853 | // monitors stopped, more writes are synced, compacted into level 1, and |
| 854 | // every level-0 file but the newest is expired by time; the newest |
| 855 | // snapshot is behind them, so the planner has to take the level-1 file |
| 856 | meta := filepath.Join(dir, "meta-tail") |
| 857 | db := openDB(t, path, meta, gen) |
| 858 | for i := 0; i < 3; i++ { |
| 859 | exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-tail", next)) |
| 860 | want[next] = fmt.Sprintf("r%d-tail", next) |
| 861 | next++ |
| 862 | sync(t, db) |
| 863 | pos++ |
| 864 | } |
| 865 | exec(t, path, `DELETE FROM t WHERE id = ?`, next-1) |
| 866 | delete(want, next-1) |
| 867 | sync(t, db) |
| 868 | pos++ |
| 869 | if _, err := db.Compact(ctx, 1); err != nil { |
| 870 | t.Fatal(err) |
| 871 | } |
| 872 | db.L0Retention = time.Millisecond |
| 873 | time.Sleep(20 * time.Millisecond) |
| 874 | if err := db.EnforceL0RetentionByTime(ctx); err != nil { |
| 875 | t.Fatal(err) |
| 876 | } |
| 877 | closeDB(t, db) |
| 878 | l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9) |
| 879 | t.Logf("GATE tail: position %d; L0 %s; L1 %s; L9 %s", pos, ranges(l0), ranges(l1), ranges(l9)) |
| 880 | if len(l0) != 1 || int(l0[0].MaxTXID) != pos { |
| 881 | t.Fatalf("tail: level 0 should hold the newest file alone, have %s", ranges(l0)) |
| 882 | } |
| 883 | if maxOf(9) >= pos || maxOf(1) != pos { |
| 884 | t.Fatalf("tail: the newest snapshot (%d) must be behind the level-1 file ending at %d", maxOf(9), maxOf(1)) |
| 885 | } |
| 886 | wipe(t, path, meta) |
| 887 | if err := restore(t, gen, path); err != nil { |
| 888 | t.Fatalf("tail: cold restore: %v", err) |
| 889 | } |
| 890 | mustRows(t, path, want) // the rows past the snapshot came through level 1 |
| 891 | } |
| 892 | |
| 893 | // TestFact_R7_1_UnconditionalPut: the pinned S3 client's PutObject carries |
| 894 | // no If-Match or If-None-Match, and its key is prefix, level and TXID range |
| 895 | // only, so two writers of one generation write the same key with |
| 896 | // unconditional PUTs: the R7-1 hazard (claim rev 6.6 §6.4a). Read from |
| 897 | // the pinned source so an upgrade that changes either fails here. |
| 898 | func TestFact_R7_1_UnconditionalPut(t *testing.T) { |
| 899 | pkg, err := build.Default.Import("github.com/benbjohnson/litestream/s3", "", build.FindOnly) |
| 900 | if err != nil { |
| 901 | t.Skipf("pinned s3 client source not found: %v", err) |
| 902 | } |
| 903 | src, err := os.ReadFile(filepath.Join(pkg.Dir, "replica_client.go")) |
| 904 | if err != nil { |
| 905 | t.Fatal(err) |
| 906 | } |
| 907 | text := string(src) |
| 908 | i := strings.Index(text, "s3.PutObjectInput{") |
| 909 | if i < 0 { |
| 910 | t.Fatalf("no PutObjectInput in the pinned client") |
| 911 | } |
| 912 | block := text[i:] |
| 913 | block = block[:strings.Index(block, "}")] |
| 914 | t.Logf("FACT R7-1a: the pinned PutObjectInput fields: %s", strings.Join(strings.Fields(strings.ReplaceAll(block, "\n", " ")), " ")) |
| 915 | for _, cond := range []string{"IfNoneMatch", "IfMatch"} { |
| 916 | if strings.Contains(block, cond) { |
| 917 | t.Fatalf("the pinned client's PutObject is conditional (%s): the writer-epoch premise changed, review the contract", cond) |
| 918 | } |
| 919 | } |
| 920 | if !strings.Contains(text, `fmt.Sprintf("%04x/%s", level, filename)`) || !strings.Contains(text, "ltx.FormatFilename(minTXID, maxTXID)") { |
| 921 | t.Fatalf("the pinned client's key is no longer prefix/level/TXID range") |
| 922 | } |
| 923 | t.Logf("FACT R7-1b: the key is c.Path + \"/\" + %%04x level + ltx.FormatFilename(min, max): two writers of one generation share it") |
| 924 | } |
| 925 | |
| 926 | // TestFact_R8_3_EmptyBootstrap: a created incarnation's first wake has |
| 927 | // neither a database file nor a replica (R8-3). The pinned encoder refuses |
| 928 | // the files a position-0 build would carry, a snapshot [1,0] and an anchor |
| 929 | // [0,0], so an empty generation has no LTX file at all: litestream opened |
| 930 | // on an absent path against an empty replica syncs nothing, a cold restore |
| 931 | // of the empty generation has nothing to restore, and the first |
| 932 | // transaction is the first upload, which a cold restore then holds. |
| 933 | func TestFact_R8_3_EmptyBootstrap(t *testing.T) { |
| 934 | for _, h := range []ltx.Header{ |
| 935 | {Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 1, MaxTXID: 0, Timestamp: 1, WALOffset: litestream.WALHeaderSize}, |
| 936 | {Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 0, MaxTXID: 0, Timestamp: 1}, |
| 937 | } { |
| 938 | var buf bytes.Buffer |
| 939 | enc, err := ltx.NewEncoder(&buf) |
| 940 | if err != nil { |
| 941 | t.Fatal(err) |
| 942 | } |
| 943 | err = enc.EncodeHeader(h) |
| 944 | t.Logf("FACT R8-3a: encoding an LTX header [%d,%d]: %v", h.MinTXID, h.MaxTXID, err) |
| 945 | if err == nil { |
| 946 | t.Fatalf("the encoder accepted a zero TXID [%d,%d]; the spec's empty-bootstrap rule rests on it refusing", h.MinTXID, h.MaxTXID) |
| 947 | } |
| 948 | } |
| 949 | dir := t.TempDir() |
| 950 | gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| 951 | path := filepath.Join(dir, "new.db") |
| 952 | meta := filepath.Join(dir, "meta") |
| 953 | if _, err := os.Stat(path); !os.IsNotExist(err) { |
| 954 | t.Fatalf("the database path exists before the first wake: %v", err) |
| 955 | } |
| 956 | if err := restore(t, gen, filepath.Join(dir, "nothing.db")); err == nil { |
| 957 | t.Fatalf("a restore from an empty replica succeeded; the empty generation must have nothing to restore") |
| 958 | } else { |
| 959 | t.Logf("FACT R8-3b: a cold restore of an empty generation: %v", err) |
| 960 | } |
| 961 | db := openDB(t, path, meta, gen) |
| 962 | err := db.Sync(context.Background()) |
| 963 | t.Logf("FACT R8-3c: litestream opened on an absent database against an empty replica; first sync: %v; level 0 %s, level 9 %s", err, ranges(files(t, gen, 0)), ranges(files(t, gen, 9))) |
| 964 | if err != nil { |
| 965 | t.Fatalf("the first sync of an absent database failed: %v", err) |
| 966 | } |
| 967 | mustRanges(t, gen, 0, "none") |
| 968 | mustRanges(t, gen, 9, "none") |
| 969 | write(t, path, 1) // CREATE TABLE, then one row: two SQLite transactions, one sync |
| 970 | sync(t, db) |
| 971 | l0 := ranges(files(t, gen, 0)) |
| 972 | t.Logf("FACT R8-3d: after the first transactions and one sync the replica's level 0 is %s, level 9 %s (a TXID is one sync's WAL frames, not one SQLite transaction)", l0, ranges(files(t, gen, 9))) |
| 973 | if l0 != "[1,1]" { |
| 974 | t.Fatalf("expected the first upload to be level 0 [1,1], got %s", l0) |
| 975 | } |
| 976 | closeDB(t, db) |
| 977 | out := filepath.Join(dir, "out.db") |
| 978 | if err := restore(t, gen, out); err != nil { |
| 979 | t.Fatalf("restore after the first transactions: %v", err) |
| 980 | } |
| 981 | mustCount(t, out, 1) |
| 982 | t.Logf("FACT R8-3e: a cold restore after the first transactions holds the row; the first transaction was the first upload") |
| 983 | } |
| 984 | |
amadan