feat: add IncrWorker and run to concurrent

src/concurrent.zig

@@ -155,3 +155,160 @@ test "incrBy on a fresh key starts from zero and returns the running total" {
     try std.testing.expectEqual(@as(i64, 103), try guarded.incrBy(allocator, "ctr", 100));
 }
 
+const IncrWorker = struct {
+    guarded: *GuardedDb,
+    key: []const u8,
+    iterations: usize,
+
+    fn run(self: IncrWorker) void {
+        const allocator = std.heap.page_allocator;
+        var i: usize = 0;
+        while (i < self.iterations) : (i += 1) {
+            _ = self.guarded.incrBy(allocator, self.key, 1) catch |err| {
+                std.debug.panic("incrBy failed under real thread contention: {s}", .{@errorName(err)});
+            };
+        }
+    }
+};
+
+test "incrBy survives real concurrent OS threads with zero lost updates" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var db = try openTemp(&tmp, allocator);
+    defer db.close();
+    var guarded = GuardedDb.init(&db);
+
+    const threadCount = 6;
+    const perThread = 300;
+    var threads: [threadCount]std.Thread = undefined;
+    for (&threads) |*t| {
+        t.* = try std.Thread.spawn(.{}, IncrWorker.run, .{IncrWorker{
+            .guarded = &guarded,
+            .key = "shared-counter",
+            .iterations = perThread,
+        }});
+    }
+    for (threads) |t| t.join();
+
+    const raw = try guarded.get(allocator, "shared-counter");
+    defer allocator.free(raw.?);
+    const total = std.mem.readInt(i64, raw.?[0..8], .little);
+    // A bare unsynchronized `db.get` + `db.put` read-modify-write would
+    // routinely lose updates here (two threads read the same stale value,
+    // both write back the same next value, one increment vanishes). The
+    // mutex inside GuardedDb makes every incrBy fully serialized, so the
+    // total below must land exactly on threadCount*perThread every single
+    // run - not "usually", not "close to" - or the guard has a real bug.
+    try std.testing.expectEqual(@as(i64, threadCount * perThread), total);
+}
+
+const TransferWorker = struct {
+    guarded: *GuardedDb,
+    rounds: usize,
+
+    fn run(self: TransferWorker) void {
+        const allocator = std.heap.page_allocator;
+        var i: usize = 0;
+        while (i < self.rounds) : (i += 1) {
+            const g = self.guarded.acquire();
+            defer g.release();
+            const aRaw = g.db.get(allocator, "balanceA") catch @panic("get a failed");
+            defer allocator.free(aRaw.?);
+            const bRaw = g.db.get(allocator, "balanceB") catch @panic("get b failed");
+            defer allocator.free(bRaw.?);
+            const a = std.mem.readInt(i64, aRaw.?[0..8], .little);
+            const b = std.mem.readInt(i64, bRaw.?[0..8], .little);
+            if (a < 1) continue;
+            var aBuf: [8]u8 = undefined;
+            var bBuf: [8]u8 = undefined;
+            std.mem.writeInt(i64, &aBuf, a - 1, .little);
+            std.mem.writeInt(i64, &bBuf, b + 1, .little);
+            g.db.put("balanceA", &aBuf) catch @panic("put a failed");
+            g.db.put("balanceB", &bBuf) catch @panic("put b failed");
+        }
+    }
+};
+
+const InvariantReader = struct {
+    guarded: *GuardedDb,
+    checks: usize,
+    violations: *std.atomic.Value(usize),
+
+    fn run(self: InvariantReader) void {
+        const allocator = std.heap.page_allocator;
+        var i: usize = 0;
+        while (i < self.checks) : (i += 1) {
+            const g = self.guarded.acquire();
+            const aRaw = g.db.get(allocator, "balanceA") catch @panic("get a failed");
+            const bRaw = g.db.get(allocator, "balanceB") catch @panic("get b failed");
+            const a = std.mem.readInt(i64, aRaw.?[0..8], .little);
+            const b = std.mem.readInt(i64, bRaw.?[0..8], .little);
+            g.release();
+            allocator.free(aRaw.?);
+            allocator.free(bRaw.?);
+            if (a + b != 1000) _ = self.violations.fetchAdd(1, .monotonic);
+        }
+    }
+};
+
+test "acquire lets a writer move value between two keys atomically under a concurrent reader" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var db = try openTemp(&tmp, allocator);
+    defer db.close();
+    var guarded = GuardedDb.init(&db);
+
+    var initA: [8]u8 = undefined;
+    var initB: [8]u8 = undefined;
+    std.mem.writeInt(i64, &initA, 1000, .little);
+    std.mem.writeInt(i64, &initB, 0, .little);
+    try guarded.put("balanceA", &initA);
+    try guarded.put("balanceB", &initB);
+
+    var violations = std.atomic.Value(usize).init(0);
+    const writerThread = try std.Thread.spawn(.{}, TransferWorker.run, .{TransferWorker{
+        .guarded = &guarded,
+        .rounds = 1000,
+    }});
+    const readerThread = try std.Thread.spawn(.{}, InvariantReader.run, .{InvariantReader{
+        .guarded = &guarded,
+        .checks = 1000,
+        .violations = &violations,
+    }});
+    writerThread.join();
+    readerThread.join();
+
+    // Every single reader observation is required to see the invariant
+    // balanceA + balanceB == 1000 hold, because each transfer's two puts
+    // happen inside one `acquire()`/`release()` pair the reader's own
+    // acquire() must wait out entirely - a real race would show up here
+    // as a nonzero violation count, not as a crash, which is why this
+    // asserts the counter is exactly zero instead of merely "did not hang".
+    try std.testing.expectEqual(@as(usize, 0), violations.load(.monotonic));
+
+    const finalARaw = try guarded.get(allocator, "balanceA");
+    defer allocator.free(finalARaw.?);
+    const finalBRaw = try guarded.get(allocator, "balanceB");
+    defer allocator.free(finalBRaw.?);
+    const finalA = std.mem.readInt(i64, finalARaw.?[0..8], .little);
+    const finalB = std.mem.readInt(i64, finalBRaw.?[0..8], .little);
+    try std.testing.expectEqual(@as(i64, 1000), finalA + finalB);
+    try std.testing.expectEqual(@as(i64, 0), finalA);
+    try std.testing.expectEqual(@as(i64, 1000), finalB);
+}
+
+test "apply is usable concurrently: disjoint batches from different threads never clobber each other" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var db = try openTemp(&tmp, allocator);
+    defer db.close();
+    var guarded = GuardedDb.init(&db);
+
+    const ApplyWorker = struct {
+        guarded: *GuardedDb,
+        prefix: u8,
+        count: usize,
+