From 638c66a0aba3a0d04f45e34d6d571665a16912ca Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sat, 12 Sep 2026 15:40:23 +0200 Subject: [PATCH] feat: add incrWorker and run to server --- src/server.zig | 158 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/src/server.zig b/src/server.zig index 8dc692f..d11376b 100644 --- a/src/server.zig +++ b/src/server.zig @@ -6753,3 +6753,161 @@ test "coll_op survives many real concurrent connections hammering the same count ok: bool = false, }; + const incrWorker = struct { + fn run(ctx: *IncrCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < perThread) : (i += 1) { + var reply = client.collOp(ctx.allocator, .counter_incrby, "shared-counter", &[_][]const u8{"1"}) catch return; + reply.deinit(ctx.allocator); + } + ctx.ok = true; + } + }.run; + + var incrContexts: [threadCount]IncrCtx = undefined; + var incrThreads: [threadCount]std.Thread = undefined; + for (0..threadCount) |i| { + incrContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() }; + incrThreads[i] = try std.Thread.spawn(.{}, incrWorker, .{&incrContexts[i]}); + } + for (0..threadCount) |i| incrThreads[i].join(); + for (0..threadCount) |i| try std.testing.expect(incrContexts[i].ok); + + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + var total = try client.collOp(allocator, .counter_get, "shared-counter", &.{}); + defer total.deinit(allocator); + // Every increment is a separate request from a separate connection, so + // a missing lock around `runColl`'s read-modify-write would show up + // here as a total strictly less than threadCount*perThread (lost + // updates), not as a crash - the mutex in `Server.dispatch` must make + // this land on the exact total every single run. + try std.testing.expectEqual(@as(i64, threadCount * perThread), total.int); +} + +// Same shape as the counter test above but for a structural collection +// (`Set`) instead of a single scalar: many connections each add their OWN +// disjoint members to one shared set, so a race would show up as a +// missing member (an add silently lost to a concurrent writer) rather +// than a wrong total, which is a different failure mode than incrBy's +// lost-update and worth its own proof. +test "coll_op survives many real concurrent connections adding disjoint members to the same set" { + 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 perThread = 50; + + const AddCtx = struct { + allocator: Allocator, + addr: std.net.Address, + thread: usize, + ok: bool = false, + }; + + const addWorker = struct { + fn run(ctx: *AddCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < perThread) : (i += 1) { + var buf: [32]u8 = undefined; + const member = std.fmt.bufPrint(&buf, "t{d}-m{d}", .{ ctx.thread, i }) catch return; + var reply = client.collOp(ctx.allocator, .set_add, "shared-set", &[_][]const u8{member}) catch return; + reply.deinit(ctx.allocator); + } + ctx.ok = true; + } + }.run; + + var addContexts: [threadCount]AddCtx = undefined; + var addThreads: [threadCount]std.Thread = undefined; + for (0..threadCount) |i| { + addContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i }; + addThreads[i] = try std.Thread.spawn(.{}, addWorker, .{&addContexts[i]}); + } + for (0..threadCount) |i| addThreads[i].join(); + for (0..threadCount) |i| try std.testing.expect(addContexts[i].ok); + + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + var card = try client.collOp(allocator, .set_card, "shared-set", &.{}); + defer card.deinit(allocator); + try std.testing.expectEqual(@as(i64, threadCount * perThread), card.int); + + // Spot-check a member from every thread, not just the aggregate count - + // a bug that dropped exactly one member per thread but "gained" one + // duplicate elsewhere could still pass the cardinality check above. + for (0..threadCount) |t| { + var buf: [32]u8 = undefined; + const member = try std.fmt.bufPrint(&buf, "t{d}-m0", .{t}); + var isMem = try client.collOp(allocator, .set_ismember, "shared-set", &[_][]const u8{member}); + defer isMem.deinit(allocator); + try std.testing.expectEqual(@as(i64, 1), isMem.int); + } +} + +test "coll_op applies the same lazy-expiry check as coll_ttl_op before every dispatch" { + 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 collmod.Hash.set(&db, allocator, "session", "user", "ada"); + try client.collTtlExpire(.hash, "session", 0); + std.time.sleep(5 * std.time.ns_per_ms); + + // The lazy check in `handleCollOp` reaps "session" before hash_get + // ever runs, so the field reads back as gone, not as its old value. + var got = try client.collOp(allocator, .hash_get, "session", &[_][]const u8{"user"}); + defer got.deinit(allocator); + try std.testing.expect(got == .nil); + try std.testing.expect(!try collmod.collectionExists(&db, allocator, .hash, "session")); +} + +test "coll_op set_inter/union/diff lazily expire every extra set name, not just the first" { + 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(); +