feat: add Combo and ReaderCtx to server

src/server.zig

@@ -2802,3 +2802,169 @@ test "table_update survives concurrent flush and compact reshaping the underlyin
     for (0..reshapeThreads) |i| try std.testing.expect(reshapeContexts[i].ok);
     try std.testing.expect(updaterContext.ok);
 
+    // Final sanity check on a fresh connection: the updater alternated
+    // krakow/gdansk, so the last krakow write was iteration 18 (last even
+    // below 20) and the last gdansk write was iteration 19 (last odd) -
+    // warsaw was never touched by any predicate and must still read 0.
+    var client = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer client.close();
+
+    const krakowRows = try client.tableSelect(allocator, "accounts", "city eq s:krakow", 0, 0, &.{});
+    defer protocol.freeRows(allocator, krakowRows);
+    try std.testing.expectEqual(@as(usize, 10), krakowRows.len);
+    for (krakowRows) |row| try std.testing.expectEqual(@as(i64, 18), row.get("epoch").?.int);
+
+    const gdanskRows = try client.tableSelect(allocator, "accounts", "city eq s:gdansk", 0, 0, &.{});
+    defer protocol.freeRows(allocator, gdanskRows);
+    try std.testing.expectEqual(@as(usize, 10), gdanskRows.len);
+    for (gdanskRows) |row| try std.testing.expectEqual(@as(i64, 19), row.get("epoch").?.int);
+
+    const warsawRows = try client.tableSelect(allocator, "accounts", "city eq s:warsaw", 0, 0, &.{});
+    defer protocol.freeRows(allocator, warsawRows);
+    try std.testing.expectEqual(@as(usize, 10), warsawRows.len);
+    for (warsawRows) |row| try std.testing.expectEqual(@as(i64, 0), row.get("epoch").?.int);
+}
+
+// Same shape as the test above, but the index guarding the concurrent
+// reads/updates is a COMPOSITE one (`city_age`, two columns) instead of a
+// single-column `city` index - `updateWhere` reached through
+// `composite_eq` was only proven correct single-threaded so far (see
+// query.zig's "updateWhere keeps a composite index in sync" test); this
+// proves the same code path survives real `flush`/`compact` reshaping the
+// underlying sstables while other connections read and write through it.
+test "table_update survives concurrent flush and compact reshaping a composite index" {
+    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 Db.open(allocator, path, .{ .sstableTargetBytes = 512 });
+    defer db.close();
+
+    // 30 rows: 3 cities x 2 age bands x 5 rows each. Every (city, age) pair
+    // is a full-tuple equality match against `city_age`, so
+    // `Table.selectPage`/`updateWhere` both reach it through `composite_eq`
+    // (see query.zig's "createCompositeIndex backfills then full-tuple eq
+    // select" test for the same match shape single-threaded).
+    const rowCount = 30;
+    {
+        var accounts = try querymod.Table.open(&db, allocator, "accounts");
+        defer accounts.deinit();
+        try accounts.createCompositeIndex("city_age", &.{ "city", "age" });
+        var pk: i64 = 1;
+        while (pk <= rowCount) : (pk += 1) {
+            const city: []const u8 = switch (@mod(pk, 3)) {
+                0 => "warsaw",
+                1 => "krakow",
+                else => "gdansk",
+            };
+            const age: i64 = if (@mod(@divTrunc(pk - 1, 3), 2) == 0) 30 else 40;
+            try accounts.put(.{ .int = pk }, &.{
+                .{ .name = "city", .value = .{ .text = city } },
+                .{ .name = "age", .value = .{ .int = age } },
+                .{ .name = "epoch", .value = .{ .int = 0 } },
+            });
+        }
+        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 reshapeThreads = 2;
+    const reshapeIters = 15;
+    const updaterIters = 20;
+
+    const Combo = struct { city: []const u8, age: i64 };
+    const combos = [_]Combo{
+        .{ .city = "warsaw", .age = 30 },
+        .{ .city = "warsaw", .age = 40 },
+        .{ .city = "krakow", .age = 30 },
+        .{ .city = "krakow", .age = 40 },
+        .{ .city = "gdansk", .age = 30 },
+        .{ .city = "gdansk", .age = 40 },
+    };
+
+    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) {
+                for (combos) |combo| {
+                    const buf = std.fmt.allocPrint(ctx.allocator, "city eq s:{s} and age eq i:{d}", .{ combo.city, combo.age }) catch return;
+                    defer ctx.allocator.free(buf);
+                    const rows = client.tableSelect(ctx.allocator, "accounts", buf, 0, 0, &.{}) catch return;
+                    defer protocol.freeRows(ctx.allocator, rows);
+                    if (rows.len != 5) return;
+                    for (rows) |row| {
+                        const gotCity = row.get("city") orelse return;
+                        if (!std.mem.eql(u8, gotCity.text, combo.city)) return;
+                        const gotAge = row.get("age") orelse return;
+                        if (gotAge.int != combo.age) return;
+                    }
+                }
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    const ReshapeCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        thread: usize,
+        ok: bool = false,
+    };
+    const reshapeWorker = struct {
+        fn run(ctx: *ReshapeCtx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var i: usize = 0;
+            while (i < reshapeIters) : (i += 1) {
+                var buf: [32]u8 = undefined;
+                const key = std.fmt.bufPrint(&buf, "update-composite-churn-{d}-{d}", .{ ctx.thread, i }) catch return;
+                client.set(key, "x") catch return;
+                client.flush() catch return;
+                client.compact() catch return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    const UpdaterCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        ok: bool = false,
+    };
+    const updaterWorker = struct {
+        fn run(ctx: *UpdaterCtx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var i: i64 = 0;
+            while (i < updaterIters) : (i += 1) {
+                // Alternates between two full-tuple matches on the composite
+                // index, never touching `city`/`age` themselves, so the
+                // (5, 5, 5, 5, 5, 5) population split is a checkable
+                // invariant on every single reader iteration in flight.
+                const combo = if (@mod(i, 2) == 0) Combo{ .city = "krakow", .age = 30 } else Combo{ .city = "gdansk", .age = 40 };
+                const expr = std.fmt.allocPrint(ctx.allocator, "city eq s:{s} and age eq i:{d}", .{ combo.city, combo.age }) catch return;
+                defer ctx.allocator.free(expr);
+                const sets = [_]rowmod.Field{.{ .name = "epoch", .value = .{ .int = i } }};
+                const updated = client.tableUpdate("accounts", &sets, expr) catch return;
+                if (updated != 5) return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+