From 0beb78c9423b3ef8bb7fea1cca2234e2dc83b3b6 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sat, 12 Sep 2026 10:12:56 +0200 Subject: [PATCH] feat: add expectedCount and expectedSum to server --- src/server.zig | 162 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) diff --git a/src/server.zig b/src/server.zig index b25a23f..969b56b 100644 --- a/src/server.zig +++ b/src/server.zig @@ -4694,3 +4694,165 @@ test "server table_window reports a malformed expr as bad_request, not a crash" const path = try tmp.dir.realpathAlloc(allocator, "."); defer allocator.free(path); + var db = try testDb(allocator, path); + defer db.close(); + try seedAggregateOrders(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 specs = [_]windowmod.WindowSpec{.{ .field = "", .kind = .count }}; + const err = client.tableWindow(allocator, "orders", &specs, &.{}, &.{}, 0, 0, "city bogus s:warsaw", .rows); + try std.testing.expectError(client_mod.ClientError.ServerBadRequest, err); +} + +test "server table_window reports a non-numeric sum field as 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 seedAggregateOrders(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 specs = [_]windowmod.WindowSpec{.{ .field = "city", .kind = .sum }}; + const err = client.tableWindow(allocator, "orders", &specs, &.{}, &.{}, 0, 0, "", .rows); + try std.testing.expectError(client_mod.ClientError.ServerBadRequest, err); +} + +test "server table_window reports a sum spec with no field as 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 seedAggregateOrders(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(); + + // The CLI's `parseWindowSpec` would reject this before ever sending it, + // but nothing on the wire itself stops a client from asking for it - + // `window.computeMany`'s `MissingField` must come back as + // `bad_request`, not an internal server error. + const specs = [_]windowmod.WindowSpec{.{ .field = "", .kind = .sum }}; + const err = client.tableWindow(allocator, "orders", &specs, &.{}, &.{}, 0, 0, "", .rows); + try std.testing.expectError(client_mod.ClientError.ServerBadRequest, err); +} + +// `table_aggregate` shares the exact same underlying-sstable-reshaping +// hazard as `table_select`/`table_join` above (see the `table_select` +// concurrency test's header comment): `flush`/`compact` swap `Db.sstables` +// out from under any in-flight reader while `Server.dispatch`'s `self.mutex` +// is what keeps an aggregate's full-table scan (and the grouping/windowing +// on top of it in `aggregate.zig`) from tearing. Thirty rows split evenly +// three ways by `@mod(pk, 3)` so every group lands on an independently +// checkable count and sum: pk%3==0 -> "warsaw" (pks 3,6,..,30, count 10, sum +// 165), pk%3==1 -> "krakow" (pks 1,4,..,28, count 10, sum 145), pk%3==2 -> +// "gdansk" (pks 2,5,..,29, count 10, sum 155) - global count 30, sum 465. +test "table_aggregate 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); + + // Small target size so a handful of `flush`/`compact` calls do real + // structural work instead of needing megabytes of data first - same + // trick as the `table_select`/`table_join` concurrency tests above. + var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 }); + defer db.close(); + + const rowCount = 30; + { + var orders = try querymod.Table.open(&db, allocator, "orders"); + defer orders.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 orders.put(.{ .int = pk }, &.{ + .{ .name = "city", .value = .{ .text = city } }, + .{ .name = "total", .value = .{ .int = pk } }, + }); + } + 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 = 20; + const mutatorThreads = 2; + const mutatorIters = 15; + + const expectedCount = struct { + fn get(city: []const u8) i64 { + return if (std.mem.eql(u8, city, "warsaw") or std.mem.eql(u8, city, "krakow") or std.mem.eql(u8, city, "gdansk")) 10 else -1; + } + }.get; + const expectedSum = struct { + fn get(city: []const u8) i64 { + if (std.mem.eql(u8, city, "warsaw")) return 165; + if (std.mem.eql(u8, city, "krakow")) return 145; + if (std.mem.eql(u8, city, "gdansk")) return 155; + return -1; + } + }.get; + + const ReaderCtx = struct { + allocator: Allocator, + addr: std.net.Address, + ok: bool = false, + }; + const readerWorker = struct { + fn run(ctx: *ReaderCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < readerIters) : (i += 1) { + const specs = [_]aggregatemod.AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "total", .kind = .sum }, + }; + const groups = client.tableAggregateGroups(ctx.allocator, "orders", "city", &specs, 0, 0, &.{}, "", "") catch return; + defer protocol.freeAggGroupRows(ctx.allocator, groups); + if (groups.len != 3) return; + for (groups) |g| { + if (g.values[0].int != expectedCount(g.group.text)) return; + if (g.values[1].int != expectedSum(g.group.text)) return; + } +