diff --git a/src/server.zig b/src/server.zig index 14edeeb..208d27e 100644 --- a/src/server.zig +++ b/src/server.zig @@ -5329,3 +5329,162 @@ test "table_join survives concurrent flush and compact reshaping the underlying 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` concurrency test above. + var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 }); + defer db.close(); + try seedJoinConcurrencyTables(allocator, &db); + 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 mutatorThreads = 2; + const mutatorIters = 15; + const matchedCount = 8; + const customerCount = 10; + + 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) { + const inner = client.tableJoin(ctx.allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return; + defer protocol.freeJoinPairs(ctx.allocator, inner); + if (inner.len != matchedCount) return; + for (inner) |pair| { + const left = pair.left orelse return; + const right = pair.right orelse return; + const pk = left.get(querymod.pkField) orelse return; + const total = right.get("total") orelse return; + if (total.int != pk.int * 10) return; + } + + const left = client.tableJoin(ctx.allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return; + defer protocol.freeJoinPairs(ctx.allocator, left); + if (left.len != customerCount) return; + + const right = client.tableJoin(ctx.allocator, .right, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return; + defer protocol.freeJoinPairs(ctx.allocator, right); + if (right.len != matchedCount + 2) return; + + const full = client.tableJoin(ctx.allocator, .full, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return; + defer protocol.freeJoinPairs(ctx.allocator, full); + if (full.len != customerCount + 2) return; + } + ctx.ok = true; + } + }.run; + + const MutatorCtx = struct { + allocator: Allocator, + addr: std.net.Address, + thread: usize, + ok: bool = false, + }; + const mutatorWorker = struct { + fn run(ctx: *MutatorCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < mutatorIters) : (i += 1) { + var buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&buf, "join-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; + + 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 mutatorContexts: [mutatorThreads]MutatorCtx = undefined; + var mutatorHandles: [mutatorThreads]std.Thread = undefined; + for (0..mutatorThreads) |i| { + mutatorContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i }; + mutatorHandles[i] = try std.Thread.spawn(.{}, mutatorWorker, .{&mutatorContexts[i]}); + } + + for (0..readerThreads) |i| readerHandles[i].join(); + for (0..mutatorThreads) |i| mutatorHandles[i].join(); + for (0..readerThreads) |i| try std.testing.expect(readerContexts[i].ok); + for (0..mutatorThreads) |i| try std.testing.expect(mutatorContexts[i].ok); + + // Final sanity check on a fresh connection: all four join shapes are + // still exactly what they were before the server ever started - the + // concurrent churn wrote only unrelated `join-churn-*` keys and + // reshaped physical sstable layout, never touched `customers`/`orders` + // values. + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + const finalInner = try client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", ""); + defer protocol.freeJoinPairs(allocator, finalInner); + try std.testing.expectEqual(@as(usize, matchedCount), finalInner.len); + const finalFull = try client.tableJoin(allocator, .full, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", ""); + defer protocol.freeJoinPairs(allocator, finalFull); + try std.testing.expectEqual(@as(usize, customerCount + 2), finalFull.len); +} + +test "server without enableAuth accepts any auth attempt and every op" { + 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(); + + 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"); + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + + try client.authenticate("nobody", "wrong"); + try client.set("k", "v"); +} + +test "server with auth enabled rejects unauthenticated requests, ping excepted" { + 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(); + const store = authmod.Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readwrite, 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(); +