feat: add CreateCtx and createWorker to server
This commit is contained in:
dualfroz committed
1 parent
5de9a26f07
commit
cf4cfe5660
1 file changed
+163
+163
@@ -7229,3 +7229,166 @@ test "coll_ttl_op survives many real concurrent connections racing writes agains
|
||||
for (0..writerCount) |i| try std.testing.expect(writeContexts[i].ok);
|
||||
for (0..sweeperCount) |i| try std.testing.expect(sweepContexts[i].ok);
|
||||
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
|
||||
const finalReaped = try client.collTtlSweep();
|
||||
|
||||
var total: u64 = finalReaped;
|
||||
for (0..sweeperCount) |i| total += sweepContexts[i].reaped.load(.monotonic);
|
||||
try std.testing.expectEqual(@as(u64, writerCount * perWriter), total);
|
||||
|
||||
// Spot-check one collection from every writer thread: the WHOLE
|
||||
// counter namespace must be gone, not merely its TTL marker.
|
||||
for (0..writerCount) |t| {
|
||||
var buf: [32]u8 = undefined;
|
||||
const name = try std.fmt.bufPrint(&buf, "ttlcoll-conc-{d}-0", .{t});
|
||||
try std.testing.expect(!try collmod.collectionExists(&db, allocator, .counter, name));
|
||||
try std.testing.expectEqual(ttlmod.TtlResult.noKey, try client.collTtlQuery(.counter, name));
|
||||
}
|
||||
}
|
||||
|
||||
// `coll_op`/`ttl_op`/`coll_ttl_op`/`table_select` above all have their own
|
||||
// concurrency proof now (#31/#32/#33's end-of-session notes tracked this
|
||||
// gap explicitly); `user_admin` was the one opcode still missing one. Its
|
||||
// shared mutable state is `auth.Store` (auth.zig), itself just a thin
|
||||
// get/put layer over the same `*Db` - `createUser` is a plain
|
||||
// check-then-write (`db.get` then `db.put`, no compare-and-swap), so a
|
||||
// missing lock around it would let two concurrent `create` calls for the
|
||||
// SAME username both observe "absent" and both write, silently doubling
|
||||
// the account (or worse, interleaving two different `UserRecord.encode()`
|
||||
// outputs into one key). `self.mutex` in `Server.dispatch` is the only
|
||||
// thing standing between that and reality here, same as every other
|
||||
// opcode's concurrency test in this file.
|
||||
test "user_admin survives many real concurrent connections racing to create the same username" {
|
||||
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");
|
||||
const threadCount = 8;
|
||||
|
||||
const CreateCtx = struct {
|
||||
allocator: Allocator,
|
||||
addr: std.net.Address,
|
||||
thread: usize,
|
||||
created: bool = false,
|
||||
conflicted: bool = false,
|
||||
ok: bool = false,
|
||||
};
|
||||
|
||||
const createWorker = struct {
|
||||
fn run(ctx: *CreateCtx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var buf: [16]u8 = undefined;
|
||||
const password = std.fmt.bufPrint(&buf, "pw-{d}", .{ctx.thread}) catch return;
|
||||
if (client.adminCreateUser("shared-user", password, .readwrite, 4)) |_| {
|
||||
ctx.created = true;
|
||||
} else |err| {
|
||||
if (err != client_mod.ClientError.ServerConflict) return;
|
||||
ctx.conflicted = true;
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.run;
|
||||
|
||||
var contexts: [threadCount]CreateCtx = undefined;
|
||||
var threads: [threadCount]std.Thread = undefined;
|
||||
for (0..threadCount) |i| {
|
||||
contexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
|
||||
threads[i] = try std.Thread.spawn(.{}, createWorker, .{&contexts[i]});
|
||||
}
|
||||
for (0..threadCount) |i| threads[i].join();
|
||||
|
||||
var createdCount: usize = 0;
|
||||
var conflictedCount: usize = 0;
|
||||
for (0..threadCount) |i| {
|
||||
try std.testing.expect(contexts[i].ok);
|
||||
if (contexts[i].created) createdCount += 1;
|
||||
if (contexts[i].conflicted) conflictedCount += 1;
|
||||
}
|
||||
// A missing lock around the check-then-write in `createUser` would let
|
||||
// more than one racer observe "absent" and both succeed, so this must
|
||||
// land on EXACTLY one winner - not "at least one".
|
||||
try std.testing.expectEqual(@as(usize, 1), createdCount);
|
||||
try std.testing.expectEqual(@as(usize, threadCount - 1), conflictedCount);
|
||||
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
const users = try client.adminListUsers(allocator);
|
||||
defer authmod.freeUserList(allocator, users);
|
||||
// Guards against the OTHER failure mode a lost lock could produce: the
|
||||
// two racing `db.put` calls interleaving their encoded bytes into one
|
||||
// corrupted record instead of cleanly overwriting each other, which
|
||||
// would still leave exactly one key but might fail to decode, or a
|
||||
// duplicate row under the same key if compaction ever saw two
|
||||
// physically distinct writes land as one logically doubled entry.
|
||||
try std.testing.expectEqual(@as(usize, 1), users.len);
|
||||
try std.testing.expectEqualStrings("shared-user", users[0].username);
|
||||
}
|
||||
|
||||
// Same shared-mutex proof as the create-race test above, but for the
|
||||
// read-modify-write shape (`setPassword`/`setRole`, each its own
|
||||
// `mustGet` then `db.put`) rather than the check-then-write shape of
|
||||
// `create`, plus a concurrent reader (`list`) racing the writers - the
|
||||
// third distinct access pattern `user_admin` exposes over the wire.
|
||||
test "user_admin survives many real concurrent connections mutating and listing the same account" {
|
||||
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 setup = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer setup.close();
|
||||
try setup.adminCreateUser("shared-admin", "init-pw", .readonly, 4);
|
||||
|
||||
const writerCount = 6;
|
||||
const perWriter = 40;
|
||||
const readerCount = 3;
|
||||
const readsPerReader = 200;
|
||||
|
||||
const WriteCtx = struct {
|
||||
allocator: Allocator,
|
||||
addr: std.net.Address,
|
||||
thread: usize,
|
||||
ok: bool = false,
|
||||
};
|
||||
|
||||
const writeWorker = struct {
|
||||
fn run(ctx: *WriteCtx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var i: usize = 0;
|
||||
while (i < perWriter) : (i += 1) {
|
||||
var buf: [32]u8 = undefined;
|
||||
const password = std.fmt.bufPrint(&buf, "pw-{d}-{d}", .{ ctx.thread, i }) catch return;
|
||||
client.adminSetPassword("shared-admin", password, 4) catch return;
|
||||
const role: authmod.Role = if (i % 2 == 0) .readwrite else .readonly;
|
||||
client.adminSetRole("shared-admin", role) catch return;
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.run;
|
||||
|
||||
Reference in new issue
Block a user