feat: flesh out server

src/server.zig

@@ -7552,3 +7552,164 @@ test "server txn_op rejects a CHECK constraint violation over the wire and commi
     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();
+
+    const entries = [_]protocol.TxnOpEntry{
+        .{ .verb = .put, .table = "orders", .pk = .{ .int = 1 }, .fields = &.{.{ .name = "qty", .value = .{ .int = 5 } }} },
+        .{ .verb = .put, .table = "orders", .pk = .{ .int = 2 }, .fields = &.{.{ .name = "qty", .value = .{ .int = -1 } }} },
+    };
+    try std.testing.expectError(client_mod.ClientError.ServerInternal, client.txnOp(&entries));
+
+    // Whole transaction is atomic across both entries: the earlier,
+    // individually-valid row 1 must not have been committed either.
+    var orders = try querymod.Table.open(&db, allocator, "orders");
+    defer orders.deinit();
+    try std.testing.expect((try orders.get(.{ .int = 1 })) == null);
+    try std.testing.expect((try orders.get(.{ .int = 2 })) == null);
+}
+
+// `handleTxnOp` runs every entry through `TableTxn.put`/`TableTxn.delete`
+// (see txn_table.zig), which now enforce ON DELETE RESTRICT exactly like
+// CHECK/UNIQUE/FOREIGN KEY on put above - never separately wired for this
+// opcode either. The interesting case a plain `table_delete` test cannot
+// cover: both the child insert AND the restricted parent delete are only
+// staged within this ONE `txn_op` batch, never durable on their own -
+// `TableTxn.delete`'s RESTRICT check has to see the child through
+// `Transaction.scanPrefix`'s same-transaction visibility (see
+// txn_table.zig's own unit test for this), not a plain Db scan, or this
+// would wrongly let the delete through.
+test "server txn_op rejects an ON DELETE RESTRICT violation over the wire, seeing a child staged earlier in the same batch" {
+    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();
+
+    {
+        var customers = try querymod.Table.open(&db, allocator, "customers");
+        defer customers.deinit();
+        try customers.put(.{ .int = 1 }, &.{});
+
+        var orders = try querymod.Table.open(&db, allocator, "orders");
+        defer orders.deinit();
+        try orders.createForeignKey("customer_id", "customers", querymod.pkField, .restrict);
+    }
+
+    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();
+
+    const entries = [_]protocol.TxnOpEntry{
+        .{ .verb = .put, .table = "orders", .pk = .{ .int = 100 }, .fields = &.{.{ .name = "customer_id", .value = .{ .int = 1 } }} },
+        .{ .verb = .del, .table = "customers", .pk = .{ .int = 1 } },
+    };
+    try std.testing.expectError(client_mod.ClientError.ServerInternal, client.txnOp(&entries));
+
+    // Nothing committed: neither the staged order nor the rejected delete.
+    var orders = try querymod.Table.open(&db, allocator, "orders");
+    defer orders.deinit();
+    try std.testing.expect((try orders.get(.{ .int = 100 })) == null);
+
+    var customers = try querymod.Table.open(&db, allocator, "customers");
+    defer customers.deinit();
+    var stillThere = (try customers.get(.{ .int = 1 })).?;
+    stillThere.deinit();
+}
+
+test "server txn_op rejects an empty operation list with bad_request" {
+    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 std.testing.expectError(client_mod.ClientError.ServerBadRequest, client.txnOp(&.{}));
+}
+
+test "server rejects a malformed txn_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 u32 entry count of 1 promises one full entry, but the payload ends
+    // right after it - not even a verb byte follows.
+    var payload: [4]u8 = undefined;
+    std.mem.writeInt(u32, &payload, 1, .little);
+    try protocol.writeFrame(stream.writer(), @intFromEnum(protocol.Op.txn_op), &payload);
+    var frame = try protocol.readFrame(allocator, stream.reader());
+    defer frame.deinit();
+    try std.testing.expectEqual(@intFromEnum(protocol.Status.bad_request), frame.tag);
+
+    // The connection must still be usable afterward - a malformed frame is
+    // a client mistake, not a fatal error for the server.
+    try protocol.writeFrame(stream.writer(), @intFromEnum(protocol.Op.ping), "");
+    var pingFrame = try protocol.readFrame(allocator, stream.reader());
+    defer pingFrame.deinit();
+    try std.testing.expectEqual(@intFromEnum(protocol.Status.ok), pingFrame.tag);
+}
+
+// Proves the `txn_op` code path (decode -> `Transaction`/`TableTxn` ->
+// index maintenance -> commit, all under `Server.dispatch`'s single
+// `self.mutex`) holds up under real concurrent OS threads each on their
+// own real TCP connection - the same rigor `coll_op`'s/`table_select`'s
+// own concurrency tests apply, but for the table+index layer instead of
+// the flat keyspace or the Redis-like collections, which is a genuinely
+// different code path (this is the first concurrency test to drive
+// `TableTxn`/index maintenance through a live socket at all;
+// `concurrent.zig`'s own `TableTxn` concurrency test bypasses the network
+// entirely via `GuardedDb`). Each worker writes into two disjoint rows per
+// round (its own row in `widgets`, indexed by `owner`, and a companion row
+// in `tags`) so a lost or duplicated write from bad interleaving would
+// show up as a wrong row count or a wrong index membership, not just a
+// wrong number.
+test "server txn_op survives many real concurrent connections writing disjoint rows across two tables" {
+    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();
+
+    {
+        var widgets = try querymod.Table.open(&db, allocator, "widgets");
+        defer widgets.deinit();
+        try widgets.createIndex("owner");
+    }
+