feat: add writeWorker and run to server

src/server.zig

@@ -7066,3 +7066,166 @@ test "ttl_op survives many real concurrent connections racing writes against swe
         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 key = std.fmt.bufPrint(&buf, "ttl-conc-{d}-{d}", .{ ctx.thread, i }) catch return;
+                client.set(key, "v") catch return;
+                client.ttlExpire(key, 0) catch return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    const SweepCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        reaped: std.atomic.Value(u64) = std.atomic.Value(u64).init(0),
+        ok: bool = false,
+    };
+
+    const sweepWorker = struct {
+        fn run(ctx: *SweepCtx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var i: usize = 0;
+            while (i < sweepsPerSweeper) : (i += 1) {
+                const n = client.ttlSweep() catch return;
+                _ = ctx.reaped.fetchAdd(n, .monotonic);
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    var writeContexts: [writerCount]WriteCtx = undefined;
+    var writeThreads: [writerCount]std.Thread = undefined;
+    var sweepContexts: [sweeperCount]SweepCtx = undefined;
+    var sweepThreads: [sweeperCount]std.Thread = undefined;
+
+    for (0..sweeperCount) |i| {
+        sweepContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
+        sweepThreads[i] = try std.Thread.spawn(.{}, sweepWorker, .{&sweepContexts[i]});
+    }
+    for (0..writerCount) |i| {
+        writeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
+        writeThreads[i] = try std.Thread.spawn(.{}, writeWorker, .{&writeContexts[i]});
+    }
+    for (0..writerCount) |i| writeThreads[i].join();
+    for (0..sweeperCount) |i| sweepThreads[i].join();
+    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();
+
+    // Any key that lost the race against every sweeper (written after the
+    // last sweep call landed) is mopped up by one final sweep here.
+    const finalReaped = try client.ttlSweep();
+
+    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 key from every writer thread: all must be truly gone
+    // (not just marker-cleared), proving the sweeps were real deletes.
+    for (0..writerCount) |t| {
+        var buf: [32]u8 = undefined;
+        const key = try std.fmt.bufPrint(&buf, "ttl-conc-{d}-0", .{t});
+        try std.testing.expectEqual(@as(?[]u8, null), try client.get(allocator, key));
+        try std.testing.expectEqual(ttlmod.TtlResult.noKey, try client.ttlQuery(key));
+    }
+}
+
+// Mirrors the `ttl_op` concurrency test above but against `coll_ttl_op`
+// and a real collection (`Counter`) instead of a flat key - the sibling
+// gap the same end-of-#32 note called out explicitly ("ttl_op dziala na
+// plaskim keyspace... analogiczny test... jest wykonalny, ale odlozony").
+// `collttl.sweepExpired` (collttl.zig) shares the exact same
+// scan-then-delete-within-one-call shape as `ttl.sweepExpired`, so the
+// same exactly-once accounting argument applies.
+test "coll_ttl_op survives many real concurrent connections racing writes against sweeps" {
+    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 writerCount = 6;
+    const perWriter = 40;
+    const sweeperCount = 3;
+    const sweepsPerSweeper = 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 name = std.fmt.bufPrint(&buf, "ttlcoll-conc-{d}-{d}", .{ ctx.thread, i }) catch return;
+                var reply = client.collOp(ctx.allocator, .counter_incrby, name, &[_][]const u8{"1"}) catch return;
+                reply.deinit(ctx.allocator);
+                client.collTtlExpire(.counter, name, 0) catch return;
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    const SweepCtx = struct {
+        allocator: Allocator,
+        addr: std.net.Address,
+        reaped: std.atomic.Value(u64) = std.atomic.Value(u64).init(0),
+        ok: bool = false,
+    };
+
+    const sweepWorker = struct {
+        fn run(ctx: *SweepCtx) void {
+            var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
+            defer client.close();
+            var i: usize = 0;
+            while (i < sweepsPerSweeper) : (i += 1) {
+                const n = client.collTtlSweep() catch return;
+                _ = ctx.reaped.fetchAdd(n, .monotonic);
+            }
+            ctx.ok = true;
+        }
+    }.run;
+
+    var writeContexts: [writerCount]WriteCtx = undefined;
+    var writeThreads: [writerCount]std.Thread = undefined;
+    var sweepContexts: [sweeperCount]SweepCtx = undefined;
+    var sweepThreads: [sweeperCount]std.Thread = undefined;
+
+    for (0..sweeperCount) |i| {
+        sweepContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
+        sweepThreads[i] = try std.Thread.spawn(.{}, sweepWorker, .{&sweepContexts[i]});
+    }
+    for (0..writerCount) |i| {
+        writeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
+        writeThreads[i] = try std.Thread.spawn(.{}, writeWorker, .{&writeContexts[i]});
+    }
+    for (0..writerCount) |i| writeThreads[i].join();
+    for (0..sweeperCount) |i| sweepThreads[i].join();
+    for (0..writerCount) |i| try std.testing.expect(writeContexts[i].ok);
+    for (0..sweeperCount) |i| try std.testing.expect(sweepContexts[i].ok);
+