storage/db: XxHash3 record integrity, garbage-ratio compaction

Two streams of work land together: they are interleaved in storage.zig
and db.zig and only build as a unit.

Already in the working tree before this session:
  - ReleaseFast as the default zig build (Debug was 10-200x slower)
  - group commit: one fsync per write command instead of per document
  - plan_id returned a pointer to a stack temporary; ReleaseFast read
    garbage and silently broke findOne({_id: ObjectId})
  - perf suite: big.js, compare.js, compare-run.sh, e2e6.js

Phase 1 performance work:

Record integrity hash CRC32 -> XxHash3. std.hash.Crc32 is the
table-driven byte-at-a-time Crc32IsoHdlc, measured at 408 MB/s against
XxHash3's 31 GB/s: 38us versus 0.5us on a 16 KiB document, which was
about two thirds of the entire bulk-insert cost. The record header
grows from u32 crc to u64 hash (header_len 20 -> 24), a breaking
format change. Bulk insert 260 -> 700 MB/s, reopen 1.1 -> 0.5s.

Compaction fsynced once per live document, because Log.open leaves
defer_sync false and compact never set it. It now issues one sync for
the whole rewrite, before the rename that publishes it.

Compaction triggers on the share of the log that is garbage
(live_docs/dead_docs, maintained at evict_doc, the single point where
a document dies) rather than on bytes appended. A fixed byte count is
wrong in both directions: a 1 GB bulk load holds no garbage at all yet
would compact ~64 times under the 16 MiB default, rewriting 1 GB each
time, while a small collection rewritten in place accumulates garbage
indefinitely without ever reaching the count. Pure inserts now never
compact, and the file stays near 1.25x the live data. Bulk load at the
default threshold: 41.6 -> 702.7 MB/s.

remove() never called maybe_compact, so a delete-heavy workload grew
the log without bound.

e2e6's compaction check required the file to bloat past 30 MiB before
being reclaimed, which encoded the old policy and failed on strictly
better behaviour (ends at 15.8 MB against ~12 MB live, was ~30 MB). It
now asserts the file ends near the live size and peaked well above it,
which does not depend on when the trigger fires. Sampling interval 50
-> 10ms: the operations now finish inside the old window.

Verified: 70 unit tests under both ReleaseFast and ReleaseSafe, e2e
29/16/17/3/2 checks, the crash-a/kill -9/crash-b pair, and e2e6 72/72
across three consecutive runs.

src/db.zig

@@ -55,7 +55,15 @@ pub const Engine = struct {
     log: storage.Log,
     dbs: std.StringHashMapUnmanaged(Db),
     seq: u64,
+    /// Floor for the compaction trigger. The real trigger also scales with
+    /// the live data size — see `maybe_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 `maybe_compact`.
+    live_docs: u64 = 0,
+    dead_docs: u64 = 0,
     /// 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.
@@ -97,6 +105,9 @@ pub const Engine = struct {
     /// 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.
+        self.live_docs -= coll.docs.count();
+        self.dead_docs += coll.docs.count();
         for (coll.indexes.items) |*ix| ix.deinit(self.gpa);
         coll.indexes.deinit(self.gpa);
         var doc_it = coll.docs.iterator();
@@ -129,6 +140,9 @@ pub const Engine = struct {
         old.value.*.deinit();
         self.gpa.destroy(old.value);
         self.gpa.free(old.key);
+        // This document's log record just became garbage.
+        self.live_docs -= 1;
+        self.dead_docs += 1;
     }
 
     // -- commands (callers must hold the matching lock) ---------------------
@@ -152,6 +166,21 @@ pub const Engine = struct {
         self.rwlock.unlockShared(self.io);
     }
 
+    /// Group commit: defer per-record fsyncs until end_batch. Callers must
+    /// hold the write lock and pair every begin with an end (the command's
+    /// defer). Every document published before end_batch is fsynced by it,
+    /// so an acknowledged multi-write command is durable as a unit — the
+    /// same crash guarantee as the old fsync-per-record, with one sync per
+    /// command instead of one per document.
+    pub fn begin_batch(self: *Engine) void {
+        self.log.defer_sync = true;
+    }
+
+    pub fn end_batch(self: *Engine) !void {
+        self.log.defer_sync = false;
+        try self.log.sync();
+    }
+
     /// 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 {
@@ -240,6 +269,7 @@ pub const Engine = struct {
 
         // 7. Publish the document and its entries.
         try coll.docs.put(self.gpa, id_key, owned);
+        self.live_docs += 1;
         for (built_list.items) |*b| {
             if (b.built.multikey) b.ix.multikey = true;
             b.ix.insert_entries(&b.built);
@@ -273,6 +303,9 @@ pub const Engine = struct {
         try self.log.append_delete(db_name, coll_name, id_doc.items, self.seq);
 
         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.
+        try self.maybe_compact();
         return true;
     }
 
@@ -477,8 +510,34 @@ pub const Engine = struct {
         return owned;
     }
 
+    /// 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.
     fn maybe_compact(self: *Engine) !void {
-        if (self.log.log_bytes < self.compact_threshold) return;
+        if (self.log.end_pos < 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;
         try self.compact();
     }
 
@@ -490,6 +549,10 @@ pub const Engine = struct {
         std.Io.Dir.cwd().deleteFile(self.io, tmp_path) catch {};
         var new_log = try storage.Log.open(self.gpa, self.io, tmp_path);
         defer new_log.close();
+        // One fsync for the whole rewrite, not one per document. The
+        // rewrite's durability comes from the rename below, which is only
+        // safe to publish after a single sync of the finished file.
+        new_log.defer_sync = true;
 
         var db_it = self.dbs.iterator();
         while (db_it.next()) |db_entry| {
@@ -513,6 +576,8 @@ pub const Engine = struct {
             }
         }
         const new_end_pos = new_log.end_pos;
+        // Durable before the rename makes it the database.
+        try new_log.sync();
 
         try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io);
         // Persist the rename: fsync the parent directory so the new
@@ -523,11 +588,18 @@ pub const Engine = struct {
         try dir_file.sync(self.io);
 
         const old_path = try self.gpa.dupe(u8, self.log.path);
+        // A batch may span a compaction (a large insert crossing the
+        // threshold): keep the deferred-sync mode on the swapped-in log so
+        // the remaining batch records stay grouped with the same command.
+        const deferred = self.log.defer_sync;
         self.log.close();
         self.log = try storage.Log.open(self.gpa, self.io, old_path);
+        self.log.defer_sync = deferred;
         // Log.open starts at end_pos 0 and does not replay; continue appending
         // where the compacted file actually ends.
         self.log.end_pos = new_end_pos;
+        // The rewritten log holds only live documents.
+        self.dead_docs = 0;
         self.gpa.free(old_path);
     }
 
@@ -637,6 +709,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
         storage.record_type_upsert => {
             self.evict_doc(coll, id_key);
             try coll.docs.put(self.gpa, id_key, doc);
+            self.live_docs += 1;
             key_owned = true;
             stored = true;
         },
@@ -708,6 +781,98 @@ test "insert, query, remove" {
     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 testing.expectEqual(@as(u64, 0), engine.dead_docs);
+    const after_insert = engine.log.end_pos;
+    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);
+        }
+        if (engine.log.end_pos < after_insert * 2) break;
+    }
+    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.end_pos < after_insert * 2);
+}
+
 test "reopen replays log" {
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();