feat: add ReaderCtx and readerWorker to server

src/server.zig

@@ -2645,3 +2645,160 @@ test "server rejects a malformed table_update frame (bad set fields) with bad_re
     try std.testing.expectEqual(@intFromEnum(protocol.Status.ok), pingFrame.tag);
 }
 
+// `table_update` shares the same underlying-sstable-reshaping hazard as
+// `table_select`/`table_join`/`table_aggregate` above, but with a sharper
+// edge: `updateWhere` both READS (via `selectAllMatches`) and WRITES (via
+// `put`) in the same request, while those three only read. `self.mutex` in
+// `Server.dispatch` still serializes every request end-to-end including
+// `flush`/`compact`, so no single request can observe a torn sstable set -
+// what this test actually proves is that real structural reshaping,
+// interleaved via real OS threads through the real TCP/dispatch path, never
+// desyncs `updateWhere`'s read-then-write from the index it must keep in
+// sync, and that a single updater thread's own sequence of writes lands
+// exactly where expected once every other thread has stopped.
+//
+// Thirty rows split evenly three ways by `@mod(pk, 3)` (same helper as the
+// `table_aggregate` concurrency test above) so the "city" index always
+// holds exactly 10 rows per bucket - that count must never move, because
+// this test's single updater thread only ever touches the "epoch" field,
+// never "city" itself, so index population is an invariant a reader can
+// check at every single snapshot, not just at the end.
+test "table_update 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`/`table_aggregate`
+    // concurrency tests above.
+    var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 });
+    defer db.close();
+
+    const rowCount = 30;
+    {
+        var accounts = try querymod.Table.open(&db, allocator, "accounts");
+        defer accounts.deinit();
+        try accounts.createIndex("city");
+        var pk: i64 = 1;
+        while (pk <= rowCount) : (pk += 1) {
+            const city: []const u8 = switch (@mod(pk, 3)) {
+                0 => "warsaw",
+                1 => "krakow",
+                else => "gdansk",
+            };
+            try accounts.put(.{ .int = pk }, &.{
+                .{ .name = "city", .value = .{ .text = city } },
+                .{ .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 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) {
+                inline for (.{ "warsaw", "krakow", "gdansk" }) |city| {
+                    const buf = std.fmt.allocPrint(ctx.allocator, "city eq s:{s}", .{city}) 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 != 10) return;
+                    for (rows) |row| {
+                        const got = row.get("city") orelse return;
+                        if (!std.mem.eql(u8, got.text, city)) 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-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) {
+                const target: []const u8 = if (@mod(i, 2) == 0) "krakow" else "gdansk";
+                const expr = std.fmt.allocPrint(ctx.allocator, "city eq s:{s}", .{target}) 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 != 10) return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    var readerContexts: [readerThreads]ReaderCtx = undefined;
+    var readerHandles: [readerThreads]std.Thread = undefined;
+    for (0..readerThreads) |i| {
+        readerContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
+        readerHandles[i] = try std.Thread.spawn(.{}, readerWorker, .{&readerContexts[i]});
+    }
+    var reshapeContexts: [reshapeThreads]ReshapeCtx = undefined;
+    var reshapeHandles: [reshapeThreads]std.Thread = undefined;
+    for (0..reshapeThreads) |i| {
+        reshapeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
+        reshapeHandles[i] = try std.Thread.spawn(.{}, reshapeWorker, .{&reshapeContexts[i]});
+    }
+    var updaterContext = UpdaterCtx{ .allocator = allocator, .addr = server.boundAddress() };
+    const updaterHandle = try std.Thread.spawn(.{}, updaterWorker, .{&updaterContext});
+
+    for (0..readerThreads) |i| readerHandles[i].join();
+    for (0..reshapeThreads) |i| reshapeHandles[i].join();
+    updaterHandle.join();
+    for (0..readerThreads) |i| try std.testing.expect(readerContexts[i].ok);
+    for (0..reshapeThreads) |i| try std.testing.expect(reshapeContexts[i].ok);
+    try std.testing.expect(updaterContext.ok);
+