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 <noreply@anthropic.com>

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",

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";

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<T> {
+  payload: T;
+  identity?: string;
+}
+
+export interface SubjectPubSub {
+  publish<T>(ctx: Context, topic: string, message: T): Promise<void>;
+  subscribe<T>(
+    pattern: string,
+    handler: (message: T, topic: string, subject?: SubjectContext) => void | Promise<void>,
+  ): () => 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<SubjectEnvelope<typeof message>>(topic, {
+        payload: message,
+        ...(identity ? { identity } : {}),
+      });
+    },
+    subscribe(pattern, handler) {
+      return bus.subscribe<SubjectEnvelope<unknown>>(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);
+      });
+    },
+  };
+}

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

packages/queue/package.json

@@ -8,6 +8,7 @@
     ".": "./src/index.ts"
   },
   "dependencies": {
-    "@wrnexus/core": "workspace:*"
+    "@wrnexus/core": "workspace:*",
+    "@wrnexus/rpc": "workspace:*"
   }
 }

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";

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<T> {
+  payload: T;
+  identity?: string;
+}
+
+export interface SubjectJob<T> extends Omit<Job<SubjectEnvelope<T>>, "data"> {
+  data: T;
+  subject?: SubjectContext;
+}
+
+export interface SubjectQueue {
+  add<T>(
+    ctx: Context,
+    name: string,
+    data: T,
+    options?: AddOptions,
+  ): Promise<Job<SubjectEnvelope<T>>>;
+  process<T>(
+    name: string,
+    handler: (job: SubjectJob<T>, context: { signal: AbortSignal }) => void | Promise<void>,
+  ): 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<SubjectEnvelope<unknown>>(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,
+        );
+      });
+    },
+  };
+}

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

packages/rpc/src/client.ts

@@ -66,7 +66,11 @@ export function serviceClient<Procedures extends AnyProcedures>(
         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]);

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,

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<ServiceResult>;
 }
 
+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<void>;
+}
+
+function targetKey(target: RpcTarget): string {
+  return `${target.app}/${target.service}/${target.procedure}`;
+}
+
+function defaultSleep(ms: number, signal?: AbortSignal): Promise<void> {
+  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<string, { failures: number; openUntil: number }>();
+  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,

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