db: add concurrent readers/writers stress test on threaded Io

src/db.zig

@@ -552,3 +552,71 @@ test "compaction rewrites log and keeps data" {
     }
     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);
+    }
+}