feat: add ReadCtx and readWorker to server

src/server.zig

@@ -7392,3 +7392,163 @@ test "user_admin survives many real concurrent connections mutating and listing
         }
     }.run;
 
+    const ReadCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        ok: bool = false,
+    };
+
+    const readWorker = struct {
+        fn run(ctx: *ReadCtx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var i: usize = 0;
+            while (i < readsPerReader) : (i += 1) {
+                // Any decode failure inside the server (a torn/corrupted
+                // record) surfaces here as `error.ServerInternal`, not a
+                // silently wrong answer, since `handleUserAdmin` maps every
+                // `UserRecord.decode` error straight to an `internal`
+                // status frame.
+                const users = client.adminListUsers(ctx.allocator) catch return;
+                defer authmod.freeUserList(ctx.allocator, users);
+                if (users.len != 1) return;
+                if (!std.mem.eql(u8, users[0].username, "shared-admin")) return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    var writeContexts: [writerCount]WriteCtx = undefined;
+    var writeThreads: [writerCount]std.Thread = undefined;
+    var readContexts: [readerCount]ReadCtx = undefined;
+    var readThreads: [readerCount]std.Thread = undefined;
+
+    for (0..readerCount) |i| {
+        readContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
+        readThreads[i] = try std.Thread.spawn(.{}, readWorker, .{&readContexts[i]});
+    }
+    for (0..writerCount) |i| {
+        writeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
+        writeThreads[i] = try std.Thread.spawn(.{}, writeWorker, .{&writeContexts[i]});
+    }
+    for (0..writerCount) |i| writeThreads[i].join();
+    for (0..readerCount) |i| readThreads[i].join();
+    for (0..writerCount) |i| try std.testing.expect(writeContexts[i].ok);
+    for (0..readerCount) |i| try std.testing.expect(readContexts[i].ok);
+
+    // A final, deterministic write from the main thread after every racer
+    // has joined, then verified through `auth.Store.verify` directly
+    // (bypassing the network entirely) against the SAME `*Db` the server
+    // just hammered concurrently: this proves the record surviving on disk
+    // is not merely "decodable" (the readers above already checked that)
+    // but functionally intact - the exact salt/rounds/hash triple that
+    // `setPassword` wrote is still paired correctly with the role
+    // `setRole` wrote after it, not a torn mix of two different racers'
+    // writes.
+    try setup.adminSetPassword("shared-admin", "final-known-pw", 4);
+    try setup.adminSetRole("shared-admin", .admin);
+
+    const store = authmod.Store.init(&db);
+    const role = try store.verify(allocator, "shared-admin", "final-known-pw");
+    try std.testing.expectEqual(authmod.Role.admin, role);
+
+    const finalUsers = try setup.adminListUsers(allocator);
+    defer authmod.freeUserList(allocator, finalUsers);
+    try std.testing.expectEqual(@as(usize, 1), finalUsers.len);
+    try std.testing.expectEqual(authmod.Role.admin, finalUsers[0].role);
+}
+
+// `txn_op` is the network counterpart of the local CLI's `txn <script>`
+// (main.zig's `runTxn`) - until now atomic multi-row/multi-table commits
+// were reachable only from a process holding the `Db` directly, the same
+// gap `table_select`/`coll_op`/`ttl_op` already closed for their own
+// subsystems. This round trip exercises both verbs across two tables in
+// one commit, including a pre-existing secondary index, to prove the field
+// list and primary key survive the wire encoding (`Row.encode`/
+// `rowmod.encodeValue`) intact enough for `TableTxn.put`'s index
+// maintenance to run correctly on the decoded side, not just for the raw
+// bytes to round-trip in isolation the way protocol.zig's own unit tests
+// already check.
+test "server txn_op commits put and del atomically across two tables, keeping an index in sync" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    {
+        var accounts = try querymod.Table.open(&db, allocator, "accounts");
+        defer accounts.deinit();
+        try accounts.put(.{ .int = 1 }, &.{.{ .name = "balance", .value = .{ .int = 100 } }});
+        try accounts.put(.{ .int = 2 }, &.{.{ .name = "balance", .value = .{ .int = 5 } }});
+
+        var people = try querymod.Table.open(&db, allocator, "people");
+        defer people.deinit();
+        try people.createIndex("city");
+        try people.put(.{ .int = 10 }, &.{.{ .name = "city", .value = .{ .text = "warsaw" } }});
+    }
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    var client = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer client.close();
+
+    const peopleFields = [_]rowmod.Field{.{ .name = "city", .value = .{ .text = "warsaw" } }};
+    const entries = [_]protocol.TxnOpEntry{
+        .{ .verb = .put, .table = "accounts", .pk = .{ .int = 2 }, .fields = &.{.{ .name = "balance", .value = .{ .int = 999 } }} },
+        .{ .verb = .put, .table = "people", .pk = .{ .int = 11 }, .fields = &peopleFields },
+        .{ .verb = .del, .table = "accounts", .pk = .{ .int = 1 } },
+    };
+    const applied = try client.txnOp(&entries);
+    try std.testing.expectEqual(@as(u32, 3), applied);
+
+    var accounts = try querymod.Table.open(&db, allocator, "accounts");
+    defer accounts.deinit();
+    try std.testing.expect((try accounts.get(.{ .int = 1 })) == null);
+    var row2 = (try accounts.get(.{ .int = 2 })).?;
+    defer row2.deinit();
+    try std.testing.expectEqual(@as(i64, 999), row2.get("balance").?.int);
+
+    var people = try querymod.Table.open(&db, allocator, "people");
+    defer people.deinit();
+    const warsaw = try people.select(.{ .eq = .{ .field = "city", .value = .{ .text = "warsaw" } } });
+    defer querymod.freeRows(allocator, warsaw);
+    try std.testing.expectEqual(@as(usize, 2), warsaw.len);
+}
+
+// Closes a gap left open by the CHECK constraint's introduction alongside
+// UNIQUE/FOREIGN KEY: `TableTxn.put` enforces all three identically (see
+// txn_table.zig), and this server's only write path (`handleTxnOp`) already
+// runs every entry through `TableTxn.put` - so a remote `txn_op` rejecting a
+// CHECK violation was already true the moment `TableTxn.put` gained the
+// check, never separately wired. This test is the same kind of "prove the
+// wire path actually exercises it, not just the in-process unit test" round
+// trip the FK/UNIQUE `txn_op` test above already is.
+test "server txn_op rejects a CHECK constraint violation over the wire and commits nothing" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    {
+        var orders = try querymod.Table.open(&db, allocator, "orders");
+        defer orders.deinit();
+        try orders.createCheckConstraint("qty_positive", "qty gt i:0");
+    }
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+