From f39d19a8ba5c3530e0365a558fdd7b98c62b8bbf Mon Sep 17 00:00:00 2001 From: Kinflou Date: Wed, 2 Sep 2026 16:51:06 +0800 Subject: [PATCH] feat(runtime): framing, transport, Client, Server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The second layer of @comline/runtime — the pieces a generated Client / dispatcher plug into: - Framing interface + DatagramFraming (compact `[call_id:u16][request_id:u64] [params]` request, `[request_id:u64][tag-byte envelope]` response) and JsonRpcFraming (`{"jsonrpc":"2.0",...}`), both wire-compatible with comline-runtime — the datagram header layout and the JSON-RPC frame wording are byte-checked in the tests. - envelope.ts — the `[0]payload` / `[1]id:u16 body` tag-byte form. - Transport interface + duplex() — a connected in-memory pair; close() ends a serve loop. - Client — connect() runs the handshake; call() frames / sends / awaits / returns the raw Envelope; notify() is fire-and-forget. - Server — serve() / serveHandshaked(): decode request, resolve the call against Dispatch.calls(), dispatch into a Reply, frame the response by outcome (none -> silent, ok, err+ordinal). Tests add a full hand-written Chat client ⇆ provider round-trip over duplex() — request/response, a raised typed error, a one-way notify — run against both framings, plus a handshake wire-format-mismatch rejection. --- runtime/README.md | 36 ++++--- runtime/src/client.ts | 61 ++++++++++++ runtime/src/contract.ts | 34 +++++++ runtime/src/envelope.ts | 43 ++++++++ runtime/src/framing/datagram.ts | 60 ++++++++++++ runtime/src/framing/jsonrpc.ts | 80 +++++++++++++++ runtime/src/index.ts | 20 +++- runtime/src/server.ts | 73 ++++++++++++++ runtime/src/transport.ts | 81 +++++++++++++++ runtime/test/framing.test.ts | 93 ++++++++++++++++++ runtime/test/roundtrip.test.ts | 169 ++++++++++++++++++++++++++++++++ 11 files changed, 735 insertions(+), 15 deletions(-) create mode 100644 runtime/src/client.ts create mode 100644 runtime/src/envelope.ts create mode 100644 runtime/src/framing/datagram.ts create mode 100644 runtime/src/framing/jsonrpc.ts create mode 100644 runtime/src/server.ts create mode 100644 runtime/src/transport.ts create mode 100644 runtime/test/framing.test.ts create mode 100644 runtime/test/roundtrip.test.ts diff --git a/runtime/README.md b/runtime/README.md index a2c0011..3f87dd9 100644 --- a/runtime/README.md +++ b/runtime/README.md @@ -5,31 +5,41 @@ the counterpart of the Rust `comline-runtime` crate. ## Status -First cut: the framing-agnostic **contract** plus a JSON codec. No `Framing`, -`Transport`, `Client`, or `Server` yet — those, and a generator that emits a -`Client` / dispatcher against this package, follow. +The contract, two framings, an in-memory transport, and a `Client` / `Server`. +Next: a generator that emits a `Client` / dispatcher against this +package, and a stream transport. | Piece | State | |---|---| -| `RuntimeError`, `Kind` / `resolveKind`, `Call`, `Envelope`, `Outcome`, `Reply`, `CallError` | ✅ | +| `RuntimeError`, `Kind` / `resolveKind`, `Call`, `Envelope`, `Outcome`, `Reply`, `CallError`, `Dispatch` | ✅ | | `Codec` interface + `JsonCodec` (`name === "json"`) | ✅ | | `Handshake` — 31-byte frame, FNV-1a `nameHash`, `check` — byte-compatible with `comline-runtime` | ✅ | -| `Dispatch` interface | ✅ (shape only) | -| `Framing` (datagram + JSON-RPC), `Transport`, `Client`, `Server` | — | +| `Framing` + `DatagramFraming` + `JsonRpcFraming` — wire-compatible with `comline-runtime` | ✅ | +| `Transport` interface + `duplex()` in-memory pair | ✅ | +| `Client` (`connect` / `call` / `notify`) and `Server` (`serve` / `serveHandshaked`) | ✅ | +| A stream `Transport`; a MessagePack `Codec` | — | -The `Handshake` wire format and `nameHash` are cross-checked against +The `Handshake` and framing wire formats are cross-checked against `comline-runtime`'s reference vectors, so a TypeScript peer and a Rust peer -generated from the same schema negotiate the same frame. +generated from the same schema negotiate the same frame and speak the same +request / response bytes. ## Layout ``` src/ - contract.ts RuntimeError, Kind, Call, Envelope, Outcome, Reply, CallError, Codec, Dispatch - handshake.ts Handshake, nameHash, FRAMING_DATAGRAM - codec.ts JsonCodec - index.ts public surface -test/ node:test, zero runtime deps + contract.ts RuntimeError, Kind, Call, Envelope, Reply, CallError, Codec, Dispatch, Framing + handshake.ts Handshake, nameHash, FRAMING_DATAGRAM + codec.ts JsonCodec + envelope.ts the datagram tag-byte Envelope form + framing/ + datagram.ts DatagramFraming + jsonrpc.ts JsonRpcFraming + transport.ts Transport, duplex() + client.ts Client + server.ts Server + index.ts public surface +test/ node:test, zero runtime deps ``` ## Develop diff --git a/runtime/src/client.ts b/runtime/src/client.ts new file mode 100644 index 0000000..0298417 --- /dev/null +++ b/runtime/src/client.ts @@ -0,0 +1,61 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import type { Call, Codec, Envelope, Framing } from "./contract.js"; +import { RuntimeError } from "./contract.js"; +import { Handshake } from "./handshake.js"; +import type { Transport } from "./transport.js"; +import { DatagramFraming } from "./framing/datagram.js"; + +/** + * The consumer side. Frames a call, sends it, waits for the matching response, + * and hands the generated stub the raw {@link Envelope} to decode. Generic over + * the {@link Framing}; defaults to {@link DatagramFraming}. + */ +export class Client { + private nextId = 0n; + + constructor( + private readonly transport: Transport, + readonly codec: Codec, + readonly framing: Framing = new DatagramFraming(), + ) {} + + /** + * Bind and run the connection {@link Handshake}: send `local`, read the + * peer's, refuse (`RuntimeError("handshake")`) on a mismatch. + */ + static async connect( + transport: Transport, + codec: Codec, + local: Handshake, + framing: Framing = new DatagramFraming(), + ): Promise { + await transport.send(local.encode()); + const peer = Handshake.decode(await transport.recv()); + if (!peer) throw RuntimeError.handshake(); + local.check(peer); + return new Client(transport, codec, framing); + } + + /** Make `call` with `params`, block for the response, return its {@link Envelope}. */ + async call(call: Call, params: unknown): Promise { + const requestId = this.nextId++; + await this.transport.send( + this.framing.encodeRequest(call, requestId, this.codec.encode(params)), + ); + const res = this.framing.decodeResponse(await this.transport.recv()); + if (!res) throw RuntimeError.framing(); + if (res.requestId !== requestId) throw RuntimeError.framing(); + return res.envelope; + } + + /** Fire-and-forget: send the call, expect no response (a one-way function). */ + async notify(call: Call, params: unknown): Promise { + const requestId = this.nextId++; + await this.transport.send( + this.framing.encodeRequest(call, requestId, this.codec.encode(params)), + ); + } +} diff --git a/runtime/src/contract.ts b/runtime/src/contract.ts index 15ab1f5..1c93447 100644 --- a/runtime/src/contract.ts +++ b/runtime/src/contract.ts @@ -141,3 +141,37 @@ export interface Dispatch { calls(): readonly string[]; dispatch(call: Kind, params: Uint8Array, codec: Codec, reply: Reply): Promise; } + +/** Whichever call address a framing put on the wire. */ +export type RequestCall = { readonly id: number } | { readonly name: string }; + +/** A decoded request frame. */ +export interface DecodedRequest { + readonly call: RequestCall; + readonly requestId: bigint; + /** The params sub-frame, independently decodable with the peer's {@link Codec}. */ + readonly params: Uint8Array; +} + +/** A decoded response frame: the correlation id and its {@link Envelope}. */ +export interface DecodedResponse { + readonly requestId: bigint; + readonly envelope: Envelope; +} + +/** + * How a call becomes bytes and back — the axis orthogonal to {@link Codec} + * (which serializes the *parts*). {@link DatagramFraming} is the default; + * {@link JsonRpcFraming} is the name-oriented alternative. `params` / `payload` + * / `body` arrive already {@link Codec}-encoded; the framing only positions + * them. Both ends of a connection must agree — the {@link Handshake} carries + * `name`, hashed. + */ +export interface Framing { + readonly name: string; + encodeRequest(call: Call, requestId: bigint, params: Uint8Array): Uint8Array; + decodeRequest(frame: Uint8Array): DecodedRequest | undefined; + encodeResponseOk(requestId: bigint, payload: Uint8Array): Uint8Array; + encodeResponseErr(requestId: bigint, id: number, body: Uint8Array): Uint8Array; + decodeResponse(frame: Uint8Array): DecodedResponse | undefined; +} diff --git a/runtime/src/envelope.ts b/runtime/src/envelope.ts new file mode 100644 index 0000000..517dd4a --- /dev/null +++ b/runtime/src/envelope.ts @@ -0,0 +1,43 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +/** + * The tag-byte {@link Envelope} form the Comline datagram framing wraps a + * response body in — `[0] payload` for ok, `[1] id:u16 LE body` for err. + * Matches `comline_runtime::contract::Envelope`. + */ + +import type { Envelope } from "./contract.js"; + +const TAG_OK = 0; +const TAG_ERR = 1; + +export function encodeEnvelopeOk(payload: Uint8Array): Uint8Array { + const out = new Uint8Array(1 + payload.length); + out[0] = TAG_OK; + out.set(payload, 1); + return out; +} + +export function encodeEnvelopeErr(id: number, body: Uint8Array): Uint8Array { + const out = new Uint8Array(3 + body.length); + out[0] = TAG_ERR; + out[1] = id & 0xff; + out[2] = (id >> 8) & 0xff; + out.set(body, 3); + return out; +} + +export function decodeEnvelope(frame: Uint8Array): Envelope | undefined { + const tag = frame[0]; + if (tag === TAG_OK) { + return { ok: frame.subarray(1) }; + } + if (tag === TAG_ERR) { + if (frame.length < 3) return undefined; + const id = frame[1]! | (frame[2]! << 8); + return { err: { id, body: frame.subarray(3) } }; + } + return undefined; +} diff --git a/runtime/src/framing/datagram.ts b/runtime/src/framing/datagram.ts new file mode 100644 index 0000000..126d14d --- /dev/null +++ b/runtime/src/framing/datagram.ts @@ -0,0 +1,60 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import type { Call, DecodedRequest, DecodedResponse, Framing } from "../contract.js"; +import { FRAMING_DATAGRAM } from "../handshake.js"; +import { decodeEnvelope, encodeEnvelopeErr, encodeEnvelopeOk } from "../envelope.js"; + +const HEAD = 10; // [call_id:u16][request_id:u64], both LE + +/** + * The Comline datagram framing — compact, one frame per message. Matches + * `comline_runtime::contract::DatagramFraming`: + * request `[call_id:u16 LE][request_id:u64 LE][params]`, + * response `[request_id:u64 LE][envelope]`. + */ +export class DatagramFraming implements Framing { + readonly name = FRAMING_DATAGRAM; + + encodeRequest(call: Call, requestId: bigint, params: Uint8Array): Uint8Array { + const out = new Uint8Array(HEAD + params.length); + const view = new DataView(out.buffer); + view.setUint16(0, call.id, true); + view.setBigUint64(2, requestId, true); + out.set(params, HEAD); + return out; + } + + decodeRequest(frame: Uint8Array): DecodedRequest | undefined { + if (frame.length < HEAD) return undefined; + const view = new DataView(frame.buffer, frame.byteOffset, HEAD); + return { + call: { id: view.getUint16(0, true) }, + requestId: view.getBigUint64(2, true), + params: frame.subarray(HEAD), + }; + } + + encodeResponseOk(requestId: bigint, payload: Uint8Array): Uint8Array { + return withRequestId(requestId, encodeEnvelopeOk(payload)); + } + + encodeResponseErr(requestId: bigint, id: number, body: Uint8Array): Uint8Array { + return withRequestId(requestId, encodeEnvelopeErr(id, body)); + } + + decodeResponse(frame: Uint8Array): DecodedResponse | undefined { + if (frame.length < 8) return undefined; + const requestId = new DataView(frame.buffer, frame.byteOffset, 8).getBigUint64(0, true); + const envelope = decodeEnvelope(frame.subarray(8)); + return envelope && { requestId, envelope }; + } +} + +function withRequestId(requestId: bigint, rest: Uint8Array): Uint8Array { + const out = new Uint8Array(8 + rest.length); + new DataView(out.buffer).setBigUint64(0, requestId, true); + out.set(rest, 8); + return out; +} diff --git a/runtime/src/framing/jsonrpc.ts b/runtime/src/framing/jsonrpc.ts new file mode 100644 index 0000000..0f719f3 --- /dev/null +++ b/runtime/src/framing/jsonrpc.ts @@ -0,0 +1,80 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import type { Call, DecodedRequest, DecodedResponse, Framing } from "../contract.js"; + +const enc = new TextEncoder(); +const dec = new TextDecoder(); + +/** JSON bytes of a value, or `null` for an empty sub-frame. */ +function jsonBytes(sub: unknown): Uint8Array { + return enc.encode(sub === undefined ? "null" : JSON.stringify(sub)); +} + +/** + * [JSON-RPC 2.0](https://www.jsonrpc.org/specification) framing — name-oriented, + * human-readable. Pair with {@link JsonCodec}. Matches + * `comline_runtime::framing::JsonRpcFraming`: + * + * - request: `{"jsonrpc":"2.0","method":,"params":,"id":}` + * - ok: `{"jsonrpc":"2.0","result":,"id":}` + * - err: `{"jsonrpc":"2.0","error":{"code":,"message":...,"data":},"id":}` + * + * `params` / `payload` / `body` arrive as already-encoded JSON bytes and are + * spliced in verbatim. + */ +export class JsonRpcFraming implements Framing { + readonly name = "jsonrpc-2.0"; + + encodeRequest(call: Call, requestId: bigint, params: Uint8Array): Uint8Array { + const p = params.length === 0 ? "null" : dec.decode(params); + return enc.encode( + `{"jsonrpc":"2.0","method":${JSON.stringify(call.name)},"params":${p},"id":${requestId}}`, + ); + } + + decodeRequest(frame: Uint8Array): DecodedRequest | undefined { + let r: { method?: unknown; params?: unknown; id?: unknown }; + try { + r = JSON.parse(dec.decode(frame)); + } catch { + return undefined; + } + if (typeof r.method !== "string") return undefined; + return { + call: { name: r.method }, + requestId: r.id === undefined || r.id === null ? 0n : BigInt(r.id as number), + params: jsonBytes(r.params), + }; + } + + encodeResponseOk(requestId: bigint, payload: Uint8Array): Uint8Array { + const r = payload.length === 0 ? "null" : dec.decode(payload); + return enc.encode(`{"jsonrpc":"2.0","result":${r},"id":${requestId}}`); + } + + encodeResponseErr(requestId: bigint, id: number, body: Uint8Array): Uint8Array { + const d = body.length === 0 ? "null" : dec.decode(body); + return enc.encode( + `{"jsonrpc":"2.0","error":{"code":${id},"message":"application error","data":${d}},"id":${requestId}}`, + ); + } + + decodeResponse(frame: Uint8Array): DecodedResponse | undefined { + let r: { result?: unknown; error?: { code?: unknown; data?: unknown }; id?: unknown }; + try { + r = JSON.parse(dec.decode(frame)); + } catch { + return undefined; + } + const requestId = r.id === undefined || r.id === null ? 0n : BigInt(r.id as number); + if (r.error && typeof r.error === "object") { + return { + requestId, + envelope: { err: { id: Number(r.error.code ?? 0), body: jsonBytes(r.error.data) } }, + }; + } + return { requestId, envelope: { ok: jsonBytes(r.result) } }; + } +} diff --git a/runtime/src/index.ts b/runtime/src/index.ts index 40b1b26..04adc51 100644 --- a/runtime/src/index.ts +++ b/runtime/src/index.ts @@ -4,8 +4,10 @@ /** * `@comline/runtime` — the TypeScript runtime that Comline-generated RPC - * bindings link against. This first cut is the framing-agnostic contract plus - * a JSON codec; `Framing`, `Transport`, `Client`, and `Server` follow. + * bindings link against. The framing-agnostic contract, two framings, an + * in-memory transport, and a `Client` / `Server`. The generator emitting a + * `Client` / dispatcher against this package, and a stream transport, + * follow. */ export { @@ -21,6 +23,10 @@ export { type CallError, type Codec, type Dispatch, + type RequestCall, + type DecodedRequest, + type DecodedResponse, + type Framing, } from "./contract.js"; export { @@ -31,3 +37,13 @@ export { } from "./handshake.js"; export { JsonCodec } from "./codec.js"; + +export { encodeEnvelopeOk, encodeEnvelopeErr, decodeEnvelope } from "./envelope.js"; + +export { DatagramFraming } from "./framing/datagram.js"; +export { JsonRpcFraming } from "./framing/jsonrpc.js"; + +export { type Transport, type InMemoryTransport, duplex } from "./transport.js"; + +export { Client } from "./client.js"; +export { Server } from "./server.js"; diff --git a/runtime/src/server.ts b/runtime/src/server.ts new file mode 100644 index 0000000..a1e4cea --- /dev/null +++ b/runtime/src/server.ts @@ -0,0 +1,73 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import type { Codec, Dispatch, Framing } from "./contract.js"; +import { Reply, RuntimeError, resolveKind } from "./contract.js"; +import { Handshake } from "./handshake.js"; +import type { Transport } from "./transport.js"; +import { DatagramFraming } from "./framing/datagram.js"; + +/** + * The provider side. Reads a request frame, dispatches it, writes the response + * frame — until the transport closes. Generic over the {@link Framing}; + * defaults to {@link DatagramFraming}. + */ +export class Server { + constructor( + private readonly dispatch: Dispatch, + private readonly codec: Codec, + private readonly framing: Framing = new DatagramFraming(), + ) {} + + /** Handle one call. `true` — served; `false` — the transport closed. */ + async serveOne(transport: Transport): Promise { + let frame: Uint8Array; + try { + frame = await transport.recv(); + } catch { + return false; + } + + const req = this.framing.decodeRequest(frame); + if (!req) throw RuntimeError.framing(); + + const idx = resolveKind(req.call, this.dispatch.calls()); + if (idx === undefined) throw RuntimeError.unknownCall(); + + const reply = new Reply(); + await this.dispatch.dispatch({ id: idx }, req.params, this.codec, reply); + + switch (reply.outcome.kind) { + case "none": + return true; // one-way call: nothing to reply + case "ok": + await transport.send(this.framing.encodeResponseOk(req.requestId, reply.body)); + return true; + case "err": + await transport.send( + this.framing.encodeResponseErr(req.requestId, reply.outcome.id, reply.body), + ); + return true; + } + } + + /** Serve calls until the transport closes. No handshake. */ + async serve(transport: Transport): Promise { + while (await this.serveOne(transport)) { + /* keep serving */ + } + } + + /** + * Run the connection {@link Handshake} against the connecting peer — send + * `local`, read theirs, refuse on a mismatch — then {@link Server.serve}. + */ + async serveHandshaked(transport: Transport, local: Handshake): Promise { + await transport.send(local.encode()); + const peer = Handshake.decode(await transport.recv()); + if (!peer) throw RuntimeError.handshake(); + local.check(peer); + await this.serve(transport); + } +} diff --git a/runtime/src/transport.ts b/runtime/src/transport.ts new file mode 100644 index 0000000..787e5b5 --- /dev/null +++ b/runtime/src/transport.ts @@ -0,0 +1,81 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +import { RuntimeError } from "./contract.js"; + +/** + * A message-oriented byte pipe: each {@link Transport.send} delivers exactly + * one frame to the peer's next {@link Transport.recv}. The counterpart of the + * Rust `Transport` trait; a stream transport (length-prefixed) is a later add. + */ +export interface Transport { + send(frame: Uint8Array): Promise; + /** Resolves with the next frame; rejects `RuntimeError("transport")` once the peer is gone and the queue is drained. */ + recv(): Promise; +} + +interface Waiter { + resolve(v: Uint8Array): void; + reject(e: unknown): void; +} + +/** One direction of an in-memory pipe: a queue plus parked receivers. */ +class Channel { + private readonly queue: Uint8Array[] = []; + private readonly waiters: Waiter[] = []; + private closed = false; + + push(frame: Uint8Array): void { + const waiter = this.waiters.shift(); + if (waiter) waiter.resolve(frame); + else this.queue.push(frame); + } + + pull(): Promise { + const next = this.queue.shift(); + if (next !== undefined) return Promise.resolve(next); + if (this.closed) return Promise.reject(RuntimeError.transport()); + return new Promise((resolve, reject) => this.waiters.push({ resolve, reject })); + } + + close(): void { + if (this.closed) return; + this.closed = true; + for (const w of this.waiters.splice(0)) w.reject(RuntimeError.transport()); + } +} + +class InMemoryTransport implements Transport { + constructor( + private readonly inbox: Channel, + private readonly outbox: Channel, + ) {} + + send(frame: Uint8Array): Promise { + this.outbox.push(frame.slice()); // copy: the caller may reuse its buffer + return Promise.resolve(); + } + + recv(): Promise { + return this.inbox.pull(); + } + + /** Drop this end — the peer's pending / next `recv` rejects, ending a serve loop. */ + close(): void { + this.outbox.close(); + } +} + +/** + * A connected in-memory transport pair — the TypeScript `duplex()`. What one + * end sends, the other receives. `close()` on either end makes the peer's + * pending / next `recv` reject. + */ +export function duplex(): [InMemoryTransport, InMemoryTransport] { + const a = new Channel(); + const b = new Channel(); + return [new InMemoryTransport(a, b), new InMemoryTransport(b, a)]; +} + +export type { InMemoryTransport }; diff --git a/runtime/test/framing.test.ts b/runtime/test/framing.test.ts new file mode 100644 index 0000000..8421898 --- /dev/null +++ b/runtime/test/framing.test.ts @@ -0,0 +1,93 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + DatagramFraming, + JsonRpcFraming, + call, + decodeEnvelope, + encodeEnvelopeErr, + encodeEnvelopeOk, +} from "../src/index.js"; + +const enc = new TextEncoder(); +const dec = new TextDecoder(); + +test("Envelope tag-byte form round-trips", () => { + const ok = encodeEnvelopeOk(enc.encode("payload")); + assert.equal(ok[0], 0); + assert.deepEqual(decodeEnvelope(ok), { ok: enc.encode("payload") }); + + const err = encodeEnvelopeErr(0x0102, enc.encode("fields")); + assert.equal(err[0], 1); + const back = decodeEnvelope(err); + assert.ok(back && "err" in back); + assert.equal(back.err.id, 0x0102); + assert.equal(dec.decode(back.err.body), "fields"); + + assert.equal(decodeEnvelope(Uint8Array.of(9)), undefined); // unknown tag +}); + +test("datagram request: header layout matches the Rust framing", () => { + const f = new DatagramFraming(); + const frame = f.encodeRequest(call(3, "send"), 42n, enc.encode("args")); + + // [call_id:u16 LE][request_id:u64 LE][params] + assert.deepEqual([...frame.subarray(0, 2)], [3, 0]); + assert.deepEqual([...frame.subarray(2, 10)], [42, 0, 0, 0, 0, 0, 0, 0]); + assert.equal(dec.decode(frame.subarray(10)), "args"); + + const req = f.decodeRequest(frame); + assert.deepEqual(req?.call, { id: 3 }); + assert.equal(req?.requestId, 42n); + assert.equal(dec.decode(req!.params), "args"); + assert.equal(f.decodeRequest(Uint8Array.of(0, 0, 0)), undefined); +}); + +test("datagram response round-trips ok and err with the ordinal", () => { + const f = new DatagramFraming(); + + const ok = f.encodeResponseOk(7n, enc.encode("payload")); + assert.deepEqual(f.decodeResponse(ok), { + requestId: 7n, + envelope: { ok: enc.encode("payload") }, + }); + + const err = f.encodeResponseErr(7n, 2, enc.encode("fields")); + const back = f.decodeResponse(err); + assert.equal(back?.requestId, 7n); + assert.ok(back && "err" in back.envelope); + assert.equal(back.envelope.err.id, 2); +}); + +test("JSON-RPC request matches the spec wording byte-for-byte", () => { + const f = new JsonRpcFraming(); + const frame = f.encodeRequest(call(0, "greet"), 1n, enc.encode(JSON.stringify([7, "x"]))); + assert.equal( + dec.decode(frame), + '{"jsonrpc":"2.0","method":"greet","params":[7,"x"],"id":1}', + ); + + const req = f.decodeRequest(frame); + assert.deepEqual(req?.call, { name: "greet" }); + assert.equal(req?.requestId, 1n); + assert.equal(dec.decode(req!.params), '[7,"x"]'); +}); + +test("JSON-RPC ok / err responses carry result / code+data", () => { + const f = new JsonRpcFraming(); + + const ok = f.encodeResponseOk(9n, enc.encode('{"body":"hi"}')); + assert.equal(dec.decode(ok), '{"jsonrpc":"2.0","result":{"body":"hi"},"id":9}'); + assert.deepEqual(f.decodeResponse(ok), { + requestId: 9n, + envelope: { ok: enc.encode('{"body":"hi"}') }, + }); + + const err = f.encodeResponseErr(9n, 3, enc.encode('{"why":"no"}')); + const back = f.decodeResponse(err); + assert.equal(back?.requestId, 9n); + assert.ok(back && "err" in back.envelope); + assert.equal(back.envelope.err.id, 3); + assert.equal(dec.decode(back.envelope.err.body), '{"why":"no"}'); +}); diff --git a/runtime/test/roundtrip.test.ts b/runtime/test/roundtrip.test.ts new file mode 100644 index 0000000..61707ae --- /dev/null +++ b/runtime/test/roundtrip.test.ts @@ -0,0 +1,169 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + Client, + DatagramFraming, + FRAMING_DATAGRAM, + Handshake, + JsonCodec, + JsonRpcFraming, + type Codec, + type Dispatch, + type Envelope, + type Framing, + type Kind, + Reply, + RuntimeError, + Server, + duplex, +} from "../src/index.js"; + +// A hand-written stand-in for what the generator will emit, mirroring the Rust +// end-to-end test's `Chat`: +// +// protocol Chat { +// function send(text: str) -> Message ! Rejected; // ordinal 0, error ord 0 +// function note(text: str); // ordinal 1, one-way +// } +interface Message { + body: string; + seq: number; +} +interface Rejected { + reason: string; +} + +interface Chat { + send(text: string): Promise; // rejects with { app: Rejected } | { runtime } + note(text: string): Promise; +} + +const CHAT_CALLS = ["send", "note"] as const; + +class ChatDispatcher implements Dispatch { + constructor(private readonly svc: Chat) {} + + calls(): readonly string[] { + return CHAT_CALLS; + } + + async dispatch(call: Kind, params: Uint8Array, codec: Codec, reply: Reply): Promise { + switch ("id" in call ? call.id : -1) { + case 0: { + const { text } = codec.decode<{ text: string }>(params); + try { + reply.ok(codec.encode(await this.svc.send(text))); + } catch (e) { + if (e instanceof RejectedError) reply.err(0, codec.encode(e.data)); + else throw e; + } + return; + } + case 1: { + const { text } = codec.decode<{ text: string }>(params); + await this.svc.note(text); + return; // one-way: leave the reply as `none` + } + default: + throw RuntimeError.unknownCall(); + } + } +} + +class RejectedError extends Error { + constructor(readonly data: Rejected) { + super(data.reason); + } +} + +class ChatClient { + constructor(private readonly client: Client) {} + + async send(text: string): Promise { + const env: Envelope = await this.client.call({ id: 0, name: "send" }, { text }); + if ("ok" in env) return this.client.codec.decode(env.ok); + if (env.err.id === 0) throw new RejectedError(this.client.codec.decode(env.err.body)); + throw RuntimeError.remote(env.err.id); + } + + async note(text: string): Promise { + await this.client.notify({ id: 1, name: "note" }, { text }); + } +} + +const IR_HASH = 0xbdbe5c6fd7420bd0n; + +function stack(framing: () => Framing): { codec: Codec; framing: Framing; hs: Handshake } { + const codec = new JsonCodec(); + const f = framing(); + return { + codec, + framing: f, + hs: new Handshake({ irHash: IR_HASH, wireFormat: codec.name, framing: f.name }), + }; +} + +for (const [label, mkFraming] of [ + ["datagram", () => new DatagramFraming()], + ["jsonrpc", () => new JsonRpcFraming()], +] as const) { + test(`${label}: a client ⇆ provider round-trip over duplex()`, async () => { + const [clientSide, providerSide] = duplex(); + const notes: string[] = []; + + const svc: Chat = { + async send(text) { + if (text === "") throw new RejectedError({ reason: "empty" }); + return { body: `echo: ${text}`, seq: 1 }; + }, + async note(text) { + notes.push(text); + }, + }; + + const s = stack(mkFraming); + const provider = new Server(new ChatDispatcher(svc), s.codec, s.framing).serveHandshaked( + providerSide, + s.hs, + ); + + const c = stack(mkFraming); + const chat = new ChatClient(await Client.connect(clientSide, c.codec, c.hs, c.framing)); + + assert.equal((await chat.send("hi")).body, "echo: hi"); + + await assert.rejects( + chat.send(""), + (e: unknown) => e instanceof RejectedError && e.data.reason === "empty", + ); + + await chat.note("saved"); // one-way + + clientSide.close(); + await provider; + assert.deepEqual(notes, ["saved"]); + }); +} + +test("connect refuses a peer on a wire-format mismatch", async () => { + const [clientSide, providerSide] = duplex(); + + // provider speaks a differently-named codec + const providerHs = new Handshake({ + irHash: IR_HASH, + wireFormat: "msgpack", + framing: FRAMING_DATAGRAM, + }); + void providerSide.send(providerHs.encode()); // just the handshake frame + void providerSide.recv(); + + await assert.rejects( + Client.connect(clientSide, new JsonCodec(), new Handshake({ + irHash: IR_HASH, + wireFormat: "json", + framing: FRAMING_DATAGRAM, + })), + (e: unknown) => e instanceof RuntimeError && e.is("handshake"), + ); +});