feat: add run and TableTransferWorker to concurrent

src/concurrent.zig

@@ -312,3 +312,126 @@ test "apply is usable concurrently: disjoint batches from different threads neve
         prefix: u8,
         count: usize,
 
+        fn run(self: @This()) void {
+            var batch = WriteBatch.init(std.heap.page_allocator);
+            defer batch.deinit();
+            var i: usize = 0;
+            while (i < self.count) : (i += 1) {
+                const key = [_]u8{ self.prefix, @intCast(i) };
+                batch.put(&key, "v") catch @panic("batch put failed");
+            }
+            self.guarded.apply(&batch) catch |err| {
+                std.debug.panic("apply failed under contention: {s}", .{@errorName(err)});
+            };
+        }
+    };
+
+    var threads: [4]std.Thread = undefined;
+    for (&threads, 0..) |*t, idx| {
+        t.* = try std.Thread.spawn(.{}, ApplyWorker.run, .{ApplyWorker{
+            .guarded = &guarded,
+            .prefix = @intCast('a' + idx),
+            .count = 50,
+        }});
+    }
+    for (threads) |t| t.join();
+
+    inline for (.{ 'a', 'b', 'c', 'd' }) |prefix| {
+        var iter = try guarded.db.scanPrefix(allocator, &[_]u8{prefix});
+        defer iter.deinit();
+        var n: usize = 0;
+        while (try iter.next()) |_| n += 1;
+        try std.testing.expectEqual(@as(usize, 50), n);
+    }
+}
+
+const TableTransferWorker = struct {
+    guarded: *GuardedDb,
+    rounds: usize,
+
+    /// Runs a full `Transaction` + `TableTxn` transfer lifecycle (begin,
+    /// read both rows, stage both writes, commit) under a single
+    /// `acquire()`/`release()` pair, closing the specific gap every prior
+    /// session on this project flagged and deferred: "brak testu
+    /// wspolbieznosci wielowatkowej (dwa watki komituja TableTxn na tym
+    /// samym Db jednoczesnie)". `Db.get`/`Transaction.commit` touch shared
+    /// mutable state (the memtable, the OCC read-set bookkeeping) with no
+    /// locking of their own, so without the guard two real OS threads
+    /// racing through `TableTxn.open`/`get`/`put`/`commit` at once would
+    /// be exactly the kind of unguarded concurrent access every other
+    /// module in this project takes its own mutex to avoid.
+    fn run(self: TableTransferWorker) 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();
+
+            var txn = Transaction.begin(g.db);
+            defer txn.deinit();
+            var t = TableTxn.open(&txn, allocator, "accounts") catch @panic("TableTxn.open failed");
+            defer t.deinit();
+
+            var from = (t.get(.{ .int = 1 }) catch @panic("get 1 failed")).?;
+            defer from.deinit();
+            var to = (t.get(.{ .int = 2 }) catch @panic("get 2 failed")).?;
+            defer to.deinit();
+            if (from.get("balance").?.int < 1) continue;
+
+            t.put(.{ .int = 1 }, &.{.{ .name = "balance", .value = .{ .int = from.get("balance").?.int - 1 } }}) catch @panic("put 1 failed");
+            t.put(.{ .int = 2 }, &.{.{ .name = "balance", .value = .{ .int = to.get("balance").?.int + 1 } }}) catch @panic("put 2 failed");
+
+            // The whole sequence runs under one lock, so no external writer
+            // can ever appear mid-transfer here: commit() must never see a
+            // real error.Conflict in this test, unlike the deliberately
+            // racy scenario in txn_table.zig's own "all-or-nothing" test.
+            txn.commit() catch |err| std.debug.panic("unexpected commit failure under serialized access: {s}", .{@errorName(err)});
+        }
+    }
+};
+
+test "TableTxn commits from many real OS threads never lose or duplicate a unit of balance" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var db = try openTemp(&tmp, allocator);
+    defer db.close();
+
+    const startBalance: i64 = 2000;
+    {
+        var setup = try querymod.Table.open(&db, allocator, "accounts");
+        defer setup.deinit();
+        try setup.put(.{ .int = 1 }, &.{.{ .name = "balance", .value = .{ .int = startBalance } }});
+        try setup.put(.{ .int = 2 }, &.{.{ .name = "balance", .value = .{ .int = 0 } }});
+    }
+
+    var guarded = GuardedDb.init(&db);
+    const threadCount = 4;
+    const roundsPerThread = 200;
+    var threads: [threadCount]std.Thread = undefined;
+    for (&threads) |*t| {
+        t.* = try std.Thread.spawn(.{}, TableTransferWorker.run, .{TableTransferWorker{
+            .guarded = &guarded,
+            .rounds = roundsPerThread,
+        }});
+    }
+    for (threads) |t| t.join();
+
+    var verify = try querymod.Table.open(&db, allocator, "accounts");
+    defer verify.deinit();
+    var row1 = (try verify.get(.{ .int = 1 })).?;
+    defer row1.deinit();
+    var row2 = (try verify.get(.{ .int = 2 })).?;
+    defer row2.deinit();
+    const balance1 = row1.get("balance").?.int;
+    const balance2 = row2.get("balance").?.int;
+
+    // Every transfer moves exactly 1 unit and every transfer's two writes
+    // commit together or not at all, so the total must be exactly
+    // preserved and account 2 must have gained exactly what account 1
+    // lost - no interleaving of `get`+`put`+`commit` across threads can
+    // produce any other pair of numbers.
+    try std.testing.expectEqual(startBalance, balance1 + balance2);
+    try std.testing.expectEqual(@as(i64, threadCount * roundsPerThread), balance2);
+    try std.testing.expectEqual(startBalance - threadCount * roundsPerThread, balance1);
+}