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")); + }); +});