//! In-memory database engine backed by the append-only log. Maps
//! db -> collection -> _id(serialized) -> owned Document. All mutations are
//! logged and synced before they become visible in memory, so a crash never
//! loses a committed write. Callers must hold the write lock (`lock`) around
//! any command that mutates state, and the read lock (`lock_read`) around
//! read-only commands so reads overlap with each other.

const std = @import("std");
const bson = @import("bson.zig");
const storage = @import("storage.zig");
const index = @import("index.zig");
const pgr = @import("pager.zig");
// Always active, including in the default ReleaseFast build -- see assert.zig
// for why std.debug.assert is the wrong tool for these invariants.
const assert = @import("assert.zig").assert;
// For the durability invariants: a panic carries no expression text, so the
// message is all an operator gets.
const assert_msg = @import("assert.zig").assert_msg;

/// Pages in a standard slab extent: 8 MiB, as the old in-memory segments were.
/// Slack is bounded by one extent per collection.
const slab_extent_pages: u32 = (8 * 1024 * 1024) / pgr.page_size;

const LogKind = enum { upsert, delete, index_create, index_drop };

pub const Collection = struct {
    /// Documents live as canonical BSON bytes in the data file, in extents this
    /// collection owns; the map holds each document's offset. Those are
    /// *absolute file offsets* now, which is what makes doc_bytes a single add
    /// rather than a binary search over segment starts -- and what removes the
    /// dangling-pointer hazard the old segment list had, since the mapping's
    /// base never moves.
    ///
    /// Removed documents leave garbage bytes until a rebuild rewrites them. A
    /// checkpoint must never renumber these offsets: every index leaf holds one
    /// (PLAN amendment A3).
    docs: std.StringHashMapUnmanaged(u64),
    /// The data file this collection's documents live in.
    pager: *pgr.Pager,
    /// Extents owned by this collection's slab, in allocation order.
    slab_extents: std.ArrayListUnmanaged(pgr.Extent),
    /// Absolute file offset of the next document write, and the end of the
    /// extent it falls in.
    slab_tail: u64,
    slab_end: u64,
    /// Secondary indexes (persisted through the log). Heap-allocated, so an
    /// `*Index` handed out by `find_index` or `create_index` stays valid when
    /// a sibling index is dropped. Held by value, `orderedRemove` memmoved the
    /// whole ~5 KB struct and every live pointer into the list -- a query
    /// plan's `index` field, or a slice into an index's promoted-key buffer --
    /// silently aimed at a different index or past the end. Nothing exercised
    /// that concurrently yet; the mmap work makes it worse, since an Index
    /// will own a mapping.
    indexes: std.ArrayListUnmanaged(*index.Index),
    /// Guards this collection's docs/slab/indexes. Writers take it
    /// exclusive, readers shared; never held while taking the catalog lock,
    /// and never more than one collection lock at a time.
    lock: std.Io.RwLock = .init,
    /// The secondary index that rejected the most recent unique write
    /// (duplicate-key error path); per-collection so concurrent writers on
    /// other collections cannot clobber it mid-command.
    dup_index: ?[]const u8 = null,
    /// The implicit _id_ index: every document has an _id and it is not
    /// sparse, so entry count equals document count and a full scan of it
    /// cannot miss a document — which is what the sort planner's full-scan
    /// plan relies on. Kept out of `indexes` so the listing/drop commands
    /// and the log format are unchanged (it is rebuilt on open like
    /// everything else). `bson.encode_key` keys are canonical, so it also
    /// replaces the old serialization-guarded docs-map fast path for
    /// integer/string/etc. _id lookups.
    id_index: index.Index,

    fn init(gpa: std.mem.Allocator, pager: *pgr.Pager) !Collection {
        var self: Collection = .{
            .docs = .empty,
            .pager = pager,
            .slab_extents = .empty,
            .slab_tail = 0,
            .slab_end = 0,
            .indexes = .empty,
            .id_index = undefined,
        };
        const keys = [_]index.IndexKey{.{ .path = "_id", .descending = false }};
        // unique: the tree, not the docs map, is what enforces _id uniqueness
        // now (PLAN A3/A4). It is keyed on bson.encode_key, which is canonical
        // where serialize_value is not, so int32 1 / int64 1 / double 1.0
        // collide as they do in MongoDB -- see the migration note in
        // apply_record.
        self.id_index = try index.Index.init(gpa, pager, "_id_", &keys, true, false, null);
        return self;
    }

    /// The secondary index with this name, or null. The single by-name
    /// lookup: index lifetime (who calls Index.deinit, and when) is decided
    /// here rather than at each caller.
    pub fn find_index(self: *Collection, name: []const u8) ?*index.Index {
        for (self.indexes.items) |ix| {
            if (std.mem.eql(u8, ix.name, name)) return ix;
        }
        return null;
    }

    /// Append `bytes` to the slab, returning its flat offset. The last
    /// segment holds up to `slab_segment_size`; a full one starts the next.
    /// Make room for a document of `len` bytes, so the append that follows
    /// cannot fail.
    ///
    /// Separated from the append because the append runs *after* the log
    /// record is durable, where failure has nowhere to go: the write is already
    /// committed and reporting an error for it would be a lie the next open
    /// contradicts. Reserving first keeps the fallible half before the log.
    fn slab_reserve(self: *Collection, gpa: std.mem.Allocator, len: usize) !void {
        // A checkpoint can land in the middle of an extent, which freezes the
        // page the tail points into. Appending there would store inside the
        // durable image, so abandon the rest of the extent and start a fresh
        // one. The waste is bounded by one extent per collection per checkpoint.
        if (self.slab_tail >= self.pager.stable_bytes() and self.slab_tail + len <= self.slab_end) return;
        // A document larger than the standard extent gets one of its own; BSON
        // reaches 16 MB and the extent is 8 MiB.
        const want_pages: u32 = @intCast(@max(
            slab_extent_pages,
            (len + pgr.page_size - 1) / pgr.page_size,
        ));
        try self.pager.reserve_pages(want_pages);
        const first = self.pager.alloc_pages_assume_reserved(want_pages);
        try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages });
        self.slab_tail = @as(u64, first) << pgr.page_shift;
        self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
    }

    /// Copy `bytes` into the slab and return its absolute file offset.
    /// Infallible: slab_reserve must have run for at least this many bytes.
    fn slab_append(self: *Collection, bytes: []const u8) u64 {
        assert_msg(
            self.slab_tail + bytes.len <= self.slab_end,
            "document append overran the slab reservation",
        );
        const off = self.slab_tail;
        @memcpy(self.pager.bytes_mut(off, bytes.len), bytes);
        self.slab_tail += bytes.len;
        return off;
    }

    /// The canonical bytes of the document stored at `off` — a slice into a
    /// segment, stable until the collection is freed or rebuilt.
    pub fn doc_bytes(self: *const Collection, off: u64) []const u8 {
        // An absolute file offset, so this is base + off. The length comes from
        // the document's own BSON int32 prefix, as it always has.
        const len: usize = std.mem.readInt(u32, self.pager.bytes(off, 4)[0..4], .little);
        return self.pager.bytes(off, len);
    }

    /// Remove and free the index with this name. Returns whether it existed.
    /// `orderedRemove` now moves 8-byte pointers rather than whole Index
    /// structs, so the surviving indexes do not move and pointers to them stay
    /// valid; only the removed one dies, here.
    fn remove_index(self: *Collection, gpa: std.mem.Allocator, name: []const u8) bool {
        for (self.indexes.items, 0..) |ix, i| {
            if (std.mem.eql(u8, ix.name, name)) {
                _ = self.indexes.orderedRemove(i);
                ix.deinit(gpa);
                gpa.destroy(ix);
                return true;
            }
        }
        return false;
    }
};

pub const Db = struct {
    /// Collections are heap-allocated so their addresses are stable while a
    /// command holds a collection lock — the map may reallocate under the
    /// catalog lock, but the pointers it holds do not move.
    collections: std.StringHashMapUnmanaged(*Collection),
};

pub const Engine = struct {
    gpa: std.mem.Allocator,
    io: std.Io,
    // Legacy whole-engine lock, used by the unit tests' explicit
    // lock()/lock_read() calls. The server uses the finer-grained locks
    // below: catalog (maps), per-collection (docs/slab/indexes), and
    // log_lock (append + commit).
    rwlock: std.Io.RwLock,
    /// Guards the dbs/collections maps. Commands hold it shared for their
    /// whole duration so a concurrent DDL cannot mutate the maps under
    /// them; DDL takes it exclusive.
    catalog_lock: std.Io.RwLock = .init,
    /// Serializes log appends, seals and the commit sync.
    log_lock: std.Io.Mutex = .init,
    /// Serializes commit decisions; the group-commit leader holds it while
    /// sealing and syncing.
    commit_lock: std.Io.Mutex = .init,
    /// Sequence number covered by the last completed commit. A seq rather
    /// than a file position: an append leaves its bytes in the log's open
    /// block without moving end_pos, so a position comparison would call
    /// buffered-but-unwritten records durable.
    committed_seq: u64 = 0,
    /// Writers increment before appending and decrement after; the commit
    /// leader waits for this to reach zero so its seal covers every append
    /// in flight, coalescing many writers' fsyncs into one.
    pending_appends: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
    committing: bool = false,
    commit_done: std.Io.Condition = std.Io.Condition.init,
    /// Set when the garbage ratio crosses the compaction threshold; the
    /// write command's epilogue runs compact after releasing its locks.
    /// Atomic because it is set under a *collection* lock (see `note_compact`)
    /// but read by the epilogue holding no lock at all.
    compact_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    /// Set while a compaction runs, so only one runs at a time. Compactions
    /// share one tmp path and each ends in a rename onto the log, so two at
    /// once would publish one compaction's half-written file as the database.
    compacting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    log: storage.Log,
    /// The data file: documents live here, and the B+tree arenas follow.
    ///
    /// Heap-allocated because `open` builds an Engine on the stack and returns
    /// it by value: every Collection holds a `*Pager`, and those were taken
    /// during replay, before the move. They all dangled -- which surfaced as a
    /// corrupt docs hashmap on the *second* engine in a test, not as anything
    /// resembling its cause.
    pager: *pgr.Pager,
    dbs: std.StringHashMapUnmanaged(Db),
    seq: u64,
    /// Floor for the compaction trigger. The real trigger also scales with
    /// the live data size — see `note_compact`.
    compact_threshold: u64,
    /// Documents currently resident across every collection, and documents
    /// superseded or deleted since the last compaction. Their ratio is the
    /// share of the log that is garbage, which is what decides whether a
    /// rewrite is worth doing — see `note_compact`.
    live_docs: u64 = 0,
    dead_docs: u64 = 0,
    /// Set when the log has grown enough since the last checkpoint to be worth
    /// reclaiming. Read by the write epilogue and the TTL monitor, both of which
    /// run without holding a collection lock.
    checkpoint_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    /// Log bytes that trigger a checkpoint. Distinct from the compaction
    /// threshold: compaction is about the *garbage share* of the data, a
    /// checkpoint is about how much replay an open would otherwise have to do.
    checkpoint_threshold: u64 = 32 * 1024 * 1024,
    /// 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.
    dup_index: ?[]const u8 = null,

    pub fn open(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Engine {
        var log = try storage.Log.open(gpa, io, path);
        errdefer log.close();

        // The data file sits beside the log and is *kept*: a valid watermark in
        // it means most of the log never has to be replayed.
        const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{log.path});
        defer gpa.free(data_path);

        const pager_box = try gpa.create(pgr.Pager);
        errdefer gpa.destroy(pager_box);
        pager_box.* = try pgr.Pager.open(gpa, io, data_path, .{});

        var engine = Engine{
            .gpa = gpa,
            .io = io,
            .rwlock = .init,
            .log = log,
            .pager = pager_box,
            .dbs = .empty,
            .seq = 0,
            .compact_threshold = 16 * 1024 * 1024,
        };
        errdefer {
            engine.pager.deinit();
            engine.dbs.deinit(gpa);
        }

        // 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). 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;
    }

    pub fn deinit(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.deinit(self.gpa);
        self.pager.deinit();
        self.gpa.destroy(self.pager);
        self.log.close();
    }

    /// Free every document in a collection along with its owned _id keys
    /// and secondary indexes (whose entries alias the documents — freed
    /// first).
    fn free_collection(self: *Engine, coll: *Collection) void {
        // Dropping a collection turns all of its records into garbage. The
        // engine's live count includes every collection's documents, so it can
        // never be smaller than this one's -- and a u64 underflow here would
        // read as an astronomically large live count, permanently suppressing
        // compaction rather than crashing.
        assert_msg(self.live_docs >= coll.docs.count(), "dropping a collection would underflow the engine's live count");
        self.live_docs -= coll.docs.count();
        self.dead_docs += coll.docs.count();
        coll.id_index.deinit(self.gpa);
        for (coll.indexes.items) |ix| {
            ix.deinit(self.gpa);
            self.gpa.destroy(ix);
        }
        coll.indexes.deinit(self.gpa);
        var doc_it = coll.docs.iterator();
        while (doc_it.next()) |doc_entry| {
            self.gpa.free(doc_entry.key_ptr.*);
        }
        coll.docs.deinit(self.gpa);
        // Give the slab's pages back. They become reusable two generations
        // later, so a fallback to the previous image still finds them intact.
        for (coll.slab_extents.items) |e| {
            self.pager.free_pages(e.first, e.pages) catch {};
        }
        coll.slab_extents.deinit(self.gpa);
        self.gpa.destroy(coll);
    }

    /// Free every collection in a database along with its owned name keys.
    fn free_db(self: *Engine, db: *Db) void {
        var coll_it = db.collections.iterator();
        while (coll_it.next()) |coll_entry| {
            self.free_collection(coll_entry.value_ptr.*);
            self.gpa.free(coll_entry.key_ptr.*);
        }
        db.collections.deinit(self.gpa);
    }

    /// Drop the document stored under `id_key`, freeing it and its key.
    /// No-op when the id is absent. This is the single chokepoint where a
    /// document dies, so index entries are removed here, keyed by the slab
    /// offset the map hands back. It used to matter that the map key was still
    /// alive at this point, because entries aliased it; entries carry an
    /// offset now, so that constraint is gone.
    ///
    /// The document itself is handed to the index: entries are located by
    /// regenerating them from it, which is far cheaper than scanning.
    fn evict_doc(self: *Engine, coll: *Collection, id_key: []const u8) void {
        const old = coll.docs.fetchRemove(id_key) orelse return;
        // Resolve the bytes before any mutation; the slab is untouched by
        // index removal, so the slice is safe for the call.
        const old_bytes = coll.doc_bytes(old.value);
        // Entries are keyed by the document's slab offset now, which is exactly
        // what the map just gave us.
        coll.id_index.remove_doc(self.gpa, old_bytes, old.value);
        for (coll.indexes.items) |ix| ix.remove_doc(self.gpa, old_bytes, old.value);
        self.gpa.free(old.key);
        // This document's log record (and its slab bytes) just became garbage.
        // The fetchRemove above succeeded, so a live document was counted.
        assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count");
        self.live_docs -= 1;
        self.dead_docs += 1;
    }

    // -- commands (callers must hold the matching lock) ---------------------

    /// Exclusive lock: for commands that mutate the engine.
    pub fn lock(self: *Engine) !void {
        try self.rwlock.lock(self.io);
    }

    pub fn unlock(self: *Engine) void {
        self.rwlock.unlock(self.io);
    }

    /// Shared lock: for read-only commands (find, count, aggregate, list*).
    /// Multiple readers may hold it simultaneously; writers wait for them.
    pub fn lock_read(self: *Engine) !void {
        try self.rwlock.lockShared(self.io);
    }

    pub fn unlock_read(self: *Engine) void {
        self.rwlock.unlockShared(self.io);
    }

    // -- per-collection locking (the server's command dispatch) ------------

    /// Lock the catalog for a command's duration.
    pub fn lock_catalog(self: *Engine, exclusive: bool) !void {
        if (exclusive) {
            try self.catalog_lock.lock(self.io);
        } else {
            try self.catalog_lock.lockShared(self.io);
        }
    }

    pub fn unlock_catalog(self: *Engine, exclusive: bool) void {
        if (exclusive) self.catalog_lock.unlock(self.io) else self.catalog_lock.unlockShared(self.io);
    }

    /// With the catalog lock held, resolve the target collection and take
    /// its lock. When the collection is missing and `create` is set, the
    /// catalog lock is upgraded to exclusive to create it (then restored to
    /// shared); the collection lock is acquired before the exclusive catalog
    /// lock is dropped, so a concurrent drop can never free it underneath.
    /// Returns null when the collection does not exist (and create is off).
    pub fn lock_collection(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        write: bool,
        create: bool,
    ) !?*Collection {
        var coll = self.get_collection(db_name, coll_name);
        if (coll == null and create) {
            self.catalog_lock.unlockShared(self.io);
            try self.catalog_lock.lock(self.io);
            coll = try self.get_or_create_collection(db_name, coll_name);
            try self.lock_one(coll.?, write);
            self.catalog_lock.unlock(self.io);
            try self.catalog_lock.lockShared(self.io);
            return coll;
        }
        if (coll) |c| try self.lock_one(c, write);
        return coll;
    }

    fn lock_one(self: *Engine, coll: *Collection, write: bool) !void {
        if (write) {
            try coll.lock.lock(self.io);
        } else {
            try coll.lock.lockShared(self.io);
        }
    }

    pub fn unlock_collection(self: *Engine, coll: *Collection, write: bool) void {
        if (write) coll.lock.unlock(self.io) else coll.lock.unlockShared(self.io);
    }

    /// Ensure this command's appends are durable. The commit leader waits
    /// for writers mid-append to finish, then seals and syncs once, covering
    /// every append in flight — followers that arrived during the leader's
    /// commit find their records already covered and return without a sync
    /// of their own. Every acknowledged write is fsynced before its reply,
    /// so the crash guarantees are unchanged.
    pub fn commit(self: *Engine) !void {
        try self.commit_lock.lock(self.io);
        defer self.commit_lock.unlock(self.io);
        // A commit can never have sealed more than was ever appended.
        assert_msg(self.committed_seq <= self.seq, "commit claims to have sealed more than was appended");
        // Everything this command appended is at or below the current seq.
        // Read it before waiting, so a leader that sealed before this
        // command's appends cannot be mistaken for one that covered them.
        try self.log_lock.lock(self.io);
        const want = self.seq;
        self.log_lock.unlock(self.io);
        // A commit is in flight; wait for it, then check whether the
        // leader's seal covered this writer's append.
        //
        // The error propagates rather than being swallowed: this function
        // returning success is what tells the caller its write is on disk, so
        // reporting success after a failed wait acknowledges a write that was
        // never synced. A canceled connection has no reason to wait out
        // another writer's commit, so cancelable is right here.
        while (self.committing) try self.commit_done.wait(self.io, &self.commit_lock);
        if (self.committed_seq >= want) {
            return; // a concurrent commit already synced this writer's records
        }
        // Become the leader: the flag is set before the wait below, so any
        // commit that arrives during it waits as a follower.
        self.committing = true;
        var done = false;
        defer {
            if (!done) {
                self.committing = false;
                // Broadcast: followers sleeping on `committing` all need to
                // re-check it, not just one of them.
                self.commit_done.broadcast(self.io);
            }
        }
        // Wait for writers mid-append to finish so the seal covers them.
        //
        // Uncancelable: `committing` is set, so every other writer is now
        // parked behind this leader. Abandoning the commit here would strand
        // them for a full extra round trip, and the drain is bounded anyway --
        // an in-flight append only holds log_lock long enough to buffer its
        // record. Finishing is strictly better than bailing out.
        while (self.pending_appends.load(.acquire) > 0) {
            self.commit_done.waitUncancelable(self.io, &self.commit_lock);
        }
        // The drain is what makes the seal below cover every append in flight.
        // Not asserted as pending_appends == 0 here: a new append can start
        // at any moment (it increments without commit_lock), so a fresh
        // writer can be in flight between the drain's last check and this
        // point. The seal still covers every append that wrote bytes before
        // the sync — appends serialize with it on log_lock — and any append
        // that starts after it is sealed by its own commit.
        try self.log_lock.lock(self.io);
        defer self.log_lock.unlock(self.io);
        try self.log.sync();
        // Appends drained above, so the seal covered every record written so
        // far -- including any that arrived while this leader waited.
        self.committed_seq = self.seq;
        // This writer's own records are now durable: the postcondition the
        // caller relies on before it acknowledges the write. Paired with the
        // same check in the dispatch epilogue (see commands.zig).
        assert_msg(self.committed_seq >= want, "commit returning success without sealing this writer's records");
        done = true;
        self.committing = false;
        // Broadcast: every follower waiting on `committing` must wake to see
        // it cleared — a single signal would wake only one and strand the
        // rest.
        self.commit_done.broadcast(self.io);
    }

    /// Log an append (and its seq increment) under the log lock, marking
    /// the append as in flight so a commit leader's seal covers it.
    fn log_append(
        self: *Engine,
        comptime kind: LogKind,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
    ) !void {
        _ = self.pending_appends.fetchAdd(1, .acq_rel);
        defer {
            // The increment above pairs with this decrement on every return
            // path, so the count can never be zero here.
            assert_msg(self.pending_appends.load(.acquire) > 0, "log_append decrementing an already-zero in-flight count");
            _ = self.pending_appends.fetchSub(1, .acq_rel);
            // Wake a commit leader waiting for in-flight appends. The
            // signal must be delivered while holding commit_lock: a leader
            // between its pending_appends check and its wait() still holds
            // the lock, so a signal here can never land in that window and
            // be lost (which froze every writer once a few connections
            // committed concurrently). The leader's wait() releases the
            // lock, so this lock only blocks until it starts waiting.
            // Broadcast rather than signal: if a follower waiting on
            // `committing` snatches the single wakeup, the leader would
            // sleep forever even with pending_appends back at zero.
            //
            // lockUncancelable, not lock: this is a cleanup path, and the
            // cancelable variant can fail. Swallowing that failure and
            // unlocking anyway would release a mutex we never took, which
            // Mutex.unlock treats as `unreachable` -- a panic in ReleaseSafe
            // and silent memory corruption in the default ReleaseFast build.
            self.commit_lock.lockUncancelable(self.io);
            self.commit_done.broadcast(self.io);
            self.commit_lock.unlock(self.io);
        }
        try self.log_lock.lock(self.io);
        defer self.log_lock.unlock(self.io);
        self.seq += 1;
        // Seqs start at 1 and only ever increase; 0 means "nothing appended",
        // which is what committed_seq is compared against.
        assert_msg(self.seq > 0, "log_append produced a zero seq");
        switch (kind) {
            .upsert => try self.log.append_upsert(db, coll, doc, self.seq),
            .delete => try self.log.append_delete(db, coll, doc, self.seq),
            .index_create => try self.log.append_index_create(db, coll, doc, self.seq),
            .index_drop => try self.log.append_index_drop(db, coll, doc, self.seq),
        }
    }

    /// Insert a document. Fails with error.DuplicateKey if the _id exists.
    /// Generates an ObjectId _id when absent.
    pub fn insert(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        doc: *const bson.Document,
        oid_gen: *bson.ObjectIdGen,
    ) !void {
        return self.upsert(db_name, coll_name, doc, oid_gen, .insert);
    }

    /// Insert or replace a document by _id (upsert without existence check).
    pub fn replace(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        doc: *const bson.Document,
        oid_gen: *bson.ObjectIdGen,
    ) !void {
        return self.upsert(db_name, coll_name, doc, oid_gen, .replace);
    }

    /// One document's built entries for one index, tracked so a failure
    /// anywhere before the log append frees them all.
    const Built = struct {
        built: index.BuiltEntries,
        ix: *index.Index,
    };

    /// Shared body of `insert` and `replace`: they differ only in how an
    /// existing _id is treated. Logs (and syncs) the new document before it
    /// becomes visible in memory.
    fn upsert(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        doc: *const bson.Document,
        oid_gen: *bson.ObjectIdGen,
        mode: enum { insert, replace },
    ) !void {
        const coll = try self.get_or_create_collection(db_name, coll_name);
        const doc_bytes = try self.serialize_with_id(doc, oid_gen);
        defer self.gpa.free(doc_bytes);
        // A document _id materializes a spine; free it right after the key
        // is serialized.
        var id_arena = std.heap.ArenaAllocator.init(self.gpa);
        defer id_arena.deinit();
        const id_value = (try bson.get_at(id_arena.allocator(), doc_bytes, "_id")) orelse unreachable;
        // Ownership of the key moves to the map once `stored` is set.
        const id_key = try bson.serialize_value(self.gpa, id_value);
        var stored = false;
        errdefer if (!stored) self.gpa.free(id_key);
        coll.dup_index = null;

        // 1. Build entries for every index. ParallelArrays escapes here,
        //    before anything is logged or mutated.
        var built_list: std.ArrayListUnmanaged(Built) = .empty;
        defer {
            for (built_list.items) |*b| b.built.deinit(self.gpa);
            built_list.deinit(self.gpa);
        }
        {
            // The implicit _id_ index, through the same protocol: reserved
            // before the log append, inserted infallibly after it. Built
            // *first* so it is checked first below -- MongoDB reports _id_
            // when a write violates both it and a unique secondary.
            var built = try coll.id_index.build_entries(self.gpa, doc_bytes);
            built_list.append(self.gpa, .{ .built = built, .ix = &coll.id_index }) catch |err| {
                built.deinit(self.gpa);
                return err;
            };
        }
        for (coll.indexes.items) |ix| {
            var built = try ix.build_entries(self.gpa, doc_bytes);
            built_list.append(self.gpa, .{ .built = built, .ix = ix }) catch |err| {
                built.deinit(self.gpa);
                return err;
            };
        }

        // 2. Unique-index checks, _id_ included; a rejected write never
        //    reaches the log. `_id` uniqueness used to be a `docs.contains`
        //    probe here, which the docs map will not be around to answer
        //    (PLAN amendment A3) -- and the tree answers it better, since it
        //    is keyed on the canonical encode_key rather than serialize_value
        //    (A4). Exclude-self is null for an insert: the document has no
        //    entries yet, and passing its offset would hide precisely the
        //    same-_id collision this must catch. For a replace it is the
        //    document's *current* slab offset, since that is what its existing
        //    entries carry -- the new offset does not exist yet.
        const exclude: ?u64 = if (mode == .replace) coll.docs.get(id_key) else null;
        for (built_list.items) |*b| {
            if (!b.ix.unique) continue;
            b.ix.check_unique(b.built.entries.items, exclude) catch {
                // The implicit index keeps its own error identity, so
                // commands.zig renders E11000 with index "_id_" exactly as
                // before and needs no change; `dup_index` stays null, which is
                // what that rendering treats as "the _id_ index".
                if (b.ix == &coll.id_index) return error.DuplicateKey;
                coll.dup_index = b.ix.name;
                return error.DuplicateKeyIndex;
            };
        }

        // 4. Reserve everything the publish step needs -- tree capacity and
        //    slab room -- as the last fallible work, so nothing after the log
        //    append can fail. The slab reservation used to be absent because
        //    appending to an in-memory ArrayList was the only failure mode; a
        //    file-backed slab can also fail on growth, and failing *after* the
        //    record is durable would report an error for a write the next open
        //    would produce anyway.
        for (built_list.items) |*b| {
            try b.ix.reserve_for(self.gpa, b.built.entries.items);
        }
        try coll.slab_reserve(self.gpa, doc_bytes.len);

        // 5. Log (and sync) before anything becomes visible. The append
        //    takes the log lock; durability (fsync) is the command's commit.
        try self.log_append(.upsert, db_name, coll_name, doc_bytes);

        // 6. Replace drops the old document (and its index entries).
        if (mode == .replace) self.evict_doc(coll, id_key);

        // 7. Publish the document and its entries: copy the bytes into the
        //    slab and record the offset. Infallible from here.
        const off = coll.slab_append(doc_bytes);
        try coll.docs.put(self.gpa, id_key, off);
        self.live_docs += 1;
        for (built_list.items) |*b| {
            if (b.built.multikey) b.ix.multikey = true;
            b.ix.insert_entries(&b.built, off);
        }
        stored = true;
        // The write is published; anything the reservations above did not claim
        // is dead. Leaving it promised would grow the file on every write.
        self.pager.release_reservation();
        self.note_compact();
        self.note_checkpoint();
    }

    /// Remove a document by its `_id` value. Returns true if it existed.
    /// The serialized-key encoding stays private to the engine.
    pub fn remove_by_id(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        id: bson.Value,
    ) !bool {
        const id_key = try bson.serialize_value(self.gpa, id);
        defer self.gpa.free(id_key);
        return self.remove(db_name, coll_name, id_key);
    }

    fn remove(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) !bool {
        const db = self.dbs.get(db_name) orelse return false;
        const coll = db.collections.get(coll_name) orelse return false;
        const off = coll.docs.get(id_key) orelse return false;

        // Log (and sync) the delete before removing it from memory, so the
        // log always describes at least as much as the in-memory state.
        // Replay only reads _id out of a delete record, so log just that
        // rather than a copy of the whole document.
        const id_bytes = coll.doc_bytes(off);
        var id_arena = std.heap.ArenaAllocator.init(self.gpa);
        defer id_arena.deinit();
        const id_pairs = [_]bson.Pair{.{ .key = "_id", .value = (try bson.get_at(id_arena.allocator(), id_bytes, "_id")) orelse unreachable }};
        var id_doc: std.ArrayListUnmanaged(u8) = .empty;
        defer id_doc.deinit(self.gpa);
        try bson.write_doc(&id_pairs, self.gpa, &id_doc);
        try self.log_append(.delete, db_name, coll_name, id_doc.items);

        self.evict_doc(coll, id_key);
        // Deletes grow the log too. Without this a delete-heavy workload
        // never compacts, because only upsert and ttl_sweep used to check.
        self.note_compact();
        return true;
    }

    pub fn get_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) ?*Collection {
        const db = self.dbs.get(db_name) orelse return null;
        return db.collections.get(coll_name);
    }

    pub fn get_doc(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        id_key: []const u8,
    ) ?[]const u8 {
        const coll = self.get_collection(db_name, coll_name) orelse return null;
        const off = coll.docs.get(id_key) orelse return null;
        return coll.doc_bytes(off);
    }

    pub fn drop_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) !bool {
        const db = self.dbs.getPtr(db_name) orelse return false;
        const removed = db.collections.fetchRemove(coll_name) orelse return false;
        self.free_collection(removed.value);
        self.gpa.free(removed.key);
        return true;
    }

    pub fn drop_database(self: *Engine, db_name: []const u8) !bool {
        var removed = self.dbs.fetchRemove(db_name) orelse return false;
        self.free_db(&removed.value);
        self.gpa.free(removed.key);
        return true;
    }

    /// Build and register a secondary index from a spec document
    /// ({key, name, unique?, sparse?}). The create record is written only
    /// after the index builds over the existing documents and passes
    /// uniqueness, so a rejected create persists nothing. Returns the new
    /// index (or the existing one when the spec matches — idempotent).
    pub fn create_index(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        spec_doc: *const bson.Document,
    ) !*index.Index {
        const coll = try self.get_or_create_collection(db_name, coll_name);
        const parsed = try index.parse_spec(self.gpa, self.pager, spec_doc);
        // Boxed before anything is built into it, so publishing is a pointer
        // append rather than a struct copy. An Index will own a mapping once
        // the arena is file-backed, and copying one then would duplicate that
        // ownership.
        const ix = self.gpa.create(index.Index) catch |err| {
            var dead = parsed;
            dead.deinit(self.gpa);
            return err;
        };
        ix.* = parsed;
        var committed = false;
        // Runs on every return path (including the idempotent no-op): the
        // parsed spec is only owned by the collection once committed.
        defer if (!committed) {
            ix.deinit(self.gpa);
            self.gpa.destroy(ix);
        };

        if (coll.find_index(ix.name)) |existing| {
            if (index.Index.spec_equal(existing, ix)) return existing;
            return error.IndexOptionsConflict;
        }

        // Build entries over the existing documents (the index is not
        // exposed until the end, so mutating it is safe). Entries are
        // appended unsorted and ordered once at the end — inserting each
        // document into a sorted array memmoves the tail every time, which
        // is what made this quadratic. On any failure the deferred
        // ix.deinit frees every appended key. Nothing is persisted.
        var doc_it = coll.docs.iterator();
        while (doc_it.next()) |entry| {
            try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.value_ptr.*), entry.value_ptr.*);
        }
        _ = try ix.finish_bulk(self.gpa, true);
        self.pager.release_reservation();

        // Reserve the collection slot, then persist and publish.
        try coll.indexes.ensureUnusedCapacity(self.gpa, 1);
        var spec_bytes: std.ArrayListUnmanaged(u8) = .empty;
        defer spec_bytes.deinit(self.gpa);
        try ix.write_spec(self.gpa, &spec_bytes);
        try self.log_append(.index_create, db_name, coll_name, spec_bytes.items);

        coll.indexes.appendAssumeCapacity(ix);
        committed = true;
        return ix;
    }

    /// Remove a secondary index by name, persisting a drop record first.
    /// Returns false when no such index exists.
    pub fn drop_index(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
        index_name: []const u8,
    ) !bool {
        const db = self.dbs.get(db_name) orelse return false;
        const coll = db.collections.get(coll_name) orelse return false;
        if (coll.find_index(index_name) == null) return false;

        const name_pairs = [_]bson.Pair{.{ .key = "name", .value = .{ .string = index_name } }};
        var name_doc: std.ArrayListUnmanaged(u8) = .empty;
        defer name_doc.deinit(self.gpa);
        try bson.write_doc(&name_pairs, self.gpa, &name_doc);
        try self.log_append(.index_drop, db_name, coll_name, name_doc.items);

        _ = coll.remove_index(self.gpa, index_name);
        return true;
    }

    /// Delete every document expired as of `now_ms` (Unix milliseconds)
    /// under some TTL index, and return how many were deleted. Callers must
    /// hold the write lock; the server's monitor coroutine (src/server.zig)
    /// is the only caller in production, tests call it with a fixed clock.
    ///
    /// Each expiry goes through `remove`, so it is logged and fsynced like
    /// any other delete and survives a restart. Expiry is therefore coarse
    /// by design (as in MongoDB): an expired document stays visible until
    /// the next sweep.
    pub fn ttl_sweep(self: *Engine, now_ms: i64) !usize {
        var deleted: usize = 0;
        // Catalog lock for the whole sweep (the collection pointers stay
        // valid); each collection is swept under its own write lock, one at
        // a time, never two at once.
        try self.catalog_lock.lockShared(self.io);
        defer self.catalog_lock.unlockShared(self.io);
        var db_it = self.dbs.iterator();
        while (db_it.next()) |db_entry| {
            var coll_it = db_entry.value_ptr.collections.iterator();
            while (coll_it.next()) |coll_entry| {
                const coll = coll_entry.value_ptr.*;
                deleted += try self.ttl_sweep_coll(coll, now_ms, db_entry.key_ptr.*, coll_entry.key_ptr.*);
            }
        }
        // A TTL-only workload never reaches the threshold check in `upsert`,
        // so the log would otherwise grow without bound.
        if (deleted > 0) self.note_compact();
        return deleted;
    }

    /// Sweep one collection under its write lock; the lock is released on
    /// every return path. Returns how many documents were removed.
    fn ttl_sweep_coll(
        self: *Engine,
        coll: *Collection,
        now_ms: i64,
        db_name: []const u8,
        coll_name: []const u8,
    ) !usize {
        try coll.lock.lock(self.io);
        defer coll.lock.unlock(self.io);
        // Offsets, collected before any removal. They are values, so unlike
        // the id slices this used to dupe -- which aliased a docs-map key that
        // `remove` would free out from under the rest of the batch -- there is
        // nothing to own here. Collect-then-remove still matters, because the
        // iterator below aliases tree pages that removal reshapes.
        var offs: std.ArrayListUnmanaged(u64) = .empty;
        defer offs.deinit(self.gpa);

        for (coll.indexes.items) |ix| {
            const ttl = ix.ttl orelse continue;
            const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000;
            // bson compare order ranks datetime above null, numbers
            // and strings and below only timestamp and maxKey, so
            // datetimes form a contiguous band in the encoded key
            // order: seek the minimum datetime and stop when the
            // leading type changes or the cutoff is passed.
            const min_dt = [_]u8{ bson.encoded_datetime_tag, 0, 0, 0, 0, 0, 0, 0, 0 };
            var it = ix.seek(&min_dt);
            while (it.next()) |e| {
                const ms = bson.encoded_leading_datetime(e.key) orelse break;
                if (@as(i128, ms) > cutoff) break;
                try offs.append(self.gpa, e.off);
            }
        }
        if (offs.items.len == 0) return 0;

        // One document can be expired by several entries (an array
        // of dates) or by several TTL indexes.
        std.mem.sort(u64, offs.items, {}, std.sort.asc(u64));
        var w: usize = 1;
        for (offs.items[1..]) |off| {
            if (off != offs.items[w - 1]) {
                offs.items[w] = off;
                w += 1;
            }
        }
        offs.items.len = w;

        // `remove` works by _id, so recover each one from the document its
        // offset names. get_at materializes a spine, hence the arena; the slab
        // is untouched by the removals, so the bytes stay valid throughout.
        var arena = std.heap.ArenaAllocator.init(self.gpa);
        defer arena.deinit();
        var removed: usize = 0;
        for (offs.items) |off| {
            const id_value = (try bson.get_at(arena.allocator(), coll.doc_bytes(off), "_id")) orelse continue;
            const id_key = try bson.serialize_value(arena.allocator(), id_value);
            if (try self.remove(db_name, coll_name, id_key)) removed += 1;
        }
        return removed;
    }

    pub fn database_names(self: *Engine, out: *std.ArrayListUnmanaged([]const u8)) !void {
        var it = self.dbs.iterator();
        while (it.next()) |entry| try out.append(self.gpa, entry.key_ptr.*);
    }

    pub fn collection_names(
        self: *Engine,
        db_name: []const u8,
        out: *std.ArrayListUnmanaged([]const u8),
    ) !void {
        const db = self.dbs.get(db_name) orelse return;
        var it = db.collections.iterator();
        while (it.next()) |entry| try out.append(self.gpa, entry.key_ptr.*);
    }

    // -- internals -----------------------------------------------------------

    pub fn get_or_create_collection(
        self: *Engine,
        db_name: []const u8,
        coll_name: []const u8,
    ) !*Collection {
        const db = self.dbs.getPtr(db_name) orelse {
            const db_key = try self.gpa.dupe(u8, db_name);
            errdefer self.gpa.free(db_key);
            try self.dbs.put(self.gpa, db_key, .{ .collections = .empty });
            return self.get_or_create_collection(db_name, coll_name);
        };
        if (db.collections.get(coll_name)) |coll| return coll;
        const coll_key = try self.gpa.dupe(u8, coll_name);
        errdefer self.gpa.free(coll_key);
        const new_coll = try self.gpa.create(Collection);
        errdefer self.gpa.destroy(new_coll);
        new_coll.* = try Collection.init(self.gpa, self.pager);
        errdefer new_coll.id_index.deinit(self.gpa);
        try db.collections.put(self.gpa, coll_key, new_coll);
        return new_coll;
    }

    /// Deep-copy a document into engine-owned storage, prepending a
    /// generated ObjectId `_id` when absent.
    /// The canonical bytes of `doc`, with an ObjectId `_id` generated when
    /// absent. The result is owned by the caller.
    fn serialize_with_id(
        self: *Engine,
        doc: *const bson.Document,
        oid_gen: *bson.ObjectIdGen,
    ) ![]u8 {
        if (doc.get("_id") != null) return serialize_doc(self.gpa, doc);
        var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
        defer pairs.deinit(self.gpa);
        const oid = oid_gen.new(self.io);
        try pairs.append(self.gpa, .{ .key = "_id", .value = .{ .object_id = oid } });
        try pairs.appendSlice(self.gpa, doc.pairs);

        var out: std.ArrayListUnmanaged(u8) = .empty;
        defer out.deinit(self.gpa);
        try bson.write_doc(pairs.items, self.gpa, &out);
        return out.toOwnedSlice(self.gpa);
    }

    /// Keep the log file at roughly 1.5x the live data, rather than
    /// compacting every fixed number of appended bytes.
    ///
    /// A fixed byte trigger makes total rewrite traffic quadratic: a 1 GB
    /// dataset with a 16 MiB threshold compacts ~64 times, rewriting 1 GB
    /// each time. Triggering on file size relative to the live size makes
    /// successive compactions geometric, so the total bytes rewritten over
    /// the life of the log is O(n) rather than O(n²) — and it bounds the
    /// disk footprint directly, which is what the threshold is really for.
    ///
    /// The other half of the problem is the opposite workload: a pure bulk
    /// insert has no garbage at all, so every compaction rewrites a
    /// perfectly compact file for nothing. `compact` reports how much it
    /// reclaimed; when that is little, we back the baseline off
    /// multiplicatively so a garbage-free log is left alone.
    /// Called at the end of a write command's in-lock section: when the
    /// garbage share crosses the threshold, record that a compaction is
    /// wanted. It runs in the command epilogue, after the collection lock is
    /// released — never inline, since compact takes the collection locks
    /// itself and would deadlock against the caller's.
    /// Arm a checkpoint when the log has grown past the threshold. Cheap enough
    /// to call on every write: one relaxed load and a compare.
    fn note_checkpoint(self: *Engine) void {
        if (self.log.log_bytes < self.checkpoint_threshold) return;
        self.checkpoint_pending.store(true, .release);
    }

    /// Claim a pending checkpoint, if there is one.
    pub fn take_checkpoint(self: *Engine) bool {
        return self.checkpoint_pending.swap(false, .acq_rel);
    }

    fn note_compact(self: *Engine) void {
        // The threshold counts data volume (uncompressed record bytes), not
        // the on-disk size: a compressed log would otherwise stay under any
        // byte threshold and never compact its garbage.
        if (self.log.data_bytes < self.compact_threshold) return;
        // Only rewrite when enough of the log is actually garbage. The old
        // rule fired on bytes appended, which is the wrong question twice
        // over: a 1 GB bulk load has no garbage at all yet would compact
        // ~64 times under a 16 MiB threshold (rewriting 1 GB each time,
        // hence quadratic), while a small collection rewritten in place
        // accumulates garbage indefinitely without ever hitting the count.
        //
        // Garbage share is dead / (live + dead); this fires at ~20%, so the
        // file stays near 1.25x the live data and each compaction is paid
        // for by the space it reclaims.
        if (self.dead_docs * 4 < self.live_docs) return;
        self.compact_pending.store(true, .release);
    }

    /// Whether a compaction is wanted; clears the flag atomically so only one
    /// of several concurrent writers takes the request. `compact` excludes
    /// itself besides, so a caller that wins here still yields to a rewrite
    /// already in progress.
    pub fn take_compact(self: *Engine) bool {
        return self.compact_pending.swap(false, .acq_rel);
    }

    /// Re-arm the compaction request. For a caller that took the request but
    /// could not carry it out (a failed or abandoned rewrite), so the garbage
    /// is reconsidered by a later, quieter epilogue instead of being forgotten.
    pub fn request_compact(self: *Engine) void {
        self.compact_pending.store(true, .release);
    }

    /// Rewrite the log with only live documents, atomically swapping the
    /// file. Runs outside every collection lock (called from a write
    /// command's epilogue). The snapshot takes the collection locks one at
    /// a time without the log lock — so a concurrent writer can always
    /// finish its append — then takes the log lock and retries until no
    /// writer appended during the snapshot (detected via the record seq),
    /// which makes the snapshot consistent with what the log contains.
    ///
    /// Only one compaction runs at a time; a second caller returns immediately.
    pub fn compact(self: *Engine) !void {
        // Claim the compaction, or leave it to the one already running. The
        // guard lives here rather than in `take_compact` so that every caller
        // is covered, including tests that invoke compact directly. Two at once
        // would share the tmp path below: one's delete-and-recreate unlinks the
        // other's file while it still holds the fd, and then both rename that
        // path onto the log -- publishing a half-written file as the database.
        // A caller that loses this race has nothing to do anyway: the winner's
        // rewrite covers its garbage too.
        if (self.compacting.swap(true, .acq_rel)) return;
        defer self.compacting.store(false, .release);

        const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path});
        defer self.gpa.free(tmp_path);

        // Bounded: every attempt rewrites the whole log before the seq check
        // below can reject it, so an unbounded retry livelocks under sustained
        // writes at one full rewrite per attempt. Giving up re-arms the request
        // for a quieter epilogue -- the log stays correct, just larger.
        const attempt_max: u32 = 8;
        var attempt: u32 = 0;
        while (attempt < attempt_max) : (attempt += 1) {
            // Truncating create, not open: a tmp file left by a crashed or
            // retried compaction is longer than what we are about to write, and
            // `open` would keep its tail. Those leftover blocks are intact and
            // hash-correct, so replay would apply them as live records once the
            // rename publishes this file.
            var new_log = try storage.Log.create(self.gpa, self.io, tmp_path);
            defer new_log.close();

            const snapshot_seq = self.seq;
            try self.catalog_lock.lockShared(self.io);
            var catalog_err: ?anyerror = null;
            var db_it = self.dbs.iterator();
            while (db_it.next()) |db_entry| {
                var coll_it = db_entry.value_ptr.collections.iterator();
                while (coll_it.next()) |coll_entry| {
                    const coll = coll_entry.value_ptr.*;
                    self.compact_snapshot_coll(coll, &new_log, db_entry.key_ptr.*, coll_entry.key_ptr.*) catch |err| {
                        catalog_err = err;
                        break;
                    };
                }
                if (catalog_err != null) break;
            }
            self.catalog_lock.unlockShared(self.io);
            if (catalog_err) |err| return err;

            // Swap under the log lock, which also blocks appends: verify the
            // snapshot saw no interleaved appends before replacing the file.
            try self.log_lock.lock(self.io);
            if (self.seq != snapshot_seq) {
                self.log_lock.unlock(self.io);
                continue; // a writer appended during the snapshot; retry
            }
            errdefer self.log_lock.unlock(self.io);
            assert_msg(self.seq == snapshot_seq, "compaction snapshot raced an append");
            // Durable before the rename makes it the database.
            try new_log.sync();
            // Nothing may follow the last sealed block: replay walks blocks
            // until it runs off the end, so a trailing byte range would be
            // applied as live data. Checked here, right where the file is about
            // to become the database, rather than trusting the truncating open.
            assert_msg(try new_log.file.length(self.io) == new_log.end_pos, "compacted log has bytes past its last sealed block");

            try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io);
            // Persist the rename: fsync the parent directory so the new
            // directory entry survives a power loss right after compaction.
            const parent = parent_dir(self.log.path);
            var dir_file = try std.Io.Dir.cwd().openFile(self.io, parent, .{ .mode = .read_only, .allow_directory = true });
            defer dir_file.close(self.io);
            try dir_file.sync(self.io);

            const old_path = try self.gpa.dupe(u8, self.log.path);
            self.log.close();
            self.log = try storage.Log.open(self.gpa, self.io, old_path);
            // Log.open does not replay, so it starts at an empty file's
            // end_pos; continue appending where the compacted file actually
            // ends. Read from new_log rather than a local captured earlier:
            // the sync above is what seals the last block and moves end_pos
            // past it, so a position read before it would leave appends
            // overwriting that block, which then vanishes on the next replay.
            self.log.end_pos = new_log.end_pos;
            // The rewritten file was synced before the rename, and the check
            // above proved no append slipped in, so everything up to the
            // snapshot's seq is durable.
            self.committed_seq = snapshot_seq;
            assert_msg(self.committed_seq <= self.seq, "compaction left committed_seq past the log's seq");
            // The rewritten log holds only live documents.
            self.dead_docs = 0;
            self.gpa.free(old_path);
            self.log_lock.unlock(self.io);
            return;
        }

        // Every attempt lost the race against a concurrent writer. Not an
        // error: the log is intact and still correct, only bigger than we would
        // like, so hand the request back rather than failing the write whose
        // epilogue called us.
        self.request_compact();
        std.debug.print(
            "multiforadb: compaction gave up after {d} attempts (concurrent writes); will retry\n",
            .{attempt_max},
        );
    }

    /// Re-emit one collection's index specs and documents into the compacted
    /// log, under the collection's write lock (released on every return
    /// path, including errors).
    fn compact_snapshot_coll(
        self: *Engine,
        coll: *Collection,
        new_log: *storage.Log,
        db_name: []const u8,
        coll_name: []const u8,
    ) !void {
        try coll.lock.lock(self.io);
        defer coll.lock.unlock(self.io);
        // Re-emit the index definitions first: a compacted log that dropped
        // them would resurrect the collections without indexes on replay.
        for (coll.indexes.items) |ix| {
            var spec_bytes: std.ArrayListUnmanaged(u8) = .empty;
            defer spec_bytes.deinit(self.gpa);
            try ix.write_spec(self.gpa, &spec_bytes);
            try new_log.append_index_create(db_name, coll_name, spec_bytes.items, self.seq);
        }
        var doc_it = coll.docs.iterator();
        while (doc_it.next()) |doc_entry| {
            // The slab bytes are the canonical serialization.
            const doc_bytes = coll.doc_bytes(doc_entry.value_ptr.*);
            try new_log.append_upsert(db_name, coll_name, doc_bytes, self.seq);
        }
    }

    /// Rebuild every empty index from the live documents. Runs after replay
    /// completes, so it is order-independent: a create record, the documents
    /// it indexes, and any drop record all replay first. A duplicate under a
    /// unique index logs a loud warning and keeps the index (still correct
    /// as a candidate generator; future writes are still enforced) — the
    /// database always opens, leaving dropIndexes as an in-band recovery
    /// path.
    fn build_all_indexes(self: *Engine) !void {
        var db_it = self.dbs.iterator();
        while (db_it.next()) |db_entry| {
            var coll_it = db_entry.value_ptr.collections.iterator();
            while (coll_it.next()) |coll_entry| {
                for (coll_entry.value_ptr.*.indexes.items) |ix| {
                    try self.rebuild_index(coll_entry.value_ptr.*, ix);
                }
                try self.rebuild_index(coll_entry.value_ptr.*, &coll_entry.value_ptr.*.id_index);
            }
        }
    }

    /// Rebuild one index from the live documents. Runs after replay, so it
    /// is order-independent; indexes already holding entries (maintained
    /// live) are skipped defensively. A duplicate under a unique index logs
    /// a loud warning and keeps the index (still correct as a candidate
    /// generator; future writes are still enforced) — the database always
    /// opens, leaving dropIndexes as an in-band recovery path.
    fn rebuild_index(self: *Engine, coll: *Collection, ix: *index.Index) !void {
        if (ix.count() > 0) return; // defensive
        var doc_it = coll.docs.iterator();
        while (doc_it.next()) |doc_entry| {
            ix.append_doc_entries(self.gpa, coll.doc_bytes(doc_entry.value_ptr.*), doc_entry.value_ptr.*) catch |err| switch (err) {
                error.ParallelArrays => {
                    std.debug.print(
                        "multiforadb: WARNING: index '{s}' cannot index an existing " ++
                            "document; entry skipped\n",
                        .{
                            ix.name,
                        },
                    );
                    continue;
                },
                else => return err,
            };
        }
        // Tolerated, not enforced: the database must always open.
        defer self.pager.release_reservation();
        if (try ix.finish_bulk(self.gpa, false)) {
            std.debug.print(
                "multiforadb: WARNING: unique index '{s}' has duplicate keys in existing " ++
                    "data; duplicates not enforced for existing documents\n",
                .{
                    ix.name,
                },
            );
        }
    }

    // -- 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();
            // The watermark is durable, so every record it covers is now
            // redundant. Strictly after the publish: the other order loses data
            // if a crash lands between them.
            self.log.truncate_to_header() catch |err| {
                // A failed truncation wastes space and costs replay time on the
                // next open; it does not lose anything, because the records are
                // still there and still above no watermark. Not worth failing
                // the checkpoint that already succeeded.
                std.debug.print("multiforadb: WARNING: log truncation failed: {s}\n", .{@errorName(err)});
            };
            self.committed_seq = snapshot_seq;
            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(
        self: *Engine,
        coll: *Collection,
        spec_doc: *const bson.Document,
    ) !void {
        const parsed = try index.parse_spec(self.gpa, self.pager, spec_doc);
        const ix = self.gpa.create(index.Index) catch |err| {
            var dead = parsed;
            dead.deinit(self.gpa);
            return err;
        };
        ix.* = parsed;
        var committed = false;
        defer if (!committed) {
            ix.deinit(self.gpa);
            self.gpa.destroy(ix);
        };
        if (coll.find_index(ix.name) != null) return;
        try coll.indexes.append(self.gpa, ix);
        committed = true;
    }
};

fn parent_dir(path: []const u8) []const u8 {
    const last = std.mem.lastIndexOfScalar(u8, path, '/') orelse return ".";
    if (last == 0) return "/";
    return path[0..last];
}

fn serialize_doc(gpa: std.mem.Allocator, doc: *const bson.Document) ![]u8 {
    var out: std.ArrayListUnmanaged(u8) = .empty;
    errdefer out.deinit(gpa);
    try doc.to_bytes(gpa, &out);
    return out.toOwnedSlice(gpa);
}

fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) anyerror!void {
    const self: *Engine = @ptrCast(@alignCast(ctx));
    // The document is transient: only its canonical bytes are stored in the
    // collection slab. Always owned by this frame.
    defer {
        doc.deinit();
        self.gpa.destroy(doc);
    }

    const coll = self.get_or_create_collection(record.db, record.coll) catch return;

    // Index records carry no _id — handle them before the lookup. Replay
    // registers indexes empty; Engine.open builds them from the live docs
    // after replay completes.
    switch (record.type) {
        storage.record_type_index_create => {
            self.register_index_from_spec(coll, doc) catch |err| {
                std.debug.print("multiforadb: index create record failed to apply: {s}\n", .{
                    @errorName(err),
                });
                return;
            };
            return;
        },
        storage.record_type_index_drop => {
            const name_value = doc.get("name") orelse return;
            const name = switch (name_value) {
                .string => |s| s,
                else => return,
            };
            _ = coll.remove_index(self.gpa, name);
            return;
        },
        else => {},
    }

    const id_value = doc.get("_id") orelse {
        std.debug.print("multiforadb: log record without _id, skipping\n", .{});
        return;
    };
    const id_key = try bson.serialize_value(self.gpa, id_value);
    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);
            const doc_bytes = try serialize_doc(self.gpa, doc);
            defer self.gpa.free(doc_bytes);
            try coll.slab_reserve(self.gpa, doc_bytes.len);
            const off = coll.slab_append(doc_bytes);
            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.
            //
            // Which is why making _id_ unique cannot lose a document here:
            // eviction above goes through the docs map, keyed on
            // serialize_value, so a database holding both {_id: int32 1} and
            // {_id: int64 1} keeps both. The bulk build then finds duplicate
            // canonical keys, tolerates them and warns (rule: the database
            // must always open). The commit that drops the docs map is where
            // that stops being true -- see PLAN amendment A4.
        },
        storage.record_type_delete => self.evict_doc(coll, id_key),
        else => {},
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

const testing = std.testing;

const TmpLog = storage.TmpLog;

fn test_env(threaded: *std.Io.Threaded) struct { io: std.Io, gen: bson.ObjectIdGen } {
    const io = threaded.io();
    const gen = bson.ObjectIdGen.init(io);
    return .{ .io = io, .gen = gen };
}

fn make_doc(gpa: std.mem.Allocator, id: i32, name: []const u8) !bson.Document {
    var arena = std.heap.ArenaAllocator.init(gpa);
    errdefer arena.deinit();
    const pairs = try arena.allocator().alloc(bson.Pair, 2);
    pairs[0] = .{ .key = try arena.allocator().dupe(u8, "_id"), .value = .{ .int32 = id } };
    pairs[1] = .{ .key = try arena.allocator().dupe(u8, "name"), .value = .{ .string = try arena.allocator().dupe(u8, name) } };
    return .{ .arena = arena, .pairs = pairs };
}

test "insert, query, remove" {
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();

    var d1 = try make_doc(gpa, 1, "alice");
    defer d1.deinit();
    var d2 = try make_doc(gpa, 2, "bob");
    defer d2.deinit();

    try engine.lock();
    try engine.insert("app", "users", &d1, &env.gen);
    try engine.insert("app", "users", &d2, &env.gen);
    engine.unlock();

    // duplicate key
    var d3 = try make_doc(gpa, 1, "alice2");
    defer d3.deinit();
    try engine.lock();
    try testing.expectError(error.DuplicateKey, engine.insert("app", "users", &d3, &env.gen));
    engine.unlock();

    // find by id
    const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 });
    defer gpa.free(id_key);
    try engine.lock();
    const found = engine.get_doc("app", "users", id_key).?;
    try testing.expectEqualStrings("bob", (try bson.get_at(gpa, found, "name")).?.string);
    const removed = try engine.remove_by_id("app", "users", .{ .int32 = 2 });
    try testing.expect(removed);
    engine.unlock();
}

test "live/dead doc accounting drives compaction" {
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();
    // Keep compaction from firing and resetting dead_docs mid-test.
    engine.compact_threshold = std.math.maxInt(u64);

    var d1 = try make_doc(gpa, 1, "alice");
    defer d1.deinit();
    var d2 = try make_doc(gpa, 2, "bob");
    defer d2.deinit();

    try engine.lock();
    defer engine.unlock();

    try engine.insert("app", "users", &d1, &env.gen);
    try engine.insert("app", "users", &d2, &env.gen);
    try testing.expectEqual(@as(u64, 2), engine.live_docs);
    try testing.expectEqual(@as(u64, 0), engine.dead_docs);

    // A replace supersedes one record: live is unchanged, garbage grows.
    var d1b = try make_doc(gpa, 1, "alice2");
    defer d1b.deinit();
    try engine.replace("app", "users", &d1b, &env.gen);
    try testing.expectEqual(@as(u64, 2), engine.live_docs);
    try testing.expectEqual(@as(u64, 1), engine.dead_docs);

    // A delete drops a live doc and leaves its record behind as garbage.
    try testing.expect(try engine.remove_by_id("app", "users", .{ .int32 = 2 }));
    try testing.expectEqual(@as(u64, 1), engine.live_docs);
    try testing.expectEqual(@as(u64, 2), engine.dead_docs);

    // Removing something absent must not move either counter.
    try testing.expect(!try engine.remove_by_id("app", "users", .{ .int32 = 99 }));
    try testing.expectEqual(@as(u64, 1), engine.live_docs);
    try testing.expectEqual(@as(u64, 2), engine.dead_docs);

    // Dropping the collection accounts for everything it still held, and
    // must leave live_docs at zero rather than wrapping.
    try testing.expect(try engine.drop_collection("app", "users"));
    try testing.expectEqual(@as(u64, 0), engine.live_docs);
    try testing.expectEqual(@as(u64, 3), engine.dead_docs);
}

test "compaction reclaims garbage but leaves a garbage-free log alone" {
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();
    engine.compact_threshold = 4096; // small enough to be crossed here

    try engine.lock();
    defer engine.unlock();

    // Pure inserts produce no garbage, so the log must never be rewritten.
    for (0..200) |i| {
        var d = try make_doc(gpa, @intCast(i), "x");
        defer d.deinit();
        try engine.insert("app", "c", &d, &env.gen);
    }
    try engine.commit();
    try testing.expectEqual(@as(u64, 0), engine.dead_docs);
    // The compaction threshold counts data volume (uncompressed bytes), not
    // the compressed on-disk size.
    const after_insert = engine.log.data_bytes;
    try testing.expect(after_insert > engine.compact_threshold);

    // Rewriting every document makes the log mostly garbage; compaction
    // must fire and bring the file back down near the live size.
    for (0..200) |round| {
        for (0..200) |i| {
            var d = try make_doc(gpa, @intCast(i), if (round % 2 == 0) "yy" else "z");
            defer d.deinit();
            try engine.replace("app", "c", &d, &env.gen);
        }
        try engine.commit();
        if (engine.log.data_bytes >= after_insert * 2) break;
    }
    // The server runs compaction in a write command's epilogue; the tests
    // drive it directly.
    if (engine.take_compact()) try engine.compact();
    try engine.commit();
    try testing.expectEqual(@as(u64, 200), engine.live_docs);
    // Bounded well below the ~40x of record bytes those rewrites wrote.
    try testing.expect(engine.log.data_bytes < after_insert * 2);
}

test "reopen replays log" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        var d1 = try make_doc(gpa, 1, "alice");
        defer d1.deinit();
        var d2 = try make_doc(gpa, 2, "bob");
        defer d2.deinit();
        try engine.lock();
        try engine.insert("app", "users", &d1, &env.gen);
        try engine.insert("app", "users", &d2, &env.gen);
        _ = try engine.remove_by_id("app", "users", .{ .int32 = 2 });
        engine.unlock();
    }

    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 });
    defer gpa.free(id_key);
    try testing.expect(engine2.get_doc("app", "users", id_key) == null);
    const id_key1 = try bson.serialize_value(gpa, bson.Value{ .int32 = 1 });
    defer gpa.free(id_key1);
    try testing.expectEqualStrings("alice", (try bson.get_at(gpa, engine2.get_doc("app", "users", id_key1).?, "name")).?.string);
    engine2.unlock();
}

test "auto _id generation survives reopen" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();

        var doc = try make_doc(gpa, 0, "no-id-here");
        defer doc.deinit();
        // strip _id
        const stripped = doc.pairs[1..];

        var arena = std.heap.ArenaAllocator.init(gpa);
        defer arena.deinit();
        var d2 = try bson.Document.alloc(gpa, try arena.allocator().dupe(bson.Pair, stripped));
        defer d2.deinit();

        try engine.lock();
        try engine.insert("app", "no_ids", &d2, &env.gen);
        engine.unlock();
    }

    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    const coll = engine2.get_collection("app", "no_ids").?;
    var it = coll.docs.iterator();
    var count: usize = 0;
    while (it.next()) |entry| {
        count += 1;
        const b = coll.doc_bytes(entry.value_ptr.*);
        try testing.expect((try bson.get_at(gpa, b, "_id")).?.object_id.len == 12);
    }
    try testing.expectEqual(@as(usize, 1), count);
}

test "compaction rewrites log and keeps data" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        engine.compact_threshold = 1; // always compact
        defer engine.deinit();

        var docs: [4]bson.Document = undefined;
        defer for (&docs) |*d| d.deinit();
        try engine.lock();
        for (0..4) |i| {
            docs[i] = try make_doc(gpa, @intCast(i + 1), "user-{d}");
            try engine.insert("app", "users", &docs[i], &env.gen);
        }
        try engine.commit();
        // threshold 1 makes every write want a compaction; run it.
        if (engine.take_compact()) try engine.compact();
        engine.unlock();
    }

    // Reopen after compaction and keep writing: with the log reopened at
    // end_pos 0, appends would clobber the compacted records.
    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    var extra = try make_doc(gpa, 5, "eve");
    defer extra.deinit();
    try engine2.insert("app", "users", &extra, &env.gen);
    try engine2.commit();
    engine2.unlock();

    var engine3 = try Engine.open(gpa, io, tmp.path);
    defer engine3.deinit();
    try engine3.lock();
    for (1..6) |i| {
        const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) });
        defer gpa.free(id_key);
        try testing.expect(engine3.get_doc("app", "users", id_key) != null);
    }
    engine3.unlock();
}

test "concurrent readers and writers on a threaded Io" {
    // Real worker threads: writers hold the exclusive lock, readers the
    // shared lock. Proves the RwLock split keeps committed writes visible
    // to concurrent readers and never corrupts the maps.
    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();

    const writers = 4;
    const readers = 4;
    const per_writer: i32 = 200;
    const total: i32 = writers * per_writer;
    var next_id = std.atomic.Value(i32).init(1);
    var remaining = std.atomic.Value(usize).init(@intCast(total));

    const Worker = struct {
        fn writer(
            e: *Engine,
            id_counter: *std.atomic.Value(i32),
            pending: *std.atomic.Value(usize),
            alloc: std.mem.Allocator,
        ) error{Canceled}!void {
            while (true) {
                const id = id_counter.fetchAdd(1, .monotonic);
                if (id > total) return;
                var doc = make_doc(alloc, id, "user") catch return error.Canceled;
                defer doc.deinit();
                e.lock() catch return error.Canceled;
                defer e.unlock();
                e.insert("app", "users", &doc, undefined) catch return error.Canceled;
                _ = pending.fetchSub(1, .monotonic);
            }
        }

        fn reader(e: *Engine, pending: *std.atomic.Value(usize)) error{Canceled}!void {
            while (pending.load(.acquire) > 0) {
                e.lock_read() catch return error.Canceled;
                defer e.unlock_read();
                if (e.get_collection("app", "users")) |coll| {
                    var n: usize = 0;
                    var it = coll.docs.iterator();
                    while (it.next()) |_| n += 1;
                    // A reader must never observe more docs than can exist.
                    if (n > @as(usize, @intCast(total))) return error.Canceled;
                }
            }
        }
    };

    var group: std.Io.Group = .init;
    defer group.cancel(io);
    for (0..readers) |_| group.async(io, Worker.reader, .{ &engine, &remaining });
    for (0..writers) |_| group.async(io, Worker.writer, .{ &engine, &next_id, &remaining, gpa });
    try group.await(io);

    // Every committed write must be visible once all writers finish.
    try engine.lock_read();
    defer engine.unlock_read();
    const coll = engine.get_collection("app", "users") orelse return error.TestUnexpectedResult;
    try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count());
    for (1..total + 1) |i| {
        const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) });
        defer gpa.free(id_key);
        try testing.expect(engine.get_doc("app", "users", id_key) != null);
    }
}

test "compact yields to a compaction already in flight" {
    // The guard's contract, checked deterministically. Two compactions at once
    // share one tmp path and each ends in a rename onto the log, so the second
    // truncates and rewrites the file the first is about to publish -- and the
    // first then renames whatever the second left there over the live log.
    //
    // The real interleaving is hard to force: `compact_snapshot_coll` holds each
    // collection's write lock while writing its snapshot, so two compactions
    // serialize there, and an insert cannot re-arm `compact_pending` while that
    // lock is held either. The overlap window is only between the end of one
    // snapshot and its rename. Rather than race for it, drive the flag directly
    // and pin what the guard promises.
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();
    engine.compact_threshold = 1;

    // Leave real garbage behind, so a compaction that ran would be visible:
    // `compact` resets dead_docs to zero and nothing else does.
    try engine.lock();
    var doc = try make_doc(gpa, 1, "alice");
    defer doc.deinit();
    try engine.insert("app", "users", &doc, &env.gen);
    var doc2 = try make_doc(gpa, 1, "alice-again");
    defer doc2.deinit();
    // A replace supersedes the first record, leaving it behind as garbage.
    try engine.replace("app", "users", &doc2, &env.gen);
    try engine.commit();
    engine.unlock();
    try testing.expect(engine.dead_docs > 0);
    const dead_before = engine.dead_docs;

    // With a compaction "in flight", compact must return without rewriting.
    engine.compacting.store(true, .release);
    try engine.compact();
    try testing.expectEqual(dead_before, engine.dead_docs);

    // With the slot free, the same call does the work -- proving the assertion
    // above came from the guard and not from there being nothing to do.
    engine.compacting.store(false, .release);
    try engine.compact();
    try testing.expectEqual(@as(u64, 0), engine.dead_docs);
}

test "concurrent writers compacting: the log survives a reopen" {
    // Real worker threads driving compaction while other writers append, each
    // following the lock sequence the server's dispatch uses (catalog ->
    // collection -> release both -> commit -> compact). Every other compaction
    // test is single-threaded, so this is the only coverage of the whole write
    // path under genuine contention.
    //
    // What it proves: concurrent compaction leaves a log that replays to
    // exactly the right documents. The count is checked exactly -- too few
    // means a rewrite was published half-written, too many means a stale tmp
    // tail was replayed as live data.
    //
    // What it does not prove: that either specific race is fixed. Both windows
    // are too narrow to hit reliably (see the test above), and this test passes
    // with the `compacting` guard removed. It is a smoke test for the path, not
    // a regression test for the guard; the deterministic tests above and in
    // storage.zig are what pin those two invariants.
    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);

    const writers = 4;
    const per_writer: i32 = 60;
    const total: i32 = writers * per_writer;

    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        // Every write wants a compaction, so writers pile into compact() with
        // maximum overlap -- the point of the test.
        engine.compact_threshold = 1;

        var next_id = std.atomic.Value(i32).init(1);

        const Worker = struct {
            /// The in-lock half of a write command: the collection lock is
            /// taken under the catalog lock, and both are released on return --
            /// so the caller's commit runs holding neither, exactly as the
            /// server's dispatch epilogue does.
            fn insert_locked(e: *Engine, doc: *bson.Document) !void {
                try e.lock_catalog(false);
                defer e.unlock_catalog(false);
                const coll = try e.lock_collection("app", "users", true, true);
                if (coll) |c| {
                    defer e.unlock_collection(c, true);
                    try e.insert("app", "users", doc, undefined);
                }
            }

            fn writer(
                e: *Engine,
                id_counter: *std.atomic.Value(i32),
                alloc: std.mem.Allocator,
            ) error{Canceled}!void {
                while (true) {
                    const id = id_counter.fetchAdd(1, .monotonic);
                    if (id > total) return;
                    var doc = make_doc(alloc, id, "user") catch return error.Canceled;
                    defer doc.deinit();
                    // Ids come from the shared counter, so no insert here can
                    // legitimately fail; any error is a real defect.
                    insert_locked(e, &doc) catch return error.Canceled;
                    e.commit() catch return error.Canceled;
                    if (e.take_compact()) e.compact() catch return error.Canceled;
                }
            }
        };

        var group: std.Io.Group = .init;
        defer group.cancel(io);
        for (0..writers) |_| group.async(io, Worker.writer, .{ &engine, &next_id, gpa });
        try group.await(io);
    }

    // Reopen from disk: this replays the log that compaction left behind, which
    // is the only place the races above are observable.
    var reopened = try Engine.open(gpa, io, tmp.path);
    defer reopened.deinit();
    try reopened.lock_read();
    defer reopened.unlock_read();
    const coll = reopened.get_collection("app", "users") orelse return error.TestUnexpectedResult;
    try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count());
    for (1..total + 1) |i| {
        const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) });
        defer gpa.free(id_key);
        try testing.expect(reopened.get_doc("app", "users", id_key) != null);
    }
}

// -- index tests -----------------------------------------------------------

/// A spec document for a single-path index, built by serializing and
/// re-parsing so the pairs are arena-owned.
fn index_spec(
    gpa: std.mem.Allocator,
    path: []const u8,
    name: []const u8,
    unique: bool,
    sparse: bool,
    ttl: ?i64,
) !bson.Document {
    var out: std.ArrayListUnmanaged(u8) = .empty;
    defer out.deinit(gpa);
    var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
    defer pairs.deinit(gpa);
    try pairs.appendSlice(gpa, &.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = path, .value = .{ .int32 = 1 } }} } },
        .{ .key = "name", .value = .{ .string = name } },
        .{ .key = "unique", .value = .{ .bool = unique } },
        .{ .key = "sparse", .value = .{ .bool = sparse } },
    });
    if (ttl) |secs| try pairs.append(gpa, .{ .key = "expireAfterSeconds", .value = .{ .int64 = secs } });
    try bson.write_doc(pairs.items, gpa, &out);
    return bson.Document.parse(gpa, out.items);
}

/// Number of entries the named index has for a single-value equality key.
fn index_count(
    gpa: std.mem.Allocator,
    engine: *Engine,
    db_name: []const u8,
    coll_name: []const u8,
    name: []const u8,
    key_value: bson.Value,
) !usize {
    const coll = engine.get_collection(db_name, coll_name) orelse return 0;
    for (coll.indexes.items) |ix| {
        if (std.mem.eql(u8, ix.name, name)) {
            var out: std.ArrayListUnmanaged(u64) = .empty;
            defer out.deinit(gpa);
            try ix.lookup_eq(gpa, &.{key_value}, &out);
            return out.items.len;
        }
    }
    return 0;
}

fn make_user(gpa: std.mem.Allocator, id: i32, email: []const u8) !bson.Document {
    var arena = std.heap.ArenaAllocator.init(gpa);
    errdefer arena.deinit();
    const pairs = try arena.allocator().alloc(bson.Pair, 2);
    pairs[0] = .{ .key = try arena.allocator().dupe(u8, "_id"), .value = .{ .int32 = id } };
    pairs[1] = .{ .key = try arena.allocator().dupe(u8, "email"), .value = .{ .string = try arena.allocator().dupe(u8, email) } };
    return .{ .arena = arena, .pairs = pairs };
}

test "unique index enforced on insert, replace, and upsert-conflict" {
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();

    var spec = try index_spec(gpa, "email", "email_1", true, false, null);
    defer spec.deinit();
    try engine.lock();
    _ = try engine.create_index("app", "users", &spec);

    var d1 = try make_user(gpa, 1, "a@x.io");
    defer d1.deinit();
    try engine.insert("app", "users", &d1, &env.gen);

    // A second doc with the same email is rejected and never logged.
    var d2 = try make_user(gpa, 2, "a@x.io");
    defer d2.deinit();
    try testing.expectError(error.DuplicateKeyIndex, engine.insert("app", "users", &d2, &env.gen));
    try testing.expectEqualStrings("email_1", engine.get_collection("app", "users").?.dup_index.?);

    // A replace that keeps its own email is fine (own entries excluded).
    var d1b = try make_user(gpa, 1, "a@x.io");
    defer d1b.deinit();
    try engine.replace("app", "users", &d1b, &env.gen);
    try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "users", "email_1", .{ .string = "a@x.io" }));

    // An update that would collide is rejected.
    var d2b = try make_user(gpa, 2, "a@x.io");
    defer d2b.deinit();
    try testing.expectError(error.DuplicateKeyIndex, engine.replace("app", "users", &d2b, &env.gen));

    // A different email still inserts.
    var d3 = try make_user(gpa, 3, "b@x.io");
    defer d3.deinit();
    try engine.insert("app", "users", &d3, &env.gen);
    engine.unlock();
}

test "index maintained across update and delete" {
    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);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();

    var spec = try index_spec(gpa, "a", "a_1", false, false, null);
    defer spec.deinit();
    try engine.lock();
    _ = try engine.create_index("app", "items", &spec);

    var d1 = try doc_with_a(gpa, 1, 10);
    defer d1.deinit();
    var d2 = try doc_with_a(gpa, 2, 20);
    defer d2.deinit();
    try engine.insert("app", "items", &d1, &env.gen);
    try engine.insert("app", "items", &d2, &env.gen);
    try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "items", "a_1", .{ .int32 = 20 }));

    // Replace doc 1 with a new value: old entry gone, new entry present.
    var d1b = try doc_with_a(gpa, 1, 30);
    defer d1b.deinit();
    try engine.replace("app", "items", &d1b, &env.gen);
    try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "items", "a_1", .{ .int32 = 10 }));
    try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "items", "a_1", .{ .int32 = 30 }));

    // Delete doc 2: its entry is removed.
    _ = try engine.remove_by_id("app", "items", .{ .int32 = 2 });
    try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "items", "a_1", .{ .int32 = 20 }));
    engine.unlock();
}

/// A document with an integer `a` field (on top of _id + name).
fn doc_with_a(gpa: std.mem.Allocator, id: i32, a: i32) !bson.Document {
    var arena = std.heap.ArenaAllocator.init(gpa);
    errdefer arena.deinit();
    const pairs = try arena.allocator().alloc(bson.Pair, 3);
    pairs[0] = .{ .key = try arena.allocator().dupe(u8, "_id"), .value = .{ .int32 = id } };
    pairs[1] = .{ .key = try arena.allocator().dupe(u8, "name"), .value = .{ .string = try arena.allocator().dupe(u8, "x") } };
    pairs[2] = .{ .key = try arena.allocator().dupe(u8, "a"), .value = .{ .int32 = a } };
    return .{ .arena = arena, .pairs = pairs };
}

test "index survives reopen and compaction" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        engine.compact_threshold = 1; // every write compacts
        var spec = try index_spec(gpa, "email", "email_1", false, false, null);
        defer spec.deinit();
        try engine.lock();
        _ = try engine.create_index("app", "users", &spec);
        var d1 = try make_user(gpa, 1, "a@x.io");
        defer d1.deinit();
        var d2 = try make_user(gpa, 2, "b@x.io");
        defer d2.deinit();
        try engine.insert("app", "users", &d1, &env.gen);
        try engine.insert("app", "users", &d2, &env.gen);
        engine.unlock();
    }

    // Reopen: the index (rebuilt from the compacted log) still finds docs.
    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine2, "app", "users", "email_1", .{ .string = "b@x.io" }));
    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 "a checkpoint reclaims the log and the data survives" {
    // The payoff of a lagging checkpoint: once the data file holds the effect of
    // a record, the record is redundant and the log can be reclaimed. Without
    // this the log only ever grows and every open pays for every write ever made.
    //
    // Mutation check, red: skipping the truncation.
    //
    // Not covered: moving the truncation *before* the publish. That is still
    // correct in the absence of a crash -- the publish follows immediately -- and
    // the hazard is precisely a crash landing between the two, with the records
    // gone from the log and not yet in any image. Catching it needs process-level
    // crash injection, which the milestone's gates cover; an in-process test
    // cannot express "stop here and die". The order stays because it is the
    // whole reason a lagging checkpoint is safe.
    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 log_after_checkpoint: u64 = 0;
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        try engine.lock();
        var i: i32 = 0;
        while (i < 200) : (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();

        // Commit first, so the records are actually on disk: appends buffer in
        // the log's open block, and only a commit seals and writes it. Measuring
        // before that reads a file that is still just its header.
        try engine.commit();
        const before = try engine.log.file.length(io);
        try testing.expect(before > storage.file_header_len);

        try engine.checkpoint();
        log_after_checkpoint = try engine.log.file.length(io);
        // The log is back to just its header.
        try testing.expect(log_after_checkpoint < before);
        try testing.expectEqual(@as(u64, storage.file_header_len), log_after_checkpoint);
        // And writing still works afterwards, at a sequence above the watermark.
        try engine.lock();
        var d = try make_user(gpa, 999, "z@x.io");
        defer d.deinit();
        try engine.insert("app", "users", &d, &env.gen);
        engine.unlock();
        try testing.expect(engine.seq > engine.pager.loaded.seq);
    }

    // Everything is still there after a reopen: 200 from the image, 1 from the
    // log records written after the truncation.
    {
        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, 201), coll.docs.count());
        try testing.expectEqual(@as(usize, 201), coll.id_index.count());
        const id_key = try bson.serialize_value(gpa, .{ .int32 = 999 });
        defer gpa.free(id_key);
        try testing.expect(engine.get_doc("app", "users", id_key) != null);
        const first_key = try bson.serialize_value(gpa, .{ .int32 = 0 });
        defer gpa.free(first_key);
        try testing.expect(engine.get_doc("app", "users", first_key) != null);
    }
}

test "index drop survives reopen" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        var spec = try index_spec(gpa, "email", "email_1", false, false, null);
        defer spec.deinit();
        try engine.lock();
        _ = try engine.create_index("app", "users", &spec);
        var d1 = try make_user(gpa, 1, "a@x.io");
        defer d1.deinit();
        try engine.insert("app", "users", &d1, &env.gen);
        try testing.expect(try engine.drop_index("app", "users", "email_1"));
        engine.unlock();
    }

    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    try testing.expectEqual(@as(usize, 0), engine2.get_collection("app", "users").?.indexes.items.len);
    engine2.unlock();
}

test "dropping an index does not move its siblings" {
    // Indexes used to be stored by value, so `orderedRemove` memmoved the
    // whole list and every `*Index` already handed out -- notably a query
    // plan's `index` field -- silently referred to a *different* index
    // afterwards. Nothing caught it: the collection's own bookkeeping stayed
    // consistent, so only a caller holding a pointer across a drop would see
    // it, and none of the tests did.
    //
    // Mutation check: restore `indexes` to ArrayListUnmanaged(index.Index)
    // (with the by-value append/remove that goes with it) and the b_1
    // assertion below reads "c_1", because slot 1 now holds what used to be in
    // slot 2.
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const env = test_env(&threaded);
    const io = env.io;
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();
    try engine.lock();
    defer engine.unlock();

    for ([_][]const u8{ "a", "b", "c" }) |field| {
        const name = try std.fmt.allocPrint(gpa, "{s}_1", .{field});
        defer gpa.free(name);
        var spec = try index_spec(gpa, field, name, false, false, null);
        defer spec.deinit();
        _ = try engine.create_index("app", "users", &spec);
    }

    const coll = engine.get_collection("app", "users").?;
    // Hold pointers across the drop, which is the whole point.
    const b_ix = coll.find_index("b_1").?;
    const c_ix = coll.find_index("c_1").?;

    try testing.expect(try engine.drop_index("app", "users", "a_1"));

    try testing.expectEqual(@as(usize, 2), coll.indexes.items.len);
    try testing.expectEqualStrings("b_1", b_ix.name);
    try testing.expectEqualStrings("c_1", c_ix.name);
    // And they are still the collection's own indexes, not detached copies.
    try testing.expectEqual(b_ix, coll.find_index("b_1").?);
    try testing.expectEqual(c_ix, coll.find_index("c_1").?);
}

test "drop_collection frees indexes; log without index records replays" {
    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);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        var spec = try index_spec(gpa, "email", "email_1", false, false, null);
        defer spec.deinit();
        try engine.lock();
        _ = try engine.create_index("app", "users", &spec);
        var d1 = try make_user(gpa, 1, "a@x.io");
        defer d1.deinit();
        try engine.insert("app", "users", &d1, &env.gen);
        // Dropped in memory; free_collection releases the index memory
        // (verified by testing.allocator at engine.deinit).
        try testing.expect(try engine.drop_collection("app", "users"));
        try testing.expect(engine.get_collection("app", "users") == null);
        // A log that only ever contained plain upserts replays fine.
        var d2 = try make_doc(gpa, 2, "bob");
        defer d2.deinit();
        try engine.insert("app", "plain", &d2, &env.gen);
        engine.unlock();
    }

    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 });
    defer gpa.free(id_key);
    try testing.expect(engine2.get_doc("app", "plain", id_key) != null);
    // Pre-existing limitation (documented in the README): drop_collection
    // writes no log record, so the collection and its index resurrect.
    const users = engine2.get_collection("app", "users").?;
    try testing.expectEqual(@as(usize, 1), users.indexes.items.len);
    try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine2, "app", "users", "email_1", .{ .string = "a@x.io" }));
    engine2.unlock();
}

/// A document with an `expireAt` field of any type (omitted when null).
fn doc_with_expire(gpa: std.mem.Allocator, id: i32, expire: ?bson.Value) !bson.Document {
    var arena = std.heap.ArenaAllocator.init(gpa);
    errdefer arena.deinit();
    const n: usize = if (expire == null) 1 else 2;
    const pairs = try arena.allocator().alloc(bson.Pair, n);
    pairs[0] = .{ .key = try arena.allocator().dupe(u8, "_id"), .value = .{ .int32 = id } };
    if (expire) |v| {
        const value = switch (v) {
            .string => |s| bson.Value{ .string = try arena.allocator().dupe(u8, s) },
            else => v,
        };
        pairs[1] = .{ .key = try arena.allocator().dupe(u8, "expireAt"), .value = value };
    }
    return .{ .arena = arena, .pairs = pairs };
}

test "ttl_sweep deletes expired documents and the deletion survives reopen" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    var env = test_env(&threaded);
    const io = env.io;
    const gpa = testing.allocator;

    // A fixed clock: the sweep takes `now` as a parameter precisely so the
    // test does not depend on the wall clock.
    const now_ms: i64 = 1_700_000_000_000;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    {
        var engine = try Engine.open(gpa, io, tmp.path);
        defer engine.deinit();
        var spec = try index_spec(gpa, "expireAt", "expireAt_1", false, false, 60);
        defer spec.deinit();
        try engine.lock();
        defer engine.unlock();
        _ = try engine.create_index("app", "sessions", &spec);

        const docs = [_]struct { id: i32, expire: ?bson.Value }{
            .{ .id = 1, .expire = .{ .datetime = now_ms - 120_000 } }, // long expired
            .{ .id = 2, .expire = .{ .datetime = now_ms - 60_000 } }, // exactly at the cutoff
            .{ .id = 3, .expire = .{ .datetime = now_ms - 30_000 } }, // not yet
            .{ .id = 4, .expire = .{ .datetime = now_ms + 3_600_000 } }, // future
            .{ .id = 5, .expire = .{ .string = "tomorrow" } }, // not a date: never expires
            .{ .id = 6, .expire = null }, // no field: indexed as null
        };
        for (docs) |d| {
            var doc = try doc_with_expire(gpa, d.id, d.expire);
            defer doc.deinit();
            try engine.insert("app", "sessions", &doc, &env.gen);
        }
        const coll = engine.get_collection("app", "sessions").?;
        try testing.expectEqual(@as(usize, 6), coll.docs.count());
        try testing.expectEqual(@as(usize, 6), coll.indexes.items[0].count());

        // The cutoff is inclusive: doc 2 goes with doc 1.
        try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms));
        try testing.expectEqual(@as(usize, 4), coll.docs.count());
        try testing.expectEqual(@as(usize, 4), coll.indexes.items[0].count());
        try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .datetime = now_ms - 120_000 }));
        // The string and the missing field are untouched by any sweep.
        try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .string = "tomorrow" }));
        try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .null));

        // Idempotent: nothing else is expired at the same instant.
        try testing.expectEqual(@as(usize, 0), try engine.ttl_sweep(now_ms));
        // An hour later doc 3 has expired too; doc 4 still has not.
        try testing.expectEqual(@as(usize, 1), try engine.ttl_sweep(now_ms + 3_000_000));
        try testing.expectEqual(@as(usize, 3), coll.docs.count());
    }

    // Sweeps go through `remove`, so they are logged: the deletions hold
    // across a restart, and the TTL index comes back with its expiry.
    var engine2 = try Engine.open(gpa, io, tmp.path);
    defer engine2.deinit();
    try engine2.lock();
    defer engine2.unlock();
    const coll = engine2.get_collection("app", "sessions").?;
    try testing.expectEqual(@as(usize, 3), coll.docs.count());
    try testing.expectEqual(@as(usize, 1), coll.indexes.items.len);
    try testing.expectEqual(@as(?i64, 60), coll.indexes.items[0].ttl);
    try testing.expectEqual(@as(usize, 3), coll.indexes.items[0].count());
    for ([_]i32{ 1, 2, 3 }) |id| {
        const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = id });
        defer gpa.free(id_key);
        try testing.expect(engine2.get_doc("app", "sessions", id_key) == null);
    }
    const alive = try bson.serialize_value(gpa, bson.Value{ .int32 = 4 });
    defer gpa.free(alive);
    try testing.expect(engine2.get_doc("app", "sessions", alive) != null);
}

test "ttl_sweep spans collections and several TTL indexes on one collection" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    var env = test_env(&threaded);
    const io = env.io;
    const gpa = testing.allocator;
    const now_ms: i64 = 1_700_000_000_000;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    var engine = try Engine.open(gpa, io, tmp.path);
    defer engine.deinit();
    try engine.lock();
    defer engine.unlock();

    // Two TTL indexes over the same collection (MongoDB allows this): one
    // document is expired by both, and must only be deleted once.
    var spec_a = try index_spec(gpa, "expireAt", "expireAt_1", false, false, 60);
    defer spec_a.deinit();
    var spec_b = try index_spec(gpa, "seenAt", "seenAt_1", false, false, 10);
    defer spec_b.deinit();
    _ = try engine.create_index("app", "sessions", &spec_a);
    _ = try engine.create_index("app", "sessions", &spec_b);

    var both = try bson.Document.alloc(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
        .{ .key = "expireAt", .value = .{ .datetime = now_ms - 120_000 } },
        .{ .key = "seenAt", .value = .{ .datetime = now_ms - 120_000 } },
    });
    defer both.deinit();
    try engine.insert("app", "sessions", &both, &env.gen);

    // A second collection with its own TTL index, and a plain collection
    // that no sweep may touch.
    var spec_c = try index_spec(gpa, "at", "at_1", false, false, 0);
    defer spec_c.deinit();
    _ = try engine.create_index("app", "events", &spec_c);
    var ev = try bson.Document.alloc(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = 2 } },
        // expireAfterSeconds 0: expires at exactly the stored instant.
        .{ .key = "at", .value = .{ .datetime = now_ms } },
    });
    defer ev.deinit();
    try engine.insert("app", "events", &ev, &env.gen);
    var plain = try make_doc(gpa, 3, "keep");
    defer plain.deinit();
    try engine.insert("other", "plain", &plain, &env.gen);

    try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms));
    try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "sessions").?.docs.count());
    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);
    }
};