db: checkpoint the engine, and open from it

`Engine.checkpoint()` publishes the current state: commit first, then snapshot
the catalog under the catalog lock, then validate the snapshot against an
unchanged `seq` under the log lock before publishing -- the same bounded-retry
shape compaction has always used. The crash-recovery invariant (PLAN D6) reduces
to that ordering, and it is asserted:
`snapshot_seq <= committed_seq`.

The catalog holds what the pages cannot say for themselves: db and collection
names, slab extents and tails, and for each index its spec, tree position,
overflow extents and id->page table. Written wholesale into fresh pages each
time, never mutated in place, so the previous copy stays valid under the previous
watermark until the new one switches over -- untearable by construction, which is
why there is no incremental update path. Every read is bounds-checked, because
the bytes come off disk and a scrambled catalog must produce an error the caller
can fall back from.

`Log.replay` takes a `from_seq` and skips below it before the BSON parse. The walk
still visits every block, because that is what leaves `end_pos` correct for the
next append; making opens *fast* is the job of truncating the log, next.

A failed catalog load warns, discards what it loaded, and replays the log in
full. The log is untouched at this commit, so that fallback is real rather than
aspirational -- which is the reason to land this before truncation.

--

The docs hashmap is deliberately *not* in the catalog. It is still the
authoritative _id lookup, but putting it there means writing a format the commit
that drops it would only delete again; it is rebuilt by walking the `_id_` tree,
which the data file already holds.

--

One real bug, and it is the interesting part. Replay does not maintain index
entries -- it puts documents in place and lets `build_all_indexes` bulk-pack
afterwards, which is O(n log n) once rather than per record. After a checkpoint
that is wrong: the indexes arrive already populated, `rebuild_index` skips a
non-empty one by design, and every record replayed on top was invisible to every
index. The symptom was a document present in the collection and absent from
`_id_` -- which, once the hashmap goes, means simply absent. Replay now maintains
entries when it opened from a checkpoint, and keeps the bulk path for a full one.

`Engine.seq` is restored, which it never was: it restarted at 0 on every open.
The mutation for it is *not* covered and the test says so rather than implying
otherwise -- the sequence is seeded from the watermark, so it only drifts by the
records replayed on top, and the catalog carries those same records in every
sequence a unit test can reach. Observing the drift needs a crash between a
duplicate-sequence append and the checkpoint that would have captured it. The
line stays because a log without monotonic sequences has no total order.

src/db.zig

@@ -231,6 +231,16 @@ pub const Engine = struct {
     /// rewrite is worth doing — see `note_compact`.
     live_docs: u64 = 0,
     dead_docs: u64 = 0,
+    /// Whether replay must maintain index entries as it goes.
+    ///
+    /// A full replay does not: it puts documents in place and lets
+    /// `build_all_indexes` bulk-pack every index afterwards, which is O(n log n)
+    /// once instead of per record. After a checkpoint that is wrong -- the
+    /// indexes arrive already populated, `rebuild_index` skips a non-empty one by
+    /// design, and the records replayed on top would be invisible to every
+    /// index. The symptom was a document present in the collection and missing
+    /// from `_id_`, which after the hashmap goes away means simply missing.
+    replay_maintains_indexes: bool = false,
     /// Set to the failing index's own stable name when an upsert is
     /// rejected by a unique secondary index (error.DuplicateKeyIndex). The
     /// command reads it while still holding the write lock.
@@ -240,16 +250,10 @@ pub const Engine = struct {
         var log = try storage.Log.open(gpa, io, path);
         errdefer log.close();
 
-        // The data file sits beside the log. It is recreated empty on every
+        // The data file sits beside the log and is *kept*: a valid watermark in
-        // open for now: the log is still replayed in full, so nothing durable
+        // it means most of the log never has to be replayed.
-        // depends on the file yet, and open/close semantics stay exactly what
-        // they were. The watermark that makes it a checkpoint comes later.
         const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{log.path});
         defer gpa.free(data_path);
-        std.Io.Dir.cwd().deleteFile(io, data_path) catch |err| switch (err) {
-            error.FileNotFound => {},
-            else => return err,
-        };
 
         const pager_box = try gpa.create(pgr.Pager);
         errdefer gpa.destroy(pager_box);
@@ -270,10 +274,40 @@ pub const Engine = struct {
             engine.dbs.deinit(gpa);
         }
 
-        try engine.log.replay(&engine, apply_record);
+        // A checkpoint, if the data file has one, decides where replay starts.
+        // Nothing below the watermark needs re-applying: the data file already
+        // holds its effect.
+        var replay_from: u64 = 0;
+        if (engine.pager.loaded.generation != 0) {
+            engine.read_catalog() catch |err| {
+                // The image is unusable but the log is not. Warn, drop
+                // everything loaded, and fall back to a full replay -- the
+                // database must always open.
+                std.debug.print(
+                    "multiforadb: WARNING: data file catalog unreadable ({s}); " ++
+                        "replaying the log in full\n",
+                    .{@errorName(err)},
+                );
+                engine.reset_after_failed_catalog();
+                replay_from = 0;
+            };
+            if (replay_from == 0 and engine.dbs.count() > 0) {
+                replay_from = engine.pager.loaded.seq;
+                engine.replay_maintains_indexes = true;
+                engine.seq = replay_from;
+                engine.committed_seq = replay_from;
+                engine.live_docs = engine.pager.loaded.live_docs;
+            }
+        }
+
+        try engine.log.replay(&engine, apply_record, replay_from);
         // Replay registers empty indexes; build them from the live docs
-        // once replay completes (order-independent).
+        // once replay completes (order-independent). A checkpointed open finds
+        // them already populated, and the guard in rebuild_index skips them.
         try engine.build_all_indexes();
+        // Everything replayed is durable by definition -- it was read back off
+        // the log -- so the commit watermark starts level with the sequence.
+        engine.committed_seq = engine.seq;
         return engine;
     }
 
@@ -1257,6 +1291,318 @@ pub const Engine = struct {
         }
     }
 
+    // -- checkpoint ---------------------------------------------------------
+
+    /// The catalog: everything about where the engine's structures live that is
+    /// not recoverable by looking at the pages themselves.
+    ///
+    /// Written wholesale into freshly allocated pages at every checkpoint, never
+    /// mutated in place, so the previous copy stays intact and referenced by the
+    /// previous watermark until the new one switches over. That is what makes it
+    /// untearable, and it is why there is no incremental catalog update path.
+    ///
+    /// Format (little-endian throughout):
+    ///   u32 magic "MFCT", u32 version, u64 live_docs, u32 db_count
+    ///   per db:    u32 name_len, name, u32 coll_count
+    ///   per coll:  u32 name_len, name, u64 slab_tail, u64 slab_end,
+    ///              u32 extent_count, (u32 first, u32 pages)*, u32 index_count
+    ///              index 0 is always the implicit _id_
+    ///   per index: u32 name_len, name, u32 key_count,
+    ///              (u32 path_len, path, u8 descending)*,
+    ///              u8 flags(unique|sparse|multikey|has_ttl), i64 ttl,
+    ///              u32 root, u32 first_leaf, u32 leaf_count, u32 depth,
+    ///              u64 entry_count, u64 ovf_tail, u64 ovf_end,
+    ///              u32 ovf_extent_count, (u32 first, u32 pages)*,
+    ///              u32 node_count, (u32 page)*
+    ///   u64 xxhash3 over everything above
+    const catalog_magic: u32 = 0x4D464354; // "MFCT"
+    const catalog_version: u32 = 1;
+
+    fn write_catalog(self: *Engine, out: *std.ArrayListUnmanaged(u8)) !void {
+        const gpa = self.gpa;
+        try put_u32(gpa, out, catalog_magic);
+        try put_u32(gpa, out, catalog_version);
+        try put_u64(gpa, out, self.live_docs);
+        try put_u32(gpa, out, @intCast(self.dbs.count()));
+        var db_it = self.dbs.iterator();
+        while (db_it.next()) |db_entry| {
+            try put_bytes(gpa, out, db_entry.key_ptr.*);
+            const colls = &db_entry.value_ptr.collections;
+            try put_u32(gpa, out, @intCast(colls.count()));
+            var coll_it = colls.iterator();
+            while (coll_it.next()) |ce| {
+                const coll = ce.value_ptr.*;
+                try put_bytes(gpa, out, ce.key_ptr.*);
+                try put_u64(gpa, out, coll.slab_tail);
+                try put_u64(gpa, out, coll.slab_end);
+                try put_u32(gpa, out, @intCast(coll.slab_extents.items.len));
+                for (coll.slab_extents.items) |e| {
+                    try put_u32(gpa, out, e.first);
+                    try put_u32(gpa, out, e.pages);
+                }
+                try put_u32(gpa, out, @intCast(coll.indexes.items.len + 1));
+                try write_index_catalog(gpa, out, &coll.id_index);
+                for (coll.indexes.items) |ix| try write_index_catalog(gpa, out, ix);
+            }
+        }
+        try put_u64(gpa, out, std.hash.XxHash3.hash(0, out.items));
+    }
+
+    fn write_index_catalog(
+        gpa: std.mem.Allocator,
+        out: *std.ArrayListUnmanaged(u8),
+        ix: *const index.Index,
+    ) !void {
+        try put_bytes(gpa, out, ix.name);
+        try put_u32(gpa, out, @intCast(ix.keys.len));
+        for (ix.keys) |k| {
+            try put_bytes(gpa, out, k.path);
+            try out.append(gpa, @intFromBool(k.descending));
+        }
+        var flags: u8 = 0;
+        if (ix.unique) flags |= 1;
+        if (ix.sparse) flags |= 2;
+        if (ix.multikey) flags |= 4;
+        if (ix.ttl != null) flags |= 8;
+        try out.append(gpa, flags);
+        try put_u64(gpa, out, @bitCast(ix.ttl orelse 0));
+        try put_u32(gpa, out, ix.root);
+        try put_u32(gpa, out, ix.first_leaf);
+        try put_u32(gpa, out, ix.leaf_count);
+        try put_u32(gpa, out, ix.depth);
+        try put_u64(gpa, out, ix.entry_count);
+        try put_u64(gpa, out, ix.ovf_tail);
+        try put_u64(gpa, out, ix.ovf_end);
+        try put_u32(gpa, out, @intCast(ix.ovf_extents.items.len));
+        for (ix.ovf_extents.items) |e| {
+            try put_u32(gpa, out, e.first);
+            try put_u32(gpa, out, e.pages);
+        }
+        try put_u32(gpa, out, @intCast(ix.node_pages.items.len));
+        for (ix.node_pages.items) |pg| try put_u32(gpa, out, pg);
+    }
+
+    /// Rebuild the catalog from the data file. On any inconsistency this returns
+    /// an error and the caller falls back to a full replay.
+    fn read_catalog(self: *Engine) !void {
+        const wm = self.pager.loaded;
+        if (wm.catalog_len == 0) return error.NoCatalog;
+        const buf = self.pager.bytes(@as(u64, wm.catalog_page) << pgr.page_shift, @intCast(wm.catalog_len));
+        if (buf.len < 8) return error.CorruptCatalog;
+        const body = buf[0 .. buf.len - 8];
+        if (std.hash.XxHash3.hash(0, body) != std.mem.readInt(u64, buf[buf.len - 8 ..][0..8], .little)) {
+            return error.CorruptCatalog;
+        }
+        var r: Reader = .{ .b = body };
+        if (try r.read_u32() != catalog_magic) return error.CorruptCatalog;
+        if (try r.read_u32() != catalog_version) return error.CorruptCatalog;
+        self.live_docs = try r.read_u64();
+        const ndbs = try r.read_u32();
+        var d: u32 = 0;
+        while (d < ndbs) : (d += 1) {
+            const db_name = try r.read_bytes();
+            const ncolls = try r.read_u32();
+            var c: u32 = 0;
+            while (c < ncolls) : (c += 1) {
+                const coll_name = try r.read_bytes();
+                const coll = try self.get_or_create_collection(db_name, coll_name);
+                coll.slab_tail = try r.read_u64();
+                coll.slab_end = try r.read_u64();
+                const nex = try r.read_u32();
+                var e: u32 = 0;
+                while (e < nex) : (e += 1) {
+                    const first = try r.read_u32();
+                    const pages = try r.read_u32();
+                    try coll.slab_extents.append(self.gpa, .{ .first = first, .pages = pages });
+                }
+                const nix = try r.read_u32();
+                // Index 0 is the implicit _id_, already created by
+                // get_or_create_collection; the rest are registered here.
+                try read_index_catalog(self.gpa, &r, &coll.id_index);
+                var i: u32 = 1;
+                while (i < nix) : (i += 1) {
+                    const boxed = try self.gpa.create(index.Index);
+                    errdefer self.gpa.destroy(boxed);
+                    boxed.* = try index.Index.init(self.gpa, self.pager, "", &.{}, false, false, null);
+                    read_index_catalog(self.gpa, &r, boxed) catch |err| {
+                        boxed.deinit(self.gpa);
+                        self.gpa.destroy(boxed);
+                        return err;
+                    };
+                    try coll.indexes.append(self.gpa, boxed);
+                }
+                // The docs map is still the authoritative _id -> offset lookup,
+                // and it is not in the catalog on purpose: the commit that drops
+                // it would only have to delete the format again. Rebuilt from the
+                // _id_ tree instead, which the data file already holds.
+                try self.rebuild_docs_map(coll);
+            }
+        }
+    }
+
+    /// Replace an index's identity and tree position from the catalog. The index
+    /// arrives freshly initialised, so its own two starting pages are discarded
+    /// in favour of what was published.
+    fn read_index_catalog(
+        gpa: std.mem.Allocator,
+        r: *Reader,
+        ix: *index.Index,
+    ) !void {
+        const name = try r.read_bytes();
+        const nkeys = try r.read_u32();
+        if (nkeys == 0 or nkeys > index.max_index_keys) return error.CorruptCatalog;
+        var keys = try gpa.alloc(index.IndexKey, nkeys);
+        var built: usize = 0;
+        errdefer {
+            for (keys[0..built]) |k| gpa.free(k.path);
+            gpa.free(keys);
+        }
+        while (built < nkeys) : (built += 1) {
+            const path = try r.read_bytes();
+            keys[built] = .{ .path = try gpa.dupe(u8, path), .descending = (try r.read_byte()) != 0 };
+        }
+        const flags = try r.read_byte();
+        const ttl_raw: i64 = @bitCast(try r.read_u64());
+
+        const new_name = try gpa.dupe(u8, name);
+        errdefer gpa.free(new_name);
+
+        // Swap in the published identity, freeing what init made.
+        for (ix.keys) |k| gpa.free(k.path);
+        gpa.free(ix.keys);
+        gpa.free(ix.name);
+        ix.name = new_name;
+        ix.keys = keys;
+        ix.unique = flags & 1 != 0;
+        ix.sparse = flags & 2 != 0;
+        ix.multikey = flags & 4 != 0;
+        ix.ttl = if (flags & 8 != 0) ttl_raw else null;
+        ix.root = try r.read_u32();
+        ix.first_leaf = try r.read_u32();
+        ix.leaf_count = try r.read_u32();
+        ix.depth = try r.read_u32();
+        ix.entry_count = @intCast(try r.read_u64());
+        ix.ovf_tail = try r.read_u64();
+        ix.ovf_end = try r.read_u64();
+        const novf = try r.read_u32();
+        var o: u32 = 0;
+        while (o < novf) : (o += 1) {
+            const first = try r.read_u32();
+            const pages = try r.read_u32();
+            try ix.ovf_extents.append(gpa, .{ .first = first, .pages = pages });
+        }
+        const nnodes = try r.read_u32();
+        if (nnodes < 2) return error.CorruptCatalog;
+        ix.node_pages.clearRetainingCapacity();
+        try ix.node_pages.ensureTotalCapacity(gpa, nnodes);
+        var n: u32 = 0;
+        while (n < nnodes) : (n += 1) ix.node_pages.appendAssumeCapacity(try r.read_u32());
+    }
+
+    /// Add a replayed document's entries to every index that is already
+    /// populated. Best effort and infallible: replay must not refuse to start,
+    /// and an index that cannot key this document is reported and left alone --
+    /// the same tolerance `rebuild_index` has always had.
+    fn index_doc_on_replay(self: *Engine, coll: *Collection, doc_bytes: []const u8, off: u64) void {
+        self.index_one(&coll.id_index, doc_bytes, off);
+        for (coll.indexes.items) |ix| self.index_one(ix, doc_bytes, off);
+    }
+
+    fn index_one(self: *Engine, ix: *index.Index, doc_bytes: []const u8, off: u64) void {
+        var built = ix.build_entries(self.gpa, doc_bytes) catch return;
+        defer built.deinit(self.gpa);
+        if (built.multikey) ix.multikey = true;
+        ix.reserve_for(self.gpa, built.entries.items) catch return;
+        ix.insert_entries(&built, off);
+    }
+
+    /// Rebuild the _id -> offset hashmap by walking the _id_ tree.
+    fn rebuild_docs_map(self: *Engine, coll: *Collection) !void {
+        var arena = std.heap.ArenaAllocator.init(self.gpa);
+        defer arena.deinit();
+        var it = coll.id_index.iter();
+        while (it.next()) |e| {
+            _ = arena.reset(.retain_capacity);
+            const id_value = (try bson.get_at(arena.allocator(), coll.doc_bytes(e.off), "_id")) orelse continue;
+            const id_key = try bson.serialize_value(self.gpa, id_value);
+            errdefer self.gpa.free(id_key);
+            try coll.docs.put(self.gpa, id_key, e.off);
+        }
+    }
+
+    /// Discard everything a failed catalog load put in place, so the caller can
+    /// replay the log into a clean engine.
+    fn reset_after_failed_catalog(self: *Engine) void {
+        var db_it = self.dbs.iterator();
+        while (db_it.next()) |db_entry| {
+            self.free_db(db_entry.value_ptr);
+            self.gpa.free(db_entry.key_ptr.*);
+        }
+        self.dbs.clearRetainingCapacity();
+        self.live_docs = 0;
+        self.dead_docs = 0;
+        self.seq = 0;
+        self.committed_seq = 0;
+    }
+
+    /// Publish the current state as a checkpoint.
+    ///
+    /// The watermark equals the sequence the log has already made durable, never
+    /// more: `commit` first, then snapshot, and the snapshot is validated against
+    /// an unchanged `seq` under the log lock -- the same bounded-retry shape
+    /// compaction has always used. That is the crash-recovery invariant (PLAN
+    /// D6) reduced to an ordering.
+    pub fn checkpoint(self: *Engine) !void {
+        try self.commit();
+
+        var buf: std.ArrayListUnmanaged(u8) = .empty;
+        defer buf.deinit(self.gpa);
+
+        var attempt: usize = 0;
+        const attempt_max = 8;
+        while (attempt < attempt_max) : (attempt += 1) {
+            buf.clearRetainingCapacity();
+            try self.catalog_lock.lockShared(self.io);
+            const snapshot_seq = self.seq;
+            self.write_catalog(&buf) catch |err| {
+                self.catalog_lock.unlockShared(self.io);
+                return err;
+            };
+            self.catalog_lock.unlockShared(self.io);
+
+            try self.log_lock.lock(self.io);
+            if (self.seq != snapshot_seq) {
+                // A writer landed mid-snapshot; the catalog describes a state
+                // that no longer matches the log. Retry rather than publish it.
+                self.log_lock.unlock(self.io);
+                continue;
+            }
+            assert_msg(
+                snapshot_seq <= self.committed_seq,
+                "checkpoint watermark past the durable log tail",
+            );
+            const pages: u32 = @intCast((buf.items.len + pgr.page_size - 1) / pgr.page_size);
+            const first = self.pager.alloc_pages(pages) catch |err| {
+                self.log_lock.unlock(self.io);
+                return err;
+            };
+            @memcpy(self.pager.bytes_mut(@as(u64, first) << pgr.page_shift, buf.items.len), buf.items);
+            self.pager.publish(.{
+                .seq = snapshot_seq,
+                .catalog_page = first,
+                .catalog_len = buf.items.len,
+                .live_docs = self.live_docs,
+            }) catch |err| {
+                self.log_lock.unlock(self.io);
+                return err;
+            };
+            self.pager.release_reservation();
+            self.log_lock.unlock(self.io);
+            return;
+        }
+        std.debug.print("multiforadb: WARNING: checkpoint gave up after {d} attempts under sustained writes\n", .{attempt_max});
+    }
+
     /// Register an (empty) index from a persisted spec document. A repeated
     /// create record for the same name is an idempotent no-op.
     fn register_index_from_spec(
@@ -1339,6 +1685,12 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
     var key_owned = false;
     defer if (!key_owned) self.gpa.free(id_key);
 
+    // Engine.seq used to restart at 0 on every open, which was harmless while
+    // the log was always replayed in full and fatal the moment a watermark
+    // exists: the first append after an open would reuse a sequence at or below
+    // it, and the *next* open would discard that record as already-checkpointed.
+    self.seq = @max(self.seq, record.seq);
+
     switch (record.type) {
         storage.record_type_upsert => {
             self.evict_doc(coll, id_key);
@@ -1346,10 +1698,13 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
             defer self.gpa.free(doc_bytes);
             try coll.slab_reserve(self.gpa, doc_bytes.len);
             const off = coll.slab_append(doc_bytes);
-            self.pager.release_reservation();
             try coll.docs.put(self.gpa, id_key, off);
             self.live_docs += 1;
             key_owned = true;
+            if (self.replay_maintains_indexes) {
+                self.index_doc_on_replay(coll, doc_bytes, off);
+            }
+            self.pager.release_reservation();
             // The _id_ entry is added after replay, in build_all_indexes,
             // together with the secondary indexes.
             //
@@ -2056,6 +2411,103 @@ test "index survives reopen and compaction" {
     engine2.unlock();
 }
 
+test "a checkpoint lets the next open skip the log it covers" {
+    // The point of the whole milestone: an open that finds a watermark loads the
+    // data file and replays only what happened after it, instead of rebuilding
+    // everything from the log.
+    //
+    // Mutation checks, red: publishing a watermark seq of 0; and removing the
+    // index maintenance in apply_record, which leaves a replayed document
+    // present in the collection and absent from `_id_` -- which, once the
+    // hashmap goes, means simply absent.
+    //
+    // Not covered, and worth stating rather than implying: removing
+    // `self.seq = @max(self.seq, record.seq)` from apply_record leaves this
+    // green. The sequence is seeded from the watermark on a checkpointed open,
+    // so it only drifts by the records replayed on top -- and every sequence
+    // reachable from here has the catalog carrying those same records, which
+    // masks the drift. Observing it needs a crash between a duplicate-sequence
+    // append and the checkpoint that would have captured it, which is the
+    // crash-injection harness's job, not this test's. The line stays because a
+    // log whose sequences are not monotonic has no total order, and
+    // `committed_seq <= seq` is asserted on every commit.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const io = env.io;
+    const gpa = testing.allocator;
+
+    var tmp = try TmpLog.init(gpa);
+    defer tmp.deinit(gpa);
+    const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{tmp.path});
+    defer gpa.free(data_path);
+    defer std.Io.Dir.cwd().deleteFile(io, data_path) catch {};
+
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        var i: i32 = 0;
+        while (i < 40) : (i += 1) {
+            var d = try make_user(gpa, i, "a@x.io");
+            defer d.deinit();
+            try engine.insert("app", "users", &d, &env.gen);
+        }
+        engine.unlock();
+        try engine.checkpoint();
+        try testing.expect(engine.pager.loaded.generation >= 1);
+        try testing.expectEqual(engine.seq, engine.pager.loaded.seq);
+
+        // Writes after the checkpoint are the ones a reopen must replay.
+        try engine.lock();
+        i = 100;
+        while (i < 105) : (i += 1) {
+            var d = try make_user(gpa, i, "b@x.io");
+            defer d.deinit();
+            try engine.insert("app", "users", &d, &env.gen);
+        }
+        engine.unlock();
+    }
+
+    // Reopen: the checkpoint is loaded, so only the five later records apply.
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        defer engine.unlock();
+        try testing.expect(engine.pager.loaded.generation >= 1);
+        const coll = engine.get_collection("app", "users").?;
+        try testing.expectEqual(@as(usize, 45), coll.docs.count());
+        try testing.expectEqual(@as(usize, 45), coll.id_index.count());
+        // And the sequence continued from the watermark rather than restarting.
+        try testing.expect(engine.seq >= engine.pager.loaded.seq);
+    }
+
+    // A second reopen, to catch a sequence that restarted: the writes made after
+    // the first reopen must survive it.
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        var d = try make_user(gpa, 500, "c@x.io");
+        defer d.deinit();
+        try engine.insert("app", "users", &d, &env.gen);
+        engine.unlock();
+        try engine.checkpoint();
+    }
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        defer engine.unlock();
+        const coll = engine.get_collection("app", "users").?;
+        try testing.expectEqual(@as(usize, 46), coll.docs.count());
+        const id_key = try bson.serialize_value(gpa, .{ .int32 = 500 });
+        defer gpa.free(id_key);
+        try testing.expect(engine.get_doc("app", "users", id_key) != null);
+    }
+}
+
 test "index drop survives reopen" {
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();
@@ -2324,3 +2776,55 @@ test "ttl_sweep spans collections and several TTL indexes on one collection" {
     try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "events").?.docs.count());
     try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.docs.count());
 }
+
+// ---------------------------------------------------------------------------
+// Catalog encoding helpers
+// ---------------------------------------------------------------------------
+
+fn put_u32(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: u32) !void {
+    var b: [4]u8 = undefined;
+    std.mem.writeInt(u32, &b, v, .little);
+    try out.appendSlice(gpa, &b);
+}
+
+fn put_u64(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: u64) !void {
+    var b: [8]u8 = undefined;
+    std.mem.writeInt(u64, &b, v, .little);
+    try out.appendSlice(gpa, &b);
+}
+
+fn put_bytes(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), v: []const u8) !void {
+    try put_u32(gpa, out, @intCast(v.len));
+    try out.appendSlice(gpa, v);
+}
+
+/// A bounds-checked cursor over the catalog. Every read is checked because the
+/// bytes come off disk: a truncated or scrambled catalog must produce an error
+/// the caller can fall back from, never a read past the end.
+const Reader = struct {
+    b: []const u8,
+    at: usize = 0,
+
+    fn take(self: *Reader, n: usize) ![]const u8 {
+        if (self.at + n > self.b.len) return error.CorruptCatalog;
+        defer self.at += n;
+        return self.b[self.at..][0..n];
+    }
+
+    fn read_byte(self: *Reader) !u8 {
+        return (try self.take(1))[0];
+    }
+
+    fn read_u32(self: *Reader) !u32 {
+        return std.mem.readInt(u32, (try self.take(4))[0..4], .little);
+    }
+
+    fn read_u64(self: *Reader) !u64 {
+        return std.mem.readInt(u64, (try self.take(8))[0..8], .little);
+    }
+
+    fn read_bytes(self: *Reader) ![]const u8 {
+        const n = try self.read_u32();
+        return self.take(n);
+    }
+};

src/storage.zig

@@ -219,7 +219,14 @@ pub const Log = struct {
     }
 
     /// Replay all valid records from the beginning of the file.
-    pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn) !void {
+    /// Replay every record with `seq > from_seq`. Zero replays everything, which
+    /// is what an engine with no checkpoint does.
+    ///
+    /// The walk still visits every block regardless, because that is what leaves
+    /// `end_pos` correct for the next append. Skipping is about not *applying*
+    /// records the data file already contains; making opens fast is the job of
+    /// truncating the log once a checkpoint covers it.
+    pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn, from_seq: u64) !void {
         var decomp: std.ArrayListUnmanaged(u8) = .empty;
         defer decomp.deinit(self.gpa);
         var pos: u64 = file_header_len;
@@ -275,7 +282,7 @@ pub const Log = struct {
 
             var idx: usize = 0;
             while (idx < decomp.items.len) {
-                idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback);
+                idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback, from_seq);
             }
             pos += total;
             self.end_pos = pos;
@@ -292,6 +299,7 @@ pub const Log = struct {
         pos: u64,
         ctx: *anyopaque,
         callback: ReplayFn,
+        from_seq: u64,
     ) !usize {
         if (bytes.len < 4) return error.InvalidLog;
         const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
@@ -312,6 +320,9 @@ pub const Log = struct {
 
         var idx: usize = header_len - 4;
         const seq: u64 = std.mem.readInt(u64, payload[8..16], .little);
+        // Below the checkpoint: the data file already holds its effect. Skipped
+        // before the BSON parse, which is the expensive part.
+        if (seq <= from_seq) return total;
         const rtype = payload[16];
         const db = read_cstring(payload, &idx) orelse return error.InvalidLog;
         const coll = read_cstring(payload, &idx) orelse return error.InvalidLog;
@@ -739,7 +750,7 @@ test "append, replay, torn tail" {
         }
     };
     var ctx = Ctx{ .seen = &seen, .gpa = gpa };
-    try log.replay(@ptrCast(&ctx), Ctx.apply);
+    try log.replay(@ptrCast(&ctx), Ctx.apply, 0);
 
     try testing.expectEqualSlices(u8, &[_]u8{ record_type_upsert, 1, record_type_delete, 2 }, seen.items);
 }
@@ -783,7 +794,7 @@ test "record larger than the read chunk replays" {
         }
     };
     var ctx = Ctx{ .count = &count, .gpa = gpa };
-    try log.replay(@ptrCast(&ctx), Ctx.apply);
+    try log.replay(@ptrCast(&ctx), Ctx.apply, 0);
     try testing.expectEqual(@as(usize, 1), count);
 }
 
@@ -832,7 +843,7 @@ test "reject corrupt interior block" {
         }
     };
     var ctx = Ctx{ .count = &count, .gpa = gpa };
-    try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply));
+    try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply, 0));
 }
 
 test "torn tail truncates cleanly and appends overwrite it" {
@@ -878,7 +889,7 @@ test "torn tail truncates cleanly and appends overwrite it" {
         }
     };
     var ctx = Ctx{ .seen = &seen, .gpa = gpa };
-    try log2.replay(@ptrCast(&ctx), Ctx.apply);
+    try log2.replay(@ptrCast(&ctx), Ctx.apply, 0);
     try testing.expectEqualSlices(u8, &[_]u8{1}, seen.items);
 
     // A new append overwrites from the replay end and replays cleanly.
@@ -888,7 +899,7 @@ test "torn tail truncates cleanly and appends overwrite it" {
     var log3 = try Log.open(gpa, io, path);
     defer log3.close();
     ctx.seen = &seen;
-    try log3.replay(@ptrCast(&ctx), Ctx.apply);
+    try log3.replay(@ptrCast(&ctx), Ctx.apply, 0);
     try testing.expectEqualSlices(u8, &[_]u8{ 1, 3 }, seen.items);
 }
 
@@ -938,7 +949,7 @@ test "Log.create discards a leftover file; Log.open keeps it" {
         var kept = try Log.open(gpa, io, path);
         defer kept.close();
         try testing.expect(try kept.file.length(io) > file_header_len);
-        try kept.replay(@ptrCast(&ctx), Ctx.apply);
+        try kept.replay(@ptrCast(&ctx), Ctx.apply, 0);
         try testing.expectEqualSlices(u8, &[_]u8{ 1, 2, 3 }, seen.items);
     }
 
@@ -959,6 +970,6 @@ test "Log.create discards a leftover file; Log.open keeps it" {
     seen.clearRetainingCapacity();
     var reopened = try Log.open(gpa, io, path);
     defer reopened.close();
-    try reopened.replay(@ptrCast(&ctx), Ctx.apply);
+    try reopened.replay(@ptrCast(&ctx), Ctx.apply, 0);
     try testing.expectEqualSlices(u8, &[_]u8{9}, seen.items);
 }