feat: add ReaderCtx to server

src/server.zig

@@ -5015,3 +5015,157 @@ test "server table_join runs left, right, and full outer variants with predicate
     }
     try std.testing.expect(carolUnmatched);
 
+    const right = try client.tableJoin(allocator, .right, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
+    defer protocol.freeJoinPairs(allocator, right);
+    try std.testing.expectEqual(@as(usize, 3), right.len);
+    var orphanOrderUnmatched = false;
+    for (right) |p| {
+        if (p.left == null) orphanOrderUnmatched = true;
+    }
+    try std.testing.expect(orphanOrderUnmatched);
+
+    const full = try client.tableJoin(allocator, .full, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
+    defer protocol.freeJoinPairs(allocator, full);
+    try std.testing.expectEqual(@as(usize, 5), full.len);
+
+    // rightExprText filters the orders side before the join runs, same as
+    // `innerJoin`'s `rightPred` - narrowing to `total gt i:40` drops one of
+    // alice's two orders.
+    const filtered = try client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "total gt i:40");
+    defer protocol.freeJoinPairs(allocator, filtered);
+    try std.testing.expectEqual(@as(usize, 1), filtered.len);
+    try std.testing.expectEqual(@as(i64, 50), filtered[0].right.?.get("total").?.int);
+}
+
+test "server table_join windows a real TCP result through offset and limit" {
+    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();
+    try seedJoinTables(allocator, &db);
+
+    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();
+
+    // Unpaged left outer join order: (alice,101) (alice,102) (bob,nil) (carol,nil).
+    const unpaged = try client.tableJoin(allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
+    defer protocol.freeJoinPairs(allocator, unpaged);
+    try std.testing.expectEqual(@as(usize, 4), unpaged.len);
+
+    const windowed = try client.tableJoin(allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 1, 2, "", "");
+    defer protocol.freeJoinPairs(allocator, windowed);
+    try std.testing.expectEqual(@as(usize, 2), windowed.len);
+    try std.testing.expectEqual(@as(i64, 30), windowed[0].right.?.get("total").?.int);
+    try std.testing.expectEqualStrings("bob", windowed[1].left.?.get("name").?.text);
+    try std.testing.expect(windowed[1].right == null);
+
+    const pastEnd = try client.tableJoin(allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 50, 0, "", "");
+    defer protocol.freeJoinPairs(allocator, pastEnd);
+    try std.testing.expectEqual(@as(usize, 0), pastEnd.len);
+}
+
+test "server table_join rejects a malformed expr with bad_request, not a crash" {
+    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();
+    try seedJoinTables(allocator, &db);
+
+    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 err = client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "name bogus s:alice", "");
+    try std.testing.expectError(client_mod.ClientError.ServerBadRequest, err);
+}
+
+// `table_select` is the only networked table opcode - rows can only be
+// written locally (CLI `table put`), never over the wire - so this cannot
+// reuse the coll_op "many writers, one shared counter" shape. What it CAN
+// share with a live server is the sstable storage underneath: `flush` and
+// `compact` (both real networked opcodes) mutate `Db.sstables` in place,
+// closing and destroying old `Reader`s and swapping in new ones while
+// `compact` in particular collapses the whole list down to a single file.
+// A `table_select` cursor concurrently mid-scan over those same readers is
+// exactly the use-after-free/torn-read shape a missing lock around
+// `Server.dispatch` would produce - `self.mutex` must serialize every
+// request regardless of which opcode it is, not just same-opcode traffic.
+test "table_select survives concurrent flush and compact reshaping the underlying sstables" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    // A small target size so a handful of `flush`/`compact` calls do real
+    // structural work instead of needing megabytes of data first.
+    var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 });
+    defer db.close();
+
+    const rowCount = 30;
+    {
+        var table = try querymod.Table.open(&db, allocator, "people");
+        defer table.deinit();
+        var pk: i64 = 1;
+        while (pk <= rowCount) : (pk += 1) {
+            const city: []const u8 = switch (@mod(pk, 3)) {
+                0 => "warsaw",
+                1 => "krakow",
+                else => "gdansk",
+            };
+            try table.put(.{ .int = pk }, &.{
+                .{ .name = "city", .value = .{ .text = city } },
+                .{ .name = "age", .value = .{ .int = pk * 2 } },
+            });
+        }
+        try db.flush();
+    }
+
+    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");
+    const readerThreads = 4;
+    const readerIters = 40;
+    const mutatorThreads = 2;
+    const mutatorIters = 15;
+
+    const ReaderCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        ok: bool = false,
+    };
+    // A multi-element order-by (city ascending, then age descending as a
+    // tiebreak within each city) - `handleTableSelect` decodes a wire
+    // orderBy count > 1 into `Table.selectPageOrdered`, which cannot
+    // early-stop mid-scan the way an unordered `selectPage` can, so it reads
+    // more of the table before sorting. This proves that path survives real
+    // `flush`/`compact` reshaping the underlying sstables mid-scan, not just
+    // the single-column order-by already covered by
+    // "orders a remote result by a field before windowing it" above.
+    const cityOrderBy = [_]querymod.OrderBy{
+        .{ .field = "city" },
+        .{ .field = "age", .descending = true },
+    };
+