From eee83a0ec72ae334e73387e7cae277111b770131 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sat, 12 Sep 2026 11:23:13 +0200 Subject: [PATCH] feat: add ReaderCtx to server --- src/server.zig | 154 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/src/server.zig b/src/server.zig index 88f6a85..f21641b 100644 --- a/src/server.zig +++ b/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 }, + }; +