From 5112cc1a620808ff5629cdaded0af2a250ad8a71 Mon Sep 17 00:00:00 2001 From: Ajay Ghanwat Date: Sun, 9 Aug 2026 10:02:36 +0530 Subject: [PATCH] docs: measure the runtime and the generated client modules Adds a per-subsystem measurement of reactive.js, made by minifying it repeatedly with one subsystem removed rather than counting source bytes. This corrects the earlier audit on both figures and on the conclusion drawn from them. Component controllers are 23,722 bytes minified / 6,660 gzipped -- 30.6% of transfer, not the "about 18%" previously claimed -- and splitting them out saves 6.6 kB gzipped on a typical page, not "3-4 kB". Measured against the example app, / and /login use none of the ten controllers and /layout uses one, so most pages download and parse the lot for nothing. The larger finding is that the runtime is not where the weight is. One page parses 490,212 decoded bytes across 11 generated client modules while transferring 21,026, and the largest module is 89.8% duplicated lines: the state-restore prologue appears 162 times because client-codegen.ts inlines the sync into every peer alias of every client function. Gzip hides it on the wire, but parse cost follows decoded bytes. Co-Authored-By: Claude Opus 5 --- bun.lock | 21 ++- docs/framework-remediation-plan.md | 142 ++++++++++++++++++ examples/inter-app-api-showcase/README.md | 19 +++ .../app/api/product-summary.ts | 60 ++++++++ .../app/example.test.ts | 11 ++ .../app/lib/contracts.ts | 34 +++++ examples/inter-app-api-showcase/package.json | 22 +++ examples/inter-app-api-showcase/tsconfig.json | 5 + packages/pubsub/package.json | 4 + packages/pubsub/src/index.ts | 2 + packages/pubsub/src/subject.ts | 49 ++++++ packages/pubsub/test/subject.test.ts | 24 +++ packages/queue/package.json | 3 +- packages/queue/src/index.ts | 2 + packages/queue/src/subject.ts | 56 +++++++ packages/queue/test/subject.test.ts | 26 ++++ packages/rpc/src/client.ts | 6 +- packages/rpc/src/index.ts | 10 +- packages/rpc/src/transport.ts | 84 +++++++++++ packages/rpc/test/retry.test.ts | 68 +++++++++ 20 files changed, 643 insertions(+), 5 deletions(-) create mode 100644 examples/inter-app-api-showcase/README.md create mode 100644 examples/inter-app-api-showcase/app/api/product-summary.ts create mode 100644 examples/inter-app-api-showcase/app/example.test.ts create mode 100644 examples/inter-app-api-showcase/app/lib/contracts.ts create mode 100644 examples/inter-app-api-showcase/package.json create mode 100644 examples/inter-app-api-showcase/tsconfig.json create mode 100644 packages/pubsub/src/subject.ts create mode 100644 packages/pubsub/test/subject.test.ts create mode 100644 packages/queue/src/subject.ts create mode 100644 packages/queue/test/subject.test.ts create mode 100644 packages/rpc/test/retry.test.ts diff --git a/bun.lock b/bun.lock index 74382c00..75d38b4b 100644 --- a/bun.lock +++ b/bun.lock @@ -105,6 +105,19 @@ "typescript": "^5.9.2", }, }, + "examples/inter-app-api-showcase": { + "name": "inter-app-api-showcase", + "version": "0.8.6", + "dependencies": { + "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*", + "@wrnexus/validation": "workspace:*", + }, + "devDependencies": { + "@types/bun": "^1.3.14", + "typescript": "^5.9.2", + }, + }, "packages/ai": { "name": "@wrnexus/ai", "version": "0.8.6", @@ -404,6 +417,10 @@ "packages/pubsub": { "name": "@wrnexus/pubsub", "version": "0.8.6", + "dependencies": { + "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*", + }, }, "packages/pwa": { "name": "@wrnexus/pwa", @@ -418,6 +435,7 @@ "version": "0.8.6", "dependencies": { "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*", }, }, "packages/reactive": { @@ -450,7 +468,6 @@ "name": "@wrnexus/rpc", "version": "0.8.6", "dependencies": { - "@wrnexus/authz": "workspace:*", "@wrnexus/core": "workspace:*", "@wrnexus/helpers": "workspace:*", "@wrnexus/jwt": "workspace:*", @@ -1058,6 +1075,8 @@ "imurmurhash": ["imurmurhash@0.1.4", "", {}, "sha512-JmXMZ6wuvDmLiHEml9ykzqO6lwFbof0GG4IkcGaENdCRDDmMVnny7s5HsIgHCbaq0w2MyPhDqkhTUgS2LU2PHA=="], + "inter-app-api-showcase": ["inter-app-api-showcase@workspace:examples/inter-app-api-showcase"], + "is-extglob": ["is-extglob@2.1.1", "", {}, "sha512-SbKbANkN603Vi4jEZv49LeVJMn4yGwsbzZworEoyEiutsN3nJYdbO36zfhGJ6QEDpOZIFkDtnq5JRxmvl3jsoQ=="], "is-glob": ["is-glob@4.0.3", "", { "dependencies": { "is-extglob": "^2.1.1" } }, "sha512-xelSayHH36ZgE7ZWhli7pW34hNbNl8Ojv5KVmkJD4hBdD3th8Tfk9vYasLM+mXWOZhFkgZfxhLSnrwRr4elSSg=="], diff --git a/docs/framework-remediation-plan.md b/docs/framework-remediation-plan.md index 4c292221..944bf04e 100644 --- a/docs/framework-remediation-plan.md +++ b/docs/framework-remediation-plan.md @@ -466,6 +466,148 @@ confusing ways. --- +## 4.6 Runtime and client-module size — measured + +A per-subsystem measurement of `reactive.js` and of the generated client +modules, made by minifying the runtime repeatedly with one subsystem removed +each time. Source-byte share was not used: it overstates code that minifies +well and understates code that does not, and the split/keep decision turns on +the real number. + +**This corrects the earlier audit**, which claimed component controllers were +"about 18%" of the runtime and concluded that splitting saves "3-4 kB gzipped". +Both figures were wrong, and the conclusion that followed from them was wrong. + +### The runtime today + +`reactive.js` is **70,101 bytes minified, 21,736 gzipped**. Removing each +subsystem and re-minifying gives its true cost: + +| Subsystem | Minified | Share | +| ----------------- | -------- | ----- | +| Select/combobox | 7,852 | 11.2% | +| PinInput | 3,734 | 5.3% | +| async boundaries | 3,151 | 4.5% | +| Navbar | 2,419 | 3.5% | +| splitters | 2,083 | 3.0% | +| roving focus | 1,646 | 2.3% | +| modal dialogs | 1,611 | 2.3% | +| csr fetch | 1,547 | 2.2% | +| anchored overlays | 1,419 | 2.0% | +| scrollspy | 1,373 | 2.0% | +| preferences | 1,045 | 1.5% | +| toast | 534 | 0.8% | + +Component-specific controllers (everything except async boundaries and csr +fetch, which are framework features) total **23,722 minified / 6,660 gzipped — +30.6% of what a visitor downloads.** The irreducible core is 46,379 minified / +15,076 gzipped: the expression engine, the scope and reactivity core, and loop +diffing. + +### How much of it a page actually uses + +Measured against the example app by checking which controller markers appear in +the served HTML: + +| Page | Controllers used | +| ------------- | ------------------ | +| `/` | **0 of 10** | +| `/login` | **0 of 10** | +| `/layout` | 1 of 10 (splitter) | +| `/navigation` | 5 of 10 | + +A typical page downloads and parses 6.6 kB gzipped of controller code it never +executes. The Select controller — the single largest item at 7,852 bytes — is +used by none of the pages above. + +**Change.** Split the component controllers out of the core runtime and load +them on demand, keyed on the marker attribute that already gates each one +(`data-wrn-select`, `data-wrn-splitter`, `data-wrn-scrollspy` and so on). The +gating logic exists; only the loading boundary is missing. Keep the core +runtime as one immutable-cached file. + +**How to test.** Assert the core bundle size, and per page assert that a +controller chunk is requested only when its marker is present: + +```bash +bun run scripts/lib/measure-runtime-size.ts # core must stay under budget +``` + +Plus a browser check on `/`: zero controller chunks requested. + +### The bigger problem: generated client modules + +The runtime is not where the weight is. On `/navigation`: + +- **490,212 bytes decoded** across 11 client modules, **21,026 transferred** — + a 23:1 compression ratio. +- The largest single module is **269,117 bytes** decoded, of which **89.8% is + duplicated lines**. +- One line appears **162 times**: `brand = context.state.brand; topLinks = +context.state.topLinks; ...` — the full state-restore prologue. + +Gzip hides this on the wire, but **parse and compile cost scales with decoded +bytes, not transferred bytes**. Half a megabyte of JavaScript is parsed to run +one page. + +**Cause.** `packages/compiler/src/client-codegen.ts:228-264` inlines the state +sync into _every peer-function alias, in every client function_. Each alias +emits `syncStateToContext` once and `syncStateFromContext` three times — the +catch path, the promise `finally`, and the synchronous path. The output is +O(functions x peers x state variables). With ~19 state variables and 81 peer +aliases in that module, that is several thousand generated assignments. + +**Change.** Hoist the sync out of the per-alias wrapper. The cheapest version +with no change to how bodies are written: emit **one** pair of closures per +client function and have every peer alias call them, instead of inlining the +sync per alias: + +```js +const __flush = () => { + context.state.brand = brand; /* ... */ +}; +const __restore = () => { + brand = context.state.brand; /* ... */ +}; +const __peer = + (name) => + (...args) => { + __flush(); + let r; + try { + r = context.functions[name](...args); + } catch (e) { + __restore(); + throw e; + } + if (r && typeof r.then === "function") return Promise.resolve(r).finally(__restore); + __restore(); + return r; + }; +const doThing = __peer("doThing"); +``` + +That removes the peer multiplier — the dominant factor — and takes the 81 +copies down to roughly one per function. It is a codegen change only, with no +change to semantics or to how anyone writes a component. + +A larger follow-up, if the first is not enough: keep state in a single object +and rewrite state identifiers in the body to reference it, which removes the +per-variable multiplier as well. That one needs the body transform and should +be measured before it is attempted. + +**How to test.** Pin decoded size, because gzip hides regressions here: + +```bash +bun run build +# assert the largest generated client module is under budget, DECODED not gzipped +``` + +Add the ratio itself as a signal: any module compressing better than about 10:1 +is duplicating itself and should fail the check. Existing behaviour is covered +by the current suite, so correctness is the 1,427 tests; this is purely a size +assertion on top. + ## 5. Order of work Ranked by return, not by size. The first item changes the cost of every item diff --git a/examples/inter-app-api-showcase/README.md b/examples/inter-app-api-showcase/README.md new file mode 100644 index 00000000..d9ceaf5a --- /dev/null +++ b/examples/inter-app-api-showcase/README.md @@ -0,0 +1,19 @@ +# Inter-app + external API showcase + +`GET /api/product-summary?sku=starter` demonstrates one request handler making: + +1. an RPC request to the `catalog` app (`getProduct`); +2. an external HTTPS request to GitHub's public REST API; and +3. an RPC request to the `audit` app (`recordLookup`). + +The handler deliberately forwards `as: ctx` only to WRNexus peer apps. The RPC package turns that into a short-lived subject/tenant token; it is never forwarded to GitHub. Each peer app must implement the same contract from `app/lib/contracts.ts` (normally a shared workspace package) under `app/services/`, and must authorize its own procedures. + +Before running this app, configure all three apps with the same private internal-origin map and a distinct, 32+ character RPC secret: + +```sh +WRNEXUS_RPC_SECRET=replace-with-a-private-32-character-minimum-secret +WRNEXUS_APP_NAME=product-summary +WRNEXUS_INTERNAL_ORIGINS={"catalog":"http://127.0.0.1:4101","audit":"http://127.0.0.1:4102"} +``` + +The peer app processes must remain private; the public gateway blocks the RPC route by design. Run with `bun run --cwd examples/inter-app-api-showcase dev`. diff --git a/examples/inter-app-api-showcase/app/api/product-summary.ts b/examples/inter-app-api-showcase/app/api/product-summary.ts new file mode 100644 index 00000000..6567754a --- /dev/null +++ b/examples/inter-app-api-showcase/app/api/product-summary.ts @@ -0,0 +1,60 @@ +import type { Context } from "@wrnexus/core"; +import { httpTransport, serviceClient } from "@wrnexus/rpc"; +import { auditService, catalogService } from "../lib/contracts.ts"; + +interface GitHubRepository { + full_name?: unknown; + stargazers_count?: unknown; +} + +function requestedSku(ctx: Context): string { + return new URL(ctx.req.url).searchParams.get("sku")?.trim() || "starter"; +} + +/** GET /api/product-summary?sku=starter */ +export async function GET(ctx: Context): Promise { + const sku = requestedSku(ctx); + const transport = httpTransport(); + + // Inter-app call #1: query the catalog app. `{ as: ctx }` forwards the + // signed subject/tenant context; catalog still authorizes independently. + const catalog = serviceClient(catalogService, { app: "catalog", as: ctx, transport }); + const product = await catalog.getProduct({ sku }); + + // External API call: GitHub's public repository endpoint. Do not send the + // user's RPC identity token or internal headers to external services. + let githubResponse: Response; + try { + githubResponse = await fetch("https://api.github.com/repos/octocat/Hello-World", { + headers: { accept: "application/vnd.github+json", "user-agent": "wrnexus-example" }, + signal: ctx.req.signal, + }); + } catch { + return Response.json({ error: "External repository lookup failed." }, { status: 502 }); + } + if (!githubResponse.ok) { + return Response.json({ error: "External repository lookup failed." }, { status: 502 }); + } + const github = (await githubResponse.json()) as GitHubRepository; + if (typeof github.full_name !== "string" || typeof github.stargazers_count !== "number") { + return Response.json( + { error: "External repository returned an unexpected response." }, + { status: 502 }, + ); + } + + // Inter-app call #2: record the completed lookup in the audit app. This is + // intentionally awaited: callers learn whether the audit record was saved. + const audit = serviceClient(auditService, { app: "audit", as: ctx, transport }); + const receipt = await audit.recordLookup({ + sku: product.sku, + repository: github.full_name, + stars: github.stargazers_count, + }); + + return Response.json({ + product, + external: { repository: github.full_name, stars: github.stargazers_count }, + audit: receipt, + }); +} diff --git a/examples/inter-app-api-showcase/app/example.test.ts b/examples/inter-app-api-showcase/app/example.test.ts new file mode 100644 index 00000000..971e7524 --- /dev/null +++ b/examples/inter-app-api-showcase/app/example.test.ts @@ -0,0 +1,11 @@ +import { expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +test("product summary composes one external request with two peer-app calls", () => { + const source = readFileSync(join(import.meta.dir, "api", "product-summary.ts"), "utf8"); + expect(source).toContain('serviceClient(catalogService, { app: "catalog", as: ctx, transport })'); + expect(source).toContain('serviceClient(auditService, { app: "audit", as: ctx, transport })'); + expect(source).toContain('fetch("https://api.github.com/repos/octocat/Hello-World"'); + expect(source).toContain("ctx.req.signal"); +}); diff --git a/examples/inter-app-api-showcase/app/lib/contracts.ts b/examples/inter-app-api-showcase/app/lib/contracts.ts new file mode 100644 index 00000000..9c608188 --- /dev/null +++ b/examples/inter-app-api-showcase/app/lib/contracts.ts @@ -0,0 +1,34 @@ +import { defineService, procedure } from "@wrnexus/rpc"; +import { v } from "@wrnexus/validation"; + +/** + * In a real multi-app workspace, put these contracts in a shared package and + * import that package from this app and each peer. They live together here so + * the example is self-contained. + */ +export const catalogService = defineService({ + name: "catalog", + procedures: { + getProduct: procedure + .input(v.object({ sku: v.string() })) + .output<{ sku: string; displayName: string; enabled: boolean }>() + .idempotent() + .build(), + }, +}); + +export const auditService = defineService({ + name: "audit", + procedures: { + recordLookup: procedure + .input( + v.object({ + sku: v.string(), + repository: v.string(), + stars: v.number(), + }), + ) + .output<{ eventId: string }>() + .build(), + }, +}); diff --git a/examples/inter-app-api-showcase/package.json b/examples/inter-app-api-showcase/package.json new file mode 100644 index 00000000..4b29d644 --- /dev/null +++ b/examples/inter-app-api-showcase/package.json @@ -0,0 +1,22 @@ +{ + "name": "inter-app-api-showcase", + "version": "0.8.6", + "private": true, + "type": "module", + "scripts": { + "dev": "bun run ../../packages/cli/src/index.ts dev .", + "build": "bun run ../../packages/cli/src/index.ts build .", + "test": "bun test", + "typecheck": "tsc --noEmit -p tsconfig.json", + "check": "bun run typecheck && bun run test && bun run build" + }, + "dependencies": { + "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*", + "@wrnexus/validation": "workspace:*" + }, + "devDependencies": { + "@types/bun": "^1.3.14", + "typescript": "^5.9.2" + } +} diff --git a/examples/inter-app-api-showcase/tsconfig.json b/examples/inter-app-api-showcase/tsconfig.json new file mode 100644 index 00000000..b077260d --- /dev/null +++ b/examples/inter-app-api-showcase/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { "lib": ["ESNext", "DOM", "DOM.Iterable"] }, + "include": ["app"] +} diff --git a/packages/pubsub/package.json b/packages/pubsub/package.json index db32a41a..f81e3202 100644 --- a/packages/pubsub/package.json +++ b/packages/pubsub/package.json @@ -4,6 +4,10 @@ "private": true, "type": "module", "main": "src/index.ts", + "dependencies": { + "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*" + }, "exports": { ".": "./src/index.ts", "./brokers": "./src/brokers.ts", diff --git a/packages/pubsub/src/index.ts b/packages/pubsub/src/index.ts index 0e3a742f..88410eef 100644 --- a/packages/pubsub/src/index.ts +++ b/packages/pubsub/src/index.ts @@ -87,3 +87,5 @@ export { createResilientPubSub, PresenceChannel } from "./resilient.ts"; export type { MessageEnvelope, ResilientPubSubOptions, PresenceMember } from "./resilient.ts"; export { natsDriver, kafkaDriver } from "./brokers.ts"; export type { NatsClient, KafkaClient } from "./brokers.ts"; +export { subjectPubSub } from "./subject.ts"; +export type { SubjectPubSub } from "./subject.ts"; diff --git a/packages/pubsub/src/subject.ts b/packages/pubsub/src/subject.ts new file mode 100644 index 00000000..1a9b4ab6 --- /dev/null +++ b/packages/pubsub/src/subject.ts @@ -0,0 +1,49 @@ +import type { Context } from "@wrnexus/core"; +import { exportSubjectContext, importSubjectContext, type SubjectContext } from "@wrnexus/rpc"; +import type { PubSub } from "./index.ts"; + +const PUBSUB_AUDIENCE = "wrnexus-pubsub"; + +interface SubjectEnvelope { + payload: T; + identity?: string; +} + +export interface SubjectPubSub { + publish(ctx: Context, topic: string, message: T): Promise; + subscribe( + pattern: string, + handler: (message: T, topic: string, subject?: SubjectContext) => void | Promise, + ): () => void; +} + +/** + * Authenticated pub/sub envelope. The token uses a fixed, purpose-specific + * audience; subscribers verify it before exposing the message to a handler. + */ +export function subjectPubSub(bus: PubSub): SubjectPubSub { + return { + async publish(ctx, topic, message) { + const identity = await exportSubjectContext(ctx, PUBSUB_AUDIENCE); + await bus.publish>(topic, { + payload: message, + ...(identity ? { identity } : {}), + }); + }, + subscribe(pattern, handler) { + return bus.subscribe>(pattern, async (envelope, topic) => { + if (!envelope || typeof envelope !== "object" || !("payload" in envelope)) return; + let subject: SubjectContext | undefined; + if (envelope.identity !== undefined) { + if (typeof envelope.identity !== "string") return; + try { + subject = await importSubjectContext(envelope.identity, PUBSUB_AUDIENCE); + } catch { + return; // Never downgrade a malformed claimed identity to anonymous. + } + } + await handler(envelope.payload as never, topic, subject); + }); + }, + }; +} diff --git a/packages/pubsub/test/subject.test.ts b/packages/pubsub/test/subject.test.ts new file mode 100644 index 00000000..916e7f39 --- /dev/null +++ b/packages/pubsub/test/subject.test.ts @@ -0,0 +1,24 @@ +import { afterEach, expect, test } from "bun:test"; +import { createPubSub, subjectPubSub } from "../src/index.ts"; + +const secret = process.env.WRNEXUS_RPC_SECRET; +const app = process.env.WRNEXUS_APP_NAME; +afterEach(() => { + if (secret === undefined) delete process.env.WRNEXUS_RPC_SECRET; + else process.env.WRNEXUS_RPC_SECRET = secret; + if (app === undefined) delete process.env.WRNEXUS_APP_NAME; + else process.env.WRNEXUS_APP_NAME = app; +}); + +test("subjectPubSub carries and verifies the publishing subject", async () => { + process.env.WRNEXUS_RPC_SECRET = "test-rpc-secret-at-least-32-chars-long"; + process.env.WRNEXUS_APP_NAME = "orders"; + const bus = subjectPubSub(createPubSub()); + let seen: string | undefined; + bus.subscribe<{ id: string }>("order:created", (message, _topic, subject) => { + expect(message.id).toBe("o1"); + seen = subject?.subjectId; + }); + await bus.publish({ user: { id: "u1" }, locals: {} } as never, "order:created", { id: "o1" }); + expect(seen).toBe("u1"); +}); diff --git a/packages/queue/package.json b/packages/queue/package.json index 76a8a4c6..542b09eb 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -8,6 +8,7 @@ ".": "./src/index.ts" }, "dependencies": { - "@wrnexus/core": "workspace:*" + "@wrnexus/core": "workspace:*", + "@wrnexus/rpc": "workspace:*" } } diff --git a/packages/queue/src/index.ts b/packages/queue/src/index.ts index 5e38a50d..e5c07482 100644 --- a/packages/queue/src/index.ts +++ b/packages/queue/src/index.ts @@ -314,4 +314,6 @@ export type { WorkflowStatus, WorkflowStore, } from "./workflow.ts"; +export { subjectQueue } from "./subject.ts"; +export type { SubjectJob, SubjectQueue } from "./subject.ts"; import { createExecutionContext, type ExecutionContext } from "@wrnexus/core"; diff --git a/packages/queue/src/subject.ts b/packages/queue/src/subject.ts new file mode 100644 index 00000000..e4f16a84 --- /dev/null +++ b/packages/queue/src/subject.ts @@ -0,0 +1,56 @@ +import type { Context } from "@wrnexus/core"; +import { exportSubjectContext, importSubjectContext, type SubjectContext } from "@wrnexus/rpc"; +import type { AddOptions, Job, Queue } from "./index.ts"; + +const QUEUE_AUDIENCE = "wrnexus-queue"; +interface SubjectEnvelope { + payload: T; + identity?: string; +} + +export interface SubjectJob extends Omit>, "data"> { + data: T; + subject?: SubjectContext; +} + +export interface SubjectQueue { + add( + ctx: Context, + name: string, + data: T, + options?: AddOptions, + ): Promise>>; + process( + name: string, + handler: (job: SubjectJob, context: { signal: AbortSignal }) => void | Promise, + ): void; +} + +/** Queue adapter that persists a signed end-user context alongside job data. */ +export function subjectQueue(queue: Queue): SubjectQueue { + return { + async add(ctx, name, data, options) { + const identity = await exportSubjectContext(ctx, QUEUE_AUDIENCE); + return queue.add(name, { payload: data, ...(identity ? { identity } : {}) }, options); + }, + process(name, handler) { + queue.process>(name, async (job, context) => { + const envelope = job.data; + if (!envelope || typeof envelope !== "object" || !("payload" in envelope)) return; + let subject: SubjectContext | undefined; + if (envelope.identity !== undefined) { + if (typeof envelope.identity !== "string") return; + try { + subject = await importSubjectContext(envelope.identity, QUEUE_AUDIENCE); + } catch { + return; + } + } + await handler( + { ...job, data: envelope.payload as never, ...(subject ? { subject } : {}) }, + context, + ); + }); + }, + }; +} diff --git a/packages/queue/test/subject.test.ts b/packages/queue/test/subject.test.ts new file mode 100644 index 00000000..445a8c77 --- /dev/null +++ b/packages/queue/test/subject.test.ts @@ -0,0 +1,26 @@ +import { afterEach, expect, test } from "bun:test"; +import { createQueue, subjectQueue } from "../src/index.ts"; + +const secret = process.env.WRNEXUS_RPC_SECRET; +const app = process.env.WRNEXUS_APP_NAME; +afterEach(() => { + if (secret === undefined) delete process.env.WRNEXUS_RPC_SECRET; + else process.env.WRNEXUS_RPC_SECRET = secret; + if (app === undefined) delete process.env.WRNEXUS_APP_NAME; + else process.env.WRNEXUS_APP_NAME = app; +}); + +test("subjectQueue supplies a verified subject to the worker", async () => { + process.env.WRNEXUS_RPC_SECRET = "test-rpc-secret-at-least-32-chars-long"; + process.env.WRNEXUS_APP_NAME = "orders"; + const queue = createQueue(); + const subjectAware = subjectQueue(queue); + let seen: string | undefined; + subjectAware.process<{ orderId: string }>("email", (job) => { + expect(job.data.orderId).toBe("o1"); + seen = job.subject?.subjectId; + }); + await subjectAware.add({ user: { id: "u1" }, locals: {} } as never, "email", { orderId: "o1" }); + await queue.drain(); + expect(seen).toBe("u1"); +}); diff --git a/packages/rpc/src/client.ts b/packages/rpc/src/client.ts index 246a4f67..39e82aff 100644 --- a/packages/rpc/src/client.ts +++ b/packages/rpc/src/client.ts @@ -66,7 +66,11 @@ export function serviceClient( const callPromise = options.transport.call( { app, service: contract.name, procedure: property }, input, - { signal: controller.signal, ...(identity ? { identity } : {}) }, + { + signal: controller.signal, + idempotent: contract.procedures[property as keyof Procedures].idempotent === true, + ...(identity ? { identity } : {}), + }, ); try { const result = await Promise.race([callPromise, timeout]); diff --git a/packages/rpc/src/index.ts b/packages/rpc/src/index.ts index 15d15c72..9190134a 100644 --- a/packages/rpc/src/index.ts +++ b/packages/rpc/src/index.ts @@ -31,8 +31,14 @@ export { } from "./identity.ts"; export type { ExportOptions, ImportOptions, SubjectContext } from "./identity.ts"; -export { inProcessTransport } from "./transport.ts"; -export type { CallOptions, InProcessHandler, RpcTarget, Transport } from "./transport.ts"; +export { inProcessTransport, retryingTransport } from "./transport.ts"; +export type { + CallOptions, + InProcessHandler, + RetryTransportOptions, + RpcTarget, + Transport, +} from "./transport.ts"; export { implement } from "./server.ts"; export type { HandlerContext, diff --git a/packages/rpc/src/transport.ts b/packages/rpc/src/transport.ts index cda9ef8d..d9f09ace 100644 --- a/packages/rpc/src/transport.ts +++ b/packages/rpc/src/transport.ts @@ -10,12 +10,96 @@ export interface RpcTarget { export interface CallOptions { signal?: AbortSignal; identity?: string; + /** Supplied from the declared procedure; only these calls may be retried. */ + idempotent?: boolean; } export interface Transport { call(target: RpcTarget, payload: unknown, options: CallOptions): Promise; } +export interface RetryTransportOptions { + /** Retries after the initial attempt. Default: 2. */ + retries?: number; + /** Initial exponential-backoff delay in milliseconds. Default: 50. */ + backoffMs?: number; + /** Consecutive retryable failures before the target circuit opens. Default: 3. */ + circuitFailureThreshold?: number; + /** How long an open circuit rejects calls before one probe is allowed. Default: 5s. */ + circuitCooldownMs?: number; + now?: () => number; + sleep?: (ms: number, signal?: AbortSignal) => Promise; +} + +function targetKey(target: RpcTarget): string { + return `${target.app}/${target.service}/${target.procedure}`; +} + +function defaultSleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + }); +} + +/** + * Add bounded retry and a per-procedure circuit breaker to any transport. + * The idempotency bit comes from the contract and is not caller-controlled. + */ +export function retryingTransport(base: Transport, options: RetryTransportOptions = {}): Transport { + const retries = options.retries ?? 2; + const backoffMs = options.backoffMs ?? 50; + const threshold = options.circuitFailureThreshold ?? 3; + const cooldownMs = options.circuitCooldownMs ?? 5_000; + const now = options.now ?? Date.now; + const sleep = options.sleep ?? defaultSleep; + if (!Number.isInteger(retries) || retries < 0) + throw new RangeError("rpc retries must be a non-negative integer"); + if (!Number.isFinite(backoffMs) || backoffMs < 0) + throw new RangeError("rpc backoffMs must be non-negative"); + if (!Number.isInteger(threshold) || threshold < 1) + throw new RangeError("rpc circuitFailureThreshold must be positive"); + if (!Number.isFinite(cooldownMs) || cooldownMs < 1) + throw new RangeError("rpc circuitCooldownMs must be positive"); + + const circuits = new Map(); + return { + async call(target, payload, callOptions) { + const key = targetKey(target); + const circuit = circuits.get(key); + if (circuit && circuit.openUntil > now()) { + return failure(RPC_ERROR_CODES.transport, "Service temporarily unavailable"); + } + if (circuit?.openUntil) circuits.delete(key); // cooldown: allow one probe + + const attempts = callOptions.idempotent ? retries + 1 : 1; + let result: ServiceResult = failure(RPC_ERROR_CODES.transport, "Service unreachable"); + for (let attempt = 0; attempt < attempts; attempt++) { + if (callOptions.signal?.aborted) return failure(RPC_ERROR_CODES.transport, "Call aborted"); + result = await base.call(target, payload, callOptions); + if (result.ok || !result.retryable) { + if (result.ok) circuits.delete(key); + return result; + } + if (attempt + 1 < attempts) await sleep(backoffMs * 2 ** attempt, callOptions.signal); + } + const failures = (circuits.get(key)?.failures ?? 0) + 1; + circuits.set(key, { + failures, + openUntil: failures >= threshold ? now() + cooldownMs : 0, + }); + return result; + }, + }; +} + export type InProcessHandler = ( payload: unknown, identity?: string, diff --git a/packages/rpc/test/retry.test.ts b/packages/rpc/test/retry.test.ts new file mode 100644 index 00000000..f679e5f3 --- /dev/null +++ b/packages/rpc/test/retry.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, test } from "bun:test"; +import { + RPC_ERROR_CODES, + failure, + retryingTransport, + success, + type Transport, +} from "../src/index.ts"; + +function failingTransport(): { transport: Transport; calls: () => number } { + let count = 0; + return { + transport: { + async call() { + count++; + return failure(RPC_ERROR_CODES.transport, "down"); + }, + }, + calls: () => count, + }; +} + +describe("retryingTransport", () => { + test("retries only declared idempotent calls", async () => { + const retryable = failingTransport(); + const write = failingTransport(); + const options = { retries: 2, backoffMs: 0 }; + await retryingTransport(retryable.transport, options).call( + { app: "billing", service: "invoice", procedure: "get" }, + {}, + { idempotent: true }, + ); + await retryingTransport(write.transport, options).call( + { app: "billing", service: "invoice", procedure: "create" }, + {}, + { idempotent: false }, + ); + expect(retryable.calls()).toBe(3); + expect(write.calls()).toBe(1); + }); + + test("opens a circuit after repeated exhausted failures and recovers after cooldown", async () => { + let clock = 0; + let calls = 0; + const base: Transport = { + async call() { + calls++; + return calls < 3 ? failure(RPC_ERROR_CODES.transport, "down") : success("ok"); + }, + }; + const transport = retryingTransport(base, { + retries: 0, + circuitFailureThreshold: 2, + circuitCooldownMs: 10, + now: () => clock, + }); + const target = { app: "billing", service: "invoice", procedure: "get" }; + await transport.call(target, {}, { idempotent: true }); + await transport.call(target, {}, { idempotent: true }); + expect(await transport.call(target, {}, { idempotent: true })).toMatchObject({ + ok: false, + code: RPC_ERROR_CODES.transport, + }); + expect(calls).toBe(2); + clock = 11; + expect(await transport.call(target, {}, { idempotent: true })).toEqual(success("ok")); + }); +});