From 6a1058fe11db5ecebe3fe2fae18c8e504d9334c5 Mon Sep 17 00:00:00 2001 From: Ryuz Date: Sun, 27 Sep 2026 17:56:06 +0200 Subject: [PATCH 1/2] fix(als): fix webcontainer polyfill --- .github/workflows/ci.yml | 24 +- bun.lock | 2 +- package.json | 3 +- packages/oxidejs/CHANGELOG.md | 33 +- packages/oxidejs/README.md | 10 +- packages/oxidejs/package.json | 2 +- packages/oxidejs/src/actions.test.ts | 4 + packages/oxidejs/src/actions.ts | 26 +- packages/oxidejs/src/context.ts | 2 + packages/oxidejs/src/core.test.ts | 11 + packages/oxidejs/src/core.ts | 12 + packages/oxidejs/src/live-query.test.ts | 32 ++ packages/oxidejs/src/live-query.ts | 54 ++- packages/oxidejs/src/plugin.ts | 18 +- packages/oxidejs/src/queue-build.ts | 9 +- packages/oxidejs/src/queue.test.ts | 127 ++++++ packages/oxidejs/src/queue.ts | 82 +++- packages/oxidejs/src/request-store.ts | 123 +++-- packages/oxidejs/src/rpc/client.test.ts | 133 ++++++ packages/oxidejs/src/rpc/client.ts | 101 ++++- packages/oxidejs/src/rpc/scrub.ts | 10 + packages/oxidejs/src/rpc/server.test.ts | 200 +++++++++ packages/oxidejs/src/rpc/server.ts | 424 ++++++++++++++---- packages/oxidejs/src/rpc/ws.ts | 3 + packages/oxidejs/src/schedule-build.ts | 9 +- packages/oxidejs/src/schedule.test.ts | 2 + packages/oxidejs/src/schedule.ts | 6 +- packages/oxidejs/src/types.ts | 10 +- packages/oxidejs/src/workflow.ts | 5 +- packages/oxidejs/src/wrapper.test.ts | 14 + packages/oxidejs/test/celld-smoke/.gitignore | 2 + .../oxidejs/test/celld-smoke/smoke.server.ts | 38 ++ .../oxidejs/test/celld-smoke/wrangler.jsonc | 10 + .../celld-smoke/wrangler.nodejs-compat.jsonc | 7 + scripts/celld-smoke.ts | 261 +++++++++++ templates/kit/package.json | 2 +- templates/simple/package.json | 2 +- 37 files changed, 1639 insertions(+), 174 deletions(-) create mode 100644 packages/oxidejs/src/rpc/client.test.ts create mode 100644 packages/oxidejs/src/rpc/server.test.ts create mode 100644 packages/oxidejs/test/celld-smoke/.gitignore create mode 100644 packages/oxidejs/test/celld-smoke/smoke.server.ts create mode 100644 packages/oxidejs/test/celld-smoke/wrangler.jsonc create mode 100644 packages/oxidejs/test/celld-smoke/wrangler.nodejs-compat.jsonc create mode 100644 scripts/celld-smoke.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d9839c4..2f5f7a9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,12 +23,34 @@ jobs: bun-version: latest - run: bun install - run: bun run typecheck - - run: bun run lint + - run: bun run check - run: bun run build - run: bun test - run: bun run docs:build - run: bunx pkg-pr-new publish './packages/*' + # Bun's test runner cannot reproduce celld's request lifecycle (a finished + # or aborted request's pending work is dropped), which is where shared + # runtimes and cross-request gates hang. Boot a real celld node on the + # action server built from source. + celld-smoke: + runs-on: ubuntu-latest + env: + CELLD_VERSION: v0.6.0 + ESBUILD_VERSION: 0.25.10 + steps: + - uses: actions/checkout@v4 + - uses: oven-sh/setup-bun@v2 + with: + bun-version: latest + - run: bun install + - name: Install celld and esbuild + run: | + curl -fsSL https://celld.dev/install.sh | sh + npm install --global "esbuild@${ESBUILD_VERSION}" + echo "$HOME/.local/bin" >> "$GITHUB_PATH" + - run: bun run smoke:celld + docs: if: github.event_name == 'push' && github.ref == 'refs/heads/main' needs: ci diff --git a/bun.lock b/bun.lock index 6b96b4f..0265532 100644 --- a/bun.lock +++ b/bun.lock @@ -32,7 +32,7 @@ }, "packages/oxidejs": { "name": "oxidejs", - "version": "0.5.3", + "version": "0.5.6", "dependencies": { "effect": "~4.0.0-rc.112", "linkedom": "^0.18.13", diff --git a/package.json b/package.json index 26a0fc9..3d011aa 100644 --- a/package.json +++ b/package.json @@ -11,9 +11,8 @@ "test": "bun run --filter './packages/*' test", "build": "bun run --filter './packages/*' build", "release": "bun scripts/release.ts", + "smoke:celld": "bun scripts/celld-smoke.ts", "typecheck": "bun run --filter './packages/*' typecheck", - "lint": "oxlint", - "fmt": "oxfmt --write .", "docs:dev": "bun run --filter website dev", "docs:build": "bun run --filter website build", "docs:verify": "bun run --filter website verify", diff --git a/packages/oxidejs/CHANGELOG.md b/packages/oxidejs/CHANGELOG.md index 7486926..751d3c3 100644 --- a/packages/oxidejs/CHANGELOG.md +++ b/packages/oxidejs/CHANGELOG.md @@ -1,6 +1,37 @@ # Changelog -## 0.5.4 +## Unreleased + +### Security + +- Concurrent actions could read another request's context. The Effect RPC runtime was built once and cached, and its server fibers kept the request context of the request that started them, so under concurrency `useRequest()` / `useEnv()` / `getRequestStore()` inside an action could return a different request (its headers, cookies, env) — measured 29 of 30 concurrent requests on Bun. Every action call now runs in its own runtime on every host + +### Added + +- `actions.timeout` (milliseconds): an action that has not answered in time is interrupted and answers with a JSON-RPC `-32603` error. Each call of a batch has its own deadline, so one slow call no longer fails the others. A stream counts as answered at its first frame. `createActionHandler` / `createWsHooks` take it as `timeoutMs` +- The generated client rejects a call 5 s after the server deadline (`actions.timeout`), for a hung network or proxy. `createClient` takes `timeout`, `retries` (default 2) and `maxReconnects` (default 8): an HTTP POST that failed in transit (network error, 502, 503, 504) is resent only when every call in it carries an `idempotencyKey`, and a WebSocket stream stops reconnecting after `maxReconnects` +- `createActionHandler({ maxBodyBytes })` refuses a larger body with HTTP 413; the worker preset passes `bodyLimit`, which before only applied to Node and dev +- `liveQuery({ mutateWaitMs })` (default 10000) and `liveQuery(...).close()` +- `bun run smoke:celld` (and a CI job) boots a celld node on the action server built from source and checks concurrency, request context, batch timeouts and recovery + +### Changed + +- A batch answers each call as soon as it ends instead of after the slowest one +- `disposeActionHandler` is a no-op: nothing is cached to dispose +- The sync request-store fallback (a module-global slot) is used on WebContainer only. Workers and celld keep the store in AsyncLocalStorage across `await`; the slot could hand one request's context to a concurrent one +- A queue consumer retries only the messages whose workflow failed to start and acks the rest, instead of retrying the whole batch +- Failed producer-side workflow starts (`producerStart`) are logged instead of swallowed + +### Fixed + +- celld: concurrent actions hung forever and then wedged the isolate until restart. celld's `node:process` answers every unknown `versions` key with a stub function, so `process.versions.webcontainer` read as truthy and every action queued behind one cross-request gate; a request the host cancelled never released it. WebContainer detection now requires a version string +- Worker hosts (Cloudflare Workers, celld) stalled on the cached runtime: the host drops a finished request's pending work +- `withRequestEntry` no longer chains Worker requests behind each other; only WebContainer serializes entry +- Two `queue.send()` calls, a `sendBatch()`, or a `workflow.start()` in one action with an idempotency key all reused that key as their id, so every message after the first was taken for the first and its workflow never started. Ids are now `key`, `key:1`, `key:2`, … per call (stable across retries of the same RPC call) +- Queue consumers and cron handlers run inside a request store, like workflow `run()`: `useEnv()`, `queue.send()` and `workflow.start()` no longer throw `request context is unavailable` there +- A `liveQuery` mutation waits at most `mutateWaitMs` for the previous one on its topic; a mutation whose request the host dropped no longer blocks the topic for good + +## 0.5.5 ### Added diff --git a/packages/oxidejs/README.md b/packages/oxidejs/README.md index 66bf8ac..dbcddd4 100644 --- a/packages/oxidejs/README.md +++ b/packages/oxidejs/README.md @@ -160,7 +160,7 @@ Stream actions do not support `bind` / `with`. Breaking the `for await` loop or ### Live queries -Use `liveQuery` / `publish` for snapshot streams (query = subscription, mutation = publish). Hubs are isolate-local Effect PubSub — D1 (or your DB) stays the source of truth. Prefer Effect `Stream` + `mutateEffect`: +Use `liveQuery` / `publish` for snapshot streams (query = subscription, mutation = publish). Hubs are isolate-local Effect PubSub — a publish reaches subscribers in the same isolate only, not other Worker isolates or celld nodes — and D1 (or your DB) stays the source of truth. Mutations on one topic run one at a time; one waits at most `mutateWaitMs` (default 10000) for the previous. `close()` shuts a topic's hub down. Prefer Effect `Stream` + `mutateEffect`: ```ts import { Effect, Stream } from "effect"; @@ -258,7 +258,7 @@ await invoices.send({ orderId: "…" }); await invoices.sendBatch([{ body: { orderId: "…" } }]); ``` -Oxide merges `queues.producers` / `queues.consumers` via `mergeDurableBindings` and attaches a same-worker `queue` handler that starts the workflow from each message (via `createBatch` when available, otherwise duplicate-aware `create`/`get`). Cloudflare does not return message ids from `send`, so oxide wraps bodies in an envelope with a client-chosen id (`{ idempotencyKey }` / request header / UUID) and returns `{ id }` from `send` (and `{ ids }` from `sendBatch`) for `workflow.status` polling. Queue transport options (`contentType` / `delaySeconds`) travel in the RPC payload; `signal` / `idempotencyKey` stay on `CallOptions`. +Oxide merges `queues.producers` / `queues.consumers` via `mergeDurableBindings` and attaches a same-worker `queue` handler that starts the workflow from each message (via `createBatch` when available, otherwise duplicate-aware `create`/`get`). Cloudflare does not return message ids from `send`, so oxide wraps bodies in an envelope with a client-chosen id (`{ idempotencyKey }` / request header / UUID) and returns `{ id }` from `send` (and `{ ids }` from `sendBatch`) for `workflow.status` polling. Queue transport options (`contentType` / `delaySeconds`) travel in the RPC payload; `signal` / `idempotencyKey` stay on `CallOptions`. Every message gets its own id: with a key, the first send in a request uses it and later sends (and each message of a `sendBatch`) use `key:1`, `key:2`, …, so a retried request reproduces the same ids. The consumer acks the messages it started and retries only the ones that failed. Queue consumers and cron handlers run inside a request store, so `useEnv()`, `queue.send()` and `workflow.start()` work there. ```ts const { id } = await invoices.send({ orderId: "…" }); @@ -352,7 +352,7 @@ export class ActionRoom { The generated worker wrapper does not create a DO — hibernation is opt-in when you own the object. -`vite dev` and `rsbuild dev` serve the endpoint via middleware. `actions: "http"` (default) serves `/__oxide/action`; `actions: "ws"` uses a WebSocket instead (`crossws` on Node/`fetch`, `WebSocketPair` on `preset: "worker"`). `actions.sameOrigin` defaults to `true` for both transports; set it to `false` only when you intentionally accept cross-origin requests. Set `actions.path` to move the endpoint. Set `actions.openrpc: true` to serve `GET /__oxide/openrpc` — an [OpenRPC](https://spec.open-rpc.org/) 1.3 document for `action()` handlers only (Effect Schema → JSON Schema; workflows/queues/schedules are omitted). OpenRPC is HTTP-only and stays off when `transport` is `"ws"`. `actionHeaders` are static headers on the shared HTTP client and are ignored for WebSocket actions. +`vite dev` and `rsbuild dev` serve the endpoint via middleware. `actions: "http"` (default) serves `/__oxide/action`; `actions: "ws"` uses a WebSocket instead (`crossws` on Node/`fetch`, `WebSocketPair` on `preset: "worker"`). `actions.sameOrigin` defaults to `true` for both transports; set it to `false` only when you intentionally accept cross-origin requests. Set `actions.path` to move the endpoint. Set `actions.openrpc: true` to serve `GET /__oxide/openrpc` — an [OpenRPC](https://spec.open-rpc.org/) 1.3 document for `action()` handlers only (Effect Schema → JSON Schema; workflows/queues/schedules are omitted). OpenRPC is HTTP-only and stays off when `transport` is `"ws"`. `actionHeaders` are static headers on the shared HTTP client and are ignored for WebSocket actions. Set `actions.timeout` (milliseconds) to bound every action: an action that has not answered in time is interrupted, and the client gets a JSON-RPC `-32603` error instead of a request that never ends. Each call of a batch has its own deadline, and a batch answers each call as soon as it ends. A stream counts as answered at its first frame. The generated client gives up 5 s after the server deadline. `bodyLimit` also bounds action request bodies on the worker preset (HTTP 413). ## Rsbuild @@ -376,7 +376,7 @@ Same factory as Vite: client stubs, `/__oxide/action`, and `dist/server.js`. | `workerEntry` | `src/server.ts` | Relative to project root. Default path is skipped when missing (actions-only). Explicit path must exist. | | `outDir` | `dist` | Output root (Node / `"fetch"`) | | `clientDir` | `client` | Must stay inside `outDir` | -| `actions` | `"http"` | `"ws"` uses WebSocket (`crossws` on Node, `WebSocketPair` with `"worker"`); object form: `{ transport, path, sameOrigin, openrpc }` (`sameOrigin: true`, `openrpc: false`) | +| `actions` | `"http"` | `"ws"` uses WebSocket (`crossws` on Node, `WebSocketPair` with `"worker"`); object form: `{ transport, path, sameOrigin, openrpc, timeout }` (`sameOrigin: true`, `openrpc: false`, no `timeout`) | | `actionHeaders` | — | Static headers on the HTTP client | | `middleware` | `[]` | Fetch middleware. On WS: honor Responses (auth 302/401/…); ignore only `@ilha/router/ssr` document Responses. Otherwise Response short-circuits before actions / server entry. | | `plugins` | `[]` | Build plugins (`beforeBuild` / `afterBuild`). Pass objects or module IDs. | @@ -447,6 +447,8 @@ The generated `__asset` function uses `path.join` — not `path.resolve` — so - Body size capped at 1 MB by default (enforced on the actual body, not just `Content-Length`). - Batch requests capped at 20 items (both HTTP and WebSocket transports). - Effect `Defect` / `Cause` payloads are scrubbed before they leave the endpoint. Clients see plain JSON-RPC errors (`code` + `message` only). Thrown messages become `Internal error` (`-32603`). Unknown methods → `-32601`; invalid params → `-32602`. +- Every action call runs in its own Effect RPC runtime, disposed when its response ends. A shared runtime would let a concurrent action read another request's context, and on Worker hosts (Cloudflare Workers, celld) it stalls, because the host drops a finished request's pending work. +- Worker code must not share a promise, timer or runtime across requests unless it is registered with `waitUntil` or its waiters are time-bounded: on celld, work a request started stops when that request answers or its client goes away. `bun run smoke:celld` checks the action server against a real celld node. - `actions.sameOrigin` defaults to `true`. HTTP actions require `Origin` and/or `Sec-Fetch-Site` and reject cross-site callers. WebSocket upgrades additionally allow requests that omit both headers when `Host` matches the request URL host (celld and some proxies strip those headers; `localhost` / `127.0.0.1` / `::1` count as the same host). - StackBlitz WebContainers do not keep `AsyncLocalStorage` across `async/await`. Oxide detects `process.versions.webcontainer` and falls back to a sync request store, capturing context before Effect schedules work and serializing handler entry so concurrent requests do not stomp that store. Stream pulls re-enter the captured store. This is a demo/dev workaround, not a concurrency model for production. diff --git a/packages/oxidejs/package.json b/packages/oxidejs/package.json index 67a0304..b709998 100644 --- a/packages/oxidejs/package.json +++ b/packages/oxidejs/package.json @@ -1,6 +1,6 @@ { "name": "oxidejs", - "version": "0.5.5", + "version": "0.5.6", "description": "Vite/Rsbuild plugin. One build → dist/server.js + optional client. Server actions via *.server.ts.", "keywords": [ "cloudflare", diff --git a/packages/oxidejs/src/actions.test.ts b/packages/oxidejs/src/actions.test.ts index e13484d..f8855e2 100644 --- a/packages/oxidejs/src/actions.test.ts +++ b/packages/oxidejs/src/actions.test.ts @@ -371,6 +371,10 @@ describe("codegen", () => { generateClientModule("http", { authorization: "Bearer x" }) ).toContain('"authorization":"Bearer x"'); expect(generateClientModule("ws")).toContain('"transport":"ws"'); + // The client deadline trails the server's by a grace period. + expect( + generateClientModule("http", undefined, "/__oxide/action", 15_000) + ).toContain('"timeout":20000'); }); test("client stub peels { signal } and keeps other last args", async () => { diff --git a/packages/oxidejs/src/actions.ts b/packages/oxidejs/src/actions.ts index 75b68dd..0512c4c 100644 --- a/packages/oxidejs/src/actions.ts +++ b/packages/oxidejs/src/actions.ts @@ -270,16 +270,25 @@ export const assetRelPath = function assetRelPath( interface ClientModuleOpts { headers?: OxidejsActionHeaders; + timeout?: number; transport?: "ws"; url: string; } +/** Client grace over the server deadline: the server's own timeout error + * should win; the client bound catches a hung network or proxy. */ +const CLIENT_TIMEOUT_GRACE_MS = 5000; + export const generateClientModule = function generateClientModule( transport: OxidejsActionTransport = "http", headers?: OxidejsActionHeaders, - actionPath: string = ACTION_PATH + actionPath: string = ACTION_PATH, + serverTimeout?: number ): string { const opts: ClientModuleOpts = { url: actionPath }; + if (serverTimeout !== undefined) { + opts.timeout = serverTimeout + CLIENT_TIMEOUT_GRACE_MS; + } if (transport === "ws") { opts.transport = "ws"; } @@ -620,6 +629,7 @@ interface WorkerWrapperOpts { actionOpenRpc?: boolean; actionPath?: string; actionSameOrigin?: boolean; + actionTimeout?: number | undefined; actions?: OxidejsActionTransport; bodyLimit?: number; /** Worker preset (`preset: "worker"`). */ @@ -847,7 +857,9 @@ const buildActionImports = function buildActionImports( ws: boolean, actionPath: string, sameOrigin: boolean, - openrpc: boolean + openrpc: boolean, + timeout: number | undefined, + bodyLimit: number ): string { if (!hasActions) { return openrpc @@ -861,15 +873,17 @@ const __openRpcEntries = []; import { actionsOpenRpcEntries as __openRpcEntries } from ${JSON.stringify(VIRTUAL_ACTIONS_ID)}; ` : ""; + const timeoutOpt = + timeout === undefined ? "" : `, timeoutMs: ${JSON.stringify(timeout)}`; if (ws) { return `${openRpcImport}import { createWsHooks } from "oxidejs/rpc"; import { actionsGroup, actionsHandlers } from ${JSON.stringify(VIRTUAL_ACTIONS_ID)}; -const __ws = createWsHooks(actionsGroup, actionsHandlers, { path: ${JSON.stringify(actionPath)}, sameOrigin: ${sameOrigin} }); +const __ws = createWsHooks(actionsGroup, actionsHandlers, { path: ${JSON.stringify(actionPath)}, sameOrigin: ${sameOrigin}${timeoutOpt} }); `; } return `${openRpcImport}import { createActionHandler } from "oxidejs/rpc"; import { actionsGroup, actionsHandlers } from ${JSON.stringify(VIRTUAL_ACTIONS_ID)}; -const __rpc = createActionHandler(actionsGroup, actionsHandlers, { path: ${JSON.stringify(actionPath)}, sameOrigin: ${sameOrigin}, createContext: (req) => req[__fetch] ?? {} }); +const __rpc = createActionHandler(actionsGroup, actionsHandlers, { path: ${JSON.stringify(actionPath)}, sameOrigin: ${sameOrigin}${timeoutOpt}, maxBodyBytes: ${JSON.stringify(bodyLimit)}, createContext: (req) => req[__fetch] ?? {} }); `; }; @@ -1081,7 +1095,9 @@ export const generateWorkerWrapper = function generateWorkerWrapper( ws, actionPath, sameOrigin, - openrpc + openrpc, + opts.actionTimeout, + bodyLimit ); const actionMatchFn = `const __actionMatch = (p) => p === ${JSON.stringify(actionPath)} || p === ${JSON.stringify(`${actionPath}/`)};`; const openRpcGate = buildOpenRpcGate(openrpc, actionPath); diff --git a/packages/oxidejs/src/context.ts b/packages/oxidejs/src/context.ts index f4ad91b..5f5929a 100644 --- a/packages/oxidejs/src/context.ts +++ b/packages/oxidejs/src/context.ts @@ -12,6 +12,8 @@ export { __setNeedsSyncRequestStoreForTests, getRequestStore, inWebcontainer, + isWebcontainerVersions, + nextRequestScopedId, needsSyncRequestStore, peekRequestStore, runWithRequest, diff --git a/packages/oxidejs/src/core.test.ts b/packages/oxidejs/src/core.test.ts index 2c1f905..11489d7 100644 --- a/packages/oxidejs/src/core.test.ts +++ b/packages/oxidejs/src/core.test.ts @@ -85,6 +85,17 @@ describe("resolveOptions", () => { expect(resolved.actionSameOrigin).toBe(false); }); + test("resolves actions.timeout and rejects a non-positive one", () => { + expect( + resolveOptions({ actions: { timeout: 15_000 } }, process.cwd()) + .actionTimeout + ).toBe(15_000); + expect(resolveOptions({}, process.cwd()).actionTimeout).toBeUndefined(); + expect(() => + resolveOptions({ actions: { timeout: 0 } }, process.cwd()) + ).toThrow("actions.timeout must be a positive number"); + }); + test("resolves actions.openrpc", () => { expect( resolveOptions({ actions: { openrpc: true } }, process.cwd()) diff --git a/packages/oxidejs/src/core.ts b/packages/oxidejs/src/core.ts index ede01f2..58d990c 100644 --- a/packages/oxidejs/src/core.ts +++ b/packages/oxidejs/src/core.ts @@ -134,6 +134,7 @@ interface ResolvedActions { openrpc: boolean; path: string; sameOrigin: boolean; + timeout: number | undefined; transport: OxidejsActionTransport; } @@ -145,6 +146,7 @@ const resolveActions = function resolveActions( openrpc: false, path: ACTION_PATH, sameOrigin: true, + timeout: undefined, transport: "http", }; } @@ -153,6 +155,7 @@ const resolveActions = function resolveActions( openrpc: false, path: ACTION_PATH, sameOrigin: true, + timeout: undefined, transport: raw, }; } @@ -171,10 +174,17 @@ const resolveActions = function resolveActions( `oxidejs: actions.path must start with "/" and contain no query string (got "${actionsPath}")` ); } + const { timeout } = raw; + if (timeout !== undefined && !(Number.isFinite(timeout) && timeout > 0)) { + throw new Error( + `oxidejs: actions.timeout must be a positive number of milliseconds (got ${String(timeout)})` + ); + } return { openrpc: transport !== "ws" && raw.openrpc === true, path: actionsPath, sameOrigin: raw.sameOrigin ?? true, + timeout, transport, }; }; @@ -270,6 +280,7 @@ export const resolveOptions = function resolveOptions( path: actionPath, sameOrigin: actionSameOrigin, openrpc: actionOpenRpc, + timeout: actionTimeout, } = resolveActions(raw?.actions); const hasClient = hasHtmlEntry(rootAbs, config); const hasPublic = fs.existsSync(path.join(rootAbs, "public")); @@ -280,6 +291,7 @@ export const resolveOptions = function resolveOptions( actionOpenRpc, actionPath, actionSameOrigin, + actionTimeout, actions, bodyLimit: raw?.bodyLimit ?? 1_048_576, clientDir, diff --git a/packages/oxidejs/src/live-query.test.ts b/packages/oxidejs/src/live-query.test.ts index 8aa15fe..96bb38d 100644 --- a/packages/oxidejs/src/live-query.test.ts +++ b/packages/oxidejs/src/live-query.test.ts @@ -86,4 +86,36 @@ describe("liveQuery", () => { } expect(values).toEqual([4]); }); + + test("a mutation whose predecessor never releases still runs after mutateWaitMs", async () => { + const q = liveQuery({ + mutateWaitMs: 50, + topic: `orphan-${crypto.randomUUID()}`, + }); + // A mutation that never settles, like one whose request the host dropped. + void q.mutate(() => Effect.runPromise(Effect.never)); + const started = Date.now(); + const value = await q.mutate(() => Promise.resolve(7)); + expect(value).toBe(7); + expect(Date.now() - started).toBeLessThan(1000); + }); + + test("close() ends open streams and frees the topic", async () => { + const topic = `close-${crypto.randomUUID()}`; + const q = liveQuery({ topic }); + const collected = Effect.runPromise( + Stream.runCollect(q.stream().pipe(Stream.take(5))) + ); + q.publish(1); + await Bun.sleep(5); + q.close(); + const values = await collected; + expect([...values].length).toBeLessThanOrEqual(1); + // A new handle on the same topic gets a fresh hub. + const again = liveQuery({ topic }); + again.publish(2); + const iter = again.values()[Symbol.asyncIterator](); + expect(await iter.next()).toEqual({ done: false, value: 2 }); + await iter.return?.(); + }); }); diff --git a/packages/oxidejs/src/live-query.ts b/packages/oxidejs/src/live-query.ts index 4414c5a..1c83958 100644 --- a/packages/oxidejs/src/live-query.ts +++ b/packages/oxidejs/src/live-query.ts @@ -4,6 +4,7 @@ import * as Stream from "effect/Stream"; import { deferred } from "./deferred"; +const DEFAULT_MUTATE_WAIT_MS = 10_000; const HUBS_KEY = Symbol.for("oxidejs.liveQuery.hubs"); const GATES_KEY = Symbol.for("oxidejs.liveQuery.gates"); @@ -45,34 +46,68 @@ const hubFor = function hubFor( }; /** FIFO gate so concurrent mutators on one topic do not interleave. */ +/** + * Serialize mutations per topic, but never wait on the previous one longer + * than `waitMs`. The previous holder can belong to another request, and a + * Worker host (celld) drops a finished or aborted request's pending work, so + * its release may never run; an unbounded wait would then block the topic + * until the isolate restarts. + */ const withTopicGate = function withTopicGate( topic: string, + waitMs: number, body: Effect.Effect ): Effect.Effect { return Effect.acquireUseRelease( Effect.promise(async () => { const map = gates(); const prev = map[topic] ?? Promise.resolve(null); - const { promise, resolve } = deferred(); + const { promise, resolve: releaseGate } = deferred(); map[topic] = promise; - await prev; - return resolve; + let timer: ReturnType | undefined; + // oxlint-disable-next-line promise/avoid-new -- a timer-backed race needs its own Promise + const expired = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), waitMs); + }); + try { + await Promise.race([prev, expired]); + } finally { + clearTimeout(timer); + } + return () => { + releaseGate(null); + if (map[topic] === promise) { + Reflect.deleteProperty(map, topic); + } + }; }), () => body, - (resolve) => Effect.sync(() => resolve(null)) + (release) => Effect.sync(release) ); }; export interface LiveQueryOptions { /** Sliding buffer size. Default 16. */ capacity?: number; + /** + * Longest time a mutation waits for the previous mutation of its topic + * before it runs anyway. Default: 10000. + */ + mutateWaitMs?: number; /** How many recent values late subscribers replay. Default 1. */ replay?: number; /** Isolate-local topic key (`"tasks"`, `"room:42"`). */ topic: string; } +/** + * A live query is local to one isolate: `publish` reaches subscribers of the + * same process only, not other Worker isolates or other celld nodes. Use a + * Durable Object (one cell per topic) to fan out across a fleet. + */ export interface LiveQuery { + /** Shut the topic's hub down and forget it; open streams end. */ + close: () => void; /** * Serialize Effect work that produces a snapshot, then publish it. * Prefer this from Effect action handlers. @@ -110,6 +145,7 @@ export const liveQuery = function liveQuery( ): LiveQuery { const capacity = options.capacity ?? 16; const replay = options.replay ?? 1; + const mutateWaitMs = options.mutateWaitMs ?? DEFAULT_MUTATE_WAIT_MS; const { topic } = options; const hub = hubFor(topic, capacity, replay); @@ -122,6 +158,7 @@ export const liveQuery = function liveQuery( ): Effect.Effect { return withTopicGate( topic, + mutateWaitMs, Effect.gen(function* mutateEffectGen() { const value = yield* fn(); publishValue(value); @@ -169,7 +206,16 @@ export const liveQuery = function liveQuery( }; }; + const close = function close() { + const map = hubs(); + if (map[topic] === hub) { + Reflect.deleteProperty(map, topic); + } + Effect.runSync(PubSub.shutdown(hub)); + }; + return { + close, mutate, mutateEffect, publish: publishValue, diff --git a/packages/oxidejs/src/plugin.ts b/packages/oxidejs/src/plugin.ts index 5ad67fa..bc8eadc 100644 --- a/packages/oxidejs/src/plugin.ts +++ b/packages/oxidejs/src/plugin.ts @@ -198,6 +198,7 @@ const actionMiddleware = function actionMiddleware( loadRpc: () => Promise, actionPath: string, sameOrigin: boolean, + timeoutMs: number | undefined, bodyLimit: number, onError?: (error: Error) => void ) { @@ -215,6 +216,7 @@ const actionMiddleware = function actionMiddleware( { path: actionPath, sameOrigin, + timeoutMs, } ); const response = await handler(await nodeToWebRequest(req, bodyLimit)); @@ -278,7 +280,8 @@ const attachActionUpgrade = function attachActionUpgrade( loadRouter: () => Promise<{ actionsHandlers: unknown; default: unknown }>, loadRpc: () => Promise, actionPath: string, - sameOrigin: boolean + sameOrigin: boolean, + timeoutMs: number | undefined ) { if (!httpServer) { return; @@ -303,6 +306,7 @@ const attachActionUpgrade = function attachActionUpgrade( { path: actionPath, sameOrigin, + timeoutMs, } ); // SAFETY: Node upgrade socket is a Duplex; crossws handleUpgrade accepts the broader net.Socket shape. @@ -511,7 +515,8 @@ const loadVirtualClient = function loadVirtualClient( return generateClientModule( transport, resolved?.actionHeaders ?? options?.actionHeaders, - resolved?.actionPath + resolved?.actionPath, + resolved?.actionTimeout ); }; @@ -605,6 +610,7 @@ const loadVirtualWorker = function loadVirtualWorker( actionOpenRpc: resolved.actionOpenRpc, actionPath: resolved.actionPath, actionSameOrigin: resolved.actionSameOrigin, + actionTimeout: resolved.actionTimeout, actions: resolved.actions, bodyLimit: resolved.bodyLimit, clientDir: resolved.clientDir, @@ -946,7 +952,8 @@ export const unpluginFactory: UnpluginFactory = ( loadRouter, loadRpcModule, opts.actionPath, - opts.actionSameOrigin + opts.actionSameOrigin, + opts.actionTimeout ); } else { server.middlewares.use( @@ -955,6 +962,7 @@ export const unpluginFactory: UnpluginFactory = ( loadRpcModule, opts.actionPath, opts.actionSameOrigin, + opts.actionTimeout, opts.bodyLimit ) ); @@ -1090,6 +1098,7 @@ export const unpluginFactory: UnpluginFactory = ( loadRpc, opts.actionPath, opts.actionSameOrigin, + opts.actionTimeout, opts.bodyLimit, logActionError ) @@ -1102,7 +1111,8 @@ export const unpluginFactory: UnpluginFactory = ( loadRouter, loadRpc, opts.actionPath, - opts.actionSameOrigin + opts.actionSameOrigin, + opts.actionTimeout ); } diff --git a/packages/oxidejs/src/queue-build.ts b/packages/oxidejs/src/queue-build.ts index 503ae3b..18f86bc 100644 --- a/packages/oxidejs/src/queue-build.ts +++ b/packages/oxidejs/src/queue-build.ts @@ -317,7 +317,7 @@ export const generateQueueHandlerModule = function generateQueueHandlerModule( return `export async function handleQueue() {\n throw new Error("oxidejs: no queues registered");\n}\n`; } const lines = [ - `import { dispatchQueueBatch, readQueueMeta } from "oxidejs";`, + `import { dispatchQueueBatch, readQueueMeta, withRequestStore } from "oxidejs";`, ]; const entries: string[] = []; for (const [i, mod] of modules.entries()) { @@ -343,7 +343,12 @@ export const generateQueueHandlerModule = function generateQueueHandlerModule( ` if (!meta) {`, ` throw new Error("oxidejs: no queue handler for " + batch.queue);`, ` }`, - ` return dispatchQueueBatch(meta, batch, env, ctx);`, + // A request store like workflow run(): useEnv() and queue/workflow + // handles work in a consumer (and the default dispatch). + ` return withRequestStore(`, + ` { env, fetchCtx: ctx, req: new Request("https://oxide.local/queue") },`, + ` () => dispatchQueueBatch(meta, batch, env, ctx)`, + ` );`, `}` ); return `${lines.join("\n")}\n`; diff --git a/packages/oxidejs/src/queue.test.ts b/packages/oxidejs/src/queue.test.ts index 1606154..5d892f9 100644 --- a/packages/oxidejs/src/queue.test.ts +++ b/packages/oxidejs/src/queue.test.ts @@ -726,4 +726,131 @@ export const demos = queue({ name: "demos", workflow: demo, maxBatchSize: 10 })\ fs.rmSync(root, { force: true, recursive: true }); } }); + + test("every send and workflow start in one keyed request gets its own id", async () => { + const sent: { id: string }[] = []; + const created: string[] = []; + const env = { + DEMO: { + create: (opts?: { id?: string }) => { + created.push(opts?.id ?? ""); + return Promise.resolve({ id: opts?.id ?? "" }); + }, + get: (id: string) => Promise.resolve({ id }), + }, + DEMOS: { + send: (body: { id: string }) => { + sent.push(body); + return Promise.resolve(); + }, + sendBatch: (messages: Iterable<{ body: { id: string } }>) => { + for (const message of messages) { + sent.push(message.body); + } + return Promise.resolve(); + }, + }, + }; + const demo = workflow({ name: "demo", run: () => Promise.resolve() }); + const demos = queue({ name: "demos", workflow: demo }); + const ids = await withRequestStore( + { + // SAFETY: test fixture stubs Queue / Workflow binding methods. + env: env as never, + idempotencyKey: "rpc-1", + req: new Request("http://localhost/"), + }, + async () => { + const first = await demos.send({ n: 1 }); + const second = await demos.send({ n: 2 }); + const batch = await demos.sendBatch([ + { body: { n: 3 } }, + { body: { n: 4 } }, + ]); + const started = await demo.start({}); + return [first.id, second.id, ...batch.ids, started.id]; + } + ); + expect(ids).toEqual(["rpc-1", "rpc-1:1", "rpc-1:2", "rpc-1:3", "rpc-1:4"]); + expect(new Set(sent.map((body) => body.id)).size).toBe(4); + expect(created).toEqual(["rpc-1:4"]); + }); + + test("a keyed sendBatch derives one id per message", async () => { + const env = { + DEMOS: { + send: () => Promise.resolve(), + sendBatch: () => Promise.resolve(), + }, + }; + const demos = queue({ name: "demos", workflow: "demo" }); + const { ids } = await withRequestStore( + // SAFETY: test fixture stubs Queue binding methods. + { env: env as never, req: new Request("http://localhost/") }, + () => + demos.sendBatch([{ body: 1 }, { body: 2 }, { body: 3 }], { + idempotencyKey: "job", + }) + ); + expect(ids).toEqual(["job", "job:1", "job:2"]); + }); + + test("dispatchQueueBatch acks started messages and retries only the failed one", async () => { + const outcome: string[] = []; + const env = { + DEMO: { + create: (opts?: { id?: string }) => + opts?.id === "bad" + ? Promise.reject(new Error("boom")) + : Promise.resolve({ id: opts?.id ?? "" }), + get: (id: string) => Promise.resolve({ id }), + }, + }; + const message = (id: string) => ({ + ack: () => outcome.push(`ack:${id}`), + body: { id, oxide: "oxidejs.queue", payload: {} }, + id, + retry: () => outcome.push(`retry:${id}`), + timestamp: new Date(), + }); + await dispatchQueueBatch( + { + binding: "DEMOS", + name: "demos", + workflowBinding: "DEMO", + workflowName: "demo", + }, + { + ackAll: () => outcome.push("ackAll"), + messages: [message("a"), message("bad"), message("c")], + queue: "demos", + retryAll: () => outcome.push("retryAll"), + }, + env, + {} + ); + expect(outcome).toEqual(["ack:a", "retry:bad", "ack:c"]); + }); + + test("queue handler module runs dispatch inside a request store", () => { + // SAFETY: a minimal scanned-module fixture; the generator reads abs, key and exports only. + const code = generateQueueHandlerModule([ + { + abs: "/app/src/jobs.server.ts", + exports: [ + { + binding: "DEMOS", + exportName: "demos", + name: "demos", + workflowBinding: "DEMO", + workflowName: "demo", + workflowRef: "demo", + }, + ], + key: "jobs", + }, + ] as never); + expect(code).toContain("withRequestStore("); + expect(code).toContain("fetchCtx: ctx"); + }); }); diff --git a/packages/oxidejs/src/queue.ts b/packages/oxidejs/src/queue.ts index ff1e830..7e4994a 100644 --- a/packages/oxidejs/src/queue.ts +++ b/packages/oxidejs/src/queue.ts @@ -3,9 +3,10 @@ import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import type { CallOptions } from "./action"; -import { useEnv, useIdempotencyKey } from "./request-store"; +import { nextRequestScopedId, useEnv } from "./request-store"; import { SchemaDecodeError } from "./with-schema"; import { + createWorkflowInstance, createWorkflowInstanceBatch, toWorkflowBinding, WORKFLOW_META, @@ -331,16 +332,25 @@ export const makeQueueEnvelope = function makeQueueEnvelope

( return { id, oxide: OXIDE_QUEUE_ENVELOPE, payload }; }; +/** + * Message id: the per-message key, else the call's key (`key`, `key:1`, … by + * position in a batch), else a distinct id derived from the RPC request's + * key, else a random UUID. Ids become workflow instance ids, so two messages + * must never share one — the second would be taken for the first and dropped. + */ const nextMessageId = function nextMessageId( options: (CallOptions & QueueSendOptions) | undefined, + position: number, perMessage?: string ): string { - return ( - perMessage ?? - options?.idempotencyKey ?? - useIdempotencyKey() ?? - crypto.randomUUID() - ); + if (perMessage !== undefined) { + return perMessage; + } + const callKey = options?.idempotencyKey; + if (callKey !== undefined) { + return position === 0 ? callKey : `${callKey}:${position}`; + } + return nextRequestScopedId() ?? crypto.randomUUID(); }; interface WorkflowEnvBinding { @@ -456,7 +466,7 @@ export const queue = function queue

( // SAFETY: send always receives body as first arg after peel. const raw = args[0] as P; const decoded = decodePayload(meta.payload, raw); - const id = nextMessageId(options); + const id = nextMessageId(options, 0); const q = requireBinding(binding); const sendOpts = { contentType: "json", ...sendOptsFrom(options) }; await q.send(makeQueueEnvelope(id, decoded), sendOpts); @@ -467,8 +477,13 @@ export const queue = function queue

( [{ id, params: decoded }], sendOpts.delaySeconds ); - } catch { - // Enqueue already succeeded — do not fail the client / cause a resend. + } catch (error) { + // Enqueue already succeeded — do not fail the client / cause a + // resend. The queue consumer still starts the workflow. + console.warn( + `oxidejs: queue "${meta.name}" producer-side workflow start failed; the consumer will start it`, + error + ); } } return { id }; @@ -491,7 +506,7 @@ export const queue = function queue

( anyDelay = true; } for (const message of messages) { - const id = nextMessageId(options, message.idempotencyKey); + const id = nextMessageId(options, ids.length, message.idempotencyKey); ids.push(id); const payload = decodePayload(meta.payload, message.body); const entry: QueueMessageSendRequest = { @@ -512,8 +527,13 @@ export const queue = function queue

( if (meta.producerStart) { try { await startWorkflowFromProducer(meta, starts, anyDelay ? 1 : undefined); - } catch { - // Enqueue already succeeded — do not fail the client / cause a resend. + } catch (error) { + // Enqueue already succeeded — do not fail the client / cause a + // resend. The queue consumer still starts the workflow. + console.warn( + `oxidejs: queue "${meta.name}" producer-side workflow start failed; the consumer will start it`, + error + ); } } return { ids }; @@ -562,11 +582,37 @@ export const dispatchQueueBatch = async function dispatchQueueBatch( `oxidejs: workflow binding "${meta.workflowBinding}" is missing for queue "${meta.name}"` ); } - const starts: WorkflowCreateOptions[] = batch.messages.map((message) => { + const starts = batch.messages.map((message) => { const { id, payload } = readQueueEnvelope(message.body, message.id); - return { id, params: payload }; + return { message, start: { id, params: payload } }; }); - // Pass the binding through — extracting methods drops `this` on celld/workerd. - // SAFETY: create/get narrowed above; keep the host binding as receiver. - await createWorkflowInstanceBatch(workflow as WorkflowBinding, starts); + try { + // SAFETY: create/get narrowed above; keep the host binding as receiver. + await createWorkflowInstanceBatch( + workflow as WorkflowBinding, + starts.map(({ start }) => start) + ); + return; + } catch (error) { + console.warn( + `oxidejs: queue "${meta.name}" batch start failed; retrying per message`, + error + ); + } + // Per message, so one bad message does not send the whole batch back: + // started ones are acked, only the failures are retried. + for (const { message, start } of starts) { + try { + // SAFETY: create/get narrowed above; keep the host binding as receiver. + // oxlint-disable-next-line eslint/no-await-in-loop -- sequential starts keep message order + await createWorkflowInstance(workflow as WorkflowBinding, start); + message.ack(); + } catch (error) { + console.warn( + `oxidejs: queue "${meta.name}" message ${start.id} failed to start its workflow; retrying`, + error + ); + message.retry(); + } + } }; diff --git a/packages/oxidejs/src/request-store.ts b/packages/oxidejs/src/request-store.ts index 066a1c2..0c9d04d 100644 --- a/packages/oxidejs/src/request-store.ts +++ b/packages/oxidejs/src/request-store.ts @@ -37,6 +37,24 @@ let webcontainerOverride: boolean | null = null; /** Override for tests. `null` uses runtime detection. */ let syncRequestStoreOverride: boolean | null = null; +/** `process.versions` as a host may shape it: a real WebContainer sets a string. */ +export interface WebcontainerVersions { + readonly [key: string]: string | (() => void) | undefined; +} + +/** + * True only for a real WebContainer version string. celld's `node:process` + * answers every unknown `versions` key with a stub function, so a truthiness + * check reads celld as a WebContainer and serializes every action behind one + * cross-request gate. + */ +export const isWebcontainerVersions = function isWebcontainerVersions( + versions: WebcontainerVersions +): boolean { + // oxlint-disable-next-line anti-slop/no-runtime-typeof -- host probe: celld answers unknown keys with a function + return typeof versions["webcontainer"] === "string"; +}; + /** StackBlitz WebContainers lose AsyncLocalStorage across `async/await`. */ export const inWebcontainer = function inWebcontainer(): boolean { if (webcontainerOverride !== null) { @@ -45,31 +63,39 @@ export const inWebcontainer = function inWebcontainer(): boolean { if (process === undefined) { return false; } - // SAFETY: Node's process.versions is a string bag; WebContainer adds an optional key. - const versions = process.versions as NodeJS.ProcessVersions & { - webcontainer?: string; - }; - return Boolean(versions.webcontainer); + return isWebcontainerVersions(process.versions); }; /** * Runtimes where request context must survive on a sync fallback because - * AsyncLocalStorage does not keep the store across `await` (WebContainer, - * Cloudflare Workers). + * AsyncLocalStorage does not keep the store across `await`: WebContainer + * only. Workers and celld need `nodejs_compat` (or `nodejs_als`) to import + * `node:async_hooks` at all, and there AsyncLocalStorage keeps the store + * across `await` (measured on celld 0.6.0). A module-global fallback on those + * hosts would hand one request's context to a concurrent request whenever + * the ALS store is empty. */ export const needsSyncRequestStore = function needsSyncRequestStore(): boolean { if (syncRequestStoreOverride !== null) { return syncRequestStoreOverride; } - if (inWebcontainer()) { - return true; - } - // Cloudflare Workers expose WebSocketPair; Node and Bun do not. - // SAFETY: Workers add WebSocketPair on globalThis; missing means a non-Worker host. - const workerApi = globalThis as typeof globalThis & { - WebSocketPair?: object; - }; - return workerApi.WebSocketPair !== undefined; + return inWebcontainer(); +}; + +/** Module fallback when ALS does not survive awaits (WebContainer). */ +let syncStore: ActionContext | null = null; + +/** Serialize handler *entry* on WebContainer so syncStore is not stomped. */ +let entryTail: Promise = Promise.resolve(null); + +/** + * A test that switches the host mode starts from an empty fallback: an + * earlier test's action that never settled (an open stream) would otherwise + * leave its context in the module slot. + */ +const resetSyncFallbackForTests = function resetSyncFallbackForTests(): void { + syncStore = null; + entryTail = Promise.resolve(null); }; /** Test-only: force or clear the WebContainer detection path. */ @@ -77,20 +103,16 @@ export const __setInWebcontainerForTests = function __setInWebcontainerForTests( value: boolean | null ): void { webcontainerOverride = value; + resetSyncFallbackForTests(); }; /** Test-only: force or clear the sync request-store fallback path. */ export const __setNeedsSyncRequestStoreForTests = function __setNeedsSyncRequestStoreForTests(value: boolean | null): void { syncRequestStoreOverride = value; + resetSyncFallbackForTests(); }; -/** Module fallback when ALS does not survive awaits (WebContainer). */ -let syncStore: ActionContext | null = null; - -/** Serialize handler *entry* on WebContainer so syncStore is not stomped. */ -let entryTail: Promise = Promise.resolve(null); - const als = function als(): AsyncLocalStorage { // SAFETY: ALS_KEY is oxide-owned; the slot is only ever AsyncLocalStorage. const g = globalThis as typeof globalThis & { @@ -137,7 +159,12 @@ export const enterRequestStore = function enterRequestStore( ctx: ActionContext ): ActionContext | null { const previous = syncStore; - syncStore = ctx; + // Only where ALS cannot carry the store: a module-global slot elsewhere + // could hand this context to a concurrent request, and an interrupted + // action that never reaches its exit would leave it set for good. + if (needsSyncRequestStore()) { + syncStore = ctx; + } return previous; }; @@ -151,6 +178,11 @@ export const exitRequestStore = function exitRequestStore( } }; +/** The sync fallback store, only on hosts that need it. */ +const syncFallback = function syncFallback(): ActionContext | null { + return needsSyncRequestStore() ? syncStore : null; +}; + /** Run `fn` under ALS for `ctx` (does not touch the sync fallback). */ export const runWithAls = function runWithAls( ctx: ActionContext, @@ -161,7 +193,7 @@ export const runWithAls = function runWithAls( /** Current request context: ALS first, then the WebContainer sync fallback. */ export const getRequestStore = function getRequestStore(): ActionContext { - const current = als().getStore() ?? syncStore; + const current = als().getStore() ?? syncFallback(); if (!current) { throw new Error("oxidejs: request context is unavailable"); } @@ -172,7 +204,7 @@ export const getRequestStore = function getRequestStore(): ActionContext { export const peekRequestStore = function peekRequestStore(): | ActionContext | undefined { - return als().getStore() ?? syncStore ?? undefined; + return als().getStore() ?? syncFallback() ?? undefined; }; const store = function store(): ActionContext { @@ -190,6 +222,9 @@ export const withRequestStore = function withRequestStore( ctx: ActionContext, fn: () => T ): T { + if (!needsSyncRequestStore()) { + return als().run(ctx, fn); + } const previous = syncStore; syncStore = ctx; let deferRestore = false; @@ -221,24 +256,23 @@ export const withRequestStore = function withRequestStore( }; /** - * Serialize request entry when the sync fallback is required. - * WebContainer: hold the gate until `fn` settles. - * Workers / celld: only serialize starts so long-lived streams do not block other actions. + * Serialize request entry on WebContainer: hold the gate until `fn` settles so + * the sync store is not stomped. Every other host runs `fn` directly. On a + * Worker host the gate would chain each request to the one before it, and a + * request the host cancels mid-wait (celld drops a finished or aborted + * request's pending work) never releases its link, which stalls every action + * after it. */ export const withRequestEntry = async function withRequestEntry( fn: () => Promise ): Promise { - if (!needsSyncRequestStore()) { + if (!inWebcontainer()) { return fn(); } const { promise: gate, resolve: release } = deferred(); const previous = entryTail; entryTail = gate; await previous; - if (!inWebcontainer()) { - release(null); - return await fn(); - } try { return await fn(); } finally { @@ -274,6 +308,29 @@ export const useFetchCtx = function useFetchCtx(): return store().fetchCtx; }; +const derivedIdCounts = new WeakMap(); + +/** + * A distinct, deterministic id for the next queue message or workflow start + * of the current keyed request, or `undefined` outside a request or without a + * key. The first call keeps the key; later calls get `key:1`, `key:2`, … So a + * retried RPC call (same key, same code path) reproduces the same ids, while + * two sends in one action no longer share one id and collapse into one + * workflow instance. + */ +export const nextRequestScopedId = function nextRequestScopedId(): + | string + | undefined { + const current = peekRequestStore(); + const key = current?.idempotencyKey; + if (!(current && key)) { + return undefined; + } + const count = derivedIdCounts.get(current) ?? 0; + derivedIdCounts.set(current, count + 1); + return count === 0 ? key : `${key}:${count}`; +}; + /** Idempotency key from the client `CallOptions` / RPC headers, if present. */ export const useIdempotencyKey = function useIdempotencyKey(): | string diff --git a/packages/oxidejs/src/rpc/client.test.ts b/packages/oxidejs/src/rpc/client.test.ts new file mode 100644 index 0000000..833b3a0 --- /dev/null +++ b/packages/oxidejs/src/rpc/client.test.ts @@ -0,0 +1,133 @@ +import { afterAll, beforeAll, expect, test } from "bun:test"; +import fs from "node:fs"; +import path from "node:path"; + +import type { NestedClient, RpcClientOptions } from "./client"; +import { createClient } from "./client"; +import { createActionHandler } from "./server"; +import { writeGeneratedActions } from "./test-harness"; + +const ACTION_PATH = "/__oxide/action"; + +interface Harness { + /** Answer the next `count` POSTs with this status instead of running them. */ + failNext: (count: number, status: number) => void; + posts: () => number; + url: string; +} + +let root = ""; +let group: never; +let handlers: never; + +beforeAll(async () => { + root = fs.mkdtempSync(path.join(import.meta.dir, "oxide-client-")); + const ctx = JSON.stringify(path.join(import.meta.dir, "../context.ts")); + fs.writeFileSync( + path.join(root, "demo.server.ts"), + `import { action } from ${ctx}; +export const ping = action(async () => "pong"); +export const stall = action(async () => new Promise(() => {})); +` + ); + // SAFETY: the generated actions module exports the RpcGroup and its handler layer. + const mod = (await import(writeGeneratedActions(root))) as { + actionsHandlers: never; + default: never; + }; + group = mod.default; + handlers = mod.actionsHandlers; +}); + +afterAll(() => { + fs.rmSync(root, { force: true, recursive: true }); +}); + +const withServer = async function withServer( + run: (harness: Harness) => Promise +) { + const handler = createActionHandler(group, handlers, { + path: ACTION_PATH, + sameOrigin: false, + }); + let posts = 0; + let failures = 0; + let failStatus = 503; + const server = Bun.serve({ + fetch: (request) => { + posts += 1; + if (failures > 0) { + failures -= 1; + return new Response("upstream unavailable", { status: failStatus }); + } + return handler(request); + }, + port: 0, + }); + try { + await run({ + failNext: (count, status) => { + failures = count; + failStatus = status; + }, + posts: () => posts, + url: `http://127.0.0.1:${server.port}${ACTION_PATH}`, + }); + } finally { + await server.stop(true); + } +}; + +type Call = (...args: { idempotencyKey: string }[]) => Promise; + +const demo = function demo(client: NestedClient, name: string): Call { + const fn = client["demo"]?.[name]; + if (!fn) { + throw new Error(`expected demo.${name} on the client`); + } + // SAFETY: unary demo actions resolve to a Promise. + return fn as Call; +}; + +const clientFor = (options: RpcClientOptions) => createClient(group, options); + +test("a unary call past `timeout` rejects instead of waiting forever", async () => { + await withServer(async ({ url }) => { + const client = clientFor({ timeout: 100, url }); + const started = Date.now(); + await expect(demo(client, "stall")()).rejects.toThrow( + "demo.stall timed out after 100ms" + ); + expect(Date.now() - started).toBeLessThan(2000); + }); +}); + +test("an idempotent call is resent after a 503", async () => { + await withServer(async ({ failNext, posts, url }) => { + const client = clientFor({ retries: 2, url }); + failNext(1, 503); + const result = await demo(client, "ping")({ idempotencyKey: "k-1" }); + expect(result).toBe("pong"); + expect(posts()).toBe(2); + }); +}); + +test("a call without an idempotency key is not resent", async () => { + await withServer(async ({ failNext, posts, url }) => { + const client = clientFor({ retries: 2, url: `${url}?no-key` }); + failNext(1, 503); + await expect(demo(client, "ping")()).rejects.toThrow(); + expect(posts()).toBe(1); + }); +}); + +test("retries stop after `retries` attempts", async () => { + await withServer(async ({ failNext, posts, url }) => { + const client = clientFor({ retries: 1, url: `${url}?cap` }); + failNext(5, 502); + await expect( + demo(client, "ping")({ idempotencyKey: "k-2" }) + ).rejects.toThrow(); + expect(posts()).toBe(2); + }); +}); diff --git a/packages/oxidejs/src/rpc/client.ts b/packages/oxidejs/src/rpc/client.ts index 6cd2f83..6313784 100644 --- a/packages/oxidejs/src/rpc/client.ts +++ b/packages/oxidejs/src/rpc/client.ts @@ -21,10 +21,41 @@ import { streamToAsyncGen } from "./stream"; export interface RpcClientOptions { headers?: OxidejsActionHeaders; + /** + * WebSocket reconnect attempts for a stream action before it fails. + * Default: 8. + */ + maxReconnects?: number; + /** + * Retries for an HTTP POST that failed in transit (network error, 502, + * 503, 504). Only a POST whose every call carries an `idempotencyKey` is + * retried, because the server may have run the others. Default: 2. + */ + retries?: number; + /** + * Reject a unary call that has not answered in this many milliseconds. + * Default: no limit. + */ + timeout?: number; transport?: "http" | "ws"; url: string; } +const IDEMPOTENCY_HEADER = "x-oxide-idempotency-key"; +const DEFAULT_RETRIES = 2; +const DEFAULT_MAX_RECONNECTS = 8; +const RETRY_BASE_MS = 100; +const RETRYABLE_STATUS = new Set([502, 503, 504]); + +/** True when the request carries an idempotency key, so a resend is safe. */ +const isIdempotentRequest = function isIdempotentRequest( + request: RpcMessage.RequestEncoded +): boolean { + return request.headers.some( + ([key]) => key.toLowerCase() === IDEMPOTENCY_HEADER + ); +}; + interface CallOptions { idempotencyKey?: string; signal?: AbortSignal; @@ -130,7 +161,8 @@ const cacheKey = function cacheKey( groupId = nextGroupId; stamped.__oxideClientId = groupId; } - return `${groupId}|${options.transport ?? "http"}|${options.url}|${headerKey}`; + const resilience = `${options.timeout ?? ""},${options.retries ?? ""},${options.maxReconnects ?? ""}`; + return `${groupId}|${options.transport ?? "http"}|${options.url}|${headerKey}|${resilience}`; }; interface QueuedRequest { @@ -236,7 +268,24 @@ const makeBatchProtocol = function makeBatchProtocol( return clientId; }; - const executed = yield* Effect.exit(http.execute(request)); + // Resend only when every call is idempotent: a POST that failed in + // transit may still have run on the server. + const retries = entries.every((entry) => + isIdempotentRequest(entry.request) + ) + ? (options.retries ?? DEFAULT_RETRIES) + : 0; + let executed = yield* Effect.exit(http.execute(request)); + for (let attempt = 0; attempt < retries; attempt += 1) { + const transient = + executed._tag === "Failure" || + RETRYABLE_STATUS.has(executed.value.status); + if (!transient) { + break; + } + yield* Effect.sleep(`${RETRY_BASE_MS * 2 ** attempt} millis`); + executed = yield* Effect.exit(http.execute(request)); + } if (executed._tag === "Failure") { return yield* failQueued( entries, @@ -470,13 +519,12 @@ const sleep = function sleep(ms: number) { return Effect.runPromise(Effect.sleep(`${ms} millis`)); }; -const IDEMPOTENCY_HEADER = "x-oxide-idempotency-key"; - const callFlat = function callFlat( client: FlatClient, tag: string, args: ActionCallArg[], - callOpts?: CallOptions + callOpts?: CallOptions, + timeoutMs?: number ) { const caller = client[tag]; if (!caller) { @@ -491,16 +539,26 @@ const callFlat = function callFlat( return streamToAsyncGenerator(result, callOpts?.signal); } // SAFETY: non-stream RpcCaller results are Effects runnable via Effect.runPromise. - return Effect.runPromise(result as Effect.Effect, { - signal: callOpts?.signal, - }); + const unary = result as Effect.Effect; + const bounded = + timeoutMs === undefined + ? unary + : Effect.timeoutOrElse(unary, { + duration: `${timeoutMs} millis`, + orElse: () => + Effect.fail( + new Error(`Action ${tag} timed out after ${timeoutMs}ms`) + ), + }); + return Effect.runPromise(bounded, { signal: callOpts?.signal }); }; const callFlatStreamResilient = function callFlatStreamResilient( client: FlatClient, tag: string, args: ActionCallArg[], - callOpts?: CallOptions + callOpts: CallOptions | undefined, + maxReconnects: number ) { return (async function* resilientStream(): AsyncGenerator { let attempt = 0; @@ -519,7 +577,11 @@ const callFlatStreamResilient = function callFlatStreamResilient( return; } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); - if (isAbortError(err, callOpts?.signal) || !isTransientWsClose(err)) { + if ( + isAbortError(err, callOpts?.signal) || + !isTransientWsClose(err) || + attempt >= maxReconnects + ) { throw error; } attempt += 1; @@ -563,7 +625,12 @@ const isCallOptions = function isCallOptions( return "signal" in bag || "idempotencyKey" in bag; }; -const nestClient = function nestClient(group: ActionGroup, flat: FlatClient) { +const nestClient = function nestClient( + group: ActionGroup, + flat: FlatClient, + options: RpcClientOptions +) { + const maxReconnects = options.maxReconnects ?? DEFAULT_MAX_RECONNECTS; const nested: NestedClient = {}; for (const tag of group.requests.keys()) { const dot = tag.indexOf("."); @@ -579,9 +646,15 @@ const nestClient = function nestClient(group: ActionGroup, flat: FlatClient) { const params = hasSignal ? args.slice(0, -1) : args; const callOpts = hasSignal ? opts : undefined; if (isStreamTag(group, tag)) { - return callFlatStreamResilient(flat, tag, params, callOpts); + return callFlatStreamResilient( + flat, + tag, + params, + callOpts, + maxReconnects + ); } - return callFlat(flat, tag, params, callOpts); + return callFlat(flat, tag, params, callOpts, options.timeout); }; } return nested; @@ -598,7 +671,7 @@ export const createClient = function createClient( try { const flat = await Effect.runPromise(loadClient(group, options)); // SAFETY: RpcClient.make yields a tag-keyed service; never bridges Effect's generated client to FlatClient. - return nestClient(group, flat as never); + return nestClient(group, flat as never, options); } catch (error) { clientCache.delete(key); throw error; diff --git a/packages/oxidejs/src/rpc/scrub.ts b/packages/oxidejs/src/rpc/scrub.ts index 4760913..b840241 100644 --- a/packages/oxidejs/src/rpc/scrub.ts +++ b/packages/oxidejs/src/rpc/scrub.ts @@ -15,6 +15,8 @@ interface JsonObject { } const INTERNAL: JsonRpcErrorBody = { code: -32_603, message: "Internal error" }; +/** Cap on one logged defect, so a huge cause tree cannot flood the log. */ +const MAX_LOGGED_DEFECT_CHARS = 4000; const NDJSON_CONTENT = "application/json-rpc"; /** Application error code for Schema-tagged Fail values (not JSON-RPC parse/params). */ const APPLICATION_ERROR = -32_000; @@ -276,6 +278,14 @@ export const scrubRpcMessage = function scrubRpcMessage( } const scrubbedError = scrubError(msg["error"]); + if (scrubbedError.code === INTERNAL.code) { + // The client only sees "Internal error"; keep the cause in the server log, + // or a Worker-side defect leaves no trace at all. + console.error( + "oxidejs: action failed with an internal error:", + JSON.stringify(msg["error"]).slice(0, MAX_LOGGED_DEFECT_CHARS) + ); + } let { id } = msg; if (id === -32_603) { const next = state.remaining.values().next(); diff --git a/packages/oxidejs/src/rpc/server.test.ts b/packages/oxidejs/src/rpc/server.test.ts new file mode 100644 index 0000000..48c5949 --- /dev/null +++ b/packages/oxidejs/src/rpc/server.test.ts @@ -0,0 +1,200 @@ +/* eslint-disable func-names -- Effect.gen uses anonymous generators (AGENTS.md) */ +import { expect, test } from "bun:test"; + +import { Effect, Layer, Schema } from "effect"; +import { Rpc, RpcGroup } from "effect/unstable/rpc"; + +import { getRequestStore } from "../context"; +import { isWebcontainerVersions } from "../request-store"; +import { createActionHandler } from "./server"; + +interface Probe { + builds: number; + finalized: number; + interrupted: number; +} + +const makeProbe = (): Probe => ({ builds: 0, finalized: 0, interrupted: 0 }); + +const makeGroup = () => + RpcGroup.make( + Rpc.make("ping", { success: Schema.String }), + Rpc.make("slow", { success: Schema.String }), + Rpc.make("whoami", { success: Schema.String }) + ); + +/** Handlers whose layer counts builds and teardowns. `slow` outlives any test deadline. */ +const makeHandlers = function makeHandlers( + group: ReturnType, + probe: Probe, + slowFor: `${number} millis` = "10000 millis" +) { + const handlers = group.toLayer({ + ping: () => Effect.succeed("pong"), + slow: () => + Effect.sleep(slowFor).pipe( + Effect.as("late"), + Effect.onInterrupt(() => + Effect.sync(() => { + probe.interrupted += 1; + }) + ) + ), + // Reads the request store the way generated handlers do, then yields so + // concurrent calls interleave. + whoami: () => { + const store = getRequestStore(); + return Effect.promise(async () => { + await Bun.sleep(Math.random() * 10); + return store.req.headers.get("x-user") ?? "none"; + }); + }, + }); + const counted = Layer.effectDiscard( + Effect.gen(function* () { + probe.builds += 1; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + probe.finalized += 1; + }) + ); + }) + ); + return Layer.merge(handlers, counted); +}; + +const post = function post(body: string, headers: HeadersInit = {}) { + return new Request("http://localhost/__oxide/action", { + body, + headers, + method: "POST", + }); +}; + +const call = (method: string, id: number) => + JSON.stringify({ id, jsonrpc: "2.0", method }); + +interface Frame { + error?: { code: number; message: string }; + id: number | null; + result?: string; +} + +const frames = (text: string): Frame[] => + text + .split("\n") + .filter((line) => line !== "") + .map((line) => { + // SAFETY: the action endpoint answers JSON-RPC frames, one per line. + const frame = JSON.parse(line) as Frame; + return frame; + }); + +const handlerFor = function handlerFor( + probe: Probe, + options: Parameters[2] = {}, + slowFor?: `${number} millis` +) { + const group = makeGroup(); + // SAFETY: the test group and layer are the shapes the generated actions module passes. + return createActionHandler( + group as never, + makeHandlers(group, probe, slowFor) as never, + { sameOrigin: false, ...options } + ); +}; + +test("every request builds its own action runtime and disposes it after the body", async () => { + const probe = makeProbe(); + const handle = handlerFor(probe); + const responses = await Promise.all( + [1, 2, 3].map((id) => handle(post(call("ping", id)))) + ); + const bodies = await Promise.all(responses.map((res) => res.text())); + expect(bodies.map((body) => frames(body)[0]?.result)).toEqual([ + "pong", + "pong", + "pong", + ]); + expect(probe.builds).toBe(3); + expect(probe.finalized).toBe(3); +}); + +test("concurrent requests each see only their own request context", async () => { + const handle = handlerFor(makeProbe()); + const users = Array.from({ length: 30 }, (_, i) => `user${i}`); + const seen = await Promise.all( + users.map(async (user) => { + const res = await handle(post(call("whoami", 1), { "x-user": user })); + return frames(await res.text())[0]?.result; + }) + ); + expect(seen).toEqual(users); +}); + +test("an action past timeoutMs answers with a JSON-RPC error and is interrupted", async () => { + const probe = makeProbe(); + const handle = handlerFor(probe, { timeoutMs: 50 }); + const started = Date.now(); + const res = await handle(post(call("slow", 7))); + expect(Date.now() - started).toBeLessThan(1000); + const expected: Frame & { jsonrpc: string } = { + error: { code: -32_603, message: "Action timed out after 50ms" }, + id: 7, + jsonrpc: "2.0", + }; + expect(frames(await res.text())).toEqual([expected]); + await Bun.sleep(20); + expect(probe.interrupted).toBe(1); + expect(probe.finalized).toBe(1); +}); + +test("a batch times out per call and keeps the calls that answered", async () => { + const probe = makeProbe(); + const handle = handlerFor(probe, { timeoutMs: 50 }); + const res = await handle(post(`[${call("ping", 1)},${call("slow", 2)}]`)); + const byId = new Map(frames(await res.text()).map((f) => [f.id, f])); + expect(byId.get(1)?.result).toBe("pong"); + expect(byId.get(2)?.error?.message).toBe("Action timed out after 50ms"); +}); + +test("a batch streams each result as soon as its call ends", async () => { + const handle = handlerFor(makeProbe(), {}, "300 millis"); + const started = Date.now(); + const res = await handle(post(`[${call("slow", 1)},${call("ping", 2)}]`)); + const reader = res.body?.getReader(); + if (!reader) { + throw new Error("expected a streamed body"); + } + const decoder = new TextDecoder(); + const first = await reader.read(); + const firstAt = Date.now() - started; + let text = decoder.decode(first.value); + expect(frames(text)[0]).toMatchObject({ id: 2, result: "pong" }); + expect(firstAt).toBeLessThan(200); + for (;;) { + // oxlint-disable-next-line eslint/no-await-in-loop -- ordered stream reads + const next = await reader.read(); + if (next.done) { + break; + } + text += decoder.decode(next.value); + } + expect(frames(text).map((f) => f.id)).toEqual([2, 1]); +}); + +test("a body over maxBodyBytes is refused with 413 before any action runs", async () => { + const probe = makeProbe(); + const handle = handlerFor(probe, { maxBodyBytes: 16 }); + const res = await handle(post(call("ping", 1))); + expect(res.status).toBe(413); + expect(probe.builds).toBe(0); +}); + +test("a stub-function versions key (celld's node:process) is not a WebContainer", () => { + // celld's node:process answers every unknown versions key with a stub function. + const celldVersions = { webcontainer: () => null }; + expect(isWebcontainerVersions(celldVersions)).toBe(false); + expect(isWebcontainerVersions({ webcontainer: "1.0.0" })).toBe(true); + expect(isWebcontainerVersions({ node: "22.0.0" })).toBe(false); +}); diff --git a/packages/oxidejs/src/rpc/server.ts b/packages/oxidejs/src/rpc/server.ts index 0aa3c12..e83e850 100644 --- a/packages/oxidejs/src/rpc/server.ts +++ b/packages/oxidejs/src/rpc/server.ts @@ -6,6 +6,7 @@ import { RpcSerialization, RpcServer } from "effect/unstable/rpc"; import { ACTION_PATH, matchesActionPath } from "../actions"; import type { ActionContext } from "../context"; import { runWithRequest, withRequestEntry } from "../context"; +import type { OxidejsJson } from "../types"; import { isSameOrigin } from "./same-origin"; import { ensureNdjsonBody, @@ -14,6 +15,7 @@ import { scrubNdjsonTransform, scrubRpcJson, } from "./scrub"; +import type { JsonRpcId } from "./scrub"; export const IDEMPOTENCY_HEADER = "x-oxide-idempotency-key"; @@ -21,6 +23,18 @@ export interface ActionHandlerOptions { createContext?: (req: Request) => ActionContext | Promise; path?: string; sameOrigin?: boolean; + /** + * Reject a request body larger than this many bytes with HTTP 413 before + * any action runs. Default: no limit (the host's own limit applies). + */ + maxBodyBytes?: number | undefined; + /** + * Answer with a JSON-RPC error when an action has not produced its response + * within this many milliseconds, and interrupt it. Each call of a batch has + * its own deadline. A stream counts as answered at its first frame. + * Default: no limit. + */ + timeoutMs?: number | undefined; transport?: "http" | "ws"; } @@ -105,26 +119,8 @@ interface HandlerBundle { type ActionGroup = RpcGroup.RpcGroup; -const bundles = new Map(); -const groupIds = new WeakMap(); -let nextGroupId = 0; - const serialization = RpcSerialization.layerNdJsonRpc(); -const bundleKey = function bundleKey( - group: ActionGroup, - path: string, - transport: "http" | "ws" -) { - let id = groupIds.get(group); - if (id === undefined) { - id = nextGroupId; - nextGroupId += 1; - groupIds.set(group, id); - } - return `${id}:${path}:${transport}`; -}; - const forbidden = function forbidden(): Response { return Response.json(JSON_RPC_FORBIDDEN, { headers: { "content-type": "application/json" }, @@ -139,6 +135,27 @@ const methodNotAllowed = function methodNotAllowed(): Response { }); }; +const payloadTooLarge = function payloadTooLarge(maxBytes: number): Response { + return Response.json( + { + error: { + code: -32_600, + message: `Payload too large (limit ${maxBytes} bytes)`, + }, + id: null, + jsonrpc: "2.0", + }, + { status: 413 } + ); +}; + +/** + * One Effect RPC runtime per call. A runtime shared across requests is unsafe + * on every host: its server fibers keep the request context of the request + * that started them, so a concurrent action could read another request's + * `useRequest()` / `useEnv()`. On Worker hosts (Cloudflare Workers, celld) it + * also stalls, because the host drops a finished request's pending work. + */ const buildBundle = function buildBundle( group: ActionGroup, handlers: Layer.Layer, @@ -158,25 +175,9 @@ const buildBundle = function buildBundle( }) as HandlerBundle; }; -const bundleFor = function bundleFor( - group: ActionGroup, - handlers: Layer.Layer, - path: string, - transport: "http" | "ws" -) { - const key = bundleKey(group, path, transport); - const cached = bundles.get(key); - if (cached) { - return cached; - } - const built = buildBundle(group, handlers, path, transport); - bundles.set(key, built); - return built; -}; - const scrubJsonResponse = function scrubJsonResponse( response: Response, - requestIds: ReturnType + requestIds: JsonRpcId[] ): Response { const contentType = response.headers.get("content-type") ?? ""; if (!contentType.includes("json")) { @@ -206,7 +207,7 @@ const scrubJsonResponse = function scrubJsonResponse( const scrubBufferedJson = async function scrubBufferedJson( response: Response, - requestIds: ReturnType + requestIds: JsonRpcId[] ): Promise { const text = await response.text(); const headers = new Headers(response.headers); @@ -218,6 +219,200 @@ const scrubBufferedJson = async function scrubBufferedJson( }); }; +const INTERNAL_ERROR_CODE = -32_603; + +/** One JSON-RPC error line per request id, in the NDJSON the client reads. */ +const errorLines = function errorLines( + requestIds: JsonRpcId[], + message: string +): string { + const ids = requestIds.length > 0 ? requestIds : [null]; + return ids + .map( + (id) => + `${JSON.stringify({ + error: { code: INTERNAL_ERROR_CODE, message }, + id, + jsonrpc: "2.0", + })}\n` + ) + .join(""); +}; + +const timeoutMessage = function timeoutMessage(timeoutMs: number): string { + return `Action timed out after ${timeoutMs}ms`; +}; + +const TIMED_OUT = Symbol("oxidejs.actionTimeout"); + +/** Race `run` against `timeoutMs`; the timer is cleared either way. */ +const withTimeout = async function withTimeout( + run: Promise, + timeoutMs: number | undefined +): Promise { + if (timeoutMs === undefined) { + return await run; + } + let timer: ReturnType | undefined; + // oxlint-disable-next-line promise/avoid-new -- a timer-backed race needs its own Promise + const expired = new Promise((resolve) => { + timer = setTimeout(() => resolve(TIMED_OUT), timeoutMs); + }); + try { + return await Promise.race([run, expired]); + } finally { + clearTimeout(timer); + } +}; + +/** Tear a runtime down; a failed teardown has nobody left to tell. */ +const disposeQuietly = async function disposeQuietly( + dispose: () => Promise +): Promise { + try { + await dispose(); + } catch { + // The response is already decided; a failed teardown changes nothing. + } +}; + +/** Dispose a per-call runtime once the response body is done (or dropped). */ +const disposeAfterBody = function disposeAfterBody( + response: Response, + dispose: () => Promise +): Response { + const release = () => disposeQuietly(dispose); + if (!response.body) { + void release(); + return response; + } + const reader = response.body.getReader(); + const body = new ReadableStream({ + cancel: async (reason) => { + try { + await reader.cancel(reason); + } finally { + await release(); + } + }, + pull: async (controller) => { + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + await release(); + return; + } + controller.enqueue(value); + } catch (error) { + controller.error(error); + await release(); + } + }, + }); + return new Response(body, { + headers: response.headers, + status: response.status, + statusText: response.statusText, + }); +}; + +/** One JSON-RPC call of a request body: its NDJSON frame and its ids. */ +interface ActionCall { + body: Uint8Array; + ids: JsonRpcId[]; +} + +const encoder = new TextEncoder(); + +/** + * Split a body into its calls: a JSON array (batch) or several NDJSON lines + * become one call each. Anything that does not parse stays one call, so + * Effect answers it with its own parse error. + */ +const splitCalls = function splitCalls( + rawBody: Uint8Array +): ActionCall[] { + const whole = [{ body: rawBody, ids: extractJsonRpcRequestIds(rawBody) }]; + const text = new TextDecoder().decode(rawBody).replace(/^\uFEFF/u, ""); + const frames: string[] = []; + for (const line of text.split("\n")) { + if (line.trim() === "") { + continue; + } + let parsed: OxidejsJson; + try { + // SAFETY: JSON.parse yields JSON values; OxidejsJson is the JSON value union. + parsed = JSON.parse(line) as OxidejsJson; + } catch { + return whole; + } + if (!Array.isArray(parsed)) { + frames.push(line); + continue; + } + if (parsed.length === 0) { + return whole; + } + for (const frame of parsed) { + frames.push(JSON.stringify(frame)); + } + } + if (frames.length <= 1) { + return whole; + } + return frames.map((frame) => ({ + body: encoder.encode(`${frame}\n`), + ids: extractJsonRpcRequestIds(frame), + })); +}; + +/** An AbortController that also aborts when `parent` does. */ +const linkedAbort = function linkedAbort(parent: AbortSignal): AbortController { + const abort = new AbortController(); + if (parent.aborted) { + abort.abort(parent.reason); + } else { + parent.addEventListener("abort", () => abort.abort(parent.reason), { + once: true, + }); + } + return abort; +}; + +/** Stream each call's NDJSON lines to the client as soon as that call ends. */ +const mergeCallResponses = function mergeCallResponses( + calls: ActionCall[], + pending: Promise[] +): Response { + const body = new ReadableStream({ + start: async (controller) => { + await Promise.all( + pending.map(async (response, index) => { + let text: string; + try { + const settled = await response; + text = await settled.text(); + } catch { + text = errorLines(calls[index]?.ids ?? [], "Internal error"); + } + if (text === "") { + return; + } + controller.enqueue( + encoder.encode(text.endsWith("\n") ? text : `${text}\n`) + ); + }) + ); + controller.close(); + }, + }); + return new Response(body, { + headers: { "content-type": NDJSON_CONTENT }, + status: 200, + }); +}; + export const createActionHandler = function createActionHandler( group: ActionGroup, handlers: Layer.Layer, @@ -226,6 +421,90 @@ export const createActionHandler = function createActionHandler( const path = options.path ?? ACTION_PATH; const transport = options.transport ?? "http"; const sameOrigin = options.sameOrigin ?? true; + const { maxBodyBytes, timeoutMs } = options; + + /** Run one call in its own runtime, bounded by `timeoutMs`. */ + const runCall = async function runCall( + request: Request, + call: ActionCall, + extra: Partial + ): Promise { + const headers = new Headers(request.headers); + // Body is always NDJSON (trailing newline). Match the serialization layer. + headers.set("content-type", NDJSON_CONTENT); + headers.delete("content-length"); + // Aborts when the client goes away or the call times out. + const abort = linkedAbort(request.signal); + const forwarded = new Request(request.url, { + body: call.body, + headers, + method: request.method, + signal: abort.signal, + }); + const idempotencyKey = + extra.idempotencyKey ?? extractIdempotencyKey(call.body, request.headers); + const callExtra: Partial = { ...extra }; + if (idempotencyKey) { + callExtra.idempotencyKey = idempotencyKey; + } + + let bundle: HandlerBundle | undefined; + let outcome: Response | typeof TIMED_OUT | undefined; + try { + // Build the runtime inside this call's request context: its server + // fibers are forked during the build, and on celld they only see the + // store they were created under (without `nodejs_compat`, a runtime + // built outside it answers every action with "request context is + // unavailable"). + outcome = await withTimeout( + runWithRequest( + forwarded, + () => { + const built = buildBundle(group, handlers, path, transport); + bundle = built; + return built.handler(forwarded); + }, + callExtra + ), + timeoutMs + ); + } finally { + // A thrown handler leaves nothing to stream; release its runtime now. + if (outcome === undefined) { + await bundle?.dispose(); + } + } + + if (outcome === TIMED_OUT) { + abort.abort(new Error("oxidejs: action timed out")); + // Do not await: the runtime we are tearing down is the one that hung. + if (bundle) { + void disposeQuietly(bundle.dispose); + } + return new Response( + errorLines(call.ids, timeoutMessage(timeoutMs ?? 0)), + { + headers: { "content-type": NDJSON_CONTENT }, + status: 200, + } + ); + } + + const dispose = bundle?.dispose ?? (() => Promise.resolve()); + const contentType = outcome.headers.get("content-type") ?? ""; + if ( + contentType.includes(NDJSON_CONTENT) || + contentType.includes("ndjson") + ) { + return disposeAfterBody(scrubJsonResponse(outcome, call.ids), dispose); + } + if (contentType.includes("json")) { + const buffered = await scrubBufferedJson(outcome, call.ids); + await dispose(); + return buffered; + } + return disposeAfterBody(outcome, dispose); + }; return async function handleActionRequest( request: Request @@ -239,57 +518,50 @@ export const createActionHandler = function createActionHandler( if (sameOrigin && !isSameOrigin(request)) { return forbidden(); } + if (maxBodyBytes !== undefined) { + const declared = Number(request.headers.get("content-length") ?? "0"); + if (declared > maxBodyBytes) { + return payloadTooLarge(maxBodyBytes); + } + } return await withRequestEntry(async () => { - const rawBody = ensureNdjsonBody(await request.arrayBuffer()); - const requestIds = extractJsonRpcRequestIds(rawBody); - const headers = new Headers(request.headers); - // Body is always NDJSON (trailing newline). Match the serialization layer. - headers.set("content-type", NDJSON_CONTENT); - - const forwarded = new Request(request.url, { - body: rawBody, - headers, - method: request.method, - signal: request.signal, - }); - - // Host stamps env on the inbound Request; `forwarded` is a clone without it. + const buffer = await request.arrayBuffer(); + if (maxBodyBytes !== undefined && buffer.byteLength > maxBodyBytes) { + return payloadTooLarge(maxBodyBytes); + } + const rawBody = ensureNdjsonBody(buffer); + // Host stamps env on the inbound Request; the per-call clones lack it. const hostExtra = (await options.createContext?.(request)) ?? {}; - const idempotencyKey = extractIdempotencyKey(rawBody, request.headers); + const headerKey = request.headers.get(IDEMPOTENCY_HEADER); const extra: Partial = { ...hostExtra }; - if (idempotencyKey) { - extra.idempotencyKey = idempotencyKey; + if (headerKey) { + extra.idempotencyKey = headerKey; } - const { handler } = bundleFor(group, handlers, path, transport); - const response = await runWithRequest( - forwarded, - () => handler(forwarded), - extra - ); - const contentType = response.headers.get("content-type") ?? ""; - if ( - contentType.includes(NDJSON_CONTENT) || - contentType.includes("ndjson") - ) { - return scrubJsonResponse(response, requestIds); + const calls = splitCalls(rawBody); + const [only] = calls; + if (calls.length === 1 && only) { + return await runCall(request, only, extra); } - if (contentType.includes("json")) { - return scrubBufferedJson(response, requestIds); - } - return response; + // A batch: every call gets its own runtime and deadline, and its result + // goes out as soon as it is ready instead of waiting for the slowest. + return mergeCallResponses( + calls, + calls.map((call) => runCall(request, call, extra)) + ); }); }; }; +/** + * @deprecated Action runtimes are built per request and disposed with their + * response, so there is nothing cached to dispose. Kept for compatibility. + */ export const disposeActionHandler = function disposeActionHandler( - group: ActionGroup, - path: string = ACTION_PATH, - transport: "http" | "ws" = "http" + _group?: ActionGroup, + _path: string = ACTION_PATH, + _transport: "http" | "ws" = "http" ) { - const key = bundleKey(group, path, transport); - const bundle = bundles.get(key); - bundles.delete(key); - return bundle?.dispose() ?? Promise.resolve(); + return Promise.resolve(); }; diff --git a/packages/oxidejs/src/rpc/ws.ts b/packages/oxidejs/src/rpc/ws.ts index 15d7df9..db0a373 100644 --- a/packages/oxidejs/src/rpc/ws.ts +++ b/packages/oxidejs/src/rpc/ws.ts @@ -35,6 +35,8 @@ export interface WsHooksOptions { maxMessageSize?: number; path?: string; sameOrigin?: boolean; + /** Per-message action deadline in milliseconds. See `ActionHandlerOptions.timeoutMs`. */ + timeoutMs?: number | undefined; } /** Hostname without port; brackets stripped for IPv6. */ @@ -290,6 +292,7 @@ export const createWsHooks = function createWsHooks( const baseOptions: ActionHandlerOptions = { path, sameOrigin, + timeoutMs: options.timeoutMs, // Synthetic POST into createActionHandler — NDJSON over HTTP framing, not Effect's WS protocol. transport: "http", }; diff --git a/packages/oxidejs/src/schedule-build.ts b/packages/oxidejs/src/schedule-build.ts index 30d6cc3..a39832e 100644 --- a/packages/oxidejs/src/schedule-build.ts +++ b/packages/oxidejs/src/schedule-build.ts @@ -312,7 +312,7 @@ export const generateScheduleHandlerModule = return `export async function handleSchedule() {\n throw new Error("oxidejs: no schedules registered");\n}\n`; } const lines = [ - `import { dispatchSchedule, readScheduleMeta } from "oxidejs";`, + `import { dispatchSchedule, readScheduleMeta, withRequestStore } from "oxidejs";`, ]; const metaVars: string[] = []; for (const [i, mod] of modules.entries()) { @@ -334,7 +334,12 @@ export const generateScheduleHandlerModule = lines.push( `const __metas = [${metaVars.join(", ")}];`, `export async function handleSchedule(controller, env, ctx) {`, - ` return dispatchSchedule(__metas, controller, env, ctx);`, + // A request store like workflow run(): useEnv() and queue/workflow + // handles work in a schedule handler (and the default dispatch). + ` return withRequestStore(`, + ` { env, fetchCtx: ctx, req: new Request("https://oxide.local/schedule") },`, + ` () => dispatchSchedule(__metas, controller, env, ctx)`, + ` );`, `}` ); return `${lines.join("\n")}\n`; diff --git a/packages/oxidejs/src/schedule.test.ts b/packages/oxidejs/src/schedule.test.ts index 90ef384..2f061c8 100644 --- a/packages/oxidejs/src/schedule.test.ts +++ b/packages/oxidejs/src/schedule.test.ts @@ -132,6 +132,8 @@ describe("schedule codegen", () => { ]); expect(code).toContain("dispatchSchedule"); expect(code).toContain("handleSchedule"); + // A request store, so useEnv() and queue/workflow handles work in a tick. + expect(code).toContain("withRequestStore("); }); test("worker wrapper attaches scheduled when hasSchedules", () => { diff --git a/packages/oxidejs/src/schedule.ts b/packages/oxidejs/src/schedule.ts index 8303366..85f1249 100644 --- a/packages/oxidejs/src/schedule.ts +++ b/packages/oxidejs/src/schedule.ts @@ -388,8 +388,12 @@ export const dispatchSchedule = async function dispatchSchedule( }, { id, params } ); - } catch { + } catch (error) { // Enqueue already succeeded — do not fail the schedule tick. + console.warn( + `oxidejs: schedule "${meta.name}" producer-side workflow start failed; the queue consumer will start it`, + error + ); } } } diff --git a/packages/oxidejs/src/types.ts b/packages/oxidejs/src/types.ts index 01a8f65..4081edc 100644 --- a/packages/oxidejs/src/types.ts +++ b/packages/oxidejs/src/types.ts @@ -11,7 +11,7 @@ export type OxidejsJson = export type OxidejsActionTransport = "http" | "ws"; -/** `actions` config: transport string, or an object with `transport`, `path`, `sameOrigin`, `openrpc`. */ +/** `actions` config: transport string, or an object with `transport`, `path`, `sameOrigin`, `openrpc`, `timeout`. */ export type OxidejsActions = | OxidejsActionTransport | { @@ -26,6 +26,12 @@ export type OxidejsActions = * is HTTP). Default: false. */ openrpc?: boolean; + /** + * Milliseconds an action may run before it answers with a JSON-RPC + * error and is interrupted. A stream counts as answered at its first + * frame. Default: no limit. + */ + timeout?: number; }; /** Static headers inlined into the shared action client. Functions cannot ship to the browser. */ @@ -129,6 +135,8 @@ export interface ResolvedOptions { actionPath: string; /** Reject cross-origin action requests (CSRF defense). Default: true. */ actionSameOrigin: boolean; + /** Per-action response deadline in milliseconds, or `undefined` for none. */ + actionTimeout: number | undefined; /** Serve OpenRPC discovery for `action()` handlers at `/__oxide/openrpc`. */ actionOpenRpc: boolean; actionHeaders: OxidejsActionHeaders | undefined; diff --git a/packages/oxidejs/src/workflow.ts b/packages/oxidejs/src/workflow.ts index 8a785f0..11d635c 100644 --- a/packages/oxidejs/src/workflow.ts +++ b/packages/oxidejs/src/workflow.ts @@ -3,7 +3,7 @@ import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import type { CallOptions } from "./action"; -import { useEnv, useIdempotencyKey } from "./request-store"; +import { nextRequestScopedId, useEnv } from "./request-store"; import type { OxidejsJson } from "./types"; import { SchemaDecodeError } from "./with-schema"; @@ -413,8 +413,9 @@ export const workflow = function workflow( const raw = args[0] as P; const decoded = decodePayload(def.payload, raw); const wf = requireBinding(binding); + // Derived per start, so two starts in one keyed request stay distinct. const id = - options?.idempotencyKey ?? useIdempotencyKey() ?? crypto.randomUUID(); + options?.idempotencyKey ?? nextRequestScopedId() ?? crypto.randomUUID(); const instance = await createWorkflowInstance(wf, { id, params: decoded }); return { id: instance.id }; }; diff --git a/packages/oxidejs/src/wrapper.test.ts b/packages/oxidejs/src/wrapper.test.ts index d33461d..78ccbcd 100644 --- a/packages/oxidejs/src/wrapper.test.ts +++ b/packages/oxidejs/src/wrapper.test.ts @@ -10,6 +10,20 @@ const BASE = { }; describe("generateWorkerWrapper", () => { + test("passes actions.timeout to the action handler", () => { + const out = generateWorkerWrapper("/x/server.ts", { + ...BASE, + actionTimeout: 15_000, + hasActions: true, + }); + expect(out).toContain("timeoutMs: 15000"); + // The body limit reaches the Worker action handler, not only Node. + expect(out).toContain("maxBodyBytes: 1048576"); + expect( + generateWorkerWrapper("/x/server.ts", { ...BASE, hasActions: true }) + ).not.toContain("timeoutMs"); + }); + test("middleware handlers run before the action gate", () => { const out = generateWorkerWrapper("/x/server.ts", { ...BASE, diff --git a/packages/oxidejs/test/celld-smoke/.gitignore b/packages/oxidejs/test/celld-smoke/.gitignore new file mode 100644 index 0000000..de028e8 --- /dev/null +++ b/packages/oxidejs/test/celld-smoke/.gitignore @@ -0,0 +1,2 @@ +.build/ +.celld/ diff --git a/packages/oxidejs/test/celld-smoke/smoke.server.ts b/packages/oxidejs/test/celld-smoke/smoke.server.ts new file mode 100644 index 0000000..e3e3fee --- /dev/null +++ b/packages/oxidejs/test/celld-smoke/smoke.server.ts @@ -0,0 +1,38 @@ +/** + * Smoke actions, compiled by oxide's own codegen (as an app's `*.server.ts` + * files are), so the celld run covers the generated handlers, + * `runActionInContext` and user-level `useRequest()` after an `await`. + */ +import { Effect } from "effect"; + +import { action, useRequest } from "../../src/context"; + +const WORK_MS = 20; + +const delay = (ms: number) => + // oxlint-disable-next-line promise/avoid-new -- a timer-backed delay + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +export const ping = action(async () => { + await Promise.resolve(); + return "pong"; +}); + +export const work = action(async () => { + await delay(WORK_MS); + return "done"; +}); + +/** Reads the request after an await, the way app code reads its session. */ +export const whoami = action(async () => { + await delay(WORK_MS); + return useRequest().headers.get("x-user") ?? "none"; +}); + +// Never settles: only the action timeout can answer it. +export const stall = action(async () => { + await Effect.runPromise(Effect.never); + return "unreachable"; +}); diff --git a/packages/oxidejs/test/celld-smoke/wrangler.jsonc b/packages/oxidejs/test/celld-smoke/wrangler.jsonc new file mode 100644 index 0000000..6febb97 --- /dev/null +++ b/packages/oxidejs/test/celld-smoke/wrangler.jsonc @@ -0,0 +1,10 @@ +// celld smoke app: `scripts/celld-smoke.ts` generates the Worker wrapper and +// actions module for smoke.server.ts and bundles them into .build/. +// No compatibility flags: the stricter case, and how apps such as Noite +// deploy. wrangler.nodejs-compat.jsonc covers the flagged runtime. +{ + "name": "oxide-celld-smoke", + "main": ".build/index.js", + "compatibility_date": "2026-09-01", + "compatibility_flags": [], +} diff --git a/packages/oxidejs/test/celld-smoke/wrangler.nodejs-compat.jsonc b/packages/oxidejs/test/celld-smoke/wrangler.nodejs-compat.jsonc new file mode 100644 index 0000000..944502c --- /dev/null +++ b/packages/oxidejs/test/celld-smoke/wrangler.nodejs-compat.jsonc @@ -0,0 +1,7 @@ +// Same smoke app with `nodejs_compat` (see wrangler.jsonc). +{ + "name": "oxide-celld-smoke", + "main": ".build/index.js", + "compatibility_date": "2026-09-01", + "compatibility_flags": ["nodejs_compat"], +} diff --git a/scripts/celld-smoke.ts b/scripts/celld-smoke.ts new file mode 100644 index 0000000..a57263a --- /dev/null +++ b/scripts/celld-smoke.ts @@ -0,0 +1,261 @@ +#!/usr/bin/env bun +/** + * celld smoke test for the oxide action server. + * + * Bun's test runner cannot reproduce what broke actions on celld: celld drops + * a request's pending work once that request answers or its client goes away, + * so anything shared across requests (a runtime, a gate, a promise) can hang + * every later request. This script generates what the plugin emits for a + * worker app (the Worker wrapper and the actions module for + * `packages/oxidejs/test/celld-smoke/smoke.server.ts`), bundles it from + * source, boots `celld dev` on it, and asserts: + * + * - concurrent async actions all answer, round after round; + * - each request sees only its own request context; + * - a batch keeps its fast result when another call times out; + * - a stuck action answers with the timeout error; + * - the isolate still answers afterwards (nothing wedged). + * + * Runs twice: without compatibility flags (how apps such as Noite deploy) + * and with `nodejs_compat`, because celld's request context behaves + * differently between the two. + * + * Needs `celld` and `esbuild` on PATH (or CELLD_BIN / CELLD_ESBUILD). + * + * bun run smoke:celld + */ +import { mkdirSync, rmSync } from "node:fs"; +import path from "node:path"; + +import { + generateActionsModule, + generateWorkerWrapper, + scanServerFiles, +} from "../packages/oxidejs/src/actions"; +import { writeGeneratedActions } from "../packages/oxidejs/src/rpc/test-harness"; + +const APP = path.join(import.meta.dir, "../packages/oxidejs/test/celld-smoke"); +const PORT = Number(process.env["CELLD_SMOKE_PORT"] ?? 19_990); +const BASE = `http://127.0.0.1:${PORT}`; +const ACTION = `${BASE}/__oxide/action`; +const CONCURRENCY = 50; +const ROUNDS = 3; +const CALL_LIMIT_MS = 10_000; +const TIMEOUT_MS = 1000; +const READY_LIMIT_MS = 60_000; + +interface Frame { + error?: { message: string }; + id: number | null; + result?: string; +} + +const CONFIGS = ["wrangler.jsonc", "wrangler.nodejs-compat.jsonc"] as const; + +const failures: string[] = []; +let currentConfig = ""; +const check = (ok: boolean, label: string) => { + console.log(`${ok ? "ok " : "FAIL"} ${label}`); + if (!ok) { + failures.push(`${currentConfig}: ${label}`); + } +}; + +const frames = (text: string): Frame[] => + text + .split("\n") + .filter((line) => line !== "") + .map((line) => { + // SAFETY: the action endpoint answers JSON-RPC frames, one per line. + const frame = JSON.parse(line) as Frame; + return frame; + }); + +/** POST a JSON-RPC body; `null` when it does not answer within the limit. */ +const post = async ( + body: string, + headers: Record = {} +): Promise => { + try { + const res = await fetch(ACTION, { + body, + headers, + method: "POST", + signal: AbortSignal.timeout(CALL_LIMIT_MS), + }); + return frames(await res.text()); + } catch { + return null; + } +}; + +const call = (method: string, id = 1) => + JSON.stringify({ id, jsonrpc: "2.0", method, params: { args: [] } }); + +const SRC = path.join(import.meta.dir, "../packages/oxidejs/src"); + +/** Resolve the generated wrapper's package and virtual imports to the source tree. */ +const oxideSource = (actionsModule: string): Bun.BunPlugin => ({ + name: "oxide-source", + setup: (build) => { + const targets = new Map( + Object.entries({ + oxidejs: path.join(SRC, "context.ts"), + "oxidejs/rpc": path.join(SRC, "rpc/index.ts"), + "oxidejs/worker-dom/install": path.join(SRC, "worker-dom/install.ts"), + "virtual:oxide/actions": actionsModule, + }) + ); + build.onResolve( + { + filter: + /^(?:oxidejs(?:\/rpc|\/worker-dom\/install)?|virtual:oxide\/actions)$/u, + }, + (args) => ({ path: targets.get(args.path) ?? args.path }) + ); + }, +}); + +const bundle = async () => { + // What the plugin generates for a worker app: the actions module for + // smoke.server.ts and the Worker wrapper (worker-dom install, request + // context stamp, action gate) — with its imports pointed at the source tree. + const buildDir = path.join(APP, ".build"); + mkdirSync(buildDir, { recursive: true }); + const actionsModule = writeGeneratedActions(buildDir, { + code: generateActionsModule(scanServerFiles(APP)), + }); + const entry = path.join(buildDir, "entry.mjs"); + await Bun.write( + entry, + generateWorkerWrapper(null, { + actionSameOrigin: false, + actionTimeout: TIMEOUT_MS, + hasActions: true, + preset: "worker", + }) + ); + const built = await Bun.build({ + entrypoints: [entry], + external: ["node:*"], + format: "esm", + naming: "index.js", + outdir: path.join(APP, ".build"), + plugins: [oxideSource(actionsModule)], + target: "browser", + }); + if (!built.success) { + throw new AggregateError(built.logs, "bundling the smoke worker failed"); + } +}; + +const waitReady = async () => { + const deadline = Date.now() + READY_LIMIT_MS; + while (Date.now() < deadline) { + try { + // Any HTTP answer means the node serves; the action path wants POST. + // oxlint-disable-next-line eslint/no-await-in-loop -- readiness poll + await fetch(ACTION, { signal: AbortSignal.timeout(1000) }); + return; + } catch { + // oxlint-disable-next-line eslint/no-await-in-loop -- readiness poll + await Bun.sleep(500); + } + } + throw new Error("celld dev did not start serving in time"); +}; + +const run = async () => { + for (let round = 1; round <= ROUNDS; round += 1) { + // oxlint-disable-next-line eslint/no-await-in-loop -- rounds run one after another on purpose + const answers = await Promise.all( + Array.from({ length: CONCURRENCY }, (_, i) => post(call("smoke.work", i))) + ); + const done = answers.filter((a) => a?.[0]?.result === "done").length; + check( + done === CONCURRENCY, + `round ${round}: ${done}/${CONCURRENCY} concurrent async actions answered` + ); + } + + const users = Array.from({ length: 30 }, (_, i) => `user${i}`); + const seen = await Promise.all( + users.map((user) => post(call("smoke.whoami"), { "x-user": user })) + ); + const own = seen.filter((a, i) => a?.[0]?.result === users[i]).length; + check( + own === users.length, + `${own}/${users.length} requests saw their own context` + ); + + const batch = await post( + `[${call("smoke.ping", 1)},${call("smoke.stall", 2)}]` + ); + const byId = new Map((batch ?? []).map((f) => [f.id, f])); + check( + byId.get(1)?.result === "pong" && + Boolean(byId.get(2)?.error?.message.includes("timed out")), + "a batch keeps its fast result when another call times out" + ); + + const stalled = await post(call("smoke.stall")); + check( + Boolean(stalled?.[0]?.error?.message.includes("timed out")), + "a stuck action answers with the timeout error" + ); + + const after = await Promise.all( + [1, 2, 3].map((id) => post(call("smoke.ping", id))) + ); + check( + after.every((a) => a?.[0]?.result === "pong"), + "the isolate still answers afterwards" + ); +}; + +/** Boot `celld dev` on one config, run every check, stop the node. */ +const runConfig = async (config: string) => { + currentConfig = config; + console.log(`== ${config}`); + rmSync(path.join(APP, ".celld"), { force: true, recursive: true }); + const celld = Bun.spawn( + [ + process.env["CELLD_BIN"] ?? "celld", + "dev", + "--no-watch", + "--port", + String(PORT), + path.join(APP, config), + ], + { + env: { ...process.env, NO_COLOR: "1" }, + stderr: "inherit", + stdout: "ignore", + } + ); + try { + await waitReady(); + await run(); + } finally { + celld.kill("SIGTERM"); + await celld.exited; + } +}; + +const main = async () => { + await bundle(); + for (const config of CONFIGS) { + // oxlint-disable-next-line eslint/no-await-in-loop -- one node on the port at a time + await runConfig(config); + } + if (failures.length > 0) { + console.error(`celld smoke: ${failures.length} check(s) failed`); + for (const failure of failures) { + console.error(` ${failure}`); + } + process.exit(1); + } + console.log("celld smoke: all checks passed"); +}; + +await main(); diff --git a/templates/kit/package.json b/templates/kit/package.json index 2a34323..3305cc9 100644 --- a/templates/kit/package.json +++ b/templates/kit/package.json @@ -27,7 +27,7 @@ "@cloudflare/workers-types": "^5.20260818.1", "@tailwindcss/vite": "^4.3.3", "daisyui": "^5.7.22", - "oxidejs": "^0.5.5", + "oxidejs": "^0.5.6", "tailwindcss": "^4.3.3", "typescript": "^7", "vite": "^8.2.2", diff --git a/templates/simple/package.json b/templates/simple/package.json index 597ae33..6b62420 100644 --- a/templates/simple/package.json +++ b/templates/simple/package.json @@ -15,7 +15,7 @@ "devDependencies": { "@tailwindcss/vite": "^4.3.3", "daisyui": "^5.7.22", - "oxidejs": "^0.5.5", + "oxidejs": "^0.5.6", "tailwindcss": "^4.3.3", "typescript": "^6", "vite": "^8.2.1" From da92daba2874e5320c2bdd8fc3b58a40cd5e8e11 Mon Sep 17 00:00:00 2001 From: Ryuz Date: Sun, 27 Sep 2026 18:23:38 +0200 Subject: [PATCH 2/2] fix(pr): improvements --- packages/oxidejs/CHANGELOG.md | 4 +- packages/oxidejs/README.md | 4 +- packages/oxidejs/src/context.ts | 2 +- packages/oxidejs/src/live-query.test.ts | 37 ++++++++++++ packages/oxidejs/src/live-query.ts | 75 +++++++++++++++++++++---- packages/oxidejs/src/queue.test.ts | 18 ++++-- packages/oxidejs/src/queue.ts | 6 +- packages/oxidejs/src/request-store.ts | 23 +++++++- packages/oxidejs/src/rpc/server.test.ts | 18 ++++++ packages/oxidejs/src/rpc/server.ts | 23 ++++++-- 10 files changed, 180 insertions(+), 30 deletions(-) diff --git a/packages/oxidejs/CHANGELOG.md b/packages/oxidejs/CHANGELOG.md index 751d3c3..bc58195 100644 --- a/packages/oxidejs/CHANGELOG.md +++ b/packages/oxidejs/CHANGELOG.md @@ -27,9 +27,9 @@ - celld: concurrent actions hung forever and then wedged the isolate until restart. celld's `node:process` answers every unknown `versions` key with a stub function, so `process.versions.webcontainer` read as truthy and every action queued behind one cross-request gate; a request the host cancelled never released it. WebContainer detection now requires a version string - Worker hosts (Cloudflare Workers, celld) stalled on the cached runtime: the host drops a finished request's pending work - `withRequestEntry` no longer chains Worker requests behind each other; only WebContainer serializes entry -- Two `queue.send()` calls, a `sendBatch()`, or a `workflow.start()` in one action with an idempotency key all reused that key as their id, so every message after the first was taken for the first and its workflow never started. Ids are now `key`, `key:1`, `key:2`, … per call (stable across retries of the same RPC call) +- Two `queue.send()` calls, a `sendBatch()`, or a `workflow.start()` in one action with an idempotency key all reused that key as their id, so every message after the first was taken for the first and its workflow never started. Ids are now `key`, `key-1`, `key-2`, … per call (stable across retries of the same RPC call), valid Workflows instance ids trimmed to 100 chars - Queue consumers and cron handlers run inside a request store, like workflow `run()`: `useEnv()`, `queue.send()` and `workflow.start()` no longer throw `request context is unavailable` there -- A `liveQuery` mutation waits at most `mutateWaitMs` for the previous one on its topic; a mutation whose request the host dropped no longer blocks the topic for good +- A `liveQuery` mutation waits at most `mutateWaitMs` for the previous one on its topic; a mutation whose request the host dropped no longer blocks the topic for good. A mutation that a newer one overtook during that wait does not publish its older snapshot, and a mutation that finishes after `close()` fails with `LiveQueryClosedError` instead of reporting success ## 0.5.5 diff --git a/packages/oxidejs/README.md b/packages/oxidejs/README.md index dbcddd4..4ee52af 100644 --- a/packages/oxidejs/README.md +++ b/packages/oxidejs/README.md @@ -160,7 +160,7 @@ Stream actions do not support `bind` / `with`. Breaking the `for await` loop or ### Live queries -Use `liveQuery` / `publish` for snapshot streams (query = subscription, mutation = publish). Hubs are isolate-local Effect PubSub — a publish reaches subscribers in the same isolate only, not other Worker isolates or celld nodes — and D1 (or your DB) stays the source of truth. Mutations on one topic run one at a time; one waits at most `mutateWaitMs` (default 10000) for the previous. `close()` shuts a topic's hub down. Prefer Effect `Stream` + `mutateEffect`: +Use `liveQuery` / `publish` for snapshot streams (query = subscription, mutation = publish). Hubs are isolate-local Effect PubSub — a publish reaches subscribers in the same isolate only, not other Worker isolates or celld nodes — and D1 (or your DB) stays the source of truth. Mutations on one topic run one at a time; one waits at most `mutateWaitMs` (default 10000) for the previous, and a mutation overtaken during that wait does not publish its older snapshot. `close()` shuts a topic's hub down; a mutation that finishes after it fails with `LiveQueryClosedError`. Prefer Effect `Stream` + `mutateEffect`: ```ts import { Effect, Stream } from "effect"; @@ -258,7 +258,7 @@ await invoices.send({ orderId: "…" }); await invoices.sendBatch([{ body: { orderId: "…" } }]); ``` -Oxide merges `queues.producers` / `queues.consumers` via `mergeDurableBindings` and attaches a same-worker `queue` handler that starts the workflow from each message (via `createBatch` when available, otherwise duplicate-aware `create`/`get`). Cloudflare does not return message ids from `send`, so oxide wraps bodies in an envelope with a client-chosen id (`{ idempotencyKey }` / request header / UUID) and returns `{ id }` from `send` (and `{ ids }` from `sendBatch`) for `workflow.status` polling. Queue transport options (`contentType` / `delaySeconds`) travel in the RPC payload; `signal` / `idempotencyKey` stay on `CallOptions`. Every message gets its own id: with a key, the first send in a request uses it and later sends (and each message of a `sendBatch`) use `key:1`, `key:2`, …, so a retried request reproduces the same ids. The consumer acks the messages it started and retries only the ones that failed. Queue consumers and cron handlers run inside a request store, so `useEnv()`, `queue.send()` and `workflow.start()` work there. +Oxide merges `queues.producers` / `queues.consumers` via `mergeDurableBindings` and attaches a same-worker `queue` handler that starts the workflow from each message (via `createBatch` when available, otherwise duplicate-aware `create`/`get`). Cloudflare does not return message ids from `send`, so oxide wraps bodies in an envelope with a client-chosen id (`{ idempotencyKey }` / request header / UUID) and returns `{ id }` from `send` (and `{ ids }` from `sendBatch`) for `workflow.status` polling. Queue transport options (`contentType` / `delaySeconds`) travel in the RPC payload; `signal` / `idempotencyKey` stay on `CallOptions`. Every message gets its own id: with a key, the first send in a request uses it and later sends (and each message of a `sendBatch`) use `key-1`, `key-2`, … (valid Workflows instance ids, trimmed to 100 chars), so a retried request reproduces the same ids. The consumer acks the messages it started and retries only the ones that failed. Queue consumers and cron handlers run inside a request store, so `useEnv()`, `queue.send()` and `workflow.start()` work there. ```ts const { id } = await invoices.send({ orderId: "…" }); diff --git a/packages/oxidejs/src/context.ts b/packages/oxidejs/src/context.ts index 5f5929a..5bbce93 100644 --- a/packages/oxidejs/src/context.ts +++ b/packages/oxidejs/src/context.ts @@ -106,7 +106,7 @@ export type { ServerActionHandle, StreamActionHandle, } from "./action"; -export { liveQuery, publish } from "./live-query"; +export { LiveQueryClosedError, liveQuery, publish } from "./live-query"; export type { LiveQuery, LiveQueryOptions } from "./live-query"; export { batch } from "./rpc/batch"; export type { BatchItem, BatchResult } from "./rpc/batch"; diff --git a/packages/oxidejs/src/live-query.test.ts b/packages/oxidejs/src/live-query.test.ts index 96bb38d..41c6982 100644 --- a/packages/oxidejs/src/live-query.test.ts +++ b/packages/oxidejs/src/live-query.test.ts @@ -118,4 +118,41 @@ describe("liveQuery", () => { expect(await iter.next()).toEqual({ done: false, value: 2 }); await iter.return?.(); }); + + test("a mutation superseded during the bounded wait does not publish its older snapshot", async () => { + const q = liveQuery({ + mutateWaitMs: 20, + topic: `order-${crypto.randomUUID()}`, + }); + const seen: string[] = []; + const collected = Effect.runPromise( + Stream.runForEach(q.stream().pipe(Stream.take(1)), (value) => + Effect.sync(() => seen.push(value)) + ) + ); + // The older mutation outlives the wait; the newer one runs and publishes first. + const older = q.mutate(async () => { + await Bun.sleep(80); + return "older"; + }); + await q.mutate(() => Promise.resolve("newer")); + expect(await older).toBe("older"); + await collected; + expect(seen).toEqual(["newer"]); + const iter = q.values()[Symbol.asyncIterator](); + expect(await iter.next()).toEqual({ done: false, value: "newer" }); + await iter.return?.(); + }); + + test("a mutation that finishes after close() fails instead of reporting success", async () => { + const q = liveQuery({ topic: `closed-${crypto.randomUUID()}` }); + const pending = q.mutate(async () => { + await Bun.sleep(20); + return 7; + }); + q.close(); + await expect(pending).rejects.toMatchObject({ + _tag: "LiveQueryClosedError", + }); + }); }); diff --git a/packages/oxidejs/src/live-query.ts b/packages/oxidejs/src/live-query.ts index 1c83958..a6f385e 100644 --- a/packages/oxidejs/src/live-query.ts +++ b/packages/oxidejs/src/live-query.ts @@ -1,5 +1,6 @@ import * as Effect from "effect/Effect"; import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { deferred } from "./deferred"; @@ -7,6 +8,33 @@ import { deferred } from "./deferred"; const DEFAULT_MUTATE_WAIT_MS = 10_000; const HUBS_KEY = Symbol.for("oxidejs.liveQuery.hubs"); const GATES_KEY = Symbol.for("oxidejs.liveQuery.gates"); +const ORDER_KEY = Symbol.for("oxidejs.liveQuery.order"); + +/** A mutation finished after `close()` shut its topic down: nobody got its snapshot. */ +// oxlint-disable-next-line unicorn/throw-new-error -- Schema.TaggedError factory +export class LiveQueryClosedError extends Schema.TaggedError()( + "LiveQueryClosedError", + { message: Schema.String, topic: Schema.String } +) {} + +/** Per topic: the last ticket handed to a mutation and the newest one published. */ +interface TopicOrder { + issued: number; + published: number; +} + +interface OrderBag { + [topic: string]: TopicOrder; +} + +const orderFor = function orderFor(topic: string): TopicOrder { + // SAFETY: isolate-global topic → mutation order; Symbol.for is oxide-owned. + const g = globalThis as typeof globalThis & { [ORDER_KEY]?: OrderBag }; + g[ORDER_KEY] ??= {}; + const bag = g[ORDER_KEY]; + bag[topic] ??= { issued: 0, published: 0 }; + return bag[topic]; +}; interface HubBag { [topic: string]: PubSub.PubSub; @@ -114,7 +142,7 @@ export interface LiveQuery { */ mutateEffect: ( fn: () => Effect.Effect - ) => Effect.Effect; + ) => Effect.Effect; /** Serialize Promise work that produces a snapshot, then publish it. */ mutate: (fn: () => Promise) => Promise; publish: (value: T) => void; @@ -153,18 +181,43 @@ export const liveQuery = function liveQuery( Effect.runSync(PubSub.publish(hub, value)); }; + /** + * Each mutation takes a ticket in gate order. The bounded gate can let a + * later mutation run while an earlier one is still working, so a mutation + * publishes only if no newer ticket has published yet: an older snapshot + * never overwrites a newer one. A mutation whose topic was closed fails + * with {@link LiveQueryClosedError} instead of reporting success. + */ const mutateEffect = function mutateEffect( fn: () => Effect.Effect - ): Effect.Effect { - return withTopicGate( - topic, - mutateWaitMs, - Effect.gen(function* mutateEffectGen() { - const value = yield* fn(); - publishValue(value); - return value; - }) - ); + ): Effect.Effect { + return Effect.suspend(() => { + const order = orderFor(topic); + order.issued += 1; + const ticket = order.issued; + return withTopicGate( + topic, + mutateWaitMs, + Effect.gen(function* mutateEffectGen() { + const value = yield* fn(); + if (ticket < order.published) { + // Superseded: a newer mutation already published its snapshot. + return value; + } + const published = Effect.runSync(PubSub.publish(hub, value)); + if (!published) { + return yield* Effect.fail( + new LiveQueryClosedError({ + message: `liveQuery topic "${topic}" was closed before its snapshot was published`, + topic, + }) + ); + } + order.published = ticket; + return value; + }) + ); + }); }; const mutate = async function mutate(fn: () => Promise) { diff --git a/packages/oxidejs/src/queue.test.ts b/packages/oxidejs/src/queue.test.ts index 5d892f9..898eeb4 100644 --- a/packages/oxidejs/src/queue.test.ts +++ b/packages/oxidejs/src/queue.test.ts @@ -27,7 +27,7 @@ import { resolveQueueWorkflowRefs, scanQueueFiles, } from "./queue-build"; -import { withRequestStore } from "./request-store"; +import { deriveKeyedId, withRequestStore } from "./request-store"; import { workflow } from "./workflow"; import type { WorkflowModule } from "./workflow-build"; import { parseWorkflowExports } from "./workflow-build"; @@ -771,9 +771,9 @@ export const demos = queue({ name: "demos", workflow: demo, maxBatchSize: 10 })\ return [first.id, second.id, ...batch.ids, started.id]; } ); - expect(ids).toEqual(["rpc-1", "rpc-1:1", "rpc-1:2", "rpc-1:3", "rpc-1:4"]); + expect(ids).toEqual(["rpc-1", "rpc-1-1", "rpc-1-2", "rpc-1-3", "rpc-1-4"]); expect(new Set(sent.map((body) => body.id)).size).toBe(4); - expect(created).toEqual(["rpc-1:4"]); + expect(created).toEqual(["rpc-1-4"]); }); test("a keyed sendBatch derives one id per message", async () => { @@ -792,7 +792,17 @@ export const demos = queue({ name: "demos", workflow: demo, maxBatchSize: 10 })\ idempotencyKey: "job", }) ); - expect(ids).toEqual(["job", "job:1", "job:2"]); + expect(ids).toEqual(["job", "job-1", "job-2"]); + }); + + test("derived ids stay valid Workflows instance ids within 100 chars", () => { + const long = "k".repeat(100); + const derived = deriveKeyedId(long, 12); + expect(derived).toHaveLength(100); + expect(derived.endsWith("-12")).toBe(true); + expect(/^[a-zA-Z0-9_][a-zA-Z0-9_-]*$/u.test(deriveKeyedId("job", 3))).toBe( + true + ); }); test("dispatchQueueBatch acks started messages and retries only the failed one", async () => { diff --git a/packages/oxidejs/src/queue.ts b/packages/oxidejs/src/queue.ts index 7e4994a..53f9cc4 100644 --- a/packages/oxidejs/src/queue.ts +++ b/packages/oxidejs/src/queue.ts @@ -3,7 +3,7 @@ import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import type { CallOptions } from "./action"; -import { nextRequestScopedId, useEnv } from "./request-store"; +import { deriveKeyedId, nextRequestScopedId, useEnv } from "./request-store"; import { SchemaDecodeError } from "./with-schema"; import { createWorkflowInstance, @@ -333,7 +333,7 @@ export const makeQueueEnvelope = function makeQueueEnvelope

( }; /** - * Message id: the per-message key, else the call's key (`key`, `key:1`, … by + * Message id: the per-message key, else the call's key (`key`, `key-1`, … by * position in a batch), else a distinct id derived from the RPC request's * key, else a random UUID. Ids become workflow instance ids, so two messages * must never share one — the second would be taken for the first and dropped. @@ -348,7 +348,7 @@ const nextMessageId = function nextMessageId( } const callKey = options?.idempotencyKey; if (callKey !== undefined) { - return position === 0 ? callKey : `${callKey}:${position}`; + return deriveKeyedId(callKey, position); } return nextRequestScopedId() ?? crypto.randomUUID(); }; diff --git a/packages/oxidejs/src/request-store.ts b/packages/oxidejs/src/request-store.ts index 0c9d04d..8924fea 100644 --- a/packages/oxidejs/src/request-store.ts +++ b/packages/oxidejs/src/request-store.ts @@ -310,10 +310,29 @@ export const useFetchCtx = function useFetchCtx(): const derivedIdCounts = new WeakMap(); +/** Cloudflare Workflows instance ids: at most 100 chars of `[A-Za-z0-9_-]`. */ +const MAX_INSTANCE_ID_LENGTH = 100; + +/** + * The id at `position` derived from `key`: the key itself first, then + * `key-1`, `key-2`, … A hyphen keeps a valid key a valid Workflows instance + * id, and the key is trimmed so the suffixed id stays within 100 chars. + */ +export const deriveKeyedId = function deriveKeyedId( + key: string, + position: number +): string { + if (position === 0) { + return key; + } + const suffix = `-${position}`; + return `${key.slice(0, MAX_INSTANCE_ID_LENGTH - suffix.length)}${suffix}`; +}; + /** * A distinct, deterministic id for the next queue message or workflow start * of the current keyed request, or `undefined` outside a request or without a - * key. The first call keeps the key; later calls get `key:1`, `key:2`, … So a + * key. The first call keeps the key; later calls get `key-1`, `key-2`, … So a * retried RPC call (same key, same code path) reproduces the same ids, while * two sends in one action no longer share one id and collapse into one * workflow instance. @@ -328,7 +347,7 @@ export const nextRequestScopedId = function nextRequestScopedId(): } const count = derivedIdCounts.get(current) ?? 0; derivedIdCounts.set(current, count + 1); - return count === 0 ? key : `${key}:${count}`; + return deriveKeyedId(key, count); }; /** Idempotency key from the client `CallOptions` / RPC headers, if present. */ diff --git a/packages/oxidejs/src/rpc/server.test.ts b/packages/oxidejs/src/rpc/server.test.ts index 48c5949..21f2ae9 100644 --- a/packages/oxidejs/src/rpc/server.test.ts +++ b/packages/oxidejs/src/rpc/server.test.ts @@ -183,6 +183,24 @@ test("a batch streams each result as soon as its call ends", async () => { expect(frames(text).map((f) => f.id)).toEqual([2, 1]); }); +test("cancelling a batch body mid-stream does not break the remaining calls", async () => { + const handle = handlerFor(makeProbe(), {}, "100 millis"); + const res = await handle( + post(`[${call("ping", 1)},${call("slow", 2)},${call("slow", 3)}]`) + ); + const reader = res.body?.getReader(); + if (!reader) { + throw new Error("expected a streamed body"); + } + await reader.read(); + await reader.cancel(); + // The slow calls finish after the cancel; their results must be dropped + // quietly, and the handler keeps serving. + await Bun.sleep(200); + const after = await handle(post(call("ping", 4))); + expect(frames(await after.text())[0]?.result).toBe("pong"); +}); + test("a body over maxBodyBytes is refused with 413 before any action runs", async () => { const probe = makeProbe(); const handle = handlerFor(probe, { maxBodyBytes: 16 }); diff --git a/packages/oxidejs/src/rpc/server.ts b/packages/oxidejs/src/rpc/server.ts index e83e850..4e8c3c9 100644 --- a/packages/oxidejs/src/rpc/server.ts +++ b/packages/oxidejs/src/rpc/server.ts @@ -385,7 +385,13 @@ const mergeCallResponses = function mergeCallResponses( calls: ActionCall[], pending: Promise[] ): Response { + // Set when the client cancels the body: later results have nowhere to go, + // and enqueue on a cancelled stream throws. + let cancelled = false; const body = new ReadableStream({ + cancel: () => { + cancelled = true; + }, start: async (controller) => { await Promise.all( pending.map(async (response, index) => { @@ -396,15 +402,22 @@ const mergeCallResponses = function mergeCallResponses( } catch { text = errorLines(calls[index]?.ids ?? [], "Internal error"); } - if (text === "") { + if (text === "" || cancelled) { return; } - controller.enqueue( - encoder.encode(text.endsWith("\n") ? text : `${text}\n`) - ); + try { + controller.enqueue( + encoder.encode(text.endsWith("\n") ? text : `${text}\n`) + ); + } catch { + // The stream went away between the check and the enqueue. + cancelled = true; + } }) ); - controller.close(); + if (!cancelled) { + controller.close(); + } }, }); return new Response(body, {