feat: introduce ReaderCtx and readerWorker to server

src/server.zig

@@ -3441,3 +3441,155 @@ test "server rejects a malformed table_export frame (bad format byte) with bad_r
     defer frame.deinit();
     try std.testing.expectEqual(@intFromEnum(protocol.Status.bad_request), frame.tag);
 
+    // The connection must still be usable afterward.
+    try protocol.writeFrame(stream.writer(), @intFromEnum(protocol.Op.ping), "");
+    var pingFrame = try protocol.readFrame(allocator, stream.reader());
+    defer pingFrame.deinit();
+    try std.testing.expectEqual(@intFromEnum(protocol.Status.ok), pingFrame.tag);
+}
+
+test "table_export requires only readonly role, table_import requires readwrite" {
+    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();
+    {
+        var table = try querymod.Table.open(&db, allocator, "people");
+        defer table.deinit();
+        try table.put(.{ .int = 1 }, &.{});
+    }
+
+    const store = authmod.Store.init(&db);
+    try store.createUser(allocator, "viewer", "pass", .readonly, 16);
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    server.enableAuth(store);
+    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();
+    try client.authenticate("viewer", "pass");
+
+    var exported = try client.tableExport(allocator, "people", .csv);
+    defer exported.deinit(allocator);
+    try std.testing.expectEqual(@as(u64, 1), exported.rowCount);
+
+    try std.testing.expectError(client_mod.ClientError.ServerForbidden, client.tableImport("people", .csv, exported.content));
+}
+
+// `table_delete` shares the same underlying-sstable-reshaping hazard as
+// `table_update` above (see that test's header comment for the full
+// rationale), but where `updateWhere` reads-then-writes each matched row in
+// place, `deleteWhere` reads-then-REMOVES it - a sharper edge again, since a
+// removed row's index entries must never resurface even if a `flush`/
+// `compact` reshapes the underlying sstables mid-scan. Three buckets
+// (warsaw/krakow/gdansk) stay untouched for the whole test and are the
+// reader threads' stable invariant (always exactly 10 rows each); a fourth
+// bucket (poznan) is what the deleter thread progressively drains one row
+// at a time via a "seq" field unrelated to the "city" index, so every
+// delete is a distinct, real structural change rather than a no-op repeat.
+test "table_delete 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 `table_update`'s own concurrency test above.
+    var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 });
+    defer db.close();
+
+    const stableBucketRows = 10;
+    const poznanRows = 10;
+    {
+        var accounts = try querymod.Table.open(&db, allocator, "accounts");
+        defer accounts.deinit();
+        try accounts.createIndex("city");
+        var pk: i64 = 1;
+        var i: i64 = 0;
+        while (i < stableBucketRows) : (i += 1) {
+            inline for (.{ "warsaw", "krakow", "gdansk" }) |city| {
+                try accounts.put(.{ .int = pk }, &.{.{ .name = "city", .value = .{ .text = city } }});
+                pk += 1;
+            }
+        }
+        var seq: i64 = 0;
+        while (seq < poznanRows) : (seq += 1) {
+            try accounts.put(.{ .int = pk }, &.{
+                .{ .name = "city", .value = .{ .text = "poznan" } },
+                .{ .name = "seq", .value = .{ .int = seq } },
+            });
+            pk += 1;
+        }
+        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 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 != stableBucketRows) 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, "delete-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;
+