feat: add WriteCtx to server

src/server.zig

@@ -6911,3 +6911,158 @@ test "coll_op set_inter/union/diff lazily expire every extra set name, not just
     var client = try client_mod.Client.connect(allocator, server.boundAddress());
     defer client.close();
 
+    _ = try collmod.Set.add(&db, allocator, "a", "x");
+    _ = try collmod.Set.add(&db, allocator, "b", "x");
+    try client.collTtlExpire(.set, "b", 0);
+    std.time.sleep(5 * std.time.ns_per_ms);
+
+    // "b" is due but is only referenced through `args`, never `name` - if
+    // the handler only checked `name` (like it does for every single-name
+    // action), "b" would still look alive here and `intersect` would see
+    // its stale "x" member. It must come back empty instead.
+    var inter = try client.collOp(allocator, .set_inter, "a", &[_][]const u8{"b"});
+    defer inter.deinit(allocator);
+    try std.testing.expectEqual(@as(usize, 0), inter.stringList.len);
+}
+
+test "coll_op gates a read action as readonly and a write action as readwrite" {
+    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();
+    _ = try collmod.Hash.set(&db, allocator, "k", "f", "v");
+
+    const store = authmod.Store.init(&db);
+    try store.createUser(allocator, "reader", "pw", .readonly, 16);
+    try store.createUser(allocator, "writer", "pw", .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 writer = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer writer.close();
+    try writer.authenticate("writer", "pw");
+
+    var reader = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer reader.close();
+    try reader.authenticate("reader", "pw");
+
+    // hash_get is read-only: a readonly connection can run it.
+    var got = try reader.collOp(allocator, .hash_get, "k", &[_][]const u8{"f"});
+    defer got.deinit(allocator);
+    try std.testing.expectEqualStrings("v", got.bytes);
+
+    // hash_set mutates: forbidden for a readonly connection.
+    try std.testing.expectError(client_mod.ClientError.ServerForbidden, reader.collOp(allocator, .hash_set, "k", &[_][]const u8{ "f", "v2" }));
+
+    // The readwrite connection can do it. "f" already exists, so hash_set
+    // reports it as updated (0), not created (1) - same int-flag shape as
+    // set_add/set_rem/hash_del elsewhere in this opcode, not `.ok`.
+    var setReply = try writer.collOp(allocator, .hash_set, "k", &[_][]const u8{ "f", "v2" });
+    defer setReply.deinit(allocator);
+    try std.testing.expectEqual(@as(i64, 0), setReply.int);
+}
+
+test "server rejects a malformed coll_op frame with bad_request without crashing" {
+    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 stream = try std.net.tcpConnectToAddress(server.boundAddress());
+    defer stream.close();
+
+    // A single action byte is not enough to decode the length-prefixed
+    // name that must follow it.
+    try protocol.writeFrame(stream.writer(), @intFromEnum(protocol.Op.coll_op), "\x00");
+    var frame = try protocol.readFrame(allocator, stream.reader());
+    defer frame.deinit();
+    try std.testing.expectEqual(@intFromEnum(protocol.Status.bad_request), frame.tag);
+}
+
+test "server answers coll_op with bad_request when an action gets too few arguments" {
+    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();
+
+    // counter_set needs a value argument - a well-formed frame that omits
+    // it should come back as a clean bad_request, not an internal error or
+    // a crash from indexing into an empty args slice.
+    try std.testing.expectError(client_mod.ClientError.ServerBadRequest, client.collOp(allocator, .counter_set, "n", &.{}));
+}
+
+// Same ksztalt of proof as the two `coll_op` concurrency tests above
+// (many real OS threads, each its own TCP connection, hammering shared
+// server state) but for `ttl_op` instead - the next direction flagged at
+// the end of session solo #32 (cobalt-postep.md). `ttl_op` has no shared
+// mutable VALUE the way a counter or set does (each writer's key is its
+// own, disjoint from every other writer's), so the race worth proving
+// here is different: writer connections racing `set`+`expire` against
+// sweeper connections racing `sweep`, all against the SAME `Db.sstables`/
+// memtable underneath. `ttl.sweepExpired` reads its own prefix scan and
+// deletes what it found within a single `handleTtlOp` call, which the
+// `self.mutex` in `Server.dispatch` executes atomically end-to-end, so a
+// key can never be swept twice and no swept count can double-count - the
+// grand total across every sweep response plus one final mop-up sweep
+// must land on exactly the number of keys written, with zero lost and
+// zero duplicated, no matter how the OS interleaves the connections.
+test "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,
+    };
+