From c71d01bd7eb97b481162de9a3e5862e93cd4519e Mon Sep 17 00:00:00 2001 From: DualFroz Date: Fri, 11 Sep 2026 18:28:38 +0200 Subject: [PATCH] feat: add Combo and ReaderCtx to server --- src/server.zig | 166 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) diff --git a/src/server.zig b/src/server.zig index 1fca9a5..fee6dc5 100644 --- a/src/server.zig +++ b/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; +