db: replace engine mutex with writer-preferring RwLock, add lock_read/unlock_read

src/db.zig

@@ -1,7 +1,9 @@
 //! 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 `mutex` around a command.
+//! 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");
@@ -22,7 +24,10 @@ pub const Db = struct {
 pub const Engine = struct {
     gpa: std.mem.Allocator,
     io: std.Io,
-    mutex: std.Io.Mutex,
+    // One writer at a time (log append + fsync, map mutation); many
+    // concurrent readers (find/count/aggregate scans). Writer-preferring:
+    // a queued writer blocks new readers rather than starving.
+    rwlock: std.Io.RwLock,
     log: storage.Log,
     dbs: std.StringHashMapUnmanaged(Db),
     seq: u64,
@@ -32,7 +37,7 @@ pub const Engine = struct {
         var engine = Engine{
             .gpa = gpa,
             .io = io,
-            .mutex = std.Io.Mutex.init,
+            .rwlock = .init,
             .log = try storage.Log.open(gpa, io, path),
             .dbs = .empty,
             .seq = 0,
@@ -68,14 +73,25 @@ pub const Engine = struct {
         self.log.close();
     }
 
-    // -- commands (callers must hold the mutex) ------------------------------
+    // -- commands (callers must hold the matching lock) ---------------------
 
+    /// Exclusive lock: for commands that mutate the engine.
     pub fn lock(self: *Engine) !void {
-        try self.mutex.lock(self.io);
+        try self.rwlock.lock(self.io);
     }
 
     pub fn unlock(self: *Engine) void {
-        self.mutex.unlock(self.io);
+        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);
     }
 
     /// Insert a document. Fails with error.DuplicateKey if the _id exists.
@@ -247,7 +263,7 @@ pub const Engine = struct {
     }
 
     /// Rewrite the log with only live documents, atomically swapping the file.
-    /// Callers must hold the mutex.
+    /// Callers must hold the write lock.
     pub fn compact(self: *Engine) !void {
         const tmp_path = try std.fmt.allocPrint(self.gpa, "{s}.tmp", .{self.log.path});
         defer self.gpa.free(tmp_path);