diff --git a/.gitignore b/.gitignore index 8fa4bd6d..efe1267f 100644 --- a/.gitignore +++ b/.gitignore @@ -2,5 +2,8 @@ dist node_modules example +# hegel's local database of failing examples to replay +/.hegel/ + # Nix /.direnv/ diff --git a/.prettierignore b/.prettierignore index cd775561..1bfe8454 100644 --- a/.prettierignore +++ b/.prettierignore @@ -1,4 +1,5 @@ .cache node_modules +.claude protobuf/gen testUtil/fixtures/gen diff --git a/PROTOCOL.md b/PROTOCOL.md index 54d76da5..453b50e2 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -20,7 +20,7 @@ The River protocol enables communication between clients and servers via remote ┌────────────┐ ┌────────────┐ │ Client │ │ Server │ ├────────────┤ emitted events ├────────────┤ -│ Transport │ ─────► (message, connectionStatus, etc.) │ Transport │ +│ Transport │ ─────► (message, sessionStatus, etc.) │ Transport │ └────────────┘ └────────────┘ ▼ 1:n (transport can have multiple sessions) ▼ 1:n ┌────────────┐ ┌────────────┐ @@ -45,7 +45,7 @@ The design of the protocol emphasizes three things in descending priority: The protocol specification defines semantics around: - How clients connect to servers. -- How they negotiate a connection an start a session. +- How they negotiate a connection and start a session. - How messages in a session are serialized and deserialized. - Dealing with message retransmission and deduplication. @@ -64,7 +64,7 @@ Note that this protocol specification does NOT detail the language-level specifi 1. `subscription`: the client sends 1 message, the server responds with m messages. A server (also called a router) is made up of multiple 'services'. Each 'service' has multiple 'procedures'. -A procedure declares its type (`rpc | stream | upload | subscription`), an initial message (`Init`), a response message type (`Response`), an error type (`Error`), and the associated handler. `upload` and `stream` may define an request message type (`Request`), which means they accept further messages from the client. +A procedure declares its type (`rpc | stream | upload | subscription`), an initial message (`Init`), a response message type (`Response`), an error type (`Error`), and the associated handler. `upload` and `stream` may define a request message type (`Request`), which means they accept further messages from the client. _Note: all types in this document are expressed roughly in TypeScript._ @@ -88,7 +88,7 @@ interface BaseError { // This can be any string message: string; // Any extra metadata - extra?: any; + extras?: unknown; } ``` @@ -100,7 +100,7 @@ type Result = | { ok: false; payload: ErrorPayload }; ``` -The messages in either direction must also contain additional information so that the receiving party knows where to route the message payload. This wrapper message is referred to as a `TransportMessage` and its payload can be a `Control`, a `Result`, an `Init`, an `Request`, or an `Response`. The schema for the transport message is as follows: +The messages in either direction must also contain additional information so that the receiving party knows where to route the message payload. This wrapper message is referred to as a `TransportMessage` and its payload can be a `Control`, a `Result`, an `Init`, a `Request`, or a `Response`. The schema for the transport message is as follows: ```ts interface TransportMessage { @@ -127,6 +127,13 @@ interface TransportMessage { // a stream of TransportMessage is grouped by streamId streamId: string; + // optional W3C trace context, propagated so a procedure call + // can be stitched into a distributed trace + tracing?: { + traceparent: string; + tracestate: string; + }; + // special flags // we will cover this later controlFlags: number; @@ -155,25 +162,31 @@ All messages MUST have no control flags set (i.e., the `controlFlags` field is ` - The client must set `serviceName` and `procedureName` as the correct string for the associated service and procedure. - All further messages MAY omit `serviceName` and `procedureName` as they are implied by the first message and are constant throughout the lifetime of a stream. - It is the last message of a stream, in which case the `StreamClosedBit` MUST be set. - - If this is sent with no payload, it is a control message the payload MUST Be a `ControlClose`. + - If there is no application-level payload to send with the close, it is a control message and the payload MUST be a `ControlClose`. - It is a message cancelling the stream, in which case the `StreamCancelBit` MUST be set. - This message MUST contain a `ProtocolError` payload. - It is an explicit heartbeat, so the `AckBit` MUST be the only bit set. - The payload MUST be `{ type: 'ACK' }`. - Because this is a control message that is not associated with a specific stream, you MUST NOT set `serviceName` or `procedureName` and `streamId` can be something arbitrary (e.g. `heartbeat`). -There are 4 error payloads that are defined in the protocol sent from server to client, these codes are reserved: +There are 4 reserved error codes. Three of them are `ProtocolError`s that travel on the wire; the fourth (`UNEXPECTED_DISCONNECT`) is never transmitted and is synthesized locally by each side when the session backing an in-flight stream dies. ```ts // When a client sends a malformed request. This can be // for a variety of reasons which would be included -// in the message. +// in the message. Sent from server to client. interface InvalidRequestError extends BaseError { code: 'INVALID_REQUEST'; message: string; + // present when the request failed schema validation + extras?: { + firstValidationErrors: Array<{ path: string; message: string }>; + totalErrors: number; + }; } // This is sent when an exception happens in the handler of a stream. +// Sent from server to client. interface UncaughtError extends BaseError { code: 'UNCAUGHT_ERROR'; message: string; @@ -181,17 +194,21 @@ interface UncaughtError extends BaseError { // This is sent when one side wishes to cancel the stream // abruptly from user-space. Handling this is up to the procedure -// implementation or the caller. +// implementation or the caller. Sent in either direction. interface CancelError extends BaseError { code: 'CANCEL'; message: string; } -// This is sent when the server encounters an internal error -// i.e. an invariant has been violated -interface; - type ProtocolError = UncaughtError | InvalidRequestError | CancelError; + +// Never sent over the wire. Each side raises this locally into any stream +// that was still in flight when its session was lost (a hard disconnect), +// so waiting callers and handlers observe a result instead of hanging. +interface UnexpectedDisconnectError extends BaseError { + code: 'UNEXPECTED_DISCONNECT'; + message: string; +} ``` `ProtocolError`s, just like service-level errors, are wrapped with a `Result`, which is further wrapped with `TransportMessage` and MUST have a `StreamCancelBit` flag. Please note that these are separate from user-defined errors, which should be treated just like any response message. @@ -212,7 +229,8 @@ interface ControlAck { interface ControlHandshakeRequest { type: 'HANDSHAKE_REQ'; - protocolVersion: 'v0' | 'v1' | 'v1.1' | 'v2.0'; + // the current implementation sends 'v2.0' and accepts 'v1.1' | 'v2.0' + protocolVersion: 'v1.1' | 'v2.0'; sessionId: string; expectedSessionState: { nextExpectedSeq: number; // integer @@ -237,7 +255,9 @@ interface ControlHandshakeResponse { | 'MALFORMED_HANDSHAKE_META' | 'MALFORMED_HANDSHAKE' | 'PROTOCOL_VERSION_MISMATCH' - | 'REJECTED_BY_CUSTOM_HANDLER'; + // fatal, returned by the custom handshake handler + | 'REJECTED_BY_CUSTOM_HANDLER' + | 'REJECTED_UNSUPPORTED_CLIENT'; }; } @@ -429,7 +449,7 @@ server: - -- - ! ##### Upload -An `upload` procedure starts with the client sending a single message with `StreamOpenBit` set and remains open until the client manually closes the request stream by sending `CloseControl` message. The server MUST send a final `Result` message with the `StreamClosedBit`. +An `upload` procedure starts with the client sending a single message with `StreamOpenBit` set and remains open until the client manually closes the request stream by sending a `ControlClose` message. The server MUST send a final `Result` message with the `StreamClosedBit`. Client finalizes upload: @@ -454,7 +474,7 @@ server: ! ##### Subscription -A `subscription` procedure starts with the client sending a single message with the `StreamOpenBit` set and remains open until either side ends the stream by sending a `ControlClose` message. The party receiving the `ControlClose` message must respond with a final `CloseControl` message. If the client initiates the closing, it MUST continue to accept data until the other side sends a `ControlClose` message. +A `subscription` procedure starts with the client sending a single message with the `StreamOpenBit` set and remains open until either side ends the stream by sending a `ControlClose` message. The party receiving the `ControlClose` message must respond with a final `ControlClose` message. If the client initiates the closing, it MUST continue to accept data until the other side sends a `ControlClose` message. Client initiated close: @@ -485,13 +505,15 @@ The TypeScript implementation utilizes a `Codec` class to handle the encoding an The TypeScript implementation has two main codecs: -1. `NaiveCodec`: a simple codec that uses JSON.stringify and JSON.parse to encode and decode messages directly to utf-8 bytes. +1. `NaiveJsonCodec`: a simple codec that uses JSON.stringify and JSON.parse to encode and decode messages directly to utf-8 bytes. 2. `BinaryCodec`: a more efficient codec that uses the [`msgpack`](https://msgpack.org/) format to encode and decode messages to and from raw bytes. +It also ships `ProtoCodec` (from `@replit/river/protobuf`), which encodes transport messages as protobuf envelopes and falls back to msgpack for error results and control payloads. + Depending on whether the underlying transport does message framing, the codec may need to handle message framing and deframing as well. For example, the WebSocket protocol has built-in message framing, so the codec only needs to handle encoding and decoding messages to and from raw bytes. -On the other hand, the UDS protocol does not have built-in message framing, so the codec must handle message framing and deframing as well. -The TypeScript implementation uses `uint32`-big-endian-length-prefixed message framing. +On the other hand, a transport like UDS does not have built-in message framing, so the codec must handle message framing and deframing as well. +WebSocket is the only transport shipped with the TypeScript implementation, so none of the built-in codecs do their own framing. ## Transports, sessions, and connections @@ -526,7 +548,7 @@ The process differs slightly between the client and server: - If the server detected a session state mismatch, any previous `Session`s will be destroyed. - Otherwise, consider the handshake successful and proceed to the next step. - The client should check for an existing `Session` for the `clientId` associated with the `Connection`. - - If an existing `Session` is found and that `Session`, it is verified whether the last `sessionId` associated with the previous `Session` matches the `sessionId` in the handshake response. + - If an existing `Session` is found, it is verified whether the `sessionId` associated with the previous `Session` matches the `sessionId` in the handshake response. - A match in `sessionId` means a reconnection to the same session, and that the server still has the state for this session. - The stale `Connection` object associated with the `Session` is closed, and replaced with the new `Connection` object. - Any buffered messages are resent. @@ -562,20 +584,26 @@ The process differs slightly between the client and server: 1. SessionNoConnection ◄──┐ │ reconnect / connect attempt │ ▼ │ - 2. SessionConnecting │ - │ connect success ──────────────┤ connect failure + 2. SessionBackingOff │ + │ backoff elapsed │ ▼ │ - 3. SessionHandshaking │ + 3. SessionConnecting │ + │ connect success ──────────────┤ connect failure + ▼ │ connect timeout + 4. SessionHandshaking │ │ handshake success ┌──────╪─ connection drop - 5. WaitingForHandshake │ handshake failure ─────┤ │ + 6. WaitingForHandshake │ handshake failure ─────┤ │ handshake timeout │ handshake success ▼ │ │ connection drop - ├───────────────────────► 4. SessionConnected │ │ heartbeat misses + ├───────────────────────► 5. SessionConnected │ │ heartbeat misses │ │ invalid message ───────╫──────┘ │ ▼ │ └───────────────────────► x. Destroy Session ◄─────┘ handshake failure ``` +A session in any non-destroyed state is torn down once the `sessionDisconnectGraceMs` grace +period elapses without reconnecting. + ### Handshake The handshake is a special message that is sent immediately after the wire connection is established and before any other messages are sent. @@ -653,7 +681,7 @@ It is important to note that this implies that there are two types of 'reconnect 1. Transparent reconnects: the connection dropped and reconnected but the session metadata is intact so resending the buffered messages will restore order. At the application level, nothing happened. 2. Hard reconnect: the other transport has lost all state and current transport should invalidate all state and start from scratch. -The TypeScript implementation of the transport explicitly emits `connectionStatus` events for transparent reconnects and `sessionStatus` events for hard reconnects which the client and server can listen to. +The TypeScript implementation of the transport explicitly emits `sessionTransition` events as a session moves between connection states (a transparent reconnect shows up as a sequence of these) and `sessionStatus` events for session creation and teardown, i.e. hard reconnects. Both can be listened to by the client and the server. Both clients and servers should listen for `sessionStatus` events to do some error handling: @@ -673,8 +701,9 @@ The `seq` and `ack` of the message should match that of the session itself and o Clients SHOULD echo back a heartbeat in the same format as soon as it receives a server heartbeat. -We track the number of heartbeats that we've sent to the other side without hearing a message. When the number of heartbeat misses exceeds some threshold `heartbeatsUntilDead` (also a parameter of the transport), -close the connection in that session. See the 'On disconnect' section above for more details on how to handle this. +Both sides run a liveness watchdog over the connection. Each side records when it last received _any_ message from its peer, and closes the connection once that timestamp falls further behind than `heartbeatsUntilDead * heartbeatIntervalMs` (both parameters of the transport). See the 'On disconnect' section above for more details on how to handle this. + +The TypeScript implementation runs this as a single interval for the lifetime of the connection that compares wall-clock time against the last inbound message, rather than arming a timer per sent heartbeat. Since it measures elapsed wall-clock time, a throttled or suspended timer can only delay detection of a dead connection — it can never report a heartbeat as missed while messages are still arriving. This explicit ack serves three purposes: diff --git a/README.md b/README.md index c11f6242..e185fd34 100644 --- a/README.md +++ b/README.md @@ -453,6 +453,8 @@ for await (const msg of resReadable) { River supports client-side cancellation using AbortController. All procedure calls accept an optional `signal` parameter: ```ts +import { CANCEL_CODE } from '@replit/river'; + const controller = new AbortController(); const rpcResult = client.example.longRunning.rpc( { data: 'hello world' }, @@ -464,7 +466,7 @@ controller.abort(); // all cancelled operations will receive an error with CANCEL_CODE const result = await rpcResult; -if (!result.ok && result.payload.code === 'CANCEL_CODE') { +if (!result.ok && result.payload.code === CANCEL_CODE) { console.log('Operation was cancelled'); } ``` @@ -474,7 +476,7 @@ When a client cancels an operation, the server handler receives the cancellation ```ts const ExampleService = ServiceSchema.define({ longRunning: Procedure.rpc({ - requestInit: Type.Object({}), + requestInit: Type.Object({ data: Type.String() }), responseData: Type.Object({ result: Type.String() }), async handler({ ctx }) { ctx.signal.addEventListener('abort', () => { @@ -511,6 +513,112 @@ const ExampleService = ServiceSchema.define({ Worth noting that the `ctx.signal` is triggered regardless of the reason the procedure has ended. +#### Cleaning up after a procedure + +`ctx.signal` fires synchronously and does not await its listeners, so several async +listeners will interleave rather than each running to completion. When cleanup is +async, use `ctx.deferCleanup` instead: + +```ts +async handler({ ctx }) { + const conn = await pool.acquire(); + ctx.deferCleanup(async () => { + await conn.release(); + }); + + // ... +} +``` + +Deferred cleanups run after the handler finishes — whether it returned, threw, or +was cancelled — in reverse registration order, and each one is awaited before the +next begins. If one throws, the error is recorded on the cleanup span and the rest +still run. + +#### Backpressure + +`write` returns `false` when the underlying session's send buffer is at or above +its high-water mark (`sendBufferHighWaterMark`, a transport option). This is +advisory, exactly like node's `stream.Writable.write`: the value is still buffered +and still delivered. Ignoring it is safe, it just means you may buffer without +bound if you produce faster than the transport drains. + +To apply backpressure, await `waitForWriteReady()`: + +```ts +for (const chunk of chunks) { + if (!reqWritable.write(chunk)) { + await reqWritable.waitForWriteReady(); + } +} +``` + +`waitForWriteReady()` resolves once the buffer drains below the high-water mark, +and immediately if there is no backpressure or the writable is already closed. It +never rejects. A promise that was already pending when the writable closes stays +pending until the session drains or closes, so re-check `isWritable()` after +awaiting if you need to know whether writing is still possible. + +#### Middleware + +Middleware runs before procedure handlers and can inspect (but not modify) incoming +requests — useful for logging, metrics, and tracing: + +```ts +import type { Middleware } from '@replit/river'; + +const logRequests: Middleware = ({ ctx, reqInit, next }) => { + console.log(`${ctx.serviceName}.${ctx.procedureName}`, { + from: ctx.from, + sessionId: ctx.sessionId, + reqInit, + }); + + next(); +}; + +const server = createServer(transport, services, { + middlewares: [logRequests], +}); +``` + +Each middleware must call `next()` to continue the chain. `ctx` is the same +handler context minus `cancel`, plus `streamId`, `serviceName`, and `procedureName`. + +#### Splitting a service across files + +`ServiceSchema.scaffold` separates a service's configuration from its procedures, +which helps when a service is too large for one file: + +```ts +// scaffold.ts +export const CounterScaffold = ServiceSchema.scaffold({ + initializeState: () => ({ count: 0 }), +}); + +// increment.ts +export const incrementProcedures = CounterScaffold.procedures({ + increment: Procedure.rpc({ + requestInit: Type.Object({ amount: Type.Number() }), + responseData: Type.Object({ current: Type.Number() }), + async handler({ ctx, reqInit }) { + ctx.state.count += reqInit.amount; + + return Ok({ current: ctx.state.count }); + }, + }), +}); + +// service.ts +export const CounterService = CounterScaffold.finalize({ + ...incrementProcedures, + // you can also define procedures directly here +}); +``` + +It also works as a builder if you just prefer that shape: +`ServiceSchema.scaffold({ ... }).finalize({ ... })`. + #### Codecs River provides two built-in codecs: @@ -634,7 +742,7 @@ const client = createClient(clientTransport, 'SERVER'); River provides utilities for testing your services: ```ts -import { createMockTransportNetwork } from '@replit/river/testUtil'; +import { createMockTransportNetwork } from '@replit/river/test-util'; describe('My Service', () => { // create mock transport network @@ -642,7 +750,7 @@ describe('My Service', () => { createMockTransportNetwork(); afterEach(cleanup); - test('should add numbers correctly', async () => { + test('should divide numbers correctly', async () => { // setup server const serverTransport = getServerTransport('SERVER'); const services = { @@ -655,7 +763,7 @@ describe('My Service', () => { const client = createClient(clientTransport, 'SERVER'); // test the service - const result = await client.math.add.rpc({ a: 1, b: 2 }); + const result = await client.math.divide.rpc({ a: 6, b: 2 }); expect(result.ok).toBe(true); if (result.ok) { expect(result.payload.result).toBe(3); @@ -679,7 +787,7 @@ const ServiceSchema = createServiceSchema(); const services = { ... }; // use custom ServiceSchema builder here const handshakeSchema = Type.Object({ token: Type.String() }); -createClient(new MockClientTransport('client'), 'SERVER', { +createClient(clientTransport, 'SERVER', { eagerlyConnect: false, handshakeOptions: createClientHandshakeOptions(handshakeSchema, async () => ({ // the type of this function is @@ -688,26 +796,40 @@ createClient(new MockClientTransport('client'), 'SERVER', { })), }); -createServer(new MockServerTransport('SERVER'), services, { +createServer(serverTransport, services, { handshakeOptions: createServerHandshakeOptions( handshakeSchema, - (metadata, previousMetadata) => { + (metadata, previousMetadata, from) => { // the type of this function is - // (metadata: Static, previousMetadata?: ParsedMetadata) => - // | false | Promise (if you reject it) - // | ParsedMetadata | Promise (if you allow it) + // ( + // metadata: Static, + // previousMetadata?: ParsedMetadata, + // from?: TransportClientId, + // ) => + // | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT' (if you reject it) + // | ParsedMetadata (if you allow it) + // | a Promise of either + // // next time a connection happens on the same session, previousMetadata will - // be populated with the last returned value + // be populated with the last returned value. `from` is the client id the peer + // presented in its handshake — check it against what the metadata authorizes + // before returning parsed metadata. return { parsedToken: metadata.token }; }, ), }); ``` +`createClientHandshakeOptions` also takes an optional third `eager` argument. When set, the +client constructs handshake metadata as soon as it starts dialing, so a slow `construct` +(e.g. fetching a fresh token) overlaps establishing the connection instead of running after +it. The trade-off is that `construct` then runs on every connection attempt, including ones +that never connect, so leave it unset when constructing is expensive or rate-limited. + You can then access the `ParsedMetadata` in your procedure handlers: ```ts -async handler(ctx, ...args) { +async handler({ ctx }) { // this contains the parsed metadata console.log(ctx.metadata) } @@ -769,7 +891,6 @@ import { createServer, createClient, Ok, - ProtoCodec, } from '@replit/river/protobuf'; import { Greeter } from './gen/greeter_pb'; @@ -788,7 +909,17 @@ const client = createClient(Greeter, clientTransport, serverId); const result = await client.sayHello({ name: 'World' }); ``` -The protobuf router uses `ProtoCodec` for wire encoding (protobuf envelopes with msgpack fallback for control payloads) and supports the same features as the TypeBox router: context, disposable state, middleware, handshakes, and OpenTelemetry tracing. +The protobuf router supports the same features as the TypeBox router: context, disposable state, middleware, handshakes, and OpenTelemetry tracing. It runs over any codec, but `@replit/river/protobuf` also ships `ProtoCodec`, which encodes transport messages as protobuf envelopes (falling back to msgpack for error results and control payloads). Opt into it on the transport like any other codec: + +```ts +import { ProtoCodec } from '@replit/river/protobuf'; + +const transport = new WebSocketClientTransport( + async () => new WebSocket('ws://localhost:3000'), + 'my-client-id', + { codec: ProtoCodec }, +); +``` > **Note:** The protobuf router is experimental and its API may change. @@ -796,15 +927,15 @@ The protobuf router uses `ProtoCodec` for wire encoding (protobuf envelopes with We've also provided an end-to-end testing environment using `Next.js`, and a simple backend connected with the WebSocket transport that you can [play with on Replit](https://replit.com/@jzhao-replit/riverbed). -You can find more service examples in the [E2E test fixtures](https://github.com/replit/river/blob/main/__tests__/fixtures/services.ts) +You can find more service examples in the [E2E test fixtures](https://github.com/replit/river/blob/main/testUtil/fixtures/services.ts) ## Developing [![Run on Repl.it](https://replit.com/badge/github/replit/river)](https://replit.com/new/github/replit/river) - `npm i` -- install dependencies -- `npm run check` -- lint -- `npm run format` -- format +- `npm run check` -- typecheck, then check formatting and lint (this is what CI runs) +- `npm run fix` -- auto-fix formatting and lint errors - `npm run test` -- run tests - `npm run build` -- build the package diff --git a/__tests__/bandwidth.bench.ts b/__tests__/bandwidth.bench.ts index 6bd239a2..309e4ad8 100644 --- a/__tests__/bandwidth.bench.ts +++ b/__tests__/bandwidth.bench.ts @@ -1,4 +1,4 @@ -import { afterAll, assert, bench, describe } from 'vitest'; +import { afterAll, assert, bench, describe, vi } from 'vitest'; import { getClientSendFn, waitForMessage } from '../testUtil'; import { TestServiceSchema } from '../testUtil/fixtures/services'; import { createServer } from '../router/server'; @@ -6,6 +6,12 @@ import { createClient } from '../router/client'; import { transports } from '../testUtil/fixtures/transports'; import { nanoid } from 'nanoid'; +// The global setup installs fake timers, which fake `performance.now` -- what +// tinybench measures with. Under them every sample lands on either 0ms or a +// clock tick, so the numbers describe event-loop turns rather than elapsed +// time. Benchmarks need the real clock. +vi.useRealTimers(); + let n = 0; const dummyPayloadSmall = () => ({ streamId: 'test', diff --git a/__tests__/codec.bench.ts b/__tests__/codec.bench.ts new file mode 100644 index 00000000..bac9d0a8 --- /dev/null +++ b/__tests__/codec.bench.ts @@ -0,0 +1,98 @@ +import { bench, describe, vi } from 'vitest'; +import { BinaryCodec, CodecMessageAdapter, NaiveJsonCodec } from '../codec'; +import { ProtoCodec } from '../protobuf/codec'; +import type { OpaqueTransportMessage } from '../transport/message'; + +// see the note in bandwidth.bench.ts: tinybench measures with `performance.now`, +// which the global setup fakes +vi.useRealTimers(); + +const BENCH_DURATION = 2_000; + +const codecs = [ + { name: 'naive', codec: NaiveJsonCodec }, + { name: 'binary', codec: BinaryCodec }, + { name: 'proto', codec: ProtoCodec }, +]; + +/** A typical procedure response: nested, small, no binary. */ +const smallMessage: OpaqueTransportMessage = { + id: 'abc123def456', + from: 'client-42', + to: 'SERVER', + seq: 1234, + ack: 1233, + streamId: 'stream-abcdef', + controlFlags: 0, + serviceName: 'documents', + procedureName: 'applyOperation', + payload: { + ok: true, + payload: { + revision: 991, + ops: [{ retain: 40 }, { insert: 'hello world' }, { delete: 3 }], + author: { id: 'u_123', name: 'someone', roles: ['owner', 'editor'] }, + meta: { ts: 1700000000000, client: 'web' }, + }, + }, +}; + +/** 64KB of binary, which is where the JSON codec's base64 path shows up. */ +const binaryMessage: OpaqueTransportMessage = { + ...smallMessage, + payload: new Uint8Array(65536).map((_, i) => i % 256), +}; + +describe('codec -- encode small message', () => { + for (const { name, codec } of codecs) { + bench(name, () => void codec.toBuffer(smallMessage), { + time: BENCH_DURATION, + }); + } +}); + +describe('codec -- decode small message', () => { + for (const { name, codec } of codecs) { + const bytes = codec.toBuffer(smallMessage); + bench(name, () => void codec.fromBuffer(bytes), { time: BENCH_DURATION }); + } +}); + +describe('codec -- encode 64KB binary payload', () => { + for (const { name, codec } of codecs) { + bench(name, () => void codec.toBuffer(binaryMessage), { + time: BENCH_DURATION, + }); + } +}); + +describe('codec -- decode 64KB binary payload', () => { + for (const { name, codec } of codecs) { + const bytes = codec.toBuffer(binaryMessage); + bench(name, () => void codec.fromBuffer(bytes), { time: BENCH_DURATION }); + } +}); + +/** + * The adapter is what the session actually calls, so it carries schema + * validation on top of the codec. That validation runs per inbound message and + * is the reason servers compile their validator. + */ +describe('adapter -- decode + validate small message (binary codec)', () => { + const interpreted = new CodecMessageAdapter(BinaryCodec); + const compiled = new CodecMessageAdapter(BinaryCodec, { + precompileValidator: true, + }); + const bytes = BinaryCodec.toBuffer(smallMessage); + + bench( + 'interpreted validator (client)', + () => void interpreted.fromBuffer(bytes), + { + time: BENCH_DURATION, + }, + ); + bench('compiled validator (server)', () => void compiled.fromBuffer(bytes), { + time: BENCH_DURATION, + }); +}); diff --git a/__tests__/properties/README.md b/__tests__/properties/README.md new file mode 100644 index 00000000..09261e51 --- /dev/null +++ b/__tests__/properties/README.md @@ -0,0 +1,125 @@ +# Property-based tests + +These tests use [hegel](https://hegel.dev/) (`@hegeldev/hegel`), a Hypothesis-style +property-based testing library. Instead of asserting on hand-picked examples, each +test states an invariant that must hold for _every_ generated input, and hegel +searches for a counterexample and shrinks it to the smallest failing case. + +Run them like any other test: + +```bash +npx vitest run __tests__/properties +``` + +To reproduce a specific failure, hegel prints a seed; pass it back via the +`seed` setting on the failing test. + +## Why these properties + +The properties below are derived from the guarantees [PROTOCOL.md](../../PROTOCOL.md) +makes. They are the things that must stay true as the implementation changes — +the example-based tests in `__tests__/` cover specific scenarios, these cover the +space around them. + +### A. Codec round-trips (`codec.property.test.ts`) + +A codec's only job is to be a faithful bijection between a `TransportMessage` and +bytes. Everything above it assumes that. + +| # | Property | +| --- | ----------------------------------------------------------------------------------------------------------------------------------------- | +| A1 | For any valid `TransportMessage` `m`, `fromBuffer(toBuffer(m))` deep-equals `m`. | +| A2 | Optional fields (`serviceName`, `procedureName`, `tracing`) stay absent after a round-trip rather than being materialized as `''`/`null`. | +| A3 | Every `controlFlags` bit combination survives exactly. | +| A4 | Arbitrary nested payloads (objects, arrays, unicode text, binary, `null`) survive. | +| A5 | `seq`/`ack` survive across their full supported range. | +| A6 | Encoding is deterministic: `toBuffer(m)` twice yields identical bytes. | +| A7 | `fromBuffer` on arbitrary bytes either throws or returns an object — it never returns a non-object and never hangs. | +| A8 | `seq`/`ack` outside the wire format's range are either carried exactly or refused at encode time — never silently truncated. | + +Writing A1 surfaced three inputs that were not round-trippable. Two are now +fixed in `NaiveJsonCodec`, which is the default codec: + +- **A payload key of `$t` used to decode as binary.** `$t` is the codec's escape + marker for `Uint8Array`, so an application payload using that key was silently + decoded as binary — corruption with no error. +- **A payload key of `$b` with a non-numeric value used to throw.** `$b` is the + marker for `bigint` and the reviver called `BigInt()` unconditionally, so + `{ $b: 'x' }` made `fromBuffer` throw. A decode failure is treated as an + invalid message, which tears the connection down — reachable from ordinary + application data. + +Both are fixed by escaping: any key that could be mistaken for a marker gains an +extra `$` on the way out and loses it on the way back in. `NaiveJsonCodec marker +escaping` covers the specific cases and the older-peer compatibility edge. + +The third is still open, and the generators are scoped around it: + +- **`BinaryCodec` and `ProtoCodec` encode a `__proto__` payload key but cannot + decode it.** msgpack guards prototype pollution on the way in but not on the + way out, so these codecs produce bytes they will then reject. `NaiveJsonCodec` + round-trips it fine, so whether a payload is deliverable depends on which codec + the transport was configured with. Escaping was cheap for `NaiveJsonCodec` + (~1.6%, since `JSON.stringify`'s replacer already visits every property); + msgpack exposes no equivalent hook, so the same fix there means a second full + traversal of every payload on encode. + +A8 checks the boundary where `TransportMessageSchema` (an unbounded +`Type.Integer()`) and ProtoCodec's envelope (`uint32`) disagree about what is +representable. The good news is that the disagreement is loud: `NaiveJsonCodec` +and `BinaryCodec` carry any safe integer exactly, and ProtoCodec **throws** on an +out-of-range or negative `seq` rather than wrapping it. `CodecMessageAdapter` +catches that and turns it into a send failure, which tears the session down with +a reason — so a peer never has to reason about a truncated `seq`. + +One narrower asymmetry is pinned the same way: `ProtoCodec` decodes an +empty-string `serviceName`/`procedureName` back as absent (its envelope uses `''` +as the absent sentinel), and its envelope types `seq`/`ack` as `uint32`, so those +fields have a lower ceiling there than the `Type.Integer()` in +`TransportMessageSchema` implies. + +### B. Stream lifecycle (`streams.property.test.ts`) + +The reader/writer semantics in PROTOCOL.md — ordering, half-close, and clean +teardown — for every procedure type. + +| # | Property | +| --- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| B1 | `upload`: the server handler observes exactly the values the client wrote, in write order, and the client gets exactly one result. | +| B2 | `stream`: for any interleaving of client and server writes, each side receives the other's values in send order, with none dropped or duplicated. | +| B3 | `subscription`: the client's `resReadable` yields exactly the values the server wrote, in order, then completes. | +| B4 | Half-close: after the client closes `reqWritable`, the server's `reqReadable` completes but the server can still write, and the client receives those writes until the server closes. | +| B5 | `close(value)` delivers `value` as the final value, then closes. | +| B6 | `Writable`: `write()` after `close()` throws, `close()` is idempotent, `isWritable()` is false exactly after close. | +| B7 | `Readable`: it can only be consumed once, and `break()` resolves a pending read with `READABLE_BROKEN`. | +| B8 | Backpressure is advisory: writing while `write()` returns `false` still delivers every value, in order. | + +### C. Session and transport under faults (`session.property.test.ts`) + +This is the closest analogue to what a deterministic-simulation tool like +Antithesis would explore: generate a fault schedule, then assert the protocol's +delivery guarantees still hold. + +| # | Property | +| --- | --------------------------------------------------------------------------------------------------------------------- | +| C1 | Exactly-once, in-order delivery holds across any schedule of transparent reconnects. | +| C2 | A receiver only accepts `msg.seq === session.ack`, and sets `ack` to `seq + 1`. | +| C3 | After an ack, the send buffer retains exactly the messages with `seq >= ack` — an unacked message is never dropped. | +| C4 | No `invariant-violation`-tagged log is ever emitted, under any fault schedule. | +| C5 | A hard reconnect (server loses state) resolves every in-flight call with `UNEXPECTED_DISCONNECT` rather than hanging. | +| C6 | A transparent reconnect preserves the session id. | +| C7 | After any fault schedule, cleanup leaves no open sessions or connections. | +| C8 | Concurrent streams sharing one session never mix up their values, across any fault schedule. | +| C9a | A connection whose peer is still sending survives arbitrarily long — elapsed time alone is never a missed heartbeat. | +| C9b | A silent peer is detected and the connection closed within `heartbeatsUntilDead * heartbeatIntervalMs`. | +| C10 | The server's stored handshake metadata converges on the client's current credential after refreshes and reconnects. | + +C2 and C3 are not asserted by reaching into session internals. The transport +already checks them itself and logs an `invariant-violation`-tagged error when +they break — `assertSendOrdering` on the way out, and the `msg.seq !== session.ack` +branch on the way in. Every property in that file (and in +`streams.property.test.ts`) collects those logs and asserts none were emitted, +which makes C4 the oracle for all three. Testing them through the behavior the +transport already guards is more durable than asserting on private fields. + + diff --git a/__tests__/properties/codec.property.test.ts b/__tests__/properties/codec.property.test.ts new file mode 100644 index 00000000..15857f44 --- /dev/null +++ b/__tests__/properties/codec.property.test.ts @@ -0,0 +1,392 @@ +import { describe, expect, test } from 'vitest'; +import * as hegel from '@hegeldev/hegel'; +import * as gs from '@hegeldev/hegel/generators'; +import { + BinaryCodec, + Codec, + CodecMessageAdapter, + NaiveJsonCodec, +} from '../../codec'; +import { ProtoCodec } from '../../protobuf/codec'; +import { OpaqueTransportMessage } from '../../transport/message'; + +/** Round-trip properties for the codecs. See ./README.md for the catalog. */ + +const codecs: Array<{ name: string; codec: Codec }> = [ + { name: 'naive', codec: NaiveJsonCodec }, + { name: 'binary', codec: BinaryCodec }, + { name: 'proto', codec: ProtoCodec }, +]; + +// Every constraint on the generators below is a real limitation, each pinned by +// a test in 'documented codec limitations'. Widening them should fail. +// +// `$t`/`$b` used to live here too -- NaiveJsonCodec's markers collided with +// application data -- until the codec started escaping them. +const UNROUNDTRIPPABLE_KEYS = ['__proto__']; +const MAX_UINT32 = 0xffffffff; + +const identifiers = gs.text({ codec: 'utf-8', minSize: 1, maxSize: 24 }); +const seqNumbers = gs.integers({ minValue: 0, maxValue: MAX_UINT32 }); + +const payloadKeys = identifiers.filter( + (key) => !UNROUNDTRIPPABLE_KEYS.includes(key), +); + +const payloadLeaves = gs.oneOf( + gs.integers({ minValue: -1_000_000, maxValue: 1_000_000 }), + gs.floats({ minValue: -1e6, maxValue: 1e6 }).filter((n) => !Object.is(n, -0)), + gs.booleans(), + gs.text({ codec: 'utf-8', maxSize: 24 }), + gs.just(null), + // hegel hands back a node Buffer, and toStrictEqual distinguishes it from a + // plain Uint8Array by prototype + gs.binary({ maxSize: 8 }).map((buf) => new Uint8Array(buf)), +); + +function payloadValues(depth: number): gs.Generator { + if (depth <= 0) { + return payloadLeaves; + } + + const inner = payloadValues(depth - 1); + + return gs.oneOf( + payloadLeaves, + gs.arrays(inner, { maxSize: 4 }), + gs + .arrays(gs.tuples(payloadKeys, inner), { maxSize: 4 }) + .map((entries) => Object.fromEntries(entries)), + ); +} + +const payloads = gs.oneOf( + payloadValues(3), + // control-message-shaped payloads, which is most real traffic + gs + .sampledFrom(['ACK', 'CLOSE', 'HANDSHAKE_REQ', 'REHANDSHAKE_REQ']) + .map((type) => ({ type })), +); + +const transportMessages: gs.Generator = gs.composite( + (tc) => { + const msg: OpaqueTransportMessage = { + id: tc.draw(identifiers), + from: tc.draw(identifiers), + to: tc.draw(identifiers), + seq: tc.draw(seqNumbers), + ack: tc.draw(seqNumbers), + streamId: tc.draw(identifiers), + controlFlags: tc.draw(gs.integers({ minValue: 0, maxValue: 0b1111 })), + payload: tc.draw(payloads), + }; + + // present-or-absent, not present-or-null: PROTOCOL.md lets every message + // after the first omit these, which is what A2 checks + if (tc.draw(gs.booleans())) { + msg.serviceName = tc.draw(identifiers); + } + + if (tc.draw(gs.booleans())) { + msg.procedureName = tc.draw(identifiers); + } + + if (tc.draw(gs.booleans())) { + msg.tracing = { + traceparent: tc.draw(gs.text({ codec: 'utf-8', maxSize: 55 })), + tracestate: tc.draw(gs.text({ codec: 'utf-8', maxSize: 55 })), + }; + } + + return msg; + }, +); + +describe.each(codecs)('codec properties -- $name', ({ codec }) => { + test('A1: encoding then decoding any transport message is the identity', () => + hegel.testAsync((tc) => { + const msg = tc.draw(transportMessages); + tc.note(`message: ${JSON.stringify(msg, bigintSafe)}`); + + expect(codec.fromBuffer(codec.toBuffer(msg))).toStrictEqual(msg); + })); + + test('A2: absent optional fields stay absent', () => + hegel.testAsync((tc) => { + const msg = tc.draw(transportMessages); + delete msg.serviceName; + delete msg.procedureName; + delete msg.tracing; + + const decoded = codec.fromBuffer(codec.toBuffer(msg)); + + expect(decoded).not.toHaveProperty('serviceName'); + expect(decoded).not.toHaveProperty('procedureName'); + expect(decoded).not.toHaveProperty('tracing'); + })); + + test('A3: every control flag combination survives', () => + hegel.testAsync((tc) => { + const msg = tc.draw(transportMessages); + msg.controlFlags = tc.draw( + gs.integers({ minValue: 0, maxValue: 0b1111 }), + ); + + const decoded = codec.fromBuffer( + codec.toBuffer(msg), + ) as OpaqueTransportMessage; + + expect(decoded.controlFlags).toBe(msg.controlFlags); + })); + + test('A5: seq and ack survive their full supported range', () => + hegel.testAsync((tc) => { + const msg = tc.draw(transportMessages); + msg.seq = tc.draw(seqNumbers); + msg.ack = tc.draw(seqNumbers); + + const decoded = codec.fromBuffer( + codec.toBuffer(msg), + ) as OpaqueTransportMessage; + + expect(decoded.seq).toBe(msg.seq); + expect(decoded.ack).toBe(msg.ack); + })); + + test('A6: encoding is deterministic', () => + hegel.testAsync((tc) => { + const msg = tc.draw(transportMessages); + + expect(codec.toBuffer(msg)).toStrictEqual(codec.toBuffer(msg)); + })); + + test('A7: decoding arbitrary bytes throws rather than returning a non-object', () => + hegel.testAsync((tc) => { + const bytes = tc.draw(gs.binary({ maxSize: 64 })); + + let decoded: unknown; + try { + decoded = codec.fromBuffer(bytes); + } catch { + // the expected outcome for almost all inputs; the transport treats a + // decode failure as an invalid message + return; + } + + // if it did decode, the transport will index into it without checking + expect(typeof decoded).toBe('object'); + expect(decoded).not.toBeNull(); + })); +}); + +describe('bigint payloads (naive and binary only)', () => { + // ProtoCodec msgpacks non-binary payloads without BinaryCodec's bigint + // extension, so bigints are out of scope for it + test.each([ + { name: 'naive', codec: NaiveJsonCodec }, + { name: 'binary', codec: BinaryCodec }, + ])('A4: bigints survive a round-trip -- $name', ({ codec }) => + hegel.testAsync((tc) => { + const value = tc.draw( + gs.bigIntegers({ + minValue: -(2n ** 80n), + maxValue: 2n ** 80n, + }), + ); + const msg = { ...tc.draw(transportMessages), payload: { value } }; + + const decoded = codec.fromBuffer( + codec.toBuffer(msg), + ) as OpaqueTransportMessage; + + expect((decoded.payload as { value: bigint }).value).toBe(value); + }), + ); +}); + +/** + * `TransportMessageSchema` types seq/ack as an unbounded `Type.Integer()` while + * ProtoCodec's envelope types them as `uint32`. What matters is that the + * disagreement is loud: a peer that decoded a truncated seq would treat correct + * traffic as out-of-order. + */ +describe('A8: seq and ack outside the wire format range', () => { + const outOfRange = gs.oneOf( + // just past the ceiling, where truncation would be most tempting + gs.integers({ minValue: 2 ** 32, maxValue: 2 ** 32 + 1_000 }), + gs.integers({ minValue: 2 ** 33, maxValue: Number.MAX_SAFE_INTEGER }), + // negative: a valid Type.Integer(), not a valid uint32 + gs.integers({ minValue: -1_000_000, maxValue: -1 }), + ); + + test.each([ + { name: 'naive', codec: NaiveJsonCodec }, + { name: 'binary', codec: BinaryCodec }, + ])('$name carries any safe integer exactly', ({ codec }) => + hegel.testAsync((tc) => { + const seq = tc.draw(outOfRange); + const ack = tc.draw(outOfRange); + const msg = { ...tc.draw(transportMessages), seq, ack }; + + const decoded = codec.fromBuffer( + codec.toBuffer(msg), + ) as OpaqueTransportMessage; + + expect(decoded.seq).toBe(seq); + expect(decoded.ack).toBe(ack); + }), + ); + + test('ProtoCodec refuses to encode rather than truncating', () => + hegel.testAsync((tc) => { + const seq = tc.draw(outOfRange); + const msg = { ...tc.draw(transportMessages), seq }; + + // the important half: no bytes carrying a wrapped seq + expect(() => ProtoCodec.toBuffer(msg)).toThrow(/cannot encode field/); + })); + + test('a codec that refuses to encode surfaces as a clean send failure', () => { + // the adapter is what the session talks to; it turns a throwing codec into a + // Result, so the session is torn down with a reason rather than exploding + const adapter = new CodecMessageAdapter(ProtoCodec); + + const result = adapter.toBuffer({ + ...messageWithPayload({ type: 'ACK' }), + seq: 2 ** 32, + }); + + expect(result.ok).toBe(false); + if (result.ok) return; + + expect(result.reason).toMatch(/cannot encode field/); + }); +}); + +/** + * A1 covers marker escaping across generated payloads. These are the specific + * cases that used to be broken, plus the compatibility edges that a property + * over well-formed messages can't reach. + */ +describe('NaiveJsonCodec marker escaping', () => { + test('a payload key of `$t` survives instead of decoding as binary', () => { + const msg = { payload: { $t: 'aGVsbG8=' } }; + + expect( + NaiveJsonCodec.fromBuffer(NaiveJsonCodec.toBuffer(msg)), + ).toStrictEqual(msg); + }); + + test('a payload key of `$b` survives instead of throwing', () => { + const msg = { payload: { $b: 'not-a-number' } }; + + expect( + NaiveJsonCodec.fromBuffer(NaiveJsonCodec.toBuffer(msg)), + ).toStrictEqual(msg); + }); + + test('escaping is stable under nesting', () => { + // a key that is already escape-shaped must not collide with the escaping + const msg = { + payload: { $$t: 1, $$$b: 2, $t: { $b: 'x' }, plain: '$t' }, + }; + + expect( + NaiveJsonCodec.fromBuffer(NaiveJsonCodec.toBuffer(msg)), + ).toStrictEqual(msg); + }); + + test('real binary and bigint values still round-trip', () => { + const msg = { + payload: { bin: Uint8Array.from([0, 42, 255]), big: 2n ** 70n }, + }; + + expect( + NaiveJsonCodec.fromBuffer(NaiveJsonCodec.toBuffer(msg)), + ).toStrictEqual(msg); + }); + + test('a marker written by an older peer still decodes as binary', () => { + // an unescaped `{ $t: }` on the wire is what pre-escaping river + // emitted for a Uint8Array, so it has to keep decoding that way + const legacy = new TextEncoder().encode( + JSON.stringify({ payload: { $t: 'aGVsbG8=' } }), + ); + + const decoded = NaiveJsonCodec.fromBuffer(legacy) as { payload: unknown }; + + expect(decoded.payload).toBeInstanceOf(Uint8Array); + }); +}); + +/** + * Bugs, not endorsements. Pinned so that changing any of them is a visible, + * intentional act, and so the generators above have something to point at. + */ +describe('documented codec limitations', () => { + test('msgpack-based codecs encode a `__proto__` payload key but cannot decode it', () => { + // msgpack guards prototype pollution on decode but not encode, so these + // produce bytes they then reject -- on the wire, a torn-down connection. + // + // Left unfixed deliberately. NaiveJsonCodec could escape its markers for + // ~1.6% because JSON.stringify's replacer already visits every property; + // msgpack exposes no equivalent hook and its `__proto__` check runs before + // `mapKeyConverter`, so the only fix is a second full traversal of every + // payload on encode -- in the codec chosen for throughput, to defend a key + // that does not appear in real payloads. + const msg = messageWithPayload(Object.fromEntries([['__proto__', 'x']])); + + for (const codec of [BinaryCodec, ProtoCodec]) { + const encoded = codec.toBuffer(msg); + expect(() => codec.fromBuffer(encoded)).toThrow( + /__proto__ is not allowed/, + ); + } + + // the default codec round-trips it fine, so deliverability depends on which + // codec the transport was configured with + expect( + NaiveJsonCodec.fromBuffer(NaiveJsonCodec.toBuffer(msg)), + ).toStrictEqual(msg); + }); + + test('ProtoCodec decodes an empty serviceName back as absent', () => { + // the envelope uses '' as the absent sentinel for serviceName/procedureName + const msg: OpaqueTransportMessage = { + id: 'id', + from: 'client', + to: 'SERVER', + seq: 0, + ack: 0, + streamId: 'stream', + controlFlags: 0, + serviceName: '', + procedureName: '', + payload: { type: 'ACK' }, + }; + + const decoded = ProtoCodec.fromBuffer(ProtoCodec.toBuffer(msg)); + + expect(decoded).not.toHaveProperty('serviceName'); + expect(decoded).not.toHaveProperty('procedureName'); + }); +}); + +/** ProtoCodec rejects anything that isn't a full `OpaqueTransportMessage`. */ +function messageWithPayload(payload: unknown): OpaqueTransportMessage { + return { + id: 'id', + from: 'client', + to: 'SERVER', + seq: 0, + ack: 0, + streamId: 'stream', + controlFlags: 0, + payload, + }; +} + +/** JSON.stringify replacer so tc.note() can print bigint-bearing messages. */ +function bigintSafe(_key: string, value: unknown) { + return typeof value === 'bigint' ? `${value.toString()}n` : value; +} diff --git a/__tests__/properties/session.property.test.ts b/__tests__/properties/session.property.test.ts new file mode 100644 index 00000000..b0a8df7a --- /dev/null +++ b/__tests__/properties/session.property.test.ts @@ -0,0 +1,661 @@ +import { describe, expect, test, vi } from 'vitest'; +import * as hegel from '@hegeldev/hegel'; +import * as gs from '@hegeldev/hegel/generators'; +import { Type, type Static } from 'typebox'; +import { + type MaybeDisposable, + Ok, + Procedure, + UNEXPECTED_DISCONNECT_CODE, + createServiceSchema, +} from '../../router'; +import { + createClientHandshakeOptions, + createServerHandshakeOptions, +} from '../../router/handshake'; +import { createClient } from '../../router/client'; +import { createServer } from '../../router/server'; +import { closeAllConnections, numberOfConnections } from '../../testUtil'; +import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport'; +import type { TestTransportOptions } from '../../testUtil/fixtures/transports'; +import { + advanceFakeTimersByConnectionBackoff, + advanceFakeTimersBySessionGrace, + cleanupTransports, + waitFor, +} from '../../testUtil/fixtures/cleanup'; + +/** + * Delivery guarantees under generated fault schedules. See ./README.md. + * + * C2 (seq/ack discipline) and C3 (send-buffer trimming) are not asserted + * against session internals: the transport already self-checks both and logs an + * `invariant-violation` when they break. Every property here asserts no such log + * fired, which makes C4 the oracle for all three. + */ + +const FAULT_CASES = { testCases: 15 }; +const FAULT_TIMEOUT_MS = 120_000; + +/** + * Reconnect policy for these properties, which are about delivery rather than + * about the reconnect policy itself (`transport/rateLimit.test.ts` covers that). + * + * The budget defaults to 5 attempts, so a 5-disconnect schedule legitimately + * exhausts it and the client stops redialing -- correct, but it would silently + * turn a delivery property into a rate-limiter property. Backoff jitter is + * `Math.random()`, which makes a failing case unreplayable; zeroing it is what + * lets hegel shrink these reliably. + */ +const deterministicReconnects = { + attemptBudgetCapacity: 100, + maxJitterMs: 0, + baseIntervalMs: 10, + maxBackoffMs: 100, +}; + +const ServiceSchema = createServiceSchema(); + +const PropertyService = ServiceSchema.define({ + /** Accumulates every request value and reports them in one response. */ + collect: Procedure.upload({ + requestInit: Type.Object({}), + requestData: Type.Object({ value: Type.Number() }), + responseData: Type.Object({ values: Type.Array(Type.Number()) }), + async handler({ reqReadable }) { + const values: Array = []; + for await (const msg of reqReadable) { + if (!msg.ok) break; + values.push(msg.payload.value); + } + + return Ok({ values }); + }, + }), + /** Never returns on its own -- used to have something in flight during a fault. */ + hang: Procedure.rpc({ + requestInit: Type.Object({}), + responseData: Type.Object({}), + async handler({ ctx }) { + await new Promise((resolve) => { + ctx.signal.addEventListener('abort', () => { + resolve(); + }); + }); + + return Ok({}); + }, + }), +}); + +const services = { svc: PropertyService }; + +const values = gs.integers({ minValue: -1_000, maxValue: 1_000 }); + +/** + * A generated schedule of writes with disconnects interleaved between them. + * `null` means "drop the connection here"; a number means "write this value". + */ +const writeSchedules = gs.arrays(gs.optional(values), { + minSize: 1, + maxSize: 14, +}); + +interface MultiplexedSchedule { + /** The values each concurrent stream is expected to deliver, in order. */ + perStream: Array>; + /** A generated interleaving of those writes across the streams. */ + writes: Array<{ stream: number; value: number }>; + /** Indices into `writes` before which the connection is dropped. */ + faultAt: Set; +} + +/** Built in two passes -- interleave, then insert faults -- so it terminates. */ +const multiplexedSchedules: gs.Generator = gs.composite( + (tc) => { + const streamCount = tc.draw(gs.integers({ minValue: 2, maxValue: 4 })); + const perStream: Array> = []; + for (let i = 0; i < streamCount; i++) { + perStream.push(tc.draw(gs.arrays(values, { maxSize: 6 }))); + } + + // a cursor per stream, picking which advances next: a real interleaving + // rather than a fixed round-robin + const cursors = perStream.map(() => 0); + const writes: Array<{ stream: number; value: number }> = []; + let outstanding = perStream.reduce((n, stream) => n + stream.length, 0); + while (outstanding > 0) { + const ready = perStream + .map((_, index) => index) + .filter((index) => cursors[index] < perStream[index].length); + const stream = + ready[ + tc.draw(gs.integers({ minValue: 0, maxValue: ready.length - 1 })) + ]; + + writes.push({ stream, value: perStream[stream][cursors[stream]] }); + cursors[stream]++; + outstanding--; + } + + const faultAt = new Set(); + const faultCount = tc.draw(gs.integers({ minValue: 0, maxValue: 3 })); + for (let i = 0; i < faultCount; i++) { + faultAt.add( + tc.draw(gs.integers({ minValue: 0, maxValue: writes.length })), + ); + } + + return { perStream, writes, faultAt }; + }, +); + +function setup(opts?: TestTransportOptions): { + network: ReturnType; + clientTransport: ReturnType< + ReturnType['getClientTransport'] + >; + serverTransport: ReturnType< + ReturnType['getServerTransport'] + >; + client: ReturnType>; + violations: Array; +} { + const network = createMockTransportNetwork({ + ...opts, + client: { ...deterministicReconnects, ...opts?.client }, + }); + const clientTransport = network.getClientTransport('client'); + const serverTransport = network.getServerTransport('SERVER'); + + const violations: Array = []; + for (const t of [clientTransport, serverTransport]) { + t.bindLogger((msg, ctx, level) => { + if (ctx?.tags?.includes('invariant-violation')) { + violations.push(`[${level}] ${msg}`); + } + }, 'debug'); + } + + createServer(serverTransport, services); + const client = createClient(clientTransport, 'SERVER'); + + return { network, clientTransport, serverTransport, client, violations }; +} + +async function teardown(ctx: ReturnType) { + await cleanupTransports([ctx.clientTransport, ctx.serverTransport]); + await ctx.network.cleanup(); +} + +describe('session properties under faults', () => { + test( + 'C1/C4: transparent reconnects preserve exactly-once, in-order delivery', + () => + hegel.testAsync(async (tc) => { + const schedule = tc.draw(writeSchedules); + const sent = schedule.filter((step): step is number => step !== null); + const disconnects = schedule.length - sent.length; + tc.note( + `${sent.length} writes, ${disconnects} disconnects interleaved`, + ); + + const ctx = setup(); + try { + const { reqWritable, finalize } = ctx.client.svc.collect.upload({}); + + for (const step of schedule) { + if (step === null) { + // the session survives the wire, so the client should reconnect + // and resend whatever was not acked + closeAllConnections(ctx.clientTransport); + continue; + } + + reqWritable.write({ value: step }); + } + + reqWritable.close(); + + const result = await finalize(); + expect(result.ok).toBe(true); + if (!result.ok) return; + + // nothing dropped, nothing duplicated, nothing reordered + expect(result.payload.values).toStrictEqual(sent); + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); + + test( + 'C6: a transparent reconnect keeps the session id', + () => + hegel.testAsync(async (tc) => { + const drops = tc.draw(gs.integers({ minValue: 1, maxValue: 4 })); + tc.note(`${drops} consecutive reconnects`); + + const ctx = setup(); + try { + // establish a session + const { reqWritable, finalize } = ctx.client.svc.collect.upload({}); + reqWritable.write({ value: 1 }); + + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(1), + ); + const originalId = ctx.clientTransport.sessions.get('SERVER')?.id; + expect(originalId).toBeDefined(); + + for (let i = 0; i < drops; i++) { + closeAllConnections(ctx.clientTransport); + + // backoff grows with each consecutive failure; skip it, don't race it + await advanceFakeTimersByConnectionBackoff(); + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(1), + ); + + // the session outlives the connection, so its identity must not change + expect(ctx.clientTransport.sessions.get('SERVER')?.id).toBe( + originalId, + ); + } + + reqWritable.close(); + await finalize(); + + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); + + test( + 'C5: a hard reconnect resolves in-flight calls with UNEXPECTED_DISCONNECT', + () => + hegel.testAsync(async (tc) => { + const inFlight = tc.draw(gs.integers({ minValue: 1, maxValue: 5 })); + tc.note(`${inFlight} calls in flight when the server restarts`); + + const ctx = setup(); + try { + const pending = Array.from({ length: inFlight }, () => + ctx.client.svc.hang.rpc({}), + ); + + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(1), + ); + + // hard reconnect: every waiting caller must get a result, not hang + await ctx.network.restartServer(); + + // jump the sessionDisconnectGraceMs wait rather than sleeping it + await advanceFakeTimersBySessionGrace(); + + const results = await Promise.all(pending); + for (const result of results) { + expect(result.ok).toBe(false); + if (result.ok) continue; + + expect(result.payload.code).toBe(UNEXPECTED_DISCONNECT_CODE); + } + + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); + + test( + 'C7: any fault schedule still tears down to zero sessions and connections', + () => + hegel.testAsync(async (tc) => { + const schedule = tc.draw(writeSchedules); + tc.note(`${schedule.length} steps`); + + const ctx = setup(); + try { + const { reqWritable, finalize } = ctx.client.svc.collect.upload({}); + for (const step of schedule) { + if (step === null) { + closeAllConnections(ctx.clientTransport); + continue; + } + + reqWritable.write({ value: step }); + } + + reqWritable.close(); + await finalize(); + } finally { + await teardown(ctx); + } + + // closing a transport must drop everything it owned, whatever happened + // to the wire along the way + for (const t of [ctx.clientTransport, ctx.serverTransport]) { + await waitFor(() => expect(numberOfConnections(t)).toBe(0)); + } + + expect(ctx.violations).toStrictEqual([]); + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); + + test( + 'C8: concurrent streams on one session never mix up their values across faults', + () => + hegel.testAsync(async (tc) => { + const { perStream, writes, faultAt } = tc.draw(multiplexedSchedules); + tc.note( + `${perStream.length} streams, ${writes.length} writes, ${faultAt.size} disconnects`, + ); + + const ctx = setup(); + try { + // one session, so streamId routing is all that keeps them apart + const calls = perStream.map(() => ctx.client.svc.collect.upload({})); + + for (let i = 0; i < writes.length; i++) { + if (faultAt.has(i)) { + closeAllConnections(ctx.clientTransport); + } + + const { stream, value } = writes[i]; + calls[stream].reqWritable.write({ value }); + } + + if (faultAt.has(writes.length)) { + closeAllConnections(ctx.clientTransport); + } + + for (const call of calls) { + call.reqWritable.close(); + } + + const results = await Promise.all(calls.map((c) => c.finalize())); + + for (let i = 0; i < results.length; i++) { + const result = results[i]; + expect(result.ok).toBe(true); + if (!result.ok) continue; + + // its own values, its own order, nothing belonging to a sibling + expect(result.payload.values).toStrictEqual(perStream[i]); + } + + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); +}); + +/** + * The watchdog (#395) measures elapsed time since the last inbound message + * rather than counting heartbeats sent. Both directions of that: it must not + * fire while traffic arrives, and must fire once traffic stops. + */ +describe('heartbeat watchdog properties', () => { + const heartbeatIntervals = gs.sampledFrom([250, 500, 1_000]); + const heartbeatsUntilDead = gs.integers({ minValue: 2, maxValue: 4 }); + + test( + 'C9a: a connection with a live peer survives arbitrarily long', + () => + hegel.testAsync(async (tc) => { + const heartbeatIntervalMs = tc.draw(heartbeatIntervals); + const misses = tc.draw(heartbeatsUntilDead); + // well past the deadline, several times over + const rounds = tc.draw(gs.integers({ minValue: 2, maxValue: 5 })); + tc.note( + `${heartbeatIntervalMs}ms x ${misses} misses, idling ${rounds} deadlines`, + ); + + const opts = { + heartbeatIntervalMs, + heartbeatsUntilDead: misses, + }; + const ctx = setup({ client: opts, server: opts }); + try { + const { reqWritable, finalize } = ctx.client.svc.collect.upload({}); + reqWritable.write({ value: 1 }); + + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(1), + ); + const sessionId = ctx.clientTransport.sessions.get('SERVER')?.id; + + // the peer heartbeats throughout, so elapsed time alone must never + // read as a missed heartbeat + await vi.advanceTimersByTimeAsync( + rounds * misses * heartbeatIntervalMs, + ); + + expect(numberOfConnections(ctx.clientTransport)).toBe(1); + expect(ctx.clientTransport.sessions.get('SERVER')?.id).toBe( + sessionId, + ); + + // and the stream that was open the whole time still works + reqWritable.write({ value: 2 }); + reqWritable.close(); + + const result = await finalize(); + expect(result.ok).toBe(true); + if (!result.ok) return; + + expect(result.payload.values).toStrictEqual([1, 2]); + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); + + test( + 'C9b: a silent peer is detected within heartbeatsUntilDead * heartbeatIntervalMs', + () => + hegel.testAsync(async (tc) => { + const heartbeatIntervalMs = tc.draw(heartbeatIntervals); + const misses = tc.draw(heartbeatsUntilDead); + tc.note(`deadline is ${misses * heartbeatIntervalMs}ms`); + + const opts = { + heartbeatIntervalMs, + heartbeatsUntilDead: misses, + }; + const ctx = setup({ client: opts, server: opts }); + try { + // sit on the dead connection, so we observe the watchdog and not a + // reconnect race + ctx.clientTransport.reconnectOnConnectionDrop = false; + + const { reqWritable } = ctx.client.svc.collect.upload({}); + reqWritable.write({ value: 1 }); + + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(1), + ); + + // the wire goes quiet without either side being told + ctx.network.simulatePhantomDisconnect(); + + // past the deadline, allowing for tick granularity and for however + // long ago the last inbound message was + await vi.advanceTimersByTimeAsync((misses + 2) * heartbeatIntervalMs); + + await waitFor(() => + expect(numberOfConnections(ctx.clientTransport)).toBe(0), + ); + + expect(ctx.violations).toStrictEqual([]); + } finally { + await teardown(ctx); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); +}); + +/** + * `__tests__/e2e.test.ts` covers re-handshaking on a quiet connection. This + * crosses it with faults, because the two interact: a transparent reconnect is + * itself a fresh handshake, re-running `construct` and `validate`. + * + * The invariant is convergence -- whatever `construct` would return right now, + * the server's metadata must catch up to it after any step, refresh or reconnect. + */ +describe('re-handshake under faults', () => { + /** + * Advances fake time until `predicate` holds. + * + * Reconnect backoff grows with each attempt and carries random jitter, so + * polling wall-clock time makes this genuinely nondeterministic. This still + * fails if convergence never happens, it just doesn't care how many backoff + * windows it took. + */ + async function settleUntil(predicate: () => boolean, what: string) { + for (let slice = 0; slice < 200; slice++) { + if (predicate()) return; + await vi.advanceTimersByTimeAsync(50); + } + + throw new Error(`timed out waiting for ${what}`); + } + + const isConnected = ( + transport: ReturnType< + ReturnType['getClientTransport'] + >, + ) => numberOfConnections(transport) === 1; + + const handshakeSchema = Type.Object({ token: Type.String() }); + + type HandshakeMetadata = Static; + + const MetadataServiceSchema = createServiceSchema< + MaybeDisposable, + HandshakeMetadata + >(); + + const metadataServices = { + svc: MetadataServiceSchema.define({ + getToken: Procedure.rpc({ + requestInit: Type.Object({}), + responseData: Type.Object({ token: Type.String() }), + handler: async ({ ctx }) => Ok({ token: ctx.metadata.token }), + }), + }), + }; + + const steps = gs.arrays(gs.sampledFrom(['reconnect', 'refresh'] as const), { + minSize: 1, + maxSize: 6, + }); + + test( + 'C10: metadata converges on the current credential after any mix of refreshes and reconnects', + () => + hegel.testAsync(async (tc) => { + const schedule = tc.draw(steps); + tc.note(schedule.join(' -> ')); + + let token = 'token-0'; + const network = createMockTransportNetwork({ + client: deterministicReconnects, + }); + const clientTransport = network.getClientTransport( + 'client', + createClientHandshakeOptions(handshakeSchema, () => ({ token })), + ); + const serverTransport = network.getServerTransport< + typeof handshakeSchema, + HandshakeMetadata + >( + 'SERVER', + createServerHandshakeOptions< + typeof handshakeSchema, + HandshakeMetadata + >(handshakeSchema, (metadata) => ({ token: metadata.token })), + ); + + const violations: Array = []; + for (const t of [clientTransport, serverTransport]) { + t.bindLogger((msg, ctx, level) => { + if (ctx?.tags?.includes('invariant-violation')) { + violations.push(`[${level}] ${msg}`); + } + }, 'debug'); + } + + createServer(serverTransport, metadataServices); + const client = createClient( + clientTransport, + 'SERVER', + ); + + try { + // establish the session with the initial credential + const first = await client.svc.getToken.rpc({}); + expect(first).toStrictEqual({ ok: true, payload: { token } }); + + const sessionId = clientTransport.sessions.get('SERVER')?.id; + expect(sessionId).toBeDefined(); + + for (let i = 0; i < schedule.length; i++) { + // the credential rotates underneath both paths + token = `token-${i + 1}`; + + if (schedule[i] === 'reconnect') { + closeAllConnections(clientTransport); + } else { + await settleUntil( + () => isConnected(clientTransport), + 'a connection to re-handshake over', + ); + expect(serverTransport.requestRehandshake('client')).toBe(true); + } + + await settleUntil( + () => isConnected(clientTransport), + 'the connection to come back', + ); + + // either way the server must end up holding the current credential + await settleUntil( + () => + serverTransport.sessionHandshakeMetadata.get('client') + ?.token === token, + `metadata to converge on ${token}`, + ); + + // and none of this is allowed to be a hard reconnect + expect(clientTransport.sessions.get('SERVER')?.id).toBe(sessionId); + } + + // ctx.metadata is live, so a handler run now sees the latest value + const last = await client.svc.getToken.rpc({}); + expect(last).toStrictEqual({ ok: true, payload: { token } }); + + expect(violations).toStrictEqual([]); + } finally { + await cleanupTransports([clientTransport, serverTransport]); + await network.cleanup(); + } + }, FAULT_CASES), + FAULT_TIMEOUT_MS, + ); +}); diff --git a/__tests__/properties/streams.property.test.ts b/__tests__/properties/streams.property.test.ts new file mode 100644 index 00000000..77dbd00a --- /dev/null +++ b/__tests__/properties/streams.property.test.ts @@ -0,0 +1,401 @@ +import { describe, expect, test } from 'vitest'; +import * as hegel from '@hegeldev/hegel'; +import * as gs from '@hegeldev/hegel/generators'; +import { Type } from 'typebox'; +import { Ok, Procedure, createServiceSchema } from '../../router'; +import { createClient } from '../../router/client'; +import { createServer } from '../../router/server'; +import { + ReadableBrokenError, + ReadableImpl, + WritableImpl, +} from '../../router/streams'; +import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport'; +import { cleanupTransports } from '../../testUtil/fixtures/cleanup'; +import type { + ProvidedClientTransportOptions, + ProvidedServerTransportOptions, +} from '../../transport/options'; + +/** + * Reader/writer semantics from PROTOCOL.md: ordering, half-close, teardown. + * See ./README.md for the catalog. + */ + +// each case stands up a whole transport network, so fewer cases than the pure +// properties below +const TRANSPORT_CASES = { testCases: 20 }; +const TRANSPORT_TIMEOUT_MS = 60_000; + +const ServiceSchema = createServiceSchema(); + +const PropertyService = ServiceSchema.define({ + /** Echoes every request value straight back. */ + echo: Procedure.stream({ + requestInit: Type.Object({}), + requestData: Type.Object({ value: Type.Number() }), + responseData: Type.Object({ value: Type.Number() }), + async handler({ reqReadable, resWritable }) { + for await (const msg of reqReadable) { + if (!msg.ok) break; + resWritable.write(Ok({ value: msg.payload.value })); + } + + resWritable.close(); + }, + }), + /** Accumulates every request value and reports them in one response. */ + collect: Procedure.upload({ + requestInit: Type.Object({}), + requestData: Type.Object({ value: Type.Number() }), + responseData: Type.Object({ values: Type.Array(Type.Number()) }), + async handler({ reqReadable }) { + const values: Array = []; + for await (const msg of reqReadable) { + if (!msg.ok) break; + values.push(msg.payload.value); + } + + return Ok({ values }); + }, + }), + /** Writes a fixed list of values, then closes. */ + emit: Procedure.subscription({ + requestInit: Type.Object({ values: Type.Array(Type.Number()) }), + responseData: Type.Object({ value: Type.Number() }), + async handler({ reqInit, resWritable }) { + for (const value of reqInit.values) { + resWritable.write(Ok({ value })); + } + + resWritable.close(); + }, + }), + /** Drains the request side first, so its writes provably happen half-closed. */ + drainThenEmit: Procedure.stream({ + requestInit: Type.Object({ after: Type.Array(Type.Number()) }), + requestData: Type.Object({ value: Type.Number() }), + responseData: Type.Object({ value: Type.Number() }), + async handler({ reqInit, reqReadable, resWritable }) { + for await (const msg of reqReadable) { + if (!msg.ok) break; + } + + // the client has closed its writer; ours is still open + for (const value of reqInit.after) { + resWritable.write(Ok({ value })); + } + + resWritable.close(); + }, + }), +}); + +const services = { svc: PropertyService }; + +const values = gs.integers({ minValue: -1_000, maxValue: 1_000 }); +const valueLists = gs.arrays(values, { maxSize: 12 }); + +/** + * A fresh in-memory client/server pair per generated case. + * + * Collects `invariant-violation` logs -- the transport's own ordering + * self-checks -- and asserts none fired, so every property here carries C4. + */ +async function withNetwork( + run: (ctx: { + client: ReturnType>; + }) => Promise, + opts?: { + client?: ProvidedClientTransportOptions; + server?: ProvidedServerTransportOptions; + }, +) { + const network = createMockTransportNetwork(opts); + const clientTransport = network.getClientTransport('client'); + const serverTransport = network.getServerTransport('SERVER'); + + const violations: Array = []; + for (const t of [clientTransport, serverTransport]) { + t.bindLogger((msg, ctx, level) => { + if (ctx?.tags?.includes('invariant-violation')) { + violations.push(`[${level}] ${msg}`); + } + }, 'debug'); + } + + createServer(serverTransport, services); + const client = createClient(clientTransport, 'SERVER'); + + try { + await run({ client }); + expect(violations).toStrictEqual([]); + } finally { + await cleanupTransports([clientTransport, serverTransport]); + await network.cleanup(); + } +} + +describe('stream lifecycle properties', () => { + test( + 'B1: upload delivers exactly the values written, in write order', + () => + hegel.testAsync(async (tc) => { + const sent = tc.draw(valueLists); + tc.note(`sending ${sent.length} values`); + + await withNetwork(async ({ client }) => { + const { reqWritable, finalize } = client.svc.collect.upload({}); + for (const value of sent) { + reqWritable.write({ value }); + } + + reqWritable.close(); + + const result = await finalize(); + expect(result.ok).toBe(true); + if (!result.ok) return; + + expect(result.payload.values).toStrictEqual(sent); + }); + }, TRANSPORT_CASES), + TRANSPORT_TIMEOUT_MS, + ); + + test( + 'B2: stream echoes every value back in send order', + () => + hegel.testAsync(async (tc) => { + const sent = tc.draw(valueLists); + tc.note(`sending ${sent.length} values`); + + await withNetwork(async ({ client }) => { + const { reqWritable, resReadable } = client.svc.echo.stream({}); + for (const value of sent) { + reqWritable.write({ value }); + } + + reqWritable.close(); + + const received = await resReadable.collect(); + expect( + received.map((r) => (r.ok ? r.payload.value : r)), + ).toStrictEqual(sent); + }); + }, TRANSPORT_CASES), + TRANSPORT_TIMEOUT_MS, + ); + + test( + 'B3: subscription delivers exactly the values the server wrote, in order', + () => + hegel.testAsync(async (tc) => { + const emitted = tc.draw(valueLists); + tc.note(`emitting ${emitted.length} values`); + + await withNetwork(async ({ client }) => { + const { resReadable } = client.svc.emit.subscribe({ + values: emitted, + }); + + const received = await resReadable.collect(); + expect( + received.map((r) => (r.ok ? r.payload.value : r)), + ).toStrictEqual(emitted); + }); + }, TRANSPORT_CASES), + TRANSPORT_TIMEOUT_MS, + ); + + test( + 'B4: after the client half-closes, the server can still write and the client still reads', + () => + hegel.testAsync(async (tc) => { + const sent = tc.draw(valueLists); + const after = tc.draw(gs.arrays(values, { minSize: 1, maxSize: 8 })); + tc.note( + `${sent.length} up, then ${after.length} down after half-close`, + ); + + await withNetwork(async ({ client }) => { + const { reqWritable, resReadable } = client.svc.drainThenEmit.stream({ + after, + }); + + for (const value of sent) { + reqWritable.write({ value }); + } + + // half-close: our writer is done, the server's is not + reqWritable.close(); + expect(reqWritable.isWritable()).toBe(false); + + const received = await resReadable.collect(); + expect( + received.map((r) => (r.ok ? r.payload.value : r)), + ).toStrictEqual(after); + }); + }, TRANSPORT_CASES), + TRANSPORT_TIMEOUT_MS, + ); + + test( + 'B8: backpressure is advisory -- ignoring it still delivers every value in order', + () => + hegel.testAsync(async (tc) => { + // enough values to overrun a high-water mark of 1 many times over + const sent = tc.draw(gs.arrays(values, { minSize: 8, maxSize: 16 })); + + let sawBackpressure = false; + await withNetwork( + async ({ client }) => { + const { reqWritable, finalize } = client.svc.collect.upload({}); + + // deliberately ignore the signal: the contract is that the value is + // still buffered and delivered, like node's stream.Writable.write + for (const value of sent) { + if (!reqWritable.write({ value })) { + sawBackpressure = true; + } + } + + reqWritable.close(); + + const result = await finalize(); + expect(result.ok).toBe(true); + if (!result.ok) return; + + // not one value dropped, not one reordered + expect(result.payload.values).toStrictEqual(sent); + }, + { client: { sendBufferHighWaterMark: 1 } }, + ); + + // the scenario is only meaningful if pressure was reported + expect(sawBackpressure).toBe(true); + }, TRANSPORT_CASES), + TRANSPORT_TIMEOUT_MS, + ); +}); + +/** No transport in the way, so these can afford hegel's full case count. */ +describe('reader and writer contract properties', () => { + test('B6: a Writable accepts writes until close, then refuses them', () => + hegel.testAsync((tc) => { + const before = tc.draw(gs.arrays(values, { maxSize: 8 })); + const closeWithValue = tc.draw(gs.optional(values)); + + const written: Array = []; + let closeCalls = 0; + const writable = new WritableImpl({ + writeCb: (v) => written.push(v), + closeCb: () => { + closeCalls++; + }, + }); + + for (const value of before) { + expect(writable.isWritable()).toBe(true); + writable.write(value); + } + + writable.close(closeWithValue ?? undefined); + + // close(value) delivers the value as the final write (property B5) + const expected = + closeWithValue === null ? before : [...before, closeWithValue]; + expect(written).toStrictEqual(expected); + + expect(writable.isWritable()).toBe(false); + expect(() => writable.write(0)).toThrow(/closed Writable/); + + // close is idempotent -- repeated calls neither write nor re-notify + const extraCloses = tc.draw(gs.integers({ minValue: 0, maxValue: 3 })); + for (let i = 0; i < extraCloses; i++) { + writable.close(); + } + + expect(closeCalls).toBe(1); + expect(written).toStrictEqual(expected); + })); + + test('B7: a Readable yields every pushed value once, in push order', () => + hegel.testAsync(async (tc) => { + const pushed = tc.draw(gs.arrays(values, { maxSize: 12 })); + + const readable = new ReadableImpl(); + for (const value of pushed) { + readable._pushValue(Ok(value)); + } + + readable._triggerClose(); + + expect(readable.isReadable()).toBe(true); + const collected = await readable.collect(); + expect(collected.map((r) => (r.ok ? r.payload : r))).toStrictEqual( + pushed, + ); + + // consuming locks the Readable for the rest of its life + expect(readable.isReadable()).toBe(false); + })); + + test('B8: write() reports the advisory signal, and closing releases the waiter', () => + hegel.testAsync(async (tc) => { + const pressured = tc.draw(gs.booleans()); + + const writable = new WritableImpl({ + writeCb: () => undefined, + closeCb: () => undefined, + backpressure: { + isSendBufferFull: () => pressured, + // while pressured and open a waiter really does hang; never awaited + // before close + waitForSendBufferDrain: () => + pressured ? new Promise(() => undefined) : Promise.resolve(), + }, + }); + + // surfaces pressure verbatim, and accepts the value either way + expect(writable.write(tc.draw(values))).toBe(!pressured); + + // a closed writable must never strand a caller + writable.close(); + await expect(writable.waitForWriteReady()).resolves.toBeUndefined(); + })); + + test('B7: break() resolves a pending read with READABLE_BROKEN', () => + hegel.testAsync(async (tc) => { + const pushed = tc.draw(gs.arrays(values, { maxSize: 6 })); + + const readable = new ReadableImpl(); + for (const value of pushed) { + readable._pushValue(Ok(value)); + } + + const iterator = readable[Symbol.asyncIterator](); + + // drain what was queued, checking order on the way through + for (const expected of pushed) { + const next = await iterator.next(); + expect(next.done).toBe(false); + if (next.done) return; + + expect(next.value.ok ? next.value.payload : next.value).toBe(expected); + } + + // now there is a reader waiting with nothing to read + const pending = iterator.next(); + readable.break(); + + const result = await pending; + expect(result.done).toBe(false); + if (result.done) return; + + expect(result.value.ok).toBe(false); + if (result.value.ok) return; + + expect(result.value.payload).toStrictEqual(ReadableBrokenError); + expect(readable.isReadable()).toBe(false); + })); +}); diff --git a/codec/adapter.ts b/codec/adapter.ts index c79d63b3..4ef7daef 100644 --- a/codec/adapter.ts +++ b/codec/adapter.ts @@ -1,19 +1,73 @@ import { Value } from 'typebox/value'; +import { Compile } from 'typebox/compile'; +// deep import: the transport barrel pulls in the client and server transports, +// which are themselves built on a codec import { - OpaqueTransportMessage, + type OpaqueTransportMessage, OpaqueTransportMessageSchema, -} from '../transport'; +} from '../transport/message'; import { Codec } from './types'; import { DeserializeResult, SerializeResult } from '../transport/results'; import { coerceErrorString } from '../transport/stringifyError'; +function interpretedCheck(msg: unknown): msg is OpaqueTransportMessage { + return Value.Check(OpaqueTransportMessageSchema, msg); +} + +type MessageCheck = (msg: unknown) => msg is OpaqueTransportMessage; + +let compiled: MessageCheck | null | undefined; + +/** + * A JIT-compiled version of {@link interpretedCheck}, or null where codegen + * isn't available. Compiling costs a `new Function`, which a strict CSP + * forbids, so this is opt-in rather than the default -- see + * {@link CodecMessageAdapterOptions.precompileValidator}. + */ +function compiledCheck(): MessageCheck | null { + if (compiled === undefined) { + try { + const validator = Compile(OpaqueTransportMessageSchema); + compiled = (msg: unknown): msg is OpaqueTransportMessage => + validator.Check(msg); + } catch { + compiled = null; + } + } + + return compiled; +} + +export interface CodecMessageAdapterOptions { + /** + * Validate inbound messages with a compiled schema validator rather than + * walking the schema on every message. Roughly three orders of magnitude + * faster, and since validation runs per inbound message it otherwise + * dominates the receive path. + * + * Off by default because compiling generates code at runtime, which a strict + * Content-Security-Policy blocks. Servers turn it on; clients (which may be + * browsers) don't. If compiling fails anyway, this silently falls back. + */ + precompileValidator?: boolean; +} + /** * Adapts a {@link Codec} to the {@link OpaqueTransportMessage} format, * accounting for fallibility of toBuffer and fromBuffer and wrapping * it with a Result type. */ export class CodecMessageAdapter { - constructor(private readonly codec: Codec) {} + private readonly check: MessageCheck; + + constructor( + private readonly codec: Codec, + options?: CodecMessageAdapterOptions, + ) { + this.check = + (options?.precompileValidator ? compiledCheck() : null) ?? + interpretedCheck; + } toBuffer(msg: OpaqueTransportMessage): SerializeResult { try { @@ -32,7 +86,7 @@ export class CodecMessageAdapter { fromBuffer(buf: Uint8Array): DeserializeResult { try { const parsedMsg = this.codec.fromBuffer(buf); - if (!Value.Check(OpaqueTransportMessageSchema, parsedMsg)) { + if (!this.check(parsedMsg)) { return { ok: false, reason: 'transport message schema mismatch', diff --git a/codec/binary.ts b/codec/binary.ts index 74e6a381..28d6454f 100644 --- a/codec/binary.ts +++ b/codec/binary.ts @@ -1,4 +1,11 @@ -import { DecodeError, ExtensionCodec, decode, encode } from '@msgpack/msgpack'; +import { + DecodeError, + Decoder, + Encoder, + ExtensionCodec, + decode, + encode, +} from '@msgpack/msgpack'; import { Codec } from './types'; const BIGINT_EXT_TYPE = 0; @@ -33,16 +40,25 @@ extensionCodec.register({ * Binary codec, uses [msgpack](https://www.npmjs.com/package/@msgpack/msgpack) under the hood * @type {Codec} */ +// msgpack's top-level encode/decode build a fresh Encoder/Decoder per call, and +// the Encoder constructor allocates a backing ArrayBuffer every time. Reusing +// one of each drops that per-message allocation. Both classes guard reentrancy +// by cloning themselves, so this stays correct under nested use, and +// Encoder.encode (unlike encodeSharedRef) returns a copy -- which the send +// buffer needs anyway, since it holds onto the bytes for retransmission. +const encoder = new Encoder({ + ignoreUndefined: true, + initialBufferSize: 512, + extensionCodec, +}); +const decoder = new Decoder({ extensionCodec }); + export const BinaryCodec: Codec = { toBuffer(obj) { - return encode(obj, { - ignoreUndefined: true, - initialBufferSize: 512, - extensionCodec, - }); + return encoder.encode(obj); }, fromBuffer: (buff: Uint8Array) => { - const res = decode(buff, { extensionCodec }); + const res = decoder.decode(buff); if (typeof res !== 'object' || res === null) { throw new Error('unpacked msg is not an object'); } diff --git a/codec/json.ts b/codec/json.ts index 505074e0..35a64159 100644 --- a/codec/json.ts +++ b/codec/json.ts @@ -3,18 +3,68 @@ import { Codec } from './types'; const encoder = new TextEncoder(); const decoder = new TextDecoder(); +/** + * JSON can't represent a Uint8Array or a bigint, so this codec encodes them as + * single-key marker objects: `{ $t: }` and `{ $b: }`. + * + * That puts the markers in the same namespace as application data, so a payload + * that happens to use those keys is ambiguous. To keep the round-trip faithful, + * any key that could be mistaken for a marker gains an extra `$` on the way out + * and loses it on the way back in. + * + * Note this only holds when both peers escape. Against an older peer, a payload + * key of `$t`/`$b` behaves as it did before: `$t` decodes as binary, and `$b` + * with a non-numeric value is rejected as a malformed message. + */ +const MARKER_BINARY = '$t'; +const MARKER_BIGINT = '$b'; +/** `$t`, `$$t`, `$b`, `$$$b`, ... -- anything that unescapes toward a marker. */ +const AMBIGUOUS_KEY = /^\$+[tb]$/; +const DOLLAR = '$'.charCodeAt(0); +/** What `bigint.toString()` can produce, so a marker is never a guess. */ +const BIGINT_DIGITS = /^-?\d+$/; + +/** + * `btoa`/`atob` dominate base64 conversion, and building the intermediate + * binary string a character at a time makes it worse -- on a 64KB payload that + * combination cost ~1.4ms per direction. Node's Buffer does the same work in + * single-digit microseconds, so use it where it exists and keep a chunked + * `btoa` path for browsers. + */ +const hasBuffer = typeof Buffer !== 'undefined'; +// how many bytes to hand String.fromCharCode at once without risking the stack +const FROM_CHAR_CODE_CHUNK = 0x8000; + // Convert Uint8Array to base64 function uint8ArrayToBase64(uint8Array: Uint8Array) { + if (hasBuffer) { + return Buffer.from( + uint8Array.buffer, + uint8Array.byteOffset, + uint8Array.byteLength, + ).toString('base64'); + } + let binary = ''; - uint8Array.forEach((byte) => { - binary += String.fromCharCode(byte); - }); + for (let i = 0; i < uint8Array.length; i += FROM_CHAR_CODE_CHUNK) { + binary += String.fromCharCode( + ...uint8Array.subarray(i, i + FROM_CHAR_CODE_CHUNK), + ); + } return btoa(binary); } // Convert base64 to Uint8Array function base64ToUint8Array(base64: string) { + if (hasBuffer) { + const buf = Buffer.from(base64, 'base64'); + + // a view, not a copy -- but typed as Uint8Array rather than Buffer, since + // callers (and deep-equality in tests) distinguish the two by prototype + return new Uint8Array(buf.buffer, buf.byteOffset, buf.byteLength); + } + const binaryString = atob(base64); const uint8Array = new Uint8Array(binaryString.length); for (let i = 0; i < binaryString.length; i++) { @@ -32,6 +82,88 @@ interface BigIntEncodedValue { $b: string; } +function isPlainObject(val: unknown): val is Record { + return typeof val === 'object' && val !== null && !Array.isArray(val); +} + +function isAmbiguous(key: string): boolean { + // the charCode guard keeps the regex off the hot path for ordinary keys + return key.charCodeAt(0) === DOLLAR && AMBIGUOUS_KEY.test(key); +} + +/** An already-escaped key, i.e. two or more `$` before the marker letter. */ +function isEscaped(key: string): boolean { + return key.length > 2 && isAmbiguous(key); +} + +function isOwn(obj: Record, key: string): boolean { + return Object.prototype.hasOwnProperty.call(obj, key); +} + +// for-in rather than Object.keys: these run per object on both hot paths, and +// Object.keys allocates an array every time +function someKey( + obj: Record, + predicate: (key: string) => boolean, +): boolean { + for (const key in obj) { + if (isOwn(obj, key) && predicate(key)) return true; + } + + return false; +} + +/** The single own key of `obj`, or undefined if it doesn't have exactly one. */ +function soleKey(obj: Record): string | undefined { + let only: string | undefined; + for (const key in obj) { + if (!isOwn(obj, key)) continue; + if (only !== undefined) return undefined; + only = key; + } + + return only; +} + +function rekey( + obj: Record, + rename: (key: string) => string, +): Record { + const out: Record = {}; + for (const key of Object.keys(obj)) { + out[rename(key)] = obj[key]; + } + + return out; +} + +/** + * Decodes a marker object, or returns undefined if this isn't one. Markers we + * write always have exactly one key and a well-formed value, so anything else + * is application data that happens to look similar. + */ +function decodeMarker(val: Record): unknown { + const key = soleKey(val); + if (key === undefined) return undefined; + + const encoded = val[key]; + if (typeof encoded !== 'string') return undefined; + + if (key === MARKER_BINARY) { + try { + return base64ToUint8Array(encoded); + } catch { + return undefined; + } + } + + if (key === MARKER_BIGINT && BIGINT_DIGITS.test(encoded)) { + return BigInt(encoded); + } + + return undefined; +} + /** * Naive JSON codec implementation using JSON.stringify and JSON.parse. * @type {Codec} @@ -47,6 +179,11 @@ export const NaiveJsonCodec: Codec = { return { $t: uint8ArrayToBase64(val) } satisfies Base64EncodedValue; } else if (typeof val === 'bigint') { return { $b: val.toString() } satisfies BigIntEncodedValue; + } else if (isPlainObject(val) && someKey(val, isAmbiguous)) { + // returning a copy is what renames the keys: stringify recurses into + // what the replacer hands back. markers we just built never come + // through here, so they are never double-escaped. + return rekey(val, (k) => (isAmbiguous(k) ? `$${k}` : k)); } else { return val; } @@ -57,13 +194,14 @@ export const NaiveJsonCodec: Codec = { const parsed = JSON.parse( decoder.decode(buff), function reviver(_key, val: unknown) { - if ((val as Base64EncodedValue | undefined)?.$t !== undefined) { - return base64ToUint8Array((val as Base64EncodedValue).$t); - } else if ((val as BigIntEncodedValue | undefined)?.$b !== undefined) { - return BigInt((val as BigIntEncodedValue).$b); - } else { - return val; - } + if (!isPlainObject(val)) return val; + + const marker = decodeMarker(val); + if (marker !== undefined) return marker; + + return someKey(val, isEscaped) + ? rekey(val, (k) => (isEscaped(k) ? k.slice(1) : k)) + : val; }, ) as unknown; if (typeof parsed !== 'object' || parsed === null) { diff --git a/package-lock.json b/package-lock.json index 5b4a7070..5bc65561 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,6 +16,7 @@ }, "devDependencies": { "@bufbuild/buf": "^1.67.0", + "@hegeldev/hegel": "^0.4.5", "@opentelemetry/api": "^1.7.0", "@opentelemetry/context-async-hooks": "^1.26.0", "@opentelemetry/core": "^1.7.0", @@ -689,6 +690,105 @@ "node": "^12.22.0 || ^14.17.0 || >=16.0.0" } }, + "node_modules/@hegeldev/hegel": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel/-/hegel-0.4.5.tgz", + "integrity": "sha512-JnCLfQT9kgQFf128efFziu9FqvOlYxZl9NzFVAq7zx1iYb8yNs3vyqZApXTSDvvCpOVPkhUcNrEX5EOc7zx7BA==", + "dev": true, + "dependencies": { + "koffi": "^3.0.2" + }, + "engines": { + "node": ">=20.11.0" + }, + "optionalDependencies": { + "@hegeldev/hegel-darwin-arm64": "0.4.5", + "@hegeldev/hegel-linux-arm64": "0.4.5", + "@hegeldev/hegel-linux-x64": "0.4.5", + "@hegeldev/hegel-win32-arm64": "0.4.5", + "@hegeldev/hegel-win32-x64": "0.4.5" + } + }, + "node_modules/@hegeldev/hegel-darwin-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-darwin-arm64/-/hegel-darwin-arm64-0.4.5.tgz", + "integrity": "sha512-YxVIYThJpCGmpYwT11dXe6RS/8VhP0hR5ZvRoRJrX1foMS5ZnU7fp1UqDBL4mU+yTEY9zCJFX/wuCxJiV4BExw==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=20.11.0" + } + }, + "node_modules/@hegeldev/hegel-linux-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-linux-arm64/-/hegel-linux-arm64-0.4.5.tgz", + "integrity": "sha512-DCoIWPJ0TkUIUn62a5OF78VBjYbEp6dF3AmSNJ/ySEHZXShxZoChUS4bLqXHBbcYZnM9R554QOyGANzc/Q8NgA==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=20.11.0" + } + }, + "node_modules/@hegeldev/hegel-linux-x64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-linux-x64/-/hegel-linux-x64-0.4.5.tgz", + "integrity": "sha512-MhBvsfizvPKOot7U6H19W65G2KEPeOFFNv4F/+E7ubO75QJGWaBPvufJ96xPv/nfgBtLbYeu3sFQxvTNluwu1g==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=20.11.0" + } + }, + "node_modules/@hegeldev/hegel-win32-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-win32-arm64/-/hegel-win32-arm64-0.4.5.tgz", + "integrity": "sha512-TDOrfNgPXeEJv4ONuqG+rgzQ+7yhqPIMVBDl3kGtFi7f6aA4cQFjKAVdinX8hHyZzFQeIU0IMPvtt0dpCHD9hA==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=20.11.0" + } + }, + "node_modules/@hegeldev/hegel-win32-x64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-win32-x64/-/hegel-win32-x64-0.4.5.tgz", + "integrity": "sha512-ZWh/Yd2iHSShpWdb5t4wm9bWG1kxjhQOFW5fsFEcVXyAKldaXgHU1Di9fkR2M+0khNjNyUIVoeoGU69i/Bt6Fg==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=20.11.0" + } + }, "node_modules/@humanwhocodes/config-array": { "version": "0.11.14", "resolved": "https://registry.npmjs.org/@humanwhocodes/config-array/-/config-array-0.11.14.tgz", @@ -822,6 +922,246 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "node_modules/@koromix/koffi-darwin-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-arm64/-/koffi-darwin-arm64-3.1.5.tgz", + "integrity": "sha512-IpqITl2fJi3QN9bTtNnygWPdK7ScSjw3xtGu8e6feYGvimCysu+spgI5KyeslY2jTnqxGS9xr8pLAbLhGJ8edA==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "darwin" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-darwin-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-x64/-/koffi-darwin-x64-3.1.5.tgz", + "integrity": "sha512-4Tia4BS5EV/+vN9eIrdToanVe+U/2VqTZCBgOzoUbPKjgky51eqM+3J4qdRUvmYJohcJNPob5/hsxeItUZrl1g==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "darwin" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-freebsd-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-arm64/-/koffi-freebsd-arm64-3.1.5.tgz", + "integrity": "sha512-bP94uzseFO79NG3flpU3WxfyvltD+jzC/kN8FDZLi8J0VUZNW1Wmu8yq87KEzjX1qT9KyX4Y+elVbGqHXHTv2Q==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "freebsd" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-freebsd-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-ia32/-/koffi-freebsd-ia32-3.1.5.tgz", + "integrity": "sha512-raFXXAPHzvCQWhaoMUF+Cc2ZWgg2UBU0RVoowHZhaw9nQYPC1pERcPRH+JA+SNIN6g4d2GFW6uPFc+QbUhsagA==", + "cpu": [ + "ia32" + ], + "dev": true, + "optional": true, + "os": [ + "freebsd" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-freebsd-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-x64/-/koffi-freebsd-x64-3.1.5.tgz", + "integrity": "sha512-h6RyBZmPMBIDWTABkJIhzDdYwSnYAJvTacHpEjbT55Arkmw1H15Rl7CFtXuEuBrqh+uivoCrRgA6vszl9CsJ9g==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "freebsd" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-linux-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-arm64/-/koffi-linux-arm64-3.1.5.tgz", + "integrity": "sha512-u0vCmKPu4yQDhl/ri1J6U3vDnvYtYjoZaIWb+oMbRXhVZeiqdE53MGPb+q2A7Dj2n9IbYloAfEICQbL6l0pmiQ==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-linux-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-ia32/-/koffi-linux-ia32-3.1.5.tgz", + "integrity": "sha512-Xa5JbumWglwPVZgrJcLhqyC1wCWlfm7+C00p3FuOTNGp0qoYuf2/tOoAh0/q7+taVm30cMopu/6lRHwdmF6I/w==", + "cpu": [ + "ia32" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-linux-loong64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-loong64/-/koffi-linux-loong64-3.1.5.tgz", + "integrity": "sha512-F3i2CeTcqVBUQiSRUBSEzX1VgtXmLLiZb/ouZtXHkWpTLhJNd9TH7s3CizTofci0VRqlKdexGUYhK5vzPDAAHA==", + "cpu": [ + "loong64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-linux-riscv64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-riscv64/-/koffi-linux-riscv64-3.1.5.tgz", + "integrity": "sha512-2TgQuzy+4PfDg+rw3kOmN6lywEWdzKT3eaLPbOp0b/9DaN7CLBJ/QIR5GhGsMNpeI30zU0YzFeYFxoVoJ/LqFw==", + "cpu": [ + "riscv64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-linux-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-x64/-/koffi-linux-x64-3.1.5.tgz", + "integrity": "sha512-2yaIg/1V0m4CiAUMzG4CIlWmq1WJ+QBMlFfaGr9su+OH5fuIqC7V3BbMiB03IwIl1VofIZO5JA4Db4lID8tpbw==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "linux" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-openbsd-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-ia32/-/koffi-openbsd-ia32-3.1.5.tgz", + "integrity": "sha512-8/OXd+u9omMooykhvdJPEP7u6FFzzrrFo9gOmSHc9/DPt3XkVYOtSsE97PDk6zYaAzwIYHSKjHvIXsfFwFc7sg==", + "cpu": [ + "ia32" + ], + "dev": true, + "optional": true, + "os": [ + "openbsd" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-openbsd-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-x64/-/koffi-openbsd-x64-3.1.5.tgz", + "integrity": "sha512-SpeqldKkuDk2aTj5PVWumy7eq6Tr2GtBPAOI1NiDHhg8xe433KraxrA9V9UjEd+1+kSGJQGR04nQByN3MPA9PQ==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "openbsd" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-win32-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-arm64/-/koffi-win32-arm64-3.1.5.tgz", + "integrity": "sha512-uej3YAEKAhlfVPoIo5sOwtxhTLRVJ01LgtWrKGpnnAQU3C+Ilmaxdh+Oc2xc1G3NK30N5eCVJpyO9r3pjKC6Vw==", + "cpu": [ + "arm64" + ], + "dev": true, + "optional": true, + "os": [ + "win32" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-win32-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-ia32/-/koffi-win32-ia32-3.1.5.tgz", + "integrity": "sha512-d42jv2f4PwtJGNJS19Xfn/BRtGsBNNVkw0O0K5tkIGI+yNq4MnPTSUsaGbDIkCLNsCgk/LFqAaE8BExyCdrujA==", + "cpu": [ + "ia32" + ], + "dev": true, + "optional": true, + "os": [ + "win32" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, + "node_modules/@koromix/koffi-win32-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-x64/-/koffi-win32-x64-3.1.5.tgz", + "integrity": "sha512-Pyo1WEHEP6Ek2NEn2pquwJzSPLOdY4vymoPSz0an1DgvFWSkOyBYYkGVoxL1ajfj5I1pPdXysYqixtVTzAtOfQ==", + "cpu": [ + "x64" + ], + "dev": true, + "optional": true, + "os": [ + "win32" + ], + "funding": { + "url": "https://liberapay.com/Koromix" + } + }, "node_modules/@msgpack/msgpack": { "version": "3.1.2", "resolved": "https://registry.npmjs.org/@msgpack/msgpack/-/msgpack-3.1.2.tgz", @@ -3021,6 +3361,33 @@ "json-buffer": "3.0.1" } }, + "node_modules/koffi": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/koffi/-/koffi-3.1.5.tgz", + "integrity": "sha512-XVwwrxg0Ca6IEUQF4YtGIU4XN0LSselFYpYvgfhh8wafCunhEEx5hPr7LZhp5QyeFA/LcRsKHTqncCdjWjWAlg==", + "dev": true, + "hasInstallScript": true, + "funding": { + "url": "https://liberapay.com/Koromix" + }, + "optionalDependencies": { + "@koromix/koffi-darwin-arm64": "3.1.5", + "@koromix/koffi-darwin-x64": "3.1.5", + "@koromix/koffi-freebsd-arm64": "3.1.5", + "@koromix/koffi-freebsd-ia32": "3.1.5", + "@koromix/koffi-freebsd-x64": "3.1.5", + "@koromix/koffi-linux-arm64": "3.1.5", + "@koromix/koffi-linux-ia32": "3.1.5", + "@koromix/koffi-linux-loong64": "3.1.5", + "@koromix/koffi-linux-riscv64": "3.1.5", + "@koromix/koffi-linux-x64": "3.1.5", + "@koromix/koffi-openbsd-ia32": "3.1.5", + "@koromix/koffi-openbsd-x64": "3.1.5", + "@koromix/koffi-win32-arm64": "3.1.5", + "@koromix/koffi-win32-ia32": "3.1.5", + "@koromix/koffi-win32-x64": "3.1.5" + } + }, "node_modules/levn": { "version": "0.4.1", "resolved": "https://registry.npmjs.org/levn/-/levn-0.4.1.tgz", @@ -5380,6 +5747,55 @@ "integrity": "sha512-Ys+3g2TaW7gADOJzPt83SJtCDhMjndcDMFVQ/Tj9iA1BfJzFKD9mAUXT3OenpuPHbI6P/myECxRJrofUsDx/5g==", "dev": true }, + "@hegeldev/hegel": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel/-/hegel-0.4.5.tgz", + "integrity": "sha512-JnCLfQT9kgQFf128efFziu9FqvOlYxZl9NzFVAq7zx1iYb8yNs3vyqZApXTSDvvCpOVPkhUcNrEX5EOc7zx7BA==", + "dev": true, + "requires": { + "@hegeldev/hegel-darwin-arm64": "0.4.5", + "@hegeldev/hegel-linux-arm64": "0.4.5", + "@hegeldev/hegel-linux-x64": "0.4.5", + "@hegeldev/hegel-win32-arm64": "0.4.5", + "@hegeldev/hegel-win32-x64": "0.4.5", + "koffi": "^3.0.2" + } + }, + "@hegeldev/hegel-darwin-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-darwin-arm64/-/hegel-darwin-arm64-0.4.5.tgz", + "integrity": "sha512-YxVIYThJpCGmpYwT11dXe6RS/8VhP0hR5ZvRoRJrX1foMS5ZnU7fp1UqDBL4mU+yTEY9zCJFX/wuCxJiV4BExw==", + "dev": true, + "optional": true + }, + "@hegeldev/hegel-linux-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-linux-arm64/-/hegel-linux-arm64-0.4.5.tgz", + "integrity": "sha512-DCoIWPJ0TkUIUn62a5OF78VBjYbEp6dF3AmSNJ/ySEHZXShxZoChUS4bLqXHBbcYZnM9R554QOyGANzc/Q8NgA==", + "dev": true, + "optional": true + }, + "@hegeldev/hegel-linux-x64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-linux-x64/-/hegel-linux-x64-0.4.5.tgz", + "integrity": "sha512-MhBvsfizvPKOot7U6H19W65G2KEPeOFFNv4F/+E7ubO75QJGWaBPvufJ96xPv/nfgBtLbYeu3sFQxvTNluwu1g==", + "dev": true, + "optional": true + }, + "@hegeldev/hegel-win32-arm64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-win32-arm64/-/hegel-win32-arm64-0.4.5.tgz", + "integrity": "sha512-TDOrfNgPXeEJv4ONuqG+rgzQ+7yhqPIMVBDl3kGtFi7f6aA4cQFjKAVdinX8hHyZzFQeIU0IMPvtt0dpCHD9hA==", + "dev": true, + "optional": true + }, + "@hegeldev/hegel-win32-x64": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@hegeldev/hegel-win32-x64/-/hegel-win32-x64-0.4.5.tgz", + "integrity": "sha512-ZWh/Yd2iHSShpWdb5t4wm9bWG1kxjhQOFW5fsFEcVXyAKldaXgHU1Di9fkR2M+0khNjNyUIVoeoGU69i/Bt6Fg==", + "dev": true, + "optional": true + }, "@humanwhocodes/config-array": { "version": "0.11.14", "resolved": "https://registry.npmjs.org/@humanwhocodes/config-array/-/config-array-0.11.14.tgz", @@ -5473,6 +5889,111 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "@koromix/koffi-darwin-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-arm64/-/koffi-darwin-arm64-3.1.5.tgz", + "integrity": "sha512-IpqITl2fJi3QN9bTtNnygWPdK7ScSjw3xtGu8e6feYGvimCysu+spgI5KyeslY2jTnqxGS9xr8pLAbLhGJ8edA==", + "dev": true, + "optional": true + }, + "@koromix/koffi-darwin-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-x64/-/koffi-darwin-x64-3.1.5.tgz", + "integrity": "sha512-4Tia4BS5EV/+vN9eIrdToanVe+U/2VqTZCBgOzoUbPKjgky51eqM+3J4qdRUvmYJohcJNPob5/hsxeItUZrl1g==", + "dev": true, + "optional": true + }, + "@koromix/koffi-freebsd-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-arm64/-/koffi-freebsd-arm64-3.1.5.tgz", + "integrity": "sha512-bP94uzseFO79NG3flpU3WxfyvltD+jzC/kN8FDZLi8J0VUZNW1Wmu8yq87KEzjX1qT9KyX4Y+elVbGqHXHTv2Q==", + "dev": true, + "optional": true + }, + "@koromix/koffi-freebsd-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-ia32/-/koffi-freebsd-ia32-3.1.5.tgz", + "integrity": "sha512-raFXXAPHzvCQWhaoMUF+Cc2ZWgg2UBU0RVoowHZhaw9nQYPC1pERcPRH+JA+SNIN6g4d2GFW6uPFc+QbUhsagA==", + "dev": true, + "optional": true + }, + "@koromix/koffi-freebsd-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-x64/-/koffi-freebsd-x64-3.1.5.tgz", + "integrity": "sha512-h6RyBZmPMBIDWTABkJIhzDdYwSnYAJvTacHpEjbT55Arkmw1H15Rl7CFtXuEuBrqh+uivoCrRgA6vszl9CsJ9g==", + "dev": true, + "optional": true + }, + "@koromix/koffi-linux-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-arm64/-/koffi-linux-arm64-3.1.5.tgz", + "integrity": "sha512-u0vCmKPu4yQDhl/ri1J6U3vDnvYtYjoZaIWb+oMbRXhVZeiqdE53MGPb+q2A7Dj2n9IbYloAfEICQbL6l0pmiQ==", + "dev": true, + "optional": true + }, + "@koromix/koffi-linux-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-ia32/-/koffi-linux-ia32-3.1.5.tgz", + "integrity": "sha512-Xa5JbumWglwPVZgrJcLhqyC1wCWlfm7+C00p3FuOTNGp0qoYuf2/tOoAh0/q7+taVm30cMopu/6lRHwdmF6I/w==", + "dev": true, + "optional": true + }, + "@koromix/koffi-linux-loong64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-loong64/-/koffi-linux-loong64-3.1.5.tgz", + "integrity": "sha512-F3i2CeTcqVBUQiSRUBSEzX1VgtXmLLiZb/ouZtXHkWpTLhJNd9TH7s3CizTofci0VRqlKdexGUYhK5vzPDAAHA==", + "dev": true, + "optional": true + }, + "@koromix/koffi-linux-riscv64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-riscv64/-/koffi-linux-riscv64-3.1.5.tgz", + "integrity": "sha512-2TgQuzy+4PfDg+rw3kOmN6lywEWdzKT3eaLPbOp0b/9DaN7CLBJ/QIR5GhGsMNpeI30zU0YzFeYFxoVoJ/LqFw==", + "dev": true, + "optional": true + }, + "@koromix/koffi-linux-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-linux-x64/-/koffi-linux-x64-3.1.5.tgz", + "integrity": "sha512-2yaIg/1V0m4CiAUMzG4CIlWmq1WJ+QBMlFfaGr9su+OH5fuIqC7V3BbMiB03IwIl1VofIZO5JA4Db4lID8tpbw==", + "dev": true, + "optional": true + }, + "@koromix/koffi-openbsd-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-ia32/-/koffi-openbsd-ia32-3.1.5.tgz", + "integrity": "sha512-8/OXd+u9omMooykhvdJPEP7u6FFzzrrFo9gOmSHc9/DPt3XkVYOtSsE97PDk6zYaAzwIYHSKjHvIXsfFwFc7sg==", + "dev": true, + "optional": true + }, + "@koromix/koffi-openbsd-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-x64/-/koffi-openbsd-x64-3.1.5.tgz", + "integrity": "sha512-SpeqldKkuDk2aTj5PVWumy7eq6Tr2GtBPAOI1NiDHhg8xe433KraxrA9V9UjEd+1+kSGJQGR04nQByN3MPA9PQ==", + "dev": true, + "optional": true + }, + "@koromix/koffi-win32-arm64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-arm64/-/koffi-win32-arm64-3.1.5.tgz", + "integrity": "sha512-uej3YAEKAhlfVPoIo5sOwtxhTLRVJ01LgtWrKGpnnAQU3C+Ilmaxdh+Oc2xc1G3NK30N5eCVJpyO9r3pjKC6Vw==", + "dev": true, + "optional": true + }, + "@koromix/koffi-win32-ia32": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-ia32/-/koffi-win32-ia32-3.1.5.tgz", + "integrity": "sha512-d42jv2f4PwtJGNJS19Xfn/BRtGsBNNVkw0O0K5tkIGI+yNq4MnPTSUsaGbDIkCLNsCgk/LFqAaE8BExyCdrujA==", + "dev": true, + "optional": true + }, + "@koromix/koffi-win32-x64": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/@koromix/koffi-win32-x64/-/koffi-win32-x64-3.1.5.tgz", + "integrity": "sha512-Pyo1WEHEP6Ek2NEn2pquwJzSPLOdY4vymoPSz0an1DgvFWSkOyBYYkGVoxL1ajfj5I1pPdXysYqixtVTzAtOfQ==", + "dev": true, + "optional": true + }, "@msgpack/msgpack": { "version": "3.1.2", "resolved": "https://registry.npmjs.org/@msgpack/msgpack/-/msgpack-3.1.2.tgz", @@ -6945,6 +7466,29 @@ "json-buffer": "3.0.1" } }, + "koffi": { + "version": "3.1.5", + "resolved": "https://registry.npmjs.org/koffi/-/koffi-3.1.5.tgz", + "integrity": "sha512-XVwwrxg0Ca6IEUQF4YtGIU4XN0LSselFYpYvgfhh8wafCunhEEx5hPr7LZhp5QyeFA/LcRsKHTqncCdjWjWAlg==", + "dev": true, + "requires": { + "@koromix/koffi-darwin-arm64": "3.1.5", + "@koromix/koffi-darwin-x64": "3.1.5", + "@koromix/koffi-freebsd-arm64": "3.1.5", + "@koromix/koffi-freebsd-ia32": "3.1.5", + "@koromix/koffi-freebsd-x64": "3.1.5", + "@koromix/koffi-linux-arm64": "3.1.5", + "@koromix/koffi-linux-ia32": "3.1.5", + "@koromix/koffi-linux-loong64": "3.1.5", + "@koromix/koffi-linux-riscv64": "3.1.5", + "@koromix/koffi-linux-x64": "3.1.5", + "@koromix/koffi-openbsd-ia32": "3.1.5", + "@koromix/koffi-openbsd-x64": "3.1.5", + "@koromix/koffi-win32-arm64": "3.1.5", + "@koromix/koffi-win32-ia32": "3.1.5", + "@koromix/koffi-win32-x64": "3.1.5" + } + }, "levn": { "version": "0.4.1", "resolved": "https://registry.npmjs.org/levn/-/levn-0.4.1.tgz", diff --git a/package.json b/package.json index 7496a747..c16f19e7 100644 --- a/package.json +++ b/package.json @@ -37,6 +37,7 @@ }, "devDependencies": { "@bufbuild/buf": "^1.67.0", + "@hegeldev/hegel": "^0.4.5", "@opentelemetry/api": "^1.7.0", "@opentelemetry/context-async-hooks": "^1.26.0", "@opentelemetry/core": "^1.7.0", @@ -73,7 +74,7 @@ "bench": "vitest bench" }, "engines": { - "node": ">=16" + "node": ">=20.11.0" }, "keywords": [ "rpc", diff --git a/router/services.ts b/router/services.ts index 85124c1e..9ba223d0 100644 --- a/router/services.ts +++ b/router/services.ts @@ -293,8 +293,8 @@ export function serializeSchema( * add: Procedure.rpc({ * requestInit: Type.Object({ a: Type.Number(), b: Type.Number() }), * responseData: Type.Object({ result: Type.Number() }), - * async handler(ctx, init) { - * return Ok({ result: init.a + init.b }); + * async handler({ ctx, reqInit }) { + * return Ok({ result: reqInit.a + reqInit.b }); * } * }), * getUserId: Procedure.rpc({ @@ -367,8 +367,8 @@ export function createServiceSchema< * increment: Procedure.rpc({ * requestInit: Type.Object({ amount: Type.Number() }), * responseData: Type.Object({ current: Type.Number() }), - * async handler(ctx, init) { - * ctx.state.count += init.amount; + * async handler({ ctx, reqInit }) { + * ctx.state.count += reqInit.amount; * return Ok({ current: ctx.state.count }); * } * }), @@ -394,8 +394,8 @@ export function createServiceSchema< * increment: Procedure.rpc({ * requestInit: Type.Object({ amount: Type.Number() }), * responseData: Type.Object({ current: Type.Number() }), - * async handler(ctx, init) { - * ctx.state.count += init.amount; + * async handler({ ctx, reqInit }) { + * ctx.state.count += reqInit.amount; * return Ok({ current: ctx.state.count }); * } * }), @@ -429,8 +429,8 @@ export function createServiceSchema< * increment: Procedure.rpc({ * requestInit: Type.Object({ amount: Type.Number() }), * responseData: Type.Object({ current: Type.Number() }), - * async handler(ctx, init) { - * ctx.state.count += init.amount; + * async handler({ ctx, reqInit }) { + * ctx.state.count += reqInit.amount; * return Ok({ current: ctx.state.count }); * } * }), @@ -464,8 +464,8 @@ export function createServiceSchema< * add: Procedure.rpc({ * requestInit: Type.Object({ a: Type.Number(), b: Type.Number() }), * responseData: Type.Object({ result: Type.Number() }), - * async handler(ctx, init) { - * return Ok({ result: init.a + init.b }); + * async handler({ ctx, reqInit }) { + * return Ok({ result: reqInit.a + reqInit.b }); * } * }), * }); diff --git a/tracing/index.ts b/tracing/index.ts index 273d5fd0..a911ac31 100644 --- a/tracing/index.ts +++ b/tracing/index.ts @@ -10,7 +10,9 @@ import { } from '@opentelemetry/api'; import { RIVER_VERSION, ValidProcType } from '../router'; import { ErrorPayload } from '../router/result'; -import { Connection } from '../transport'; +// deep + type-only: transport.ts imports getTracer from here, so importing the +// transport barrel as a value would cycle +import type { Connection } from '../transport/connection'; import { MessageMetadata } from '../logging'; import { ClientSession } from '../transport/sessionStateMachine/transitions'; import { IdentifiedSession } from '../transport/sessionStateMachine/common'; diff --git a/transport/index.ts b/transport/index.ts index 32eda807..9bd911e1 100644 --- a/transport/index.ts +++ b/transport/index.ts @@ -24,13 +24,13 @@ export { export { TransportMessageSchema, OpaqueTransportMessageSchema, + isStreamOpen, + isStreamClose, } from './message'; export type { TransportMessage, OpaqueTransportMessage, TransportClientId, - isStreamOpen, - isStreamClose, } from './message'; export { EventMap, diff --git a/transport/message.ts b/transport/message.ts index a55a3145..5dbd0cd2 100644 --- a/transport/message.ts +++ b/transport/message.ts @@ -1,7 +1,8 @@ import { Type, type TSchema, type Static } from 'typebox'; import { PropagationContext } from '../tracing'; import { generateId } from './id'; -import { ErrResult } from '../router'; +// type-only: a value import closes a transport <-> router require cycle +import type { ErrResult } from '../router'; import type { ErrorPayload } from '../router/result'; /** diff --git a/transport/sessionStateMachine/transitions.ts b/transport/sessionStateMachine/transitions.ts index 556d7740..41df4313 100644 --- a/transport/sessionStateMachine/transitions.ts +++ b/transport/sessionStateMachine/transitions.ts @@ -42,6 +42,12 @@ import { EncodedTransportMessage, ProtocolVersion } from '../message'; import { Tracer } from '@opentelemetry/api'; import { CodecMessageAdapter } from '../../codec'; +/** + * Only the server compiles its inbound validator: compiling generates code at + * runtime, which a strict CSP blocks, and the client may be a browser. + */ +const serverValidation = { precompileValidator: true }; + function inheritSharedSession( session: IdentifiedSession, ): Omit { @@ -128,7 +134,7 @@ export const SessionStateGraph = { options, tracer, log, - codec: new CodecMessageAdapter(options.codec), + codec: new CodecMessageAdapter(options.codec, serverValidation), }); session.log?.info(`session created in WaitingForHandshake state`, { @@ -281,7 +287,7 @@ export const SessionStateGraph = { tracer: pendingSession.tracer, log: pendingSession.log, protocolVersion, - codec: new CodecMessageAdapter(options.codec), + codec: new CodecMessageAdapter(options.codec, serverValidation), } satisfies Omit); pendingSession._handleStateExit(); diff --git a/vitest.config.ts b/vitest.config.ts index a9881d20..6e9de16f 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -3,9 +3,19 @@ import { configDefaults, coverageConfigDefaults } from 'vitest/config'; export default defineConfig({ test: { - exclude: [...configDefaults.exclude, '**/.direnv/**'], + // .claude holds scratch git worktrees: stale checkouts of this same repo + exclude: [...configDefaults.exclude, '**/.direnv/**', '**/.claude/**'], coverage: { - exclude: [...coverageConfigDefaults.exclude, '**/.direnv/**'], + exclude: [ + ...coverageConfigDefaults.exclude, + '**/.direnv/**', + '**/.claude/**', + ], + }, + // benchmark globbing is separate from test globbing, so it needs the same + // exclusions spelled out again + benchmark: { + exclude: [...configDefaults.exclude, '**/.direnv/**', '**/.claude/**'], }, sequence: { hooks: 'stack',