db/commands: acknowledged writes reach the disk again

Three defects, each of which made the database lose data that had already
been acknowledged, or answer a client with a malformed reply.

- Engine.commit decided a writer was "already covered" by comparing
  log.end_pos with the position of the last completed commit. Under block
  framing an append leaves its bytes in the log's open in-memory block and
  does not move end_pos -- only sealing does. So once the first commit had
  set committed_end = end_pos, every later write command found itself
  covered and returned without sealing or syncing anything. A no-op
  deleteMany followed by insertMany(50) was acknowledged with the file
  still 16 bytes (its header) and lost all 50 documents on kill -9, which
  is precisely what e2e2's crash pair does. Coverage is now decided by
  sequence number, which counts records rather than bytes on disk.

- Compaction read the new log's end position before syncing it, but the
  sync is what seals the open block, and the seal is what moves end_pos
  past it. Appends after a compaction therefore started inside the
  compacted file's last block and overwrote it, so those documents were
  gone at the next replay: e2e6's phase 2 ended with 1000 documents in
  memory and 996 after a graceful restart.

- cmd_find returned early on a missing namespace without putting anything
  in the reply, so a find on an unknown collection arrived at the driver as
  a response with no `ok` field ("MongoServerError: n/a") instead of an
  empty cursor. The other commands' missing-namespace paths were fine.

Verified with the unit suite in ReleaseFast/ReleaseSafe/Debug, the split
fuzzer, all six e2e suites (e2e6 back to 72/72) and the kill -9 crash pair
-- none of which passed beforehand -- plus 13 kill -9 runs over 1/2/8
connections with 1200 acknowledged inserts each and nothing lost.

tests/e2e/results/phase7.txt records the benchmark with the fixes in place:
no regression against phase6 (bulk 739 -> 753 MB/s, updateMany 1.9 -> 2.0
ms, RSS 547 -> 546 MB), and concurrent durable writes now measurable at
7.1k/15.0k/21.8k docs/s over 1/8/32 connections.

src/commands.zig

@@ -624,7 +624,14 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
 
     // Sorting and emitting need the documents as trees; materialize the
     // matched page into the reply arena (the slab itself is never copied).
-    const coll = ctx.engine.get_collection(db_name, coll_name) orelse return;
+    const coll = ctx.engine.get_collection(db_name, coll_name) orelse {
+        // A find on a namespace that does not exist is an empty cursor, not
+        // an error -- and above all not a reply with no `ok` at all, which
+        // is what returning here without one sends.
+        const none: []const *const bson.Document = &.{};
+        try emit_docs_tree(reply, db_name, coll_name, proj_pairs, none);
+        return reply.put_ok();
+    };
     // Lives in the reply arena; freed with it.
     var tree_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty;
     const arena = reply.arena_alloc();

src/db.zig

@@ -137,8 +137,11 @@ pub const Engine = struct {
     /// Serializes commit decisions; the group-commit leader holds it while
     /// sealing and syncing.
     commit_lock: std.Io.Mutex = .init,
-    /// Log end position covered by the last completed commit.
-    committed_end: u64 = 0,
+    /// 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.
@@ -327,13 +330,16 @@ pub const Engine = struct {
     pub fn commit(self: *Engine) !void {
         try self.commit_lock.lock(self.io);
         defer self.commit_lock.unlock(self.io);
+        // 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.
         while (self.committing) self.commit_done.wait(self.io, &self.commit_lock) catch return;
-        try self.log_lock.lock(self.io);
-        const covered = self.log.end_pos == self.committed_end;
-        self.log_lock.unlock(self.io);
-        if (covered) {
+        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
@@ -351,7 +357,9 @@ pub const Engine = struct {
         try self.log_lock.lock(self.io);
         defer self.log_lock.unlock(self.io);
         try self.log.sync();
-        self.committed_end = self.log.end_pos;
+        // 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;
         done = true;
         self.committing = false;
         self.commit_done.signal(self.io);
@@ -837,9 +845,13 @@ pub const Engine = struct {
                 continue; // a writer appended during the snapshot; retry
             }
             errdefer self.log_lock.unlock(self.io);
-            const new_end_pos = new_log.end_pos;
             // Durable before the rename makes it the database.
             try new_log.sync();
+            // After the sync, not before: sync seals the open block, and
+            // that seal is what moves end_pos past it. Reading the position
+            // first leaves appends writing over the compacted file's last
+            // block, which then vanishes on the next replay.
+            const new_end_pos = new_log.end_pos;
 
             try std.Io.Dir.renameAbsolute(tmp_path, self.log.path, self.io);
             // Persist the rename: fsync the parent directory so the new
@@ -855,7 +867,9 @@ pub const Engine = struct {
             // 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;
-            self.committed_end = new_end_pos;
+            // The rewritten file was synced before the rename, so everything
+            // applied so far is durable.
+            self.committed_seq = self.seq;
             // The rewritten log holds only live documents.
             self.dead_docs = 0;
             self.gpa.free(old_path);