From f34107ef56c90a095eda2f4005d09b5f236a7a80 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Fri, 11 Sep 2026 19:27:09 +0200 Subject: [PATCH] feat: introduce ReaderCtx and readerWorker to server --- src/server.zig | 152 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 152 insertions(+) diff --git a/src/server.zig b/src/server.zig index 05618f3..33e8c23 100644 --- a/src/server.zig +++ b/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; +