diff --git a/src/server.zig b/src/server.zig index a4264ae..ab80c1a 100644 --- a/src/server.zig +++ b/src/server.zig @@ -1086,3 +1086,160 @@ pub const Server = struct { try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); } + /// Executes a `txn_op` request: the network counterpart of the CLI's + /// local `txn ` (main.zig's `runTxn`), applying every + /// put/delete in `entries` as one atomic `Transaction` commit across + /// however many tables they touch. Runs entirely inside `dispatch`'s + /// single `self.mutex` acquisition, like every other opcode - which + /// means, unlike a raw multi-thread `Transaction` against an unguarded + /// `*Db` (see `concurrent.zig`'s `GuardedDb` doc comment), no other + /// connection's request can ever interleave between this transaction's + /// begin and its commit. `txn.commit()`'s OCC conflict check stays in + /// place and still maps to `Status.conflict` below, but under this + /// server's locking discipline nothing can actually trigger it for a + /// request that arrives, stages, and commits within one dispatch call - + /// the same reason `concurrent.zig`'s `TableTransferWorker` test treats + /// a real `error.Conflict` under `GuardedDb.acquire()` as a bug, not an + /// expected outcome. + fn handleTxnOp(self: *Server, payload: []const u8, writer: anytype) !void { + const entries = protocol.decodeTxnOp(self.allocator, payload) catch { + return writeErr(writer, .bad_request, "malformed txn_op payload"); + }; + defer self.allocator.free(entries); + if (entries.len == 0) return writeErr(writer, .bad_request, "txn_op requires at least one operation"); + + var txn = txnmod.Transaction.begin(self.db); + defer txn.deinit(); + + var tables = std.StringHashMap(txntablemod.TableTxn).init(self.allocator); + defer { + var it = tables.valueIterator(); + while (it.next()) |t| t.deinit(); + tables.deinit(); + } + + for (entries) |e| { + const gop = tables.getOrPut(e.table) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (!gop.found_existing) { + gop.value_ptr.* = txntablemod.TableTxn.open(&txn, self.allocator, e.table) catch |err| { + return writeErr(writer, .bad_request, @errorName(err)); + }; + gop.key_ptr.* = e.table; + } + const table = gop.value_ptr; + + const pk = rowmod.decodeValue(self.allocator, e.pkBytes) catch { + return writeErr(writer, .bad_request, "malformed primary key"); + }; + defer pk.free(self.allocator); + + switch (e.verb) { + .put => { + var fieldsRow = rowmod.Row.decode(self.allocator, e.fieldsBytes) catch { + return writeErr(writer, .bad_request, "malformed fields"); + }; + defer fieldsRow.deinit(); + table.put(pk, fieldsRow.fields.items) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + }, + .del => { + table.delete(pk) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + }, + } + } + + txn.commit() catch |err| { + return switch (err) { + error.Conflict => writeErr(writer, .conflict, "transaction conflict"), + else => writeErr(writer, .internal, @errorName(err)), + }; + }; + + const out = protocol.encodeTxnResult(self.allocator, @intCast(entries.len)) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(out); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out); + } + + /// Executes a `table select [expr]` remotely: opens the named + /// table over the same `*Db` the rest of the server serves (so it sees + /// exactly the indexes/rows a local CLI `table` command would), parses + /// the expression text with the same `expr.zig` grammar the CLI uses + /// (empty text means "every row", matching `Table.select(null)`), and + /// wire-encodes the resulting rows. `args.orderBy`, when present, is + /// forwarded to `Table.selectPageOrdered` so the offset/limit window + /// applies AFTER sorting, exactly like the local `table select + /// --order-by` CLI path - `null` falls through to the plain + /// `selectPage` early-stop behavior instead. A bad table name or + /// malformed expression is a client mistake (`bad_request`), not a + /// server fault. + fn handleTableSelect(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeTableSelect(self.allocator, payload) catch { + return writeErr(writer, .bad_request, "malformed table_select payload"); + }; + defer self.allocator.free(args.orderBy); + + var table = querymod.Table.open(self.db, self.allocator, args.table) catch |err| { + return writeErr(writer, .bad_request, @errorName(err)); + }; + defer table.deinit(); + + var parsed: ?exprmod.Parsed = null; + defer if (parsed) |*p| p.deinit(); + if (args.exprText.len > 0) { + parsed = exprmod.parse(self.allocator, args.exprText) catch |err| { + return writeErr(writer, .bad_request, @errorName(err)); + }; + } + const pred: ?querymod.Predicate = if (parsed) |p| p.predicate else null; + + const rows = table.selectPageOrdered(pred, .{ .offset = args.offset, .limit = args.limit }, args.orderBy) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer querymod.freeRows(self.allocator, rows); + + const out = protocol.encodeRows(self.allocator, rows) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(out); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out); + } + + /// Executes a `table_join` request against `join.zig`'s four equi-join + /// variants, chosen by `args.mode` - the network counterpart of the + /// CLI's local `table join [--left|--right|--full]`. Each variant + /// produces its own result type (`JoinedRow`/`OuterJoinedRow`/ + /// `RightOuterJoinedRow`/`FullOuterJoinedRow`), but every one of them is + /// just two optional `Row`s under the hood, so this flattens whichever + /// one came back into `protocol.JoinPair` before encoding - one wire + /// format for all four modes, exactly like `coll_op` bundles thirty + /// collection actions behind one opcode instead of thirty. Ownership of + /// each row is moved (by value) from the mode-specific slice into + /// `pairs`, then that mode-specific slice is freed as a raw allocation + /// (not through `join.freeJoined`/`freeOuterJoined`/etc, which would + /// double-deinit rows already moved into `pairs`) - `pairs`'s own defer + /// is what actually frees the row contents, once encoding is done. + /// `args.offset`/`args.limit` window the mode-specific slice through + /// `join.applyPage` right after it comes back, before the flatten - + /// `applyPage` already frees the pre-page slice and hands back a fresh + /// one, so the pattern below frees `paged`, never `joined` directly. + fn handleTableJoin(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeTableJoin(payload) catch { + return writeErr(writer, .bad_request, "malformed table_join payload"); + }; + + var left = querymod.Table.open(self.db, self.allocator, args.leftTable) catch |err| { + return writeErr(writer, .bad_request, @errorName(err)); + }; + defer left.deinit(); + var right = querymod.Table.open(self.db, self.allocator, args.rightTable) catch |err| { + return writeErr(writer, .bad_request, @errorName(err)); + }; + defer right.deinit(); +