feat: add expectedCount and expectedSum to server

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;
+                }
+