diff --git a/.changeset/ws-events-transport.md b/.changeset/ws-events-transport.md new file mode 100644 index 0000000000..15551cd10d --- /dev/null +++ b/.changeset/ws-events-transport.md @@ -0,0 +1,8 @@ +--- +'@workflow/world-vercel': minor +'@workflow/builders': patch +'@workflow/rollup': patch +'@workflow/next': patch +--- + +Add an opt-in WebSocket transport for event writes during step execution, enabled with `WORKFLOW_EVENTS_TRANSPORT=ws`. Defaults to HTTP; no behavior change unless explicitly enabled. diff --git a/.github/scripts/aggregate-e2e-results.js b/.github/scripts/aggregate-e2e-results.js index 2041fa73a9..64bd9522dd 100644 --- a/.github/scripts/aggregate-e2e-results.js +++ b/.github/scripts/aggregate-e2e-results.js @@ -260,6 +260,8 @@ function parseJobInfo(filename) { 'mongodb', 'redis', 'starter', + 'nest', + 'tanstack', ].some((app) => p.startsWith(app)) ); diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 63eb544cbc..8e63495fa1 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -604,6 +604,185 @@ jobs: retention-days: 7 if-no-files-found: ignore + e2e-vercel-ws-transport: + # Dedicated coverage for world-vercel's WebSocket events transport + # (WORKFLOW_EVENTS_TRANSPORT=ws — see events-v4.ts's + # isWsEventsTransportEnabled). Every other job in this file exercises + # the default HTTP transport only, now that the WS-on-by-default POC + # hack is gone. + # + # Mirrors e2e-vercel-prod's shape (same apps subset, same full e2e + # suite) but deploys itself via a plain `vercel deploy` (no + # `--prebuilt`) instead of waiting for the GitHub App's passive + # deployment — that's the only way to get a *second*, independent + # deployment of a project we don't own the Environment Variables UI + # for, with its own env var, without creating a whole new project. + # + # Deliberately NOT `--prebuilt`: an earlier version of this job built + # locally on the runner via `vercel build --prebuilt` and hit a string + # of environment-specific bugs that don't reproduce on a real + # deployment (missing `serverExternalPackages` tracing for + # `@workflow/world-vercel`, nitro preset auto-detection picking + # `node-server` instead of `vercel`, needing the whole monorepo + # pre-built on this runner). Plain `vercel deploy` uploads source and + # builds remotely on Vercel's own infra — the exact same pipeline the + # GitHub App uses for e2e-vercel-prod's deployments — so none of that + # applies here. + # Opt-in on PRs, unconditional on main. Three real `vercel deploy`s per + # run is too much to charge every unrelated PR in the repo for a transport + # that is off by default; main still gets the signal on every commit. + # `workflow-server-test` counts as opt-in too — that label exists to test + # workflow-server changes, which is where the WS protocol lives. + # + # `on.pull_request.types` has no `labeled`, so a label added to an already + # open PR takes effect on the next push, not immediately. + name: E2E Vercel WS Transport Test (${{ matrix.app.name }}) + runs-on: ubuntu-latest + timeout-minutes: 30 + needs: ci-scope + if: >- + ${{ needs.ci-scope.outputs.fast-path != 'true' + && (github.event_name != 'pull_request' + || contains(github.event.pull_request.labels.*.name, 'ws-transport-test') + || contains(github.event.pull_request.labels.*.name, 'workflow-server-test')) }} + permissions: + id-token: write + contents: read + strategy: + fail-fast: false + matrix: + app: + - name: "example" + project-id: "prj_xWq20Dd860HHAfzMjK2Mb6TPVxMa" + project-slug: "example-workflow" + - name: "nextjs-turbopack" + project-id: "prj_yjkM7UdHliv8bfxZ1sMJQf1pMpdi" + project-slug: "example-nextjs-workflow-turbopack" + - name: "express" + project-id: "prj_cCZjpBy92VRbKHHbarDMhOHtkuIr" + project-slug: "workbench-express-workflow" + # Vite resolves `ws`'s absent optional accelerators to a stub + # instead of failing the require, so a build check can't see the + # breakage and only a masked frame at or above 48 bytes trips it + # (see WORKFLOW_OPTIONAL_WS_NATIVE_MODULES). Every CBOR event + # frame clears that, so this lane is the coverage for it. + - name: "vite" + project-id: "prj_uLIcNZNDmETulAvj5h0IcDHi5432" + project-slug: "workbench-vite-workflow" + env: + TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} + TURBO_TEAM: ${{ vars.TURBO_TEAM }} + WORKFLOW_PUBLIC_MANIFEST: '1' + WS_TEAM_ID: "team_nO2mCG4W8IxPIeKoSsqwAxxB" + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + + - name: Setup environment + uses: ./.github/actions/setup-workflow-dev + with: + build-packages: 'false' + + - name: Build CLI + run: pnpm turbo run build --filter='@workflow/cli' + + # Pinned and installed once rather than `npx vercel@` per + # command — repeated `npx` resolution showed inconsistent behavior + # (different Node version in the CLI's own banner, a doubled project + # path) across otherwise-identical earlier CI attempts. + - name: Install vercel CLI + run: npm install -g vercel@56.2.0 + + - name: Deploy a dedicated WS-transport preview + id: wsDeploy + env: + WS_VERCEL_TOKEN: ${{ secrets.VERCEL_LABS_TOKEN }} + run: | + set -euo pipefail + # `vercel deploy` (source-upload mode, no --prebuilt) re-applies + # the linked project's own Root Directory setting internally — + # unlike `vercel build`, it expects to run from the repo root and + # narrow down itself, the same way the GitHub App's own deploy + # does. Passing `--cwd=` here double-joins that Root + # Directory underneath itself ("workbench/x/workbench/x does not + # exist"). Link and deploy from the repo root — no --cwd at all — + # and let the project's own Root Directory setting do the + # narrowing, matching how every passively-triggered deployment in + # this file already works. + vercel pull --yes --environment=preview \ + --project="${{ matrix.app.project-id }}" \ + --scope="$WS_TEAM_ID" --token="$WS_VERCEL_TOKEN" + # `vercel deploy` writes the new deployment's URL to stdout and + # progress/logs to stderr, specifically so it composes as + # `url=$(vercel deploy ...)` in scripts — do not redirect stderr + # into this capture. `--target=preview` is explicit here even + # though it's the default without `--prod` — these are shared, + # production-adjacent projects, and this shouldn't ever silently + # start deploying to their production aliases if a future edit + # changes flags around this call. + URL=$(vercel deploy --yes --target=preview \ + -e WORKFLOW_EVENTS_TRANSPORT=ws \ + --scope="$WS_TEAM_ID" --token="$WS_VERCEL_TOKEN") + echo "url=$URL" >> "$GITHUB_OUTPUT" + # Best-effort: only used by the runtime-logs capture on failure, + # which already skips gracefully if this doesn't parse. + DEPLOYMENT_ID=$(vercel inspect "$URL" --format=json \ + --scope="$WS_TEAM_ID" --token="$WS_VERCEL_TOKEN" 2>/dev/null \ + | node -e "let d='';process.stdin.on('data',c=>d+=c).on('end',()=>{try{console.log(JSON.parse(d).id||'')}catch{}})" || true) + echo "deploymentId=$DEPLOYMENT_ID" >> "$GITHUB_OUTPUT" + + - name: Record E2E start time + id: e2eStart + run: echo "ms=$(($(date +%s) * 1000))" >> "$GITHUB_OUTPUT" + + - name: Run E2E Tests + run: pnpm run test:e2e --reporter=verbose --reporter=json --reporter=./packages/core/e2e/github-reporter.ts "--outputFile=e2e-vercel-ws-transport-$APP_NAME.json" + env: + NODE_OPTIONS: "--enable-source-maps" + DEPLOYMENT_URL: ${{ steps.wsDeploy.outputs.url }} + VERCEL_DEPLOYMENT_ID: ${{ steps.wsDeploy.outputs.deploymentId }} + APP_NAME: ${{ matrix.app.name }} + WORKFLOW_VERCEL_ENV: "preview" + WORKFLOW_VERCEL_AUTH_TOKEN: ${{ secrets.VERCEL_LABS_TOKEN }} + WORKFLOW_VERCEL_TEAM: ${{ env.WS_TEAM_ID }} + WORKFLOW_VERCEL_PROJECT: ${{ matrix.app.project-id }} + WORKFLOW_VERCEL_PROJECT_SLUG: ${{ matrix.app.project-slug }} + VERCEL_WORKFLOW_SERVER_URL: ${{ github.ref != 'refs/heads/main' && secrets.VERCEL_WORKFLOW_SERVER_URL || '' }} + + - name: Capture runtime logs on failure + if: failure() + env: + APP_NAME: ${{ matrix.app.name }} + WORKFLOW_VERCEL_AUTH_TOKEN: ${{ secrets.VERCEL_LABS_TOKEN }} + WORKFLOW_VERCEL_TEAM: ${{ env.WS_TEAM_ID }} + WORKFLOW_VERCEL_PROJECT: ${{ matrix.app.project-id }} + WORKFLOW_VERCEL_ENV: "preview" + VERCEL_DEPLOYMENT_ID: ${{ steps.wsDeploy.outputs.deploymentId }} + E2E_START_MS: ${{ steps.e2eStart.outputs.ms }} + run: node .github/scripts/fetch-e2e-runtime-logs.mjs + + - name: Generate E2E summary + if: always() + env: + APP_NAME: ${{ matrix.app.name }} + run: node .github/scripts/aggregate-e2e-results.js . --job-name "E2E Vercel WS Transport ($APP_NAME)" >> $GITHUB_STEP_SUMMARY || true + + - name: Upload E2E results + if: always() + uses: actions/upload-artifact@v4 + with: + name: e2e-results-vercel-ws-transport-${{ matrix.app.name }} + path: | + e2e-vercel-ws-transport-${{ matrix.app.name }}.json + e2e-metadata-${{ matrix.app.name }}-vercel.json + e2e-failures-${{ matrix.app.name }}-vercel.json + e2e-diagnostics-${{ matrix.app.name }}-vercel.json + e2e-runtime-logs-${{ matrix.app.name }}-vercel.json + retention-days: 7 + if-no-files-found: ignore + getTestMatrix: name: Get Test Matrix runs-on: ubuntu-latest @@ -1129,7 +1308,7 @@ jobs: summary: name: E2E Summary runs-on: ubuntu-latest - needs: [ci-scope, e2e-vercel-prod, e2e-local-dev, e2e-local-prod, e2e-local-postgres, e2e-windows] + needs: [ci-scope, e2e-vercel-prod, e2e-vercel-ws-transport, e2e-local-dev, e2e-local-prod, e2e-local-postgres, e2e-windows] if: always() && !cancelled() && needs.ci-scope.outputs.fast-path != 'true' timeout-minutes: 10 @@ -1165,7 +1344,7 @@ jobs: e2e-required-check: name: E2E Required Check runs-on: ubuntu-latest - needs: [ci-scope, unit, e2e-package-build, e2e-vercel-prod, e2e-local-dev, e2e-local-prod, e2e-local-postgres, e2e-windows] + needs: [ci-scope, unit, e2e-package-build, e2e-vercel-prod, e2e-vercel-ws-transport, e2e-local-dev, e2e-local-prod, e2e-local-postgres, e2e-windows] if: always() timeout-minutes: 5 @@ -1175,6 +1354,7 @@ jobs: UNIT_STATUS: ${{ needs.unit.result }} BUILD_STATUS: ${{ needs.e2e-package-build.result }} VERCEL_STATUS: ${{ needs.e2e-vercel-prod.result }} + VERCEL_WS_STATUS: ${{ needs.e2e-vercel-ws-transport.result }} LOCAL_DEV_STATUS: ${{ needs.e2e-local-dev.result }} LOCAL_PROD_STATUS: ${{ needs.e2e-local-prod.result }} POSTGRES_STATUS: ${{ needs.e2e-local-postgres.result }} @@ -1182,6 +1362,12 @@ jobs: FAST_PATH: ${{ needs.ci-scope.outputs.fast-path }} VALIDATION_FAST_PATH: ${{ needs.ci-scope.outputs.validation-fast-path }} HAS_LABEL: ${{ contains(github.event.pull_request.labels.*.name, 'workflow-server-test') }} + # e2e-vercel-ws-transport runs on main and on labelled PRs only, so + # its status is required in exactly those cases and must be allowed + # to be `skipped` otherwise. + WS_REQUIRED: ${{ github.event_name != 'pull_request' + || contains(github.event.pull_request.labels.*.name, 'ws-transport-test') + || contains(github.event.pull_request.labels.*.name, 'workflow-server-test') }} run: | FAILED_JOBS=() @@ -1199,6 +1385,9 @@ jobs: if [[ "$HAS_LABEL" == "true" ]]; then echo "workflow-server-test label detected - only checking required jobs" [[ "$VERCEL_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-vercel-prod ($VERCEL_STATUS)") + # This label is itself opt-in for the WS lane (see that job's `if`), + # so it runs here and is required. + [[ "$VERCEL_WS_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-vercel-ws-transport ($VERCEL_WS_STATUS)") # Verify other jobs were actually skipped [[ "$UNIT_STATUS" == "skipped" ]] || echo "Warning: unit was not skipped ($UNIT_STATUS)" @@ -1211,6 +1400,11 @@ jobs: [[ "$UNIT_STATUS" == "success" ]] || FAILED_JOBS+=("unit ($UNIT_STATUS)") [[ "$BUILD_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-package-build ($BUILD_STATUS)") [[ "$VERCEL_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-vercel-prod ($VERCEL_STATUS)") + if [[ "$WS_REQUIRED" == "true" ]]; then + [[ "$VERCEL_WS_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-vercel-ws-transport ($VERCEL_WS_STATUS)") + else + [[ "$VERCEL_WS_STATUS" == "skipped" ]] || echo "Warning: e2e-vercel-ws-transport ran unlabelled ($VERCEL_WS_STATUS)" + fi [[ "$LOCAL_DEV_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-local-dev ($LOCAL_DEV_STATUS)") [[ "$LOCAL_PROD_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-local-prod ($LOCAL_PROD_STATUS)") [[ "$POSTGRES_STATUS" == "success" ]] || FAILED_JOBS+=("e2e-local-postgres ($POSTGRES_STATUS)") diff --git a/docs/content/docs/v5/configuration/worlds.mdx b/docs/content/docs/v5/configuration/worlds.mdx index be63f77ded..72487e4a1f 100644 --- a/docs/content/docs/v5/configuration/worlds.mdx +++ b/docs/content/docs/v5/configuration/worlds.mdx @@ -273,3 +273,11 @@ Platform-provided values such as `VERCEL_DEPLOYMENT_ID`, `VERCEL_PROJECT_ID`, an - CLI flag: none - Default: `1000` - Maximum stream chunks written in one Vercel World request. Larger batches are split. + +### `WORKFLOW_EVENTS_TRANSPORT` + +- Factory option: none +- CLI flag: none +- Default: `http` +- Experimental. Set to `ws` to ship workflow run events to the Vercel World over a WebSocket instead of one HTTP request each. +- Ignored when the World is configured with `projectConfig` and routes through the `api-workflow` proxy — that endpoint is an HTTP-only REST gateway and does not forward a WebSocket upgrade, so events stay on HTTP. diff --git a/docs/content/worlds/v5/vercel.mdx b/docs/content/worlds/v5/vercel.mdx index 100e3973f8..2c2570c26b 100644 --- a/docs/content/worlds/v5/vercel.mdx +++ b/docs/content/worlds/v5/vercel.mdx @@ -196,6 +196,12 @@ Per-request timeout, in milliseconds, for Vercel World HTTP calls to workflow-se Maximum stream chunks written in one Vercel World request. Larger batches are split across multiple requests. Default: `1000`. Minimum: `1`. +### `WORKFLOW_EVENTS_TRANSPORT` + +Experimental. Set `WORKFLOW_EVENTS_TRANSPORT=ws` to ship workflow run events to the Vercel World over a WebSocket instead of one HTTP request each. Default: `http`. + +The setting is ignored when the World is configured with `projectConfig` and therefore routes through the `api-workflow` proxy: that endpoint is an HTTP-only REST gateway and does not forward a WebSocket upgrade, so events stay on HTTP and a warning is logged once per process. + ### Programmatic configuration `createWorld()` accepts explicit API configuration. It does not read `WORKFLOW_VERCEL_*` automatically, so pass the environment values yourself when you want a configured World module: diff --git a/packages/builders/src/index.ts b/packages/builders/src/index.ts index b9ea2adf77..0214bcee51 100644 --- a/packages/builders/src/index.ts +++ b/packages/builders/src/index.ts @@ -32,6 +32,7 @@ export { export { createNodeModuleErrorPlugin } from './node-module-esbuild-plugin.js'; export { WORKFLOW_OPTIONAL_OTEL_API_MODULE } from './optional-otel-api.js'; export { WORKFLOW_OPTIONAL_TYPESCRIPT_ALIAS } from './optional-typescript-alias.js'; +export { WORKFLOW_OPTIONAL_WS_NATIVE_MODULES } from './optional-ws-native.js'; export { createPseudoPackagePlugin, PSEUDO_PACKAGES, diff --git a/packages/builders/src/optional-ws-native.ts b/packages/builders/src/optional-ws-native.ts new file mode 100644 index 0000000000..031f43356e --- /dev/null +++ b/packages/builders/src/optional-ws-native.ts @@ -0,0 +1,32 @@ +/** + * `bufferutil` and `utf-8-validate` are optional native accelerators for the + * `ws` package, which `@workflow/world-vercel` uses for its WebSocket events + * transport. Neither is installed by default: `ws` requires each one inside a + * try/catch and falls back to a pure-JS implementation when the require throws. + * + * Marking them external keeps that fallback reachable. Two bundlers take it away + * without changing the build's outcome, so the damage only shows up at runtime, + * on the first frame big enough to reach the native masker (`ws` uses pure JS + * below 48 bytes — small-payload smoke tests pass; CBOR event frames don't): + * + * - **webpack** bundles the JS wrapper without its native `.node` binding. + * - **Vite** resolves the absent peer to its own `optional-peer-dep` stub, so + * `ws`'s require *succeeds* and the try/catch never fires. + * + * Both end at `bufferUtil.mask is not a function`. Rollup-family setups are not + * known to break: nothing substitutes a stub, and nitro traces and externalizes + * `ws` in a production build, so the require stays a real one that fails and the + * fallback engages. Externalizing there buys determinism against a change in + * externals policy, not a fix for a known break. (Turbopack already externalizes + * Node packages.) + * + * Externalized unconditionally, unlike `WORKFLOW_OPTIONAL_OTEL_API_MODULE`, + * which is only externalized when it can't be resolved. The OTEL API has to + * load for tracing to work, so a self-contained output must bundle it when + * present; these accelerators must specifically *not* load, so a failed runtime + * require is the designed path and a half-bundled native module is not. + */ +export const WORKFLOW_OPTIONAL_WS_NATIVE_MODULES = [ + 'bufferutil', + 'utf-8-validate', +] as const; diff --git a/packages/next/src/index.ts b/packages/next/src/index.ts index 58aee9c55c..98a3e465cf 100644 --- a/packages/next/src/index.ts +++ b/packages/next/src/index.ts @@ -4,6 +4,7 @@ import { dirname, isAbsolute, join } from 'node:path'; import { resolveConfiguredProjectRoot, resolveProjectRoot, + WORKFLOW_OPTIONAL_WS_NATIVE_MODULES, } from '@workflow/builders'; import type { NextConfig } from 'next'; import semver from 'semver'; @@ -321,6 +322,26 @@ function registerWorkflowDiagnosticsManifestCopy(metadata: { globalWithMarker[marker].push(metadata); } +/** + * Mark `ws`'s optional native accelerators external on the webpack server + * build, which otherwise bundles their JS wrapper without the native `.node` + * binding and throws "bufferUtil.mask is not a function" at runtime. See + * `WORKFLOW_OPTIONAL_WS_NATIVE_MODULES`; `@workflow/rollup` handles the + * Rollup/Vite/Nitro side. + */ +function externalizeWsNativeAccelerators(webpackConfig: { + externals?: unknown; +}): void { + const names = [...WORKFLOW_OPTIONAL_WS_NATIVE_MODULES]; + if (Array.isArray(webpackConfig.externals)) { + webpackConfig.externals.push(...names); + } else if (webpackConfig.externals) { + webpackConfig.externals = [webpackConfig.externals, ...names]; + } else { + webpackConfig.externals = names; + } +} + export function withWorkflow( nextConfigOrFn: | NextConfig @@ -571,6 +592,10 @@ export function withWorkflow( loader: loaderPath, }); + if (args[1]?.isServer) { + externalizeWsNativeAccelerators(webpackConfig); + } + return existingWebpackModify ? (existingWebpackModify(...args) ?? webpackConfig) : webpackConfig; diff --git a/packages/rollup/src/index.test.ts b/packages/rollup/src/index.test.ts index 2855259064..42c15f1c4f 100644 --- a/packages/rollup/src/index.test.ts +++ b/packages/rollup/src/index.test.ts @@ -1,4 +1,7 @@ -import { WORKFLOW_OPTIONAL_OTEL_API_MODULE } from '@workflow/builders'; +import { + WORKFLOW_OPTIONAL_OTEL_API_MODULE, + WORKFLOW_OPTIONAL_WS_NATIVE_MODULES, +} from '@workflow/builders'; import { describe, expect, it, vi } from 'vitest'; import { workflowTransformPlugin } from './index.js'; @@ -71,3 +74,61 @@ describe('workflowTransformPlugin resolveId — @opentelemetry/api optional peer await expect(resolveId('some-other-package')).resolves.toBeNull(); }); }); + +/** + * `ws` (world-vercel's WebSocket events transport) requires `bufferutil` and + * `utf-8-validate` inside a try/catch and falls back to pure JS when they're + * absent — which is the default, since neither is installed. The require has to + * actually fail for that fallback to engage, and under Vite it doesn't: an + * absent optional peer resolves to a stub, leaving `bufferUtil.mask` undefined + * until a frame is masked. Externalizing makes the failure the designed one on + * every bundler this plugin runs under. + * + * This is the shipped fix for what the workbench Vite/TanStack apps used to + * work around locally via `nitro.rollupConfig.external`: because every + * Nitro-based integration (and SvelteKit, and Astro) installs this plugin, a + * real user gets it without hand-editing their own bundler config. + */ +describe('workflowTransformPlugin resolveId — ws optional native accelerators', () => { + it.each([ + ...WORKFLOW_OPTIONAL_WS_NATIVE_MODULES, + ])('marks %s external so nothing substitutes for it', async (name) => { + const { resolveId } = getResolveId(() => null); + await expect(resolveId(name)).resolves.toEqual({ + id: name, + external: true, + }); + }); + + it('externalizes them even when they ARE installed', async () => { + // Deliberately the opposite of the OTEL peer above. A resolvable + // accelerator must still be externalized: bundling it pulls in the JS + // wrapper without its native `.node` binding, which is the broken + // half-module that throws "bufferUtil.mask is not a function" at runtime. + // `ws`'s own try/catch fallback makes a failed require the safe path. + const { ctx, resolveId } = getResolveId(() => ({ + id: '/node_modules/bufferutil/index.js', + external: false, + })); + await expect(resolveId('bufferutil')).resolves.toEqual({ + id: 'bufferutil', + external: true, + }); + // Short-circuits before consulting the resolver at all. + expect(ctx.resolve).not.toHaveBeenCalled(); + }); + + it('covers deep imports into the accelerators', async () => { + const { resolveId } = getResolveId(() => null); + await expect(resolveId('bufferutil/fallback')).resolves.toEqual({ + id: 'bufferutil/fallback', + external: true, + }); + }); + + it('does not intercept lookalike package names', async () => { + const { resolveId } = getResolveId(); + await expect(resolveId('bufferutil-extra')).resolves.toBeNull(); + await expect(resolveId('ws')).resolves.toBeNull(); + }); +}); diff --git a/packages/rollup/src/index.ts b/packages/rollup/src/index.ts index 0254d0d8af..921fca756d 100644 --- a/packages/rollup/src/index.ts +++ b/packages/rollup/src/index.ts @@ -6,6 +6,7 @@ import { resolveModuleSpecifier, shouldTransformFile, WORKFLOW_OPTIONAL_OTEL_API_MODULE, + WORKFLOW_OPTIONAL_WS_NATIVE_MODULES, } from '@workflow/builders'; import { resolveModulePath } from 'exsolve'; import type { Plugin } from 'rollup'; @@ -49,6 +50,21 @@ export function workflowTransformPlugin( return resolved ?? { id: source, external: true }; } + // `ws`'s optional native accelerators. Unlike the OTEL peer above these + // are externalized unconditionally, not only when unresolvable — `ws` + // requires them in a try/catch, so a failed runtime require is the + // designed path while a partially bundled native module is not. This + // plugin also runs under Vite, which substitutes a stub for an absent + // optional peer and would let that require succeed with no working + // masker. See WORKFLOW_OPTIONAL_WS_NATIVE_MODULES. + if ( + WORKFLOW_OPTIONAL_WS_NATIVE_MODULES.some( + (name) => source === name || source.startsWith(`${name}/`) + ) + ) { + return { id: source, external: true }; + } + return null; }, }, diff --git a/packages/world-vercel/package.json b/packages/world-vercel/package.json index 5d2cbc21cf..425642aa46 100644 --- a/packages/world-vercel/package.json +++ b/packages/world-vercel/package.json @@ -42,6 +42,7 @@ "cbor-x": "1.6.0", "ulid": "catalog:", "undici": "catalog:", + "ws": "^8.20.0", "zod": "catalog:" }, "peerDependencies": { @@ -58,6 +59,7 @@ "@opentelemetry/core": "1.30.1", "@opentelemetry/sdk-trace-base": "1.30.1", "@types/node": "catalog:", + "@types/ws": "^8.18.1", "@workflow/tsconfig": "workspace:*", "genversion": "3.2.0", "typescript": "catalog:", diff --git a/packages/world-vercel/src/event-retry.test.ts b/packages/world-vercel/src/event-retry.test.ts index 15c16abc45..366ad31bff 100644 --- a/packages/world-vercel/src/event-retry.test.ts +++ b/packages/world-vercel/src/event-retry.test.ts @@ -102,6 +102,62 @@ describe('isRetryableEventPostError', () => { ).toBe(true); }); + it('retries a TRANSPORT failure from either transport', () => { + // The one code that lets this policy serve HTTP and WS alike: `utils.ts` + // sets it for a `fetch` that failed transiently, `events-v4.ts` for a WS + // socket that died or produced no correlatable reply. Both mean the write + // was never acked. + expect( + isRetryableEventPostError( + new WorkflowWorldError('POST … transport failure: socket hang up', { + code: 'TRANSPORT', + }) + ) + ).toBe(true); + }); + + it('retries a TRANSPORT failure whose cause carries no known marker', () => { + // Why the code has to be read directly rather than left to the cause walk: + // a WS failure's cause is a `WsTransportError`, which carries none of the + // undici/Node markers in TRANSIENT_CODES. + const cause = Object.assign(new Error('connection closed (code 1006)'), { + name: 'WsTransportError', + }); + expect(isRetryableEventPostError(cause)).toBe(false); + expect( + isRetryableEventPostError( + new WorkflowWorldError('POST … transport failure: connection closed', { + code: 'TRANSPORT', + cause, + }) + ) + ).toBe(true); + }); + + it('does not retry a TIMEOUT that is a caller-requested abort', () => { + // `utils.ts` maps both a missed deadline and an `AbortError` onto + // `code: 'TIMEOUT'`, so — unlike TRANSPORT — that code must not be a + // blanket yes; it falls through to the marker walk, which tells them apart. + expect( + isRetryableEventPostError( + new WorkflowWorldError('request aborted', { + code: 'TIMEOUT', + cause: Object.assign(new Error('aborted'), { name: 'AbortError' }), + }) + ) + ).toBe(false); + expect( + isRetryableEventPostError( + new WorkflowWorldError('request timed out', { + code: 'TIMEOUT', + cause: Object.assign(new Error('timed out'), { + name: 'TimeoutError', + }), + }) + ) + ).toBe(true); + }); + it('retries raw transport errors by code', () => { for (const code of [ 'ECONNRESET', diff --git a/packages/world-vercel/src/event-retry.ts b/packages/world-vercel/src/event-retry.ts index 103e798b02..38db81fd0f 100644 --- a/packages/world-vercel/src/event-retry.ts +++ b/packages/world-vercel/src/event-retry.ts @@ -41,6 +41,11 @@ * * Only transient/ambiguous transport failures are retried; definitive responses * (409/410/425/429 and any other 4xx) surface immediately, exactly as before. + * + * This is the *only* retry loop on the event-write path, for both transports. + * The WebSocket transport raises `code: 'TRANSPORT'` — the shape `utils.ts` + * produces for a failed `fetch` — so it lands here rather than carrying a + * second policy blind to `EVENT_RETRY_ELIGIBILITY`. */ import { @@ -225,6 +230,22 @@ export function isRetryableEventPostError(err: unknown): boolean { // Body parsed past the response but the write may have landed — safe to // retry for eligible events (a landed original re-surfaces as 409). if (err.code === 'PARSE_ERROR') return true; + // A transport failure the world layer already classified as transient: + // `utils.ts` sets this for a `fetch` failing with a + // `TRANSIENT_TRANSPORT_ERROR_CODES` code, `events-v4.ts` for a WS socket + // that died, refused a send, produced no correlatable reply or missed its + // deadline. The frame was never acked in any of them. Keying on the code + // rather than re-deriving transience from the cause chain is what lets one + // policy serve both transports; it also collapses a drift, since + // `UND_ERR_CONNECT`, `UND_ERR_CLOSED` and `EAI_AGAIN` are in utils.ts's set + // but not in TRANSIENT_CODES below, so they counted as retryable for queue + // redelivery yet were skipped in-process. + // + // `TIMEOUT` is deliberately excluded: utils.ts maps both a missed deadline + // and a caller's `AbortError` onto that one code, so it must keep falling + // through to the marker walk, which retries the former and refuses the + // latter. + if (err.code === 'TRANSPORT') return true; if (typeof err.status === 'number') { // Transient server errors; 4xx are definitive and not retried. return err.status >= 500 && err.status <= 599; diff --git a/packages/world-vercel/src/events-v4-ws.test.ts b/packages/world-vercel/src/events-v4-ws.test.ts new file mode 100644 index 0000000000..7efa0a2361 --- /dev/null +++ b/packages/world-vercel/src/events-v4-ws.test.ts @@ -0,0 +1,439 @@ +/** + * `createWorkflowRunEventV4`'s WebSocket branch (`WORKFLOW_EVENTS_TRANSPORT=ws`). + * + * Kept out of `events-v4.test.ts` because the WS branch needs + * `./ws-transport.js` mocked at the module level, and that file's HTTP + * cases must keep exercising the real one. The transport's own socket + * lifecycle is covered in `ws-transport.test.ts`; what matters here is the + * adapter layer — how a reply frame becomes a `Response`-like result, and + * what happens when it isn't one this client understands. + */ + +import { EntityConflictError, WorkflowWorldError } from '@workflow/errors'; +import { encode } from 'cbor-x'; +import { MockAgent } from 'undici'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + isRetryableEventPostError, + MAX_EVENT_POST_RETRIES, + withEventPostRetry, +} from './event-retry.js'; +import { createWorkflowRunEventV4 } from './events-v4.js'; +import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; +import { type WsFrameReply, WsTransportError } from './ws-transport.js'; + +const WS_URL = 'wss://vercel-workflow.com/api/websockets/v1/runs/wrun_1'; + +const requestMock = vi.fn<() => Promise>(); +/** + * `resolveWsTransport` is the seam to mock, not `getWsEventsTransport`: URL + * resolution, proxy fallback and the header thunk all call each other *inside* + * `ws-transport.js`, and an ESM mock replaces a module's exports, not its own + * call sites. Those selection cases live in `ws-transport.test.ts`. + */ +const resolveWsTransportMock = vi.fn(() => ({ + transport: { request: requestMock }, + wsUrl: WS_URL, +})); + +vi.mock('./ws-transport.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + resolveWsTransport: (...args: unknown[]) => + resolveWsTransportMock(...(args as [])), + }; +}); + +const CREATED_AT = '2026-06-10T00:00:00.000Z'; + +const input = { + runId: 'wrun_1', + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', +} as const; + +/** + * The materialized CBOR body a `step_completed` write answers with. Shared by + * the WS reply frames and the HTTP interceptors below so the two transports are + * asserted against one shape — the point of the adapter is that the caller + * cannot tell them apart. + */ +const materializedBody = () => + new Uint8Array( + encode({ + event: { + eventId: 'evnt_1', + runId: 'wrun_1', + createdAt: CREATED_AT, + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', + eventData: { result: new Uint8Array() }, + }, + step: { + runId: 'wrun_1', + stepId: 'step_1', + stepName: 'step', + status: 'completed', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + }, + }) + ); + +/** A well-formed `event_ack` reply frame. */ +const ack = ( + meta: Record = {}, + body: Uint8Array = materializedBody() +): WsFrameReply => ({ + meta: { + reqId: 1, + type: 'event_ack', + status: 201, + eventId: 'evnt_1', + runId: 'wrun_1', + createdAt: CREATED_AT, + ...meta, + }, + body, +}); + +beforeEach(() => { + vi.clearAllMocks(); + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; +}); + +afterEach(() => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; +}); + +/** + * The gate is the whole safety story for this feature: everything else in + * the PR is dead code for anyone who hasn't opted in. `events-v4.test.ts` + * covers the HTTP path itself in depth, but nothing there pins the + * *choice* of path — so a future edit that flipped the default (as an + * earlier revision of this branch did deliberately, for benchmarking) + * would sail through with every HTTP assertion still green, because the + * two transports are built to be indistinguishable at the result layer. + */ +describe('transport gate', () => { + it('goes over HTTP, never touching the WS transport, when the gate is off', async () => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + const origin = + WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(origin) + .intercept({ + path: '/api/v4/runs/wrun_1/events/step_completed', + method: 'POST', + }) + .reply(200, materializedBody(), { + headers: { + 'x-wf-event-id': 'evnt_1', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': CREATED_AT, + }, + }); + + const result = await createWorkflowRunEventV4(input, { + token: 'test-token', + dispatcher: agent, + }); + + expect(result.event.eventId).toBe('evnt_1'); + expect(resolveWsTransportMock).not.toHaveBeenCalled(); + agent.assertNoPendingInterceptors(); + }); +}); + +describe('createWorkflowRunEventV4 over ws', () => { + it('maps an event_ack reply onto the same result shape as HTTP', async () => { + requestMock.mockResolvedValueOnce(ack()); + + const result = await createWorkflowRunEventV4(input, { + token: 'test-token', + }); + + expect(result.event.eventId).toBe('evnt_1'); + expect(result.event.runId).toBe('wrun_1'); + expect(result.event.eventType).toBe('step_completed'); + expect(result.step).toMatchObject({ stepId: 'step_1' }); + }); + + it('falls back to HTTP when this World has no usable WS transport', async () => { + // `resolveWsTransport` returns null for the api-workflow proxy World (see + // `ws-transport.test.ts`). The gate is still on here, so the write has to + // complete over HTTP rather than fail — that fallback is the reason the + // resolve step returns null instead of throwing. + resolveWsTransportMock.mockReturnValueOnce( + null as unknown as { + transport: { request: typeof requestMock }; + wsUrl: string; + } + ); + const origin = + WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(origin) + .intercept({ + path: '/api/v4/runs/wrun_1/events/step_completed', + method: 'POST', + }) + .reply(200, materializedBody(), { + headers: { + 'x-wf-event-id': 'evnt_1', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': CREATED_AT, + }, + }); + + const result = await createWorkflowRunEventV4(input, { + token: 'test-token', + dispatcher: agent, + }); + + expect(result.event.eventId).toBe('evnt_1'); + expect(requestMock).not.toHaveBeenCalled(); + agent.assertNoPendingInterceptors(); + }); + + it('translates a non-2xx status into the same typed error HTTP raises', async () => { + requestMock.mockResolvedValueOnce( + ack( + { status: 409, eventId: undefined }, + new TextEncoder().encode('{"message":"already applied"}') + ) + ); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(EntityConflictError); + }); + + it('fails closed when the reply carries no numeric status', async () => { + // The protocol is designed to grow new response variants, so an older + // client will meet a reply it doesn't understand. Defaulting that to + // 200 would report a write as applied when nothing confirmed it. + requestMock.mockResolvedValueOnce({ + meta: { reqId: 1, type: 'some_future_reply' }, + body: new Uint8Array(0), + }); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(/no numeric status/); + }); + + it('fails closed on an error frame rather than reading it as success', async () => { + // 403 rather than 500: a 5xx is retryable (see the retry-parity suite + // below), so it would be absorbed rather than surfaced. What this pins + // is that an `error` frame is never mistaken for an ack. + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'error', status: 403 }, + body: new TextEncoder().encode('{"message":"boom"}'), + }); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(); + }); +}); + +/** + * Retry is owned by `withEventPostRetry` (`event-retry.ts`) for both transports + * — it is the only layer that knows which event types are idempotent-on-retry, + * and an earlier revision of this branch had a *second* loop inside the WS + * adapter that sat underneath it and therefore defeated that gate entirely + * (`step_started` got retried 6 times, double-incrementing the attempt counter, + * while an eligible type got 3 × 6 attempts with the inner backoff reaching 30s + * against an outer base deliberately set to 100ms). + * + * What keeps the shared loop working for WS without a WS-specific branch is the + * error *shape*: every transport failure leaves this adapter as a + * `WorkflowWorldError` with `code: 'TRANSPORT'`, the same thing `utils.ts` + * produces for a failed `fetch`. These tests pin both halves — that the adapter + * itself attempts exactly once, and that what it throws is classified the way + * the shared policy needs. + */ +describe('retry is owned by the shared policy, not the adapter', () => { + it.each([ + 500, 502, 503, 504, + ])('attempts a %i reply exactly once — the retry lives one layer up', async (status) => { + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'error', status }, + body: new TextEncoder().encode('{"message":"transient"}'), + }); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + it('does not retry a 4xx the server will reject identically', async () => { + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'error', status: 409 }, + body: new TextEncoder().encode('{"message":"already applied"}'), + }); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(EntityConflictError); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + it('wraps a transport failure as a TRANSPORT WorkflowWorldError', async () => { + // A bare `WsTransportError` fails `WorkflowWorldError.is()` (the `@workflow/errors` + // checks are name-based), which would let it escape both the in-process + // retry and the queue's redelivery classification and surface as a + // USER_ERROR — blaming the workflow's own code for a dead socket. + const cause = new WsTransportError('connection closed (code 1006)'); + requestMock.mockRejectedValue(cause); + + const err = await createWorkflowRunEventV4(input, { + token: 'test-token', + }).catch((e: unknown) => e); + + expect(WorkflowWorldError.is(err)).toBe(true); + expect((err as WorkflowWorldError).code).toBe('TRANSPORT'); + expect((err as WorkflowWorldError).cause).toBe(cause); + expect((err as Error).message).toMatch(/connection closed \(code 1006\)/); + // And the shared policy recognizes it, which is the whole point of the shape. + expect(isRetryableEventPostError(err)).toBe(true); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + it('raises an unrecognized reply variant as a PARSE_ERROR', async () => { + // A version skew, not a transient fault — but the write may or may not have + // landed, so it is retryable for the types that can absorb a replay (a + // landed original comes back as a 409 the SDK already handles). + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'some_future_reply' }, + body: new Uint8Array(0), + }); + + const err = await createWorkflowRunEventV4(input, { + token: 'test-token', + }).catch((e: unknown) => e); + + expect((err as WorkflowWorldError).code).toBe('PARSE_ERROR'); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + it('leaves a 429 to the runtime rather than absorbing it in-process', async () => { + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'error', status: 429, retryAfter: '2' }, + body: new TextEncoder().encode('{"message":"slow down"}'), + }); + + const err = await createWorkflowRunEventV4(input, { + token: 'test-token', + }).catch((e: unknown) => e); + + expect(isRetryableEventPostError(err)).toBe(false); + expect(requestMock).toHaveBeenCalledTimes(1); + }); +}); + +/** + * The composition, end to end: the same loop the HTTP path already runs under, + * driving WS writes. These would have passed before too — but only because the + * adapter's own loop was doing the work; they pin that the *outer* loop is what + * retries now, and in particular that its per-event-type gate is back in force. + */ +describe('withEventPostRetry over ws', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + /** Drive a call to completion with the backoff timers auto-advanced. */ + const runWithTimers = async (promise: Promise): Promise => { + const settled = promise.then( + (value) => ({ ok: true as const, value }), + (error) => ({ ok: false as const, error }) + ); + await vi.runAllTimersAsync(); + const result = await settled; + if (!result.ok) throw result.error; + return result.value; + }; + + it('retries a dead socket for an idempotent-on-retry event type', async () => { + requestMock + .mockRejectedValueOnce(new WsTransportError('connection closed (1006)')) + .mockResolvedValueOnce(ack()); + + const result = await runWithTimers( + withEventPostRetry( + () => createWorkflowRunEventV4(input, { token: 'test-token' }), + 'step_completed' + ) + ); + + expect(result.event.eventId).toBe('evnt_1'); + expect(requestMock).toHaveBeenCalledTimes(2); + }); + + it('surfaces it for queue redelivery once the shared cap is reached', async () => { + requestMock.mockRejectedValue(new WsTransportError('connection closed')); + + await expect( + runWithTimers( + withEventPostRetry( + () => createWorkflowRunEventV4(input, { token: 'test-token' }), + 'step_completed' + ) + ) + ).rejects.toThrow(/connection closed/); + // MAX_EVENT_POST_RETRIES + the initial attempt — not the 6 the removed + // adapter-level loop produced, and not 3 × 6 in composition with it. + expect(requestMock).toHaveBeenCalledTimes(MAX_EVENT_POST_RETRIES + 1); + }); + + it('does NOT retry step_started, whose handler double-counts attempts', async () => { + // This is the case the adapter-level loop silently defeated: the gate lives + // in EVENT_RETRY_ELIGIBILITY, one layer above where that loop sat. + requestMock.mockRejectedValue(new WsTransportError('connection closed')); + + await expect( + runWithTimers( + withEventPostRetry( + () => + createWorkflowRunEventV4( + { ...input, eventType: 'step_started' }, + { token: 'test-token' } + ), + 'step_started' + ) + ) + ).rejects.toThrow(/connection closed/); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + it('does not retry a definitive 4xx reply', async () => { + requestMock.mockResolvedValue({ + meta: { reqId: 1, type: 'error', status: 409 }, + body: new TextEncoder().encode('{"message":"already applied"}'), + }); + + await expect( + runWithTimers( + withEventPostRetry( + () => createWorkflowRunEventV4(input, { token: 'test-token' }), + 'step_completed' + ) + ) + ).rejects.toThrow(EntityConflictError); + expect(requestMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index e59e75ca9d..ab712abf95 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -50,12 +50,15 @@ import { } from './http-client.js'; import { errorForResponse, + headersToRecord, instrumentedFetch, parseRetryAfter, } from './http-core.js'; import { hasSerializedDataFormatPrefix } from './serialized-data.js'; import { deserializeStep, StepWireSchema } from './steps.js'; import { type APIConfig, getHttpConfig } from './utils.js'; +import type { WsFrameReply } from './ws-transport.js'; +import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; /** * Issue an instrumented v4 request through the global `fetch` — NOT undici's @@ -112,16 +115,6 @@ async function fetchV4( }); } -/** Flatten a fetch `Headers` into the record shape throwForErrorResponse - * expects (it mirrors the v3 `makeRequest` error contract). */ -function headersToRecord(headers: Headers): Record { - const out: Record = {}; - headers.forEach((value, key) => { - out[key] = value; - }); - return out; -} - const EVENT_ID_HEADER = 'x-wf-event-id'; const MAX_EVENTS_HEADER = 'x-wf-max-events'; @@ -642,10 +635,23 @@ async function postWorkflowRunEventV4( ); } +/** + * The WS transport carries this materialized POST and only this one. The + * `event-stream` writes below answer with a sentinel-terminated sequence of + * frames, which has no representation in a protocol that pairs one reply frame + * with one request frame. + */ export async function createWorkflowRunEventV4( input: CreateEventV4Input & { eventType: T }, config?: APIConfig ): Promise & { event: Event }> { + if (isWsEventsTransportEnabled()) { + // Absent means no socket was resolvable for this run, not that the write + // failed — fall through to HTTP. + const reply = await postEventFrameOverWs(input, config); + if (reply) return decodeCreateEventResponse(reply, input.eventType); + } + const response = await postWorkflowRunEventV4(input, 'materialized', config); const contentType = response.headers.get('content-type'); @@ -656,8 +662,11 @@ export async function createWorkflowRunEventV4( return decodeCreateEventResponse(response, input.eventType); } +/** Takes `FrameResponseLike` rather than `Response` because the WS branch has + * none to hand over — it synthesizes one. A real `Response` satisfies the + * interface, so the HTTP callers are unaffected. */ async function decodeCreateEventResponse( - response: Response, + response: FrameResponseLike, eventType: T ): Promise & { event: Event }> { const bodyBytes = new Uint8Array(await response.arrayBuffer()); @@ -706,6 +715,135 @@ export async function createWorkflowRunStartedEventV4( return { events, ...page, maxEvents: maxEvents.data }; } +/** The only two members a decoded transport result is read for. `fetch`'s + * `Response` satisfies it structurally, so the HTTP branch returns one + * unchanged and the WS branch synthesizes the same shape. */ +interface FrameResponseLike { + headers: { get(name: string): string | null }; + arrayBuffer(): Promise; +} + +/** Flatten a reply frame's meta into the header record `errorFromV4Response` + * already reads (retry-after, x-vercel-mitigated), so the WS path reuses the + * HTTP error mapping rather than growing its own. The materialized ids come + * out of the CBOR body, so no `x-wf-*` name is mapped here. + * + * Left unmapped, deliberately: `meta.deprecated`, which the server copies from + * `X-API-Deprecated`. Inert while the v4 route's middleware chain has no + * deprecation middleware to set it — but this record is the only header source + * a WS reply has, so an unmapped key is gone rather than merely unread, which + * is not true of the real `Response` the HTTP path returns. */ +function replyMetaToHeaderRecord( + meta: Record +): Record { + const out: Record = {}; + if (typeof meta.eventId === 'string') out[EVENT_ID_HEADER] = meta.eventId; + if (typeof meta.retryAfter === 'string') out['retry-after'] = meta.retryAfter; + if (typeof meta.mitigated === 'string') + out['x-vercel-mitigated'] = meta.mitigated; + return out; +} + +/** + * Retry is owned by `withEventPostRetry` (event-retry.ts) for both transports, + * so there is deliberately no retry loop here: one would sit *inside* + * `EVENT_RETRY_ELIGIBILITY` and replay the event types whose handlers have no + * duplicate guard (a second `step_started` double-increments `attempt`), and + * would multiply that policy's backoff. Note undici's `RetryAgent` + * (http-client.ts) retries POSTs on neither transport — `RetryHandler` + * defaults `methods` to GET/HEAD/OPTIONS/PUT/DELETE/TRACE — which is why + * event-retry.ts exists at all. So `postEventFrameOverWs` makes exactly one + * attempt and reports in that policy's vocabulary; a retry from it re-enters + * `transport.request()`, which reconnects on the way through. + */ + +/** + * Read the status off a reply frame, failing closed when there isn't one: + * defaulting to 200 would report success for any frame this client doesn't + * understand, and the protocol is designed to grow new response variants (see + * the server's docs/ws-protocol.md). `PARSE_ERROR` is the code `utils.ts` uses + * for an unreadable HTTP body — the same situation — and unlike a bare `Error` + * it satisfies `WorkflowWorldError.is()` instead of surfacing as a USER_ERROR. + */ +function wsReplyStatus(reply: WsFrameReply, endpoint: string): number { + const { status } = reply.meta; + if (typeof status !== 'number') { + throw new WorkflowWorldError( + `Failed to parse response body for POST ${endpoint}: reply frame ` + + `carried no numeric status (type: ` + + `${String(reply.meta.type ?? 'absent')}). Refusing to treat an ` + + `unrecognized reply as success — the write may or may not have been ` + + `applied.`, + { url: endpoint, code: 'PARSE_ERROR' } + ); + } + return status; +} + +async function postEventFrameOverWs( + input: CreateEventV4InputBase & { + eventType: EventType; + skipPreload?: true; + }, + config: APIConfig | undefined +): Promise { + // Dynamic so `ws` initializes only on a deployment that opted in — the gate + // at the call site lives in its own import-free module for exactly this + // reason. The module is cached after the first write, and on the queue path + // the pre-warm has already paid for it. + const { resolveWsTransport } = await import('./ws-transport.js'); + const { runId } = input; + const resolved = resolveWsTransport(runId, config); + if (!resolved) return undefined; + const { transport, wsUrl } = resolved; + const endpoint = `${wsUrl}#runs/${encodeURIComponent(runId)}/events`; + + let reply: WsFrameReply; + try { + // `runId` isn't repeated here — it's already in `wsUrl`, one connection + // per run. The server's request-frame schema is a discriminated union on + // `type` with each type's payload nested under its own name, so a future + // request type is a new variant rather than a reshape of this one. + reply = await transport.request((reqId) => + encodeFrame( + { reqId, type: 'event', event: buildPostFrameMeta(input) }, + input.payload ?? new Uint8Array(0) + ) + ); + } catch (err) { + // Anything `transport.request()` throws means the frame was never acked. + // `code: 'TRANSPORT'` is the shape `utils.ts` gives a failed `fetch`, so + // one classification drives both transports — in-process retry gated by + // event type, then queue redelivery. An unwrapped `WsTransportError` + // would fail `WorkflowWorldError.is()` and classify as a USER_ERROR. + // Application errors are raised below, outside this try. + throw new WorkflowWorldError( + `POST ${endpoint} transport failure: ${ + err instanceof Error ? err.message : String(err) + }`, + { url: wsUrl, code: 'TRANSPORT', cause: err } + ); + } + + const status = wsReplyStatus(reply, endpoint); + const headerRecord = replyMetaToHeaderRecord(reply.meta); + + if (status < 200 || status >= 300) { + throw errorFromV4Response( + status, + headerRecord, + new TextDecoder().decode(reply.body), + 'createEvent', + endpoint + ); + } + + return { + headers: { get: (name) => headerRecord[name.toLowerCase()] ?? null }, + arrayBuffer: async () => reply.body.slice().buffer as ArrayBuffer, + }; +} + /** * Result of a `hook_received` POST that opted into the replay-log preload, * discriminated on `kind` (keyed on the response content type). diff --git a/packages/world-vercel/src/http-core.ts b/packages/world-vercel/src/http-core.ts index 68caf10fa2..4e0fe0a07a 100644 --- a/packages/world-vercel/src/http-core.ts +++ b/packages/world-vercel/src/http-core.ts @@ -152,6 +152,21 @@ export function parseRetryAfter( return Number.isNaN(parsed) ? undefined : parsed; } +/** + * Flatten a fetch `Headers` into the plain record both `throwForErrorResponse` + * (mirroring the v3 `makeRequest` error contract) and the WS transport's + * `getHeaders` seam expect. Lives here because both `events-v4.ts` and + * `ws-transport.ts` need it and `events-v4` already imports the transport, so + * the reverse edge would be a cycle. + */ +export function headersToRecord(headers: Headers): Record { + const out: Record = {}; + headers.forEach((value, key) => { + out[key] = value; + }); + return out; +} + /** * Build the typed error for a non-2xx response. This is the single source of * truth for the status → error-type contract the runtime branches on: diff --git a/packages/world-vercel/src/queue.ts b/packages/world-vercel/src/queue.ts index e17e860cf1..cfec381ecd 100644 --- a/packages/world-vercel/src/queue.ts +++ b/packages/world-vercel/src/queue.ts @@ -22,6 +22,7 @@ import { getDispatcher } from './http-client.js'; import { decode as decodeTaggedRunId } from './run-id/index.js'; import { isKnownRegionCode, REGION_IDS } from './run-id/regions.js'; import { type APIConfig, getHeaders, getHttpUrl } from './utils.js'; +import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; /** * CBOR-based queue transport. Encodes values with cbor-x on send and @@ -194,6 +195,53 @@ function getRunIdFromPayload(payload: QueuePayload): string | undefined { return undefined; } +/** + * Bind this run's events channel to one invocation of the flow route. This is + * the only pair of calls that opens one: nothing else in the SDK does, so every + * other writer — `start()` writing `run_created` from an arbitrary request + * handler, where a lone write would not repay a handshake — stays on HTTP. + * + * Both halves are no-ops on the HTTP default, and the gate is checked before the + * import so a deployment on the default never loads `ws`. + */ +const wsEventsChannelForInvocation = ( + runId: string | undefined, + config: APIConfig | undefined +) => { + /** This invocation's release, once the open has resolved one. */ + let claim: Promise<(() => void) | undefined> | undefined; + + return { + /** + * Unawaited and failure-proof: callers treat the handshake as free. The + * refcount therefore rises a microtask late, so a write racing the import + * finds no channel and goes over HTTP — one frame, not the invocation, + * since `close` awaits this same promise and so cannot release ahead of + * the claim it is releasing. + */ + open(): void { + if (!runId || !isWsEventsTransportEnabled()) return; + claim = import('./ws-transport.js') + .then(({ openWsChannel }) => openWsChannel(runId, config)) + .catch(() => undefined); + }, + /** + * Awaited, unlike the open: work scheduled after the handler returns is not + * guaranteed to run, and an open socket is not `unref`'d, so skipping this + * stops the process exiting and keeps a server invocation pinned. + * + * Releases the claim the open returned rather than re-resolving the run, + * which is what keeps a channel this invocation never opened — a later + * invocation's, registered under the same URL after ours was evicted — out + * of reach of our release. + */ + async close(): Promise { + const release = await claim; + release?.(); + }, + }; +}; + /** * Workflow run IDs are prefixed with `wrun_` before the underlying ULID. * Strip that prefix so the payload can be fed to the tagged-ULID decoder. @@ -481,26 +529,43 @@ export function createQueue(config?: APIConfig): Queue { const { payload, queueName, deploymentId } = MessageWrapper.parse(message); - const result = await handler(payload, { - queueName, - messageId: MessageId.parse(metadata.messageId), - attempt: metadata.deliveryCount, - requestId, - }); - - if (typeof result?.timeoutSeconds === 'number') { - // When timeoutSeconds is 0, skip delaySeconds entirely for immediate re-enqueue. - // Otherwise, clamp to one continuation hop (23h by default). Longer - // sleeps chain delayed messages until the full duration has elapsed. - const delaySeconds = - result.timeoutSeconds > 0 - ? Math.min(result.timeoutSeconds, MAX_DELAY_SECONDS) - : undefined; - - // Send new message BEFORE acknowledging current message. - // This ensures crash safety: if process dies after send but before ack, - // we may get a duplicate invocation but won't lose the scheduled wakeup. - await queue(queueName, payload, { deploymentId, delaySeconds }); + // Earliest point in an invocation where the run id is known, so the WS + // handshake happens here instead of on the runtime's first event write + // (which would record a `step_started` later than the work it + // timestamps). This path also absorbs `ws`'s module init. + const wsEvents = wsEventsChannelForInvocation( + getRunIdFromPayload(payload), + config + ); + wsEvents.open(); + + try { + const result = await handler(payload, { + queueName, + messageId: MessageId.parse(metadata.messageId), + attempt: metadata.deliveryCount, + requestId, + }); + + if (typeof result?.timeoutSeconds === 'number') { + // When timeoutSeconds is 0, skip delaySeconds entirely for immediate re-enqueue. + // Otherwise, clamp to one continuation hop (23h by default). Longer + // sleeps chain delayed messages until the full duration has elapsed. + const delaySeconds = + result.timeoutSeconds > 0 + ? Math.min(result.timeoutSeconds, MAX_DELAY_SECONDS) + : undefined; + + // Send new message BEFORE acknowledging current message. + // This ensures crash safety: if process dies after send but before ack, + // we may get a duplicate invocation but won't lose the scheduled wakeup. + await queue(queueName, payload, { deploymentId, delaySeconds }); + } + } finally { + // The only point in the SDK that knows an invocation has no writes + // left. In a `finally` so a failed handler closes too — the retry + // arrives as a new invocation and opens its own channel. + await wsEvents.close(); } }, { diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts index 7f52c59514..c071d59960 100644 --- a/packages/world-vercel/src/trace-propagation.test.ts +++ b/packages/world-vercel/src/trace-propagation.test.ts @@ -12,6 +12,7 @@ import { afterAll, afterEach, beforeAll, + beforeEach, describe, expect, it, @@ -30,6 +31,31 @@ vi.mock('@vercel/oidc', () => ({ getVercelOidcToken: vi.fn().mockRejectedValue(new Error('no OIDC')), })); +/** + * The WS transport can only carry trace context on the upgrade — frames have no + * headers — so the assertion is on what reaches the `ws` constructor. This fake + * records that and stays in CONNECTING; no test here needs a handshake. + */ +const { wsUpgrades } = vi.hoisted(() => ({ + wsUpgrades: [] as { url: string; headers: Record }[], +})); + +vi.mock('ws', () => ({ + WebSocket: class { + static OPEN = 1; + readyState = 0; + constructor(url: string, options?: { headers?: Record }) { + wsUpgrades.push({ url, headers: options?.headers ?? {} }); + } + on() { + return this; + } + send() {} + close() {} + terminate() {} + }, +})); + const exporter = new InMemorySpanExporter(); const provider = new BasicTracerProvider(); const contextManager = new AsyncLocalStorageContextManager(); @@ -287,3 +313,51 @@ describe('streamer write trace propagation', () => { ); }); }); + +describe('ws events transport upgrade trace propagation', () => { + // `openWsChannel` is gated, unlike the `resolveWsTransport` lookup it + // replaced: nothing opens a channel on the HTTP default. + beforeEach(() => { + vi.stubEnv('WORKFLOW_EVENTS_TRANSPORT', 'ws'); + }); + + afterEach(async () => { + vi.unstubAllEnvs(); + wsUpgrades.length = 0; + const { resetWsEventsTransportsForTest } = await import( + './ws-transport.js' + ); + resetWsEventsTransportsForTest(); + }); + + it('injects traceparent on the upgrade, parented to the invocation span', async () => { + const { openWsChannel } = await import('./ws-transport.js'); + + const tracer = otelTrace.getTracer('test'); + let traceId = ''; + let spanId = ''; + await tracer.startActiveSpan('flow-invocation', async (span) => { + traceId = span.spanContext().traceId; + spanId = span.spanContext().spanId; + openWsChannel('wrun_1', { token: 'test-token' }); + await vi.waitFor(() => expect(wsUpgrades).toHaveLength(1)); + span.end(); + }); + + // Every event written over this socket is parented to whoever opened it — + // there is no per-frame traceparent to fall back on, so an uninjected + // upgrade orphans the server's spans for the whole run. + const traceparent = wsUpgrades[0]?.headers.traceparent; + expect(traceparent).toBe(`00-${traceId}-${spanId}-01`); + }); + + it('opens the upgrade without traceparent when no span is active', async () => { + const { openWsChannel } = await import('./ws-transport.js'); + openWsChannel('wrun_2', { token: 'test-token' }); + await vi.waitFor(() => expect(wsUpgrades).toHaveLength(1)); + + expect(wsUpgrades[0]?.headers.traceparent).toBeUndefined(); + // The rest of the upgrade must survive an absent propagator unchanged. + expect(wsUpgrades[0]?.headers.authorization).toBe('Bearer test-token'); + }); +}); diff --git a/packages/world-vercel/src/ws-protocol-conformance.test.ts b/packages/world-vercel/src/ws-protocol-conformance.test.ts new file mode 100644 index 0000000000..deffac7c67 --- /dev/null +++ b/packages/world-vercel/src/ws-protocol-conformance.test.ts @@ -0,0 +1,477 @@ +/** + * Wire-contract conformance for the WS events protocol. + * + * `ws-transport.test.ts` drives the client against a socket that replies with + * whatever the test hands it, which proves the transport's lifecycle but not + * that the bytes it emits are the bytes workflow-server accepts — a client and + * server can each pass their own suites while disagreeing on the envelope. + * + * So this file pairs the real client stack (including + * `createWorkflowRunEventV4`) with a fixture mirroring what the server route + * does per message: decode one frame, validate the meta against a local copy of + * `WsRequestFrameSchema`, dispatch, and encode the reply the way the route's + * `replyMeta` builder does. Only the socket is faked, since + * `experimental_upgradeWebSocket` needs a real Vercel runtime. + * + * That schema copy is the one thing here that can drift from the server. The + * golden-byte fixtures are the backstop: they pin the encoding, so a one-sided + * envelope change fails `encodes a known request frame to stable bytes`. The + * matching half of this interop test still needs to land in workflow-server. + */ + +import { EntityConflictError } from '@workflow/errors'; +import { decode, encode } from 'cbor-x'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { z } from 'zod'; +import { createWorkflowRunEventV4 } from './events-v4.js'; +import { type DecodedFrame, decodeFrames, encodeFrame } from './frames.js'; + +// --------------------------------------------------------------------------- +// Mirror of workflow-server's lib/handlers/v4/frames.ts +// --------------------------------------------------------------------------- + +const WsRequestFrameSchema = z.discriminatedUnion('type', [ + z.object({ + reqId: z.number(), + type: z.literal('event'), + event: z.object({}).passthrough(), + }), +]); + +/** Reserved reqId the server replies under for an unparseable frame. */ +const MALFORMED_FRAME_REQ_ID = -1; + +/** What the server route does to one inbound message, minus the transport. */ +interface RouteResponse { + status: number; + headers?: Record; + body?: Uint8Array; +} + +const { FakeWebSocket, sockets } = vi.hoisted(() => { + const sockets: FakeSocket[] = []; + class FakeSocket { + static readonly OPEN = 1; + readyState = 0; + binaryType = ''; + /** Set by the test: how the fixture server answers a dispatched event. */ + respond: (eventMeta: Record, body: Uint8Array) => unknown = + () => ({ status: 201 }); + /** Every raw frame the client put on the wire. */ + readonly sent: Uint8Array[] = []; + private readonly listeners = new Map< + string, + Array<(...a: unknown[]) => void> + >(); + + /** Set by the fixture server: called with each frame the client sends. */ + onFrame: ((raw: Uint8Array) => void) | null = null; + + constructor(_url: string, _opts?: unknown) { + sockets.push(this); + } + on(event: string, cb: (...a: unknown[]) => void): this { + const l = this.listeners.get(event) ?? []; + l.push(cb); + this.listeners.set(event, l); + return this; + } + emit(event: string, ...args: unknown[]): void { + for (const cb of [...(this.listeners.get(event) ?? [])]) cb(...args); + } + send(data: Uint8Array, cb?: (err?: Error) => void): void { + this.sent.push(data); + cb?.(); + this.onFrame?.(data); + } + close(code = 1000): void { + if (this.readyState === 3) return; + this.readyState = 3; + this.emit('close', code); + } + open(): void { + this.readyState = 1; + this.emit('open'); + } + deliver(frame: Uint8Array): void { + this.emit('message', Buffer.from(frame)); + } + } + return { FakeWebSocket: FakeSocket, sockets }; +}); + +vi.mock('ws', () => ({ WebSocket: FakeWebSocket })); + +const { openWsChannel, resetWsEventsTransportsForTest } = await import( + './ws-transport.js' +); + +/** Decode exactly one frame, the way the server's `decodeFrame` does. */ +async function decodeOne(raw: Uint8Array): Promise { + for await (const frame of decodeFrames( + (async function* () { + yield raw; + })() + )) { + return frame; + } + throw new Error('empty frame'); +} + +function encodeErrorFrame( + reqId: number, + status: number, + message: string +): Uint8Array { + return encodeFrame( + { reqId, type: 'error', status }, + new TextEncoder().encode(JSON.stringify({ message })) + ); +} + +/** + * Mirrors the route's `replyMeta` construction: status, plus the handful of + * `x-wf-*` response headers flattened onto the frame meta. + */ +function buildReplyMeta( + reqId: number, + res: RouteResponse +): Record { + const h = res.headers ?? {}; + const byHeader: Array<[string, string]> = [ + ['x-wf-event-id', 'eventId'], + ['x-wf-run-id', 'runId'], + ['x-wf-created-at', 'createdAt'], + ['retry-after', 'retryAfter'], + ]; + const meta: Record = { + reqId, + type: 'event_ack', + status: res.status, + }; + for (const [header, key] of byHeader) { + if (h[header]) meta[key] = h[header]; + } + return meta; +} + +/** + * Wire the fixture "server" to a socket: on every frame the client sends, + * run the route's message handling and deliver the reply back. + */ +function attachFixtureServer( + socket: InstanceType, + handler: ( + eventMeta: Record, + body: Uint8Array + ) => RouteResponse +) { + const seen: Array<{ meta: Record; body: Uint8Array }> = []; + socket.onFrame = (raw) => { + void (async () => { + let frame: DecodedFrame; + try { + frame = await decodeOne(raw); + } catch (err) { + socket.deliver( + encodeErrorFrame( + MALFORMED_FRAME_REQ_ID, + 400, + `ws v1: malformed frame: ${String(err)}` + ) + ); + return; + } + + const parsed = WsRequestFrameSchema.safeParse(frame.meta); + if (!parsed.success) { + const rawReqId = (frame.meta as { reqId?: unknown }).reqId; + socket.deliver( + encodeErrorFrame( + typeof rawReqId === 'number' ? rawReqId : MALFORMED_FRAME_REQ_ID, + 400, + `ws v1 frame: ${parsed.error.message}` + ) + ); + return; + } + + seen.push({ meta: parsed.data.event, body: frame.body }); + const res = handler(parsed.data.event, frame.body); + socket.deliver( + encodeFrame( + buildReplyMeta(parsed.data.reqId, res), + res.body ?? new Uint8Array(0) + ) + ); + })(); + }; + return seen; +} + +const CREATED_AT = '2026-06-10T00:00:00.000Z'; + +const okHeaders = { + 'x-wf-event-id': 'evnt_1', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': CREATED_AT, +}; + +/** The materialized CBOR body a `step_completed` write answers with. The ids + * the caller sees come out of this body, not the reply meta. */ +const materializedBody = (eventId = 'evnt_1') => + new Uint8Array( + encode({ + event: { + eventId, + runId: 'wrun_1', + createdAt: CREATED_AT, + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', + eventData: { result: new Uint8Array() }, + }, + step: { + runId: 'wrun_1', + stepId: 'step_1', + stepName: 'step', + status: 'completed', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + }, + }) + ); + +const input = { + runId: 'wrun_1', + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', + payload: new TextEncoder().encode('"result"'), +} as const; + +beforeEach(() => { + sockets.length = 0; + resetWsEventsTransportsForTest(); + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + vi.spyOn(console, 'log').mockImplementation(() => {}); + vi.spyOn(console, 'error').mockImplementation(() => {}); +}); + +afterEach(() => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + vi.restoreAllMocks(); +}); + +/** + * Kick off a real `createWorkflowRunEventV4`, attach the fixture server to + * the socket the transport opens, then let the handshake complete. + */ +async function callThroughFixture( + handler: ( + eventMeta: Record, + body: Uint8Array + ) => RouteResponse +) { + // Stand in for the flow route: without an open channel the write resolves + // none and goes over HTTP, which is the whole point of the explicit pair. + openWsChannel(input.runId, { token: 'test-token' }); + const pending = createWorkflowRunEventV4(input, { token: 'test-token' }); + void pending.catch(() => {}); + // Let the transport construct its socket (it awaits the header thunk). + await vi.waitFor(() => expect(sockets.length).toBeGreaterThan(0)); + const socket = sockets[0]; + const seen = attachFixtureServer(socket, handler); + socket.open(); + return { pending, socket, seen }; +} + +describe('client → server frame shape', () => { + it('emits a frame the server schema accepts, with meta nested under `event`', async () => { + const { pending, seen } = await callThroughFixture(() => ({ + status: 201, + headers: okHeaders, + body: materializedBody(), + })); + + const result = await pending; + + // Round-trip succeeded, which already means the schema validated — + // a flat frame would have come back as a 400 error frame instead. + expect(result.event.eventId).toBe('evnt_1'); + expect(result.step).toMatchObject({ stepId: 'step_1' }); + + // And the meta the server unpacked is the event meta, not the envelope. + expect(seen).toHaveLength(1); + expect(seen[0].meta).toMatchObject({ + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', + }); + expect(seen[0].meta.reqId).toBeUndefined(); + expect(seen[0].meta.type).toBeUndefined(); + // runId rides the connection URL, never the frame. + expect(seen[0].meta.runId).toBeUndefined(); + }); + + it('sends the payload through untouched as the frame body', async () => { + const { pending, seen } = await callThroughFixture(() => ({ + status: 201, + headers: okHeaders, + body: materializedBody(), + })); + await pending; + + expect(new TextDecoder().decode(seen[0].body)).toBe('"result"'); + }); + + it('puts exactly one frame in each socket message (no trailing bytes)', async () => { + // The server rejects trailing bytes past the declared body as a client + // framing bug, so a message must be byte-exact. + const { pending, socket } = await callThroughFixture(() => ({ + status: 201, + headers: okHeaders, + body: materializedBody(), + })); + await pending; + + const raw = socket.sent[0]; + const metaLen = new DataView( + raw.buffer, + raw.byteOffset, + raw.byteLength + ).getUint32(0, false); + const bodyLen = new DataView( + raw.buffer, + raw.byteOffset, + raw.byteLength + ).getUint32(4 + metaLen, false); + expect(raw.byteLength).toBe(4 + metaLen + 4 + bodyLen); + }); +}); + +describe('server → client reply handling', () => { + it('surfaces a non-2xx ack as the same typed error the HTTP path raises', async () => { + const { pending } = await callThroughFixture(() => ({ + status: 409, + body: new TextEncoder().encode('{"message":"already applied"}'), + })); + + // Same class the HTTP path raises for a 409, so the runtime's + // "already applied, drop it" branch fires identically on either + // transport — that equivalence is the whole point of the WS adapter. + await expect(pending).rejects.toThrow(EntityConflictError); + await expect(pending).rejects.toThrow(/already applied/); + }); + + it('fails closed on a reply variant this client version does not know', async () => { + const { pending, socket } = await callThroughFixture(() => ({ + status: 201, + headers: okHeaders, + })); + // Pre-empt the fixture's ack with a future reply type carrying no status. + socket.onFrame = (raw) => { + void decodeOne(raw).then((f) => + socket.deliver( + encodeFrame( + { reqId: f.meta.reqId, type: 'event_ack_v2' }, + new Uint8Array(0) + ) + ) + ); + }; + + await expect(pending).rejects.toThrow(/no numeric status/); + }); + + it('correlates replies by reqId across concurrent writes on one socket', async () => { + let n = 0; + const { pending, socket } = await callThroughFixture(() => { + n++; + return { + status: 201, + headers: { ...okHeaders, 'x-wf-event-id': `evnt_${n}` }, + body: materializedBody(`evnt_${n}`), + }; + }); + const second = createWorkflowRunEventV4( + { ...input, correlationId: 'step_2' }, + { token: 'test-token' } + ); + + const [a, b] = await Promise.all([pending, second]); + expect(new Set([a.event.eventId, b.event.eventId])).toEqual( + new Set(['evnt_1', 'evnt_2']) + ); + expect(socket.sent).toHaveLength(2); + }); +}); + +describe('golden frames', () => { + // Hand-computed from the format in the spec: + // frame := u32_be(meta_len) || cbor_meta || u32_be(body_len) || body + // If this changes, the server's decoder must change with it — that's the + // point of pinning bytes rather than re-deriving them with encodeFrame. + const GOLDEN_META = { reqId: 1, type: 'event', event: { eventType: 'x' } }; + const GOLDEN_BODY = new TextEncoder().encode('hi'); + + // 0000002a meta_len = 42 + // b900 03 ... CBOR map: reqId/type/event + // 00000002 body_len = 2 + // 6869 "hi" + const GOLDEN_HEX = + '0000002a' + + 'b90003657265714964016474797065656576656e74656576656e74b90001696576656e7454797065617' + + '8' + + '00000002' + + '6869'; + + it('encodes a known request frame to stable bytes', () => { + const frame = encodeFrame(GOLDEN_META, GOLDEN_BODY); + expect(Buffer.from(frame).toString('hex')).toBe(GOLDEN_HEX); + }); + + it('decodes the golden bytes back to the golden frame', async () => { + const decoded = await decodeOne( + new Uint8Array(Buffer.from(GOLDEN_HEX, 'hex')) + ); + expect(decoded.meta).toEqual(GOLDEN_META); + expect(new TextDecoder().decode(decoded.body)).toBe('hi'); + }); + + it('round-trips through the shared decoder', async () => { + const frame = encodeFrame(GOLDEN_META, GOLDEN_BODY); + const decoded = await decodeOne(frame); + expect(decoded.meta).toEqual(GOLDEN_META); + expect(new TextDecoder().decode(decoded.body)).toBe('hi'); + }); + + it('is accepted by the server request schema', () => { + expect(WsRequestFrameSchema.safeParse(GOLDEN_META).success).toBe(true); + }); + + it('rejects the flat shape that caused the original drift', () => { + // The bug the spec doc exists to prevent: event meta hoisted onto the + // frame instead of nested under `event`. + const flat = { reqId: 1, type: 'event', eventType: 'x', specVersion: 2 }; + expect(WsRequestFrameSchema.safeParse(flat).success).toBe(false); + }); + + it('decodes a golden event_ack the way the client expects', async () => { + const ack = encodeFrame( + { + reqId: 1, + type: 'event_ack', + status: 201, + eventId: 'evnt_1', + runId: 'wrun_1', + createdAt: '2026-06-10T00:00:00.000Z', + }, + new Uint8Array(encode({ ok: true })) + ); + const decoded = await decodeOne(ack); + expect(decoded.meta).toMatchObject({ type: 'event_ack', status: 201 }); + expect(decode(decoded.body)).toEqual({ ok: true }); + }); +}); diff --git a/packages/world-vercel/src/ws-transport-enabled.ts b/packages/world-vercel/src/ws-transport-enabled.ts new file mode 100644 index 0000000000..ed370ee56f --- /dev/null +++ b/packages/world-vercel/src/ws-transport-enabled.ts @@ -0,0 +1,26 @@ +/** + * The events-transport opt-in gate, deliberately alone in a module with no + * imports. `events-v4.ts` and `queue.ts` read it on every invocation, so it has + * to be answerable without pulling in `ws-transport.js` — and with it `ws`, + * ~17 ms of module init that a deployment on the HTTP default never gets a + * return on. Both call sites `await import('./ws-transport.js')` behind a true + * result, so the cost lands only where the socket is actually used. + */ + +/** + * HTTP unless `WORKFLOW_EVENTS_TRANSPORT=ws`. Only `createWorkflowRunEventV4` + * (POST) is wired to it — GET/LIST aren't on the hot per-step path, and LIST's + * streamed, sentinel-terminated multi-frame response doesn't map onto a single + * WS message. + * + * **Known gap: WS writes open no client span.** The upgrade carries W3C trace + * context (see `resolveUpgradeHeaders`), so server spans still join the caller's + * trace, but the HTTP branch's `instrumentedFetch` also opens an OTEL CLIENT + * span per write and routes through the global `fetch` that Vercel's + * outgoing-requests view instruments; the WS branch has neither, and individual + * frames carry no `traceparent` of their own. Acceptable behind a flag — + * per-write instrumentation is a prerequisite for defaulting to it. + */ +export function isWsEventsTransportEnabled(): boolean { + return process.env.WORKFLOW_EVENTS_TRANSPORT === 'ws'; +} diff --git a/packages/world-vercel/src/ws-transport.test.ts b/packages/world-vercel/src/ws-transport.test.ts new file mode 100644 index 0000000000..3a07d47b5d --- /dev/null +++ b/packages/world-vercel/src/ws-transport.test.ts @@ -0,0 +1,1219 @@ +/** + * Unit coverage for the WS events transport's failure and lifecycle paths. + * + * These are the parts that are effectively unreachable from the e2e suite: + * they need a socket to break at a specific moment (a send that fails + * mid-flight, an error after open, a superseded socket closing late), which + * a live deployment won't reproduce on demand. The happy path is already + * covered end-to-end by the `e2e-vercel-ws-transport` jobs, so the emphasis + * here is "nothing hangs and nothing is silent". + * + * `ws` is replaced with a fake we can drive directly — the real one needs a + * live server to reach `'open'`. Fake timers throughout so the reconnect + * backoff is deterministic and a microtask flush is a single `tick()`. + */ + +import { decode } from 'cbor-x'; +import { + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { encodeFrame } from './frames.js'; +import { REQUEST_TIMEOUT_MS } from './http-core.js'; +import { injectTraceContextIntoHeaders } from './telemetry.js'; +import { + getWsEventsTransport, + isWsEventsTransportEnabled, + openWsChannel, + resetWsEventsTransportsForTest, + resolveWsTransport, + toEventsWsUrl, +} from './ws-transport.js'; + +type Listener = (...args: unknown[]) => void; + +const { FakeWebSocket, sockets } = vi.hoisted(() => { + const sockets: FakeSocket[] = []; + + class FakeSocket { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + readyState = 0; + binaryType = ''; + readonly url: string; + readonly headers: Record; + readonly sent: Uint8Array[] = []; + /** Queued failures for upcoming `send` calls. */ + readonly sendErrors: Error[] = []; + private readonly listeners = new Map(); + + constructor(url: string, options?: { headers?: Record }) { + this.url = url; + this.headers = options?.headers ?? {}; + sockets.push(this); + } + + on(event: string, cb: Listener): this { + const list = this.listeners.get(event) ?? []; + list.push(cb); + this.listeners.set(event, list); + return this; + } + + emit(event: string, ...args: unknown[]): void { + for (const cb of [...(this.listeners.get(event) ?? [])]) cb(...args); + } + + send(data: Uint8Array, cb?: (err?: Error) => void): void { + const err = this.sendErrors.shift(); + if (err) { + cb?.(err); + return; + } + this.sent.push(data); + cb?.(); + } + + close(code = 1000): void { + if (this.readyState === FakeSocket.CLOSED) return; + this.readyState = FakeSocket.CLOSED; + this.emit('close', code); + } + + // ---- test drivers ---- + + /** Complete the handshake. */ + open(): void { + this.readyState = FakeSocket.OPEN; + this.emit('open'); + } + + /** Fail the handshake the way `ws` does: 'error', then 'close'. */ + failHandshake(err = new Error('ECONNREFUSED')): void { + this.emit('error', err); + this.close(1006); + } + + /** An error on an already-open socket (broken pipe, protocol fault). */ + errorAfterOpen(err = new Error('EPIPE')): void { + this.emit('error', err); + } + + /** + * Half-dead: no longer usable for new work (so the transport will open + * a replacement) but hasn't emitted `'close'` yet — the window in which + * a superseded socket can still fire a late close. + */ + stall(): void { + this.readyState = FakeSocket.CLOSING; + } + + deliver(frame: Uint8Array): void { + this.emit('message', Buffer.from(frame)); + } + } + + return { FakeWebSocket: FakeSocket, sockets }; +}); + +vi.mock('ws', () => ({ WebSocket: FakeWebSocket })); + +const WS_URL = 'wss://vercel-workflow.com/api/websockets/v1/runs/wrun_test'; +const EMPTY = new Uint8Array(0); + +/** Flush pending microtasks (and any timers due right now). */ +const tick = () => vi.advanceTimersByTimeAsync(0); + +/** Build a request frame the way `postEventFrameOverWs` does. */ +const eventFrame = (reqId: number) => + encodeFrame( + { reqId, type: 'event', event: { eventType: 'run_created' } }, + EMPTY + ); + +/** An `event_ack` reply for `reqId`. */ +const ackFrame = (reqId: number, status = 201, body = EMPTY) => + encodeFrame({ reqId, type: 'event_ack', status }, body); + +/** Read the `reqId`s off the frames this transport put on the wire. */ +const sentReqIds = (socket: { sent: Uint8Array[] }): unknown[] => + socket.sent.map((raw) => { + const metaLen = new DataView( + raw.buffer, + raw.byteOffset, + raw.byteLength + ).getUint32(0, false); + const meta = decode(raw.subarray(4, 4 + metaLen)) as { reqId?: unknown }; + return meta.reqId; + }); + +const latest = () => sockets[sockets.length - 1]; + +/** + * Wait for the transport to construct its next socket. Header resolution is + * async (token thunk, trace-context injection, and a one-time + * `@opentelemetry/api` import on the first upgrade of the process), so the + * number of microtask turns before `new WebSocket()` is not fixed — poll rather + * than encode a count that shifts whenever the upgrade grows an await. + */ +async function nextSocket(): Promise> { + const before = sockets.length; + for (let i = 0; i < 50; i++) { + await tick(); + if (sockets.length > before) return latest(); + } + throw new Error('transport never constructed a socket'); +} + +/** + * Start a request and let its socket finish the handshake. The returned + * promise already has a no-op rejection handler attached so tests that + * don't assert on it can't trip an unhandled-rejection warning. + */ +async function connectAndSend( + transport: ReturnType +) { + const promise = transport.request(eventFrame); + void promise.catch(() => {}); + const socket = await nextSocket(); + socket.open(); + await tick(); + return { promise, socket }; +} + +let errorSpy: ReturnType; +let logSpy: ReturnType; + +beforeAll(async () => { + // The upgrade injects trace context, whose first call imports the optional + // `@opentelemetry/api` peer. Module loading needs the real event loop, which + // fake timers can't drive — settle the memoized import here so every upgrade + // under test is microtasks only. + await injectTraceContextIntoHeaders(new Headers()); +}); + +beforeEach(() => { + vi.useFakeTimers(); + sockets.length = 0; + resetWsEventsTransportsForTest(); + errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); +}); + +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); + delete process.env.WORKFLOW_REQUEST_TIMEOUT_MS; +}); + +const headers = async () => ({ authorization: 'Bearer token-1' }); + +const loggedErrors = () => + errorSpy.mock.calls.map((args) => args.join(' ')).join('\n'); + +describe('toEventsWsUrl', () => { + it('upgrades the scheme and scopes the path to one run', () => { + expect(toEventsWsUrl('https://vercel-workflow.com/api', 'wrun_abc')).toBe( + 'wss://vercel-workflow.com/api/websockets/v1/runs/wrun_abc' + ); + }); + + it('uses ws: for a plaintext base URL and encodes the runId', () => { + expect(toEventsWsUrl('http://localhost:3000/api/', 'a/b')).toBe( + 'ws://localhost:3000/api/websockets/v1/runs/a%2Fb' + ); + }); +}); + +describe('request/reply', () => { + it('resolves a request with the reply frame matched by reqId', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + expect(socket.sent).toHaveLength(1); + + socket.deliver(ackFrame(1, 201, new Uint8Array([1, 2]))); + await tick(); + + await expect(promise).resolves.toMatchObject({ + meta: { reqId: 1, type: 'event_ack', status: 201 }, + body: new Uint8Array([1, 2]), + }); + }); + + it('multiplexes concurrent requests and matches each reply to its caller', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise: first, socket } = await connectAndSend(transport); + const second = transport.request(eventFrame); + await tick(); + + expect(sentReqIds(socket)).toEqual([1, 2]); + + // Replies arrive out of order — correlation is by reqId, not arrival. + socket.deliver(ackFrame(2, 200)); + socket.deliver(ackFrame(1, 201)); + await tick(); + + await expect(second).resolves.toMatchObject({ meta: { reqId: 2 } }); + await expect(first).resolves.toMatchObject({ meta: { reqId: 1 } }); + }); +}); + +describe('per-connection reqId and pending map', () => { + it('restarts reqId at 1 on a new connection', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + socket.deliver(ackFrame(1)); + await tick(); + await promise; + + socket.close(1001); + await tick(); + + const { socket: second } = await connectAndSend(transport); + // The protocol defines reqId as a per-connection counter; a + // transport-wide counter would have put 2 on the wire here. + expect(sentReqIds(second)).toEqual([1]); + }); + + it("a superseded socket's late close cannot reject the live socket's request", async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { socket: first } = await connectAndSend(transport); + // Take the first socket out of rotation without emitting 'close' yet. + first.stall(); + + const { promise: live, socket: second } = await connectAndSend(transport); + expect(second).not.toBe(first); + + // The stale socket finally closes. Its reqId 1 is the same id the live + // socket is waiting on — one shared pending map would reject it here. + first.close(1006); + await tick(); + + second.deliver(ackFrame(1, 201)); + await tick(); + await expect(live).resolves.toMatchObject({ meta: { reqId: 1 } }); + }); + + it('fails the requests riding a socket when it closes', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + + socket.close(1001); + await tick(); + + await expect(promise).rejects.toThrow(/connection closed \(code 1001\)/); + }); +}); + +describe('failures are never silent', () => { + it('rejects the request when send fails instead of hanging', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const promise = transport.request(eventFrame); + void promise.catch(() => {}); + await tick(); + // `ws.send()` reports a non-OPEN socket through its callback rather + // than throwing; without that callback wired up, this request would + // never settle. + latest().sendErrors.push(new Error('WebSocket is not open')); + latest().open(); + await tick(); + + await expect(promise).rejects.toThrow(/WebSocket is not open/); + }); + + it('logs a post-open socket error and tears the connection down', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + + socket.errorAfterOpen(new Error('EPIPE')); + await tick(); + + expect(loggedErrors()).toContain('socket error'); + expect(loggedErrors()).toContain('EPIPE'); + // The error handler closes explicitly, so in-flight work fails fast + // instead of waiting on a dead socket. + await expect(promise).rejects.toThrow(/connection closed/); + }); + + /** + * The three ways a reply can arrive that no caller can be matched to. Each + * one used to log and return — which left the request that provoked it in + * `pending` with nothing in existence that could ever settle it, so the + * caller hung until the server's own drain (~680s from connect), typically + * past the invocation's `maxDuration`. The frame is unanswerable either way; + * the only choice is whether the waiter finds out. + */ + describe.each([ + [ + 'an undecodable frame', + () => new Uint8Array([0, 0, 0]), + 'could not decode', + ], + [ + 'the reserved malformed-frame sentinel', + () => + encodeFrame( + { reqId: -1, type: 'error', status: 400 }, + new TextEncoder().encode(JSON.stringify({ message: 'bad frame' })) + ), + 'malformed-frame error', + ], + [ + 'a reply with a non-numeric reqId', + () => + encodeFrame({ reqId: 'one', type: 'event_ack', status: 201 }, EMPTY), + 'no numeric reqId', + ], + ])('%s', (_name, frame, logged) => { + it('is logged loudly', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { socket } = await connectAndSend(transport); + + socket.deliver(frame()); + await tick(); + + expect(loggedErrors()).toContain(logged); + }); + + it('fails the in-flight request instead of leaving it to hang', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + + socket.deliver(frame()); + await tick(); + + await expect(promise).rejects.toThrow(/events WS transport/); + }); + + it('drops the connection so the next write gets a fresh socket', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { socket } = await connectAndSend(transport); + const before = sockets.length; + + socket.deliver(frame()); + await tick(); + // The `close` handler's eager reconnect fires on the backoff timer. + await vi.advanceTimersByTimeAsync(1000); + + expect(sockets.length).toBeGreaterThan(before); + expect(latest()).not.toBe(socket); + }); + }); + + it("surfaces the server's message on a malformed-frame error", async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { socket } = await connectAndSend(transport); + + socket.deliver( + encodeFrame( + { reqId: -1, type: 'error', status: 400 }, + new TextEncoder().encode(JSON.stringify({ message: 'bad frame' })) + ) + ); + await tick(); + + // Not just the sentinel id: why the server rejected our frame is the only + // thing that makes this actionable, since it is a bug on this side. + expect(loggedErrors()).toContain('bad frame'); + }); + + it('logs but tolerates a reply that matches no in-flight request', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + let settled = false; + void promise.then( + () => { + settled = true; + }, + () => { + settled = true; + } + ); + + socket.deliver(ackFrame(999)); + await tick(); + + expect(loggedErrors()).toContain('unknown reqId 999'); + // The deliberate exception to the three cases above: an id nobody is + // waiting on means that request already settled (a send error, or its own + // deadline with the reply arriving late), so nothing is orphaned — while + // the request still in flight has done nothing wrong. Failing its socket + // would punish it for a race that resolved correctly. + expect(settled).toBe(false); + expect(sockets).toHaveLength(1); + }); + + it('fails a request that never gets a reply, rather than waiting for the drain', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + expect(socket.sent).toHaveLength(1); + + // A server that acks nothing and never closes: no error, no drain, no + // frame. Before the deadline the only thing that ended this wait was the + // invocation's own SIGTERM. + await vi.advanceTimersByTimeAsync(REQUEST_TIMEOUT_MS + 1); + + await expect(promise).rejects.toThrow(/timed out after/); + }); + + it('takes the silent connection down with the timed-out request', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + void promise.catch(() => {}); + + await vi.advanceTimersByTimeAsync(REQUEST_TIMEOUT_MS + 1); + await vi.advanceTimersByTimeAsync(1000); + + // A socket that swallowed a request is not the one to send the retry on. + expect(latest()).not.toBe(socket); + }); + + it('disarms the deadline once a reply lands', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + + socket.deliver(ackFrame(1)); + await tick(); + await expect(promise).resolves.toMatchObject({ meta: { reqId: 1 } }); + + // A deadline left armed would later fail a connection the caller is long + // done with — and on a reused socket, the unrelated writes riding it. + await vi.advanceTimersByTimeAsync(REQUEST_TIMEOUT_MS + 1000); + expect(loggedErrors()).not.toContain('timed out'); + }); + + it('logs a drain notice without settling in-flight work', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + let settled = false; + void promise.then( + () => { + settled = true; + }, + () => { + settled = true; + } + ); + + socket.deliver(encodeFrame({ type: 'drain', graceMs: 10_000 }, EMPTY)); + await tick(); + + expect(logSpy.mock.calls.map((a) => a.join(' ')).join('\n')).toContain( + 'drain notice' + ); + expect(settled).toBe(false); + }); +}); + +describe('eager reconnect', () => { + it('reopens the socket after an unexpected close, with no new request', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + + socket.close(1001); + await tick(); + await expect(promise).rejects.toThrow(); + + // Nothing is asking for a socket — the reconnect runs off its own timer. + expect(sockets).toHaveLength(1); + await vi.advanceTimersByTimeAsync(200); + expect(sockets).toHaveLength(2); + }); + + it('resolves headers again on reconnect so the new socket gets a fresh token', async () => { + let n = 0; + const rotating = async () => ({ authorization: `Bearer token-${++n}` }); + const transport = getWsEventsTransport(WS_URL, rotating); + + const { promise, socket } = await connectAndSend(transport); + expect(socket.headers.authorization).toBe('Bearer token-1'); + + socket.close(1006); + await tick(); + await expect(promise).rejects.toThrow(); + + await vi.advanceTimersByTimeAsync(200); + expect(sockets).toHaveLength(2); + // The frozen-token problem is per socket: a reconnect must not replay + // the token the previous socket was opened with. + expect(sockets[1].headers.authorization).toBe('Bearer token-2'); + }); + + it('resolves headers once per socket, not once per request', async () => { + const getHeaders = vi.fn(async () => ({ authorization: 'Bearer t' })); + const transport = getWsEventsTransport(WS_URL, getHeaders); + + const { socket } = await connectAndSend(transport); + void transport.request(eventFrame).catch(() => {}); + void transport.request(eventFrame).catch(() => {}); + await tick(); + + expect(socket.sent).toHaveLength(3); + expect(getHeaders).toHaveBeenCalledTimes(1); + }); + + it('gives up after the attempt cap and leaves the retry to the next request', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + socket.close(1006); + await tick(); + await expect(promise).rejects.toThrow(); + + // Every reconnect attempt now fails its handshake. Backoff is + // exponential from 100ms, capped at 5s, so 10s per iteration covers + // each attempt with room to spare. + for (let i = 0; i < 8; i++) { + await vi.advanceTimersByTimeAsync(10_000); + if (latest().readyState === 0) { + latest().failHandshake(); + await tick(); + } + } + + // 1 original + at most RECONNECT_MAX_ATTEMPTS reconnect attempts. + expect(sockets.length).toBeLessThanOrEqual(6); + expect(loggedErrors()).toContain('giving up eager reconnect'); + }); + + it('reconnects lazily on the next request after giving up', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise, socket } = await connectAndSend(transport); + socket.close(1006); + await tick(); + await expect(promise).rejects.toThrow(); + + for (let i = 0; i < 8; i++) { + await vi.advanceTimersByTimeAsync(10_000); + if (latest().readyState === 0) { + latest().failHandshake(); + await tick(); + } + } + const afterGivingUp = sockets.length; + + void transport.request(eventFrame).catch(() => {}); + await tick(); + + expect(sockets.length).toBe(afterGivingUp + 1); + }); + + it('does not reconnect when a newer socket is already live', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { promise: first, socket } = await connectAndSend(transport); + + // Stale socket steps aside and a new one takes over... + socket.stall(); + const { socket: second } = await connectAndSend(transport); + expect(second).not.toBe(socket); + + // ...and only now does the stale one emit its close. + socket.close(1006); + await tick(); + await expect(first).rejects.toThrow(); + + await vi.advanceTimersByTimeAsync(10_000); + expect(sockets).toHaveLength(2); + }); +}); + +/** + * Nothing else in this module creates or drops a socket — the explicit + * `open`/`release` pair is the whole lifetime. Eager reconnect renews a + * connection forever and the server pins an invocation per connection, so a + * channel nobody releases is a socket, and a server invocation, held until the + * platform kills the process. + */ +describe('open/close lifecycle', () => { + /** Run one request through to its ack, as a writing invocation would. */ + async function completeRequest( + transport: ReturnType + ) { + const { promise, socket } = await connectAndSend(transport); + socket.deliver(ackFrame(sentReqIds(socket)[0] as number)); + await promise; + return socket; + } + + it('drops the socket on release', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + const socket = await completeRequest(transport); + expect(socket.readyState).toBe(1); + + transport.release('invocation complete'); + await tick(); + + expect(socket.readyState).toBe(3); + }); + + it('drops a socket that never got a write', async () => { + // An invocation can open a channel and write nothing — a delivery whose + // replay finds the run already finished. The socket is not `unref`'d, so + // stranding it holds this process and a server invocation open. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await tick(); + const socket = latest(); + socket.open(); + await tick(); + expect(socket.readyState).toBe(1); + + transport.release('invocation complete'); + await tick(); + + expect(socket.readyState).toBe(3); + }); + + it('does not reconnect the socket it just released', async () => { + // The release closes the socket, which fires the same `close` handler an + // unexpected drop would — without the `closed` guard the eager reconnect + // would immediately undo it. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await completeRequest(transport); + + transport.release('invocation complete'); + await vi.advanceTimersByTimeAsync(60_000); + + expect(sockets).toHaveLength(1); + }); + + it('evicts itself from the process-wide cache', async () => { + // Membership of that cache is what `resolveWsTransport` reads as "this run + // has a channel", so eviction is what sends later writes down the HTTP path. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await completeRequest(transport); + expect(getWsEventsTransport(WS_URL, headers)).toBe(transport); + + transport.release('invocation complete'); + + expect(getWsEventsTransport(WS_URL, headers)).not.toBe(transport); + }); + + it('holds the socket until the last holder releases', async () => { + // Fluid runs invocations concurrently in one instance and inline step + // executions ride the flow topic, so two invocations for one run share this + // transport. The first to finish must not cut the other's writes short. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + transport.open(); + const socket = await completeRequest(transport); + + transport.release('first invocation complete'); + await tick(); + expect(socket.readyState).toBe(1); + + transport.release('second invocation complete'); + await tick(); + expect(socket.readyState).toBe(3); + }); + + it('ignores a release with nothing open', async () => { + // The count floors at zero instead of going negative, which would leave the + // next holder's release one short and pin its socket for good. + const transport = getWsEventsTransport(WS_URL, headers); + transport.release('unbalanced'); + + transport.open(); + const socket = await completeRequest(transport); + transport.release('invocation complete'); + await tick(); + + expect(socket.readyState).toBe(3); + }); + + it('fails a write issued after the release rather than reviving the socket', async () => { + // Unreachable through `resolveWsTransport`, which only hands back a + // registered channel. Reviving here would resurrect exactly the socket + // nobody owns that the explicit pair exists to rule out. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await completeRequest(transport); + transport.release('invocation complete'); + + await expect(transport.request(eventFrame)).rejects.toThrow('is closed'); + expect(sockets).toHaveLength(1); + }); + + it('does not adopt a socket that opens after the release', async () => { + // `close()` can only drop the connection it can see, and a handshake still + // in flight isn't one yet — so the release lands while this socket is still + // CONNECTING, and it must not install itself on the way in. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await tick(); + const socket = latest(); + + transport.release('invocation complete'); + socket.open(); + await tick(); + + expect(socket.readyState).toBe(3); + }); +}); + +describe('open', () => { + it('opens the socket with no write to trigger it', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + + transport.open(); + await tick(); + + expect(sockets).toHaveLength(1); + expect(latest().headers.authorization).toBe('Bearer token-1'); + }); + + it('leaves the first write with no handshake left to pay for', async () => { + // The point of the whole exercise: the write reuses the socket the open + // started rather than opening a second one. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await tick(); + latest().open(); + await tick(); + + const promise = transport.request(eventFrame); + await tick(); + + expect(sockets).toHaveLength(1); + latest().deliver(ackFrame(sentReqIds(latest())[0] as number)); + await expect(promise).resolves.toBeDefined(); + }); + + it('joins an in-flight handshake instead of racing a second one', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + + transport.open(); + transport.open(); + await tick(); + // A write arriving mid-handshake joins the same connect. + void transport.request(eventFrame).catch(() => {}); + await tick(); + + expect(sockets).toHaveLength(1); + }); + + it('starts no second socket once one is live', async () => { + const transport = getWsEventsTransport(WS_URL, headers); + const { socket } = await connectAndSend(transport); + + transport.open(); + await tick(); + + expect(sockets).toHaveLength(1); + expect(socket.readyState).toBe(1); + }); + + it('closes the channel when the handshake fails, so writes take HTTP', async () => { + // A never-opened first connect is the one case `connect` declines to retry. + // Leaving the channel registered would hand every write of the invocation + // its own doomed handshake; de-registering sends them down the pooled HTTP + // path instead, which is the floor this transport must not fall below. + const transport = getWsEventsTransport(WS_URL, headers); + transport.open(); + await tick(); + latest().failHandshake(); + await tick(); + // Loud either way — an unawaited open is exactly the case where the log is + // the only signal the socket failed to come up. + expect(loggedErrors()).toContain('socket error'); + + await vi.advanceTimersByTimeAsync(30_000); + expect(sockets).toHaveLength(1); + expect(getWsEventsTransport(WS_URL, headers)).not.toBe(transport); + }); +}); + +/** + * workflow-server tags a `drain` with why it is closing. `max_duration` means + * the socket aged out and a plain reconnect is right; `auth_expiry` means the + * *bearer* ran out, and reconnecting with the same one just earns a 401. + */ +describe('drain reason', () => { + const drainFrame = (reason?: string) => + encodeFrame( + { type: 'drain', graceMs: 5_000, ...(reason ? { reason } : {}) }, + EMPTY + ); + + /** Headers thunk that records the `forceRefresh` flag it was called with + * and hands out a different bearer each time when asked to rotate. */ + const trackingHeaders = (tokens: string[]) => { + const calls: boolean[] = []; + let i = 0; + const fn = async ({ forceRefresh }: { forceRefresh: boolean }) => { + calls.push(forceRefresh); + const token = tokens[Math.min(i, tokens.length - 1)]; + i++; + return { authorization: `Bearer ${token}` }; + }; + return { fn, calls }; + }; + + it('asks for a refreshed token after an auth_expiry drain', async () => { + const { fn, calls } = trackingHeaders(['token-1', 'token-2']); + const transport = getWsEventsTransport(WS_URL, fn); + const { socket } = await connectAndSend(transport); + + socket.deliver(drainFrame('auth_expiry')); + await tick(); + socket.close(1001); + await vi.advanceTimersByTimeAsync(1_000); + + expect(calls).toEqual([false, true]); + expect(latest().headers.authorization).toBe('Bearer token-2'); + }); + + it('does not ask for a refresh on an ordinary max_duration drain', async () => { + const { fn, calls } = trackingHeaders(['token-1']); + const transport = getWsEventsTransport(WS_URL, fn); + const { socket } = await connectAndSend(transport); + + socket.deliver(drainFrame('max_duration')); + await tick(); + socket.close(1001); + await vi.advanceTimersByTimeAsync(1_000); + + expect(calls).toEqual([false, false]); + }); + + it('treats a drain with no reason as max_duration', async () => { + // Servers predating the field only ever drained for maxDuration. + const { fn, calls } = trackingHeaders(['token-1']); + const transport = getWsEventsTransport(WS_URL, fn); + const { socket } = await connectAndSend(transport); + + socket.deliver(drainFrame()); + await tick(); + socket.close(1001); + await vi.advanceTimersByTimeAsync(1_000); + + expect(calls).toEqual([false, false]); + }); + + it('stops reconnecting when the refresh yields the same token', async () => { + // Inside a Vercel function the bearer is the invocation's own and can't + // be refreshed mid-invocation. Reconnecting would 401 five times and + // burn the attempt budget; the next write (likely a new invocation with + // a new token) is the thing worth waiting for. + const { fn } = trackingHeaders(['token-1']); + const transport = getWsEventsTransport(WS_URL, fn); + const { socket } = await connectAndSend(transport); + + socket.deliver(drainFrame('auth_expiry')); + await tick(); + socket.close(1001); + await vi.advanceTimersByTimeAsync(30_000); + + expect(sockets).toHaveLength(1); + expect(loggedErrors()).toContain('same token'); + }); + + it('still reconnects lazily once a new token is available', async () => { + const tokens = ['token-1', 'token-1', 'token-2']; + const { fn } = trackingHeaders(tokens); + const transport = getWsEventsTransport(WS_URL, fn); + const { socket } = await connectAndSend(transport); + + socket.deliver(drainFrame('auth_expiry')); + await tick(); + socket.close(1001); + await vi.advanceTimersByTimeAsync(30_000); + expect(sockets).toHaveLength(1); + + // A later write, by which point the process is serving a new invocation. + const { socket: revived } = await connectAndSend(transport); + expect(revived.headers.authorization).toBe('Bearer token-2'); + }); +}); + +/** + * Which transport a write takes, and which socket it takes it on — the layer + * `createWorkflowRunEventV4` and the queue handler call into. Covered here + * rather than in `events-v4-ws.test.ts` because that file mocks + * `resolveWsTransport` wholesale (it has to: the URL resolution and the header + * thunk are intra-module calls that an export-level mock cannot intercept), so + * this is the only place the real selection code runs. + */ +describe('transport selection', () => { + afterEach(() => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + }); + + /** A World that routes through the api-workflow proxy. */ + const proxyConfig = { + token: 'test-token', + projectConfig: { projectId: 'prj_1', teamId: 'team_1' }, + }; + const directConfig = { token: 'test-token' }; + + /** + * The gate is the whole safety story for this feature: everything else is + * dead code for anyone who hasn't opted in. Nothing on the HTTP side pins + * the *choice* of path — a future edit that flipped the default (as an + * earlier revision of this branch did deliberately, for benchmarking) would + * sail through with every HTTP assertion still green, because the two + * transports are built to be indistinguishable at the result layer. + */ + describe('isWsEventsTransportEnabled', () => { + it.each([ + ['ws', true], + ['http', false], + ['', false], + ['WS', false], + ])('%o resolves to ws=%o', (value, expected) => { + process.env.WORKFLOW_EVENTS_TRANSPORT = value; + expect(isWsEventsTransportEnabled()).toBe(expected); + }); + + it('defaults to HTTP when unset', () => { + expect(isWsEventsTransportEnabled()).toBe(false); + }); + }); + + /** + * The write path's availability check. A lookup, never a create: it answers + * "did this invocation open a channel", and every `null` means write over HTTP. + */ + describe('resolveWsTransport', () => { + it('returns null for a run with no open channel', () => { + // The default for every writer that isn't the flow route — `start()` + // writing `run_created` most of all, where one write would not repay a + // handshake. + expect(resolveWsTransport('wrun_1', directConfig)).toBeNull(); + expect(sockets).toHaveLength(0); + }); + + it('scopes the channel to one run and memoizes it per URL', () => { + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + openWsChannel('wrun_1', directConfig); + openWsChannel('wrun_2', directConfig); + + const first = resolveWsTransport('wrun_1', directConfig); + const second = resolveWsTransport('wrun_1', directConfig); + + expect(first?.wsUrl).toContain('/websockets/v1/runs/wrun_1'); + expect(second?.transport).toBe(first?.transport); + expect(resolveWsTransport('wrun_2', directConfig)?.transport).not.toBe( + first?.transport + ); + }); + + it('returns null once the channel is closed', () => { + // What makes a late write — one the runtime issues after the invocation + // that opened the channel has returned — fall back instead of failing. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const release = openWsChannel('wrun_1', directConfig); + expect(resolveWsTransport('wrun_1', directConfig)).not.toBeNull(); + + release?.(); + + expect(resolveWsTransport('wrun_1', directConfig)).toBeNull(); + }); + + it('returns null for a World behind the api-workflow proxy', () => { + // That gateway does not forward a raw upgrade, so the caller has to fall + // back to HTTP rather than attempt a connection it can't serve. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + vi.spyOn(console, 'warn').mockImplementation(() => {}); + openWsChannel('wrun_1', proxyConfig); + + expect(resolveWsTransport('wrun_1', proxyConfig)).toBeNull(); + expect(sockets).toHaveLength(0); + }); + + it('logs the proxy fallback and the ws-in-use notice once per process', () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + + openWsChannel('wrun_1', proxyConfig); + openWsChannel('wrun_2', proxyConfig); + openWsChannel('wrun_1', directConfig); + openWsChannel('wrun_2', directConfig); + + // Every event takes the resolve path, so a per-request line would be noise. + expect(warnSpy).toHaveBeenCalledTimes(1); + expect( + logSpy.mock.calls.filter(([m]) => String(m).includes('using ws')) + ).toHaveLength(1); + }); + }); + + /** + * The flow route calls these on every message, for every World — including the + * HTTP default and the proxy World that can't speak WS at all. So "does + * nothing, quietly" is as much the contract as the open itself: a throw or an + * await would land on the critical path of every invocation, WS or not. + */ + describe('openWsChannel', () => { + it('opens the run’s socket when the gate is on', async () => { + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + + openWsChannel('wrun_1', directConfig); + await tick(); + + expect(sockets).toHaveLength(1); + expect(latest().url).toContain('/websockets/v1/runs/wrun_1'); + expect(latest().headers.authorization).toBe('Bearer test-token'); + }); + + it('does nothing when the gate is off', async () => { + openWsChannel('wrun_1', directConfig); + await tick(); + + expect(sockets).toHaveLength(0); + // And nothing registered, so writes resolve no channel either. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + expect(resolveWsTransport('wrun_1', directConfig)).toBeNull(); + }); + + it('does nothing for a World behind the api-workflow proxy', async () => { + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + vi.spyOn(console, 'warn').mockImplementation(() => {}); + + openWsChannel('wrun_1', proxyConfig); + await tick(); + + expect(sockets).toHaveLength(0); + }); + }); + + /** + * The release `openWsChannel` hands back. Bound to the instance it claimed, + * so it can only ever give up that claim — see the note on `openWsChannel`. + */ + describe('the channel release', () => { + it('closes the run’s socket', async () => { + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const release = openWsChannel('wrun_1', directConfig); + await tick(); + latest().open(); + await tick(); + expect(latest().readyState).toBe(1); + + release?.(); + await tick(); + + expect(latest().readyState).toBe(3); + }); + + it('leaves another run’s socket alone', async () => { + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + openWsChannel('wrun_1', directConfig); + await tick(); + const first = latest(); + first.open(); + const release2 = openWsChannel('wrun_2', directConfig); + await tick(); + latest().open(); + await tick(); + + release2?.(); + await tick(); + + expect(first.readyState).toBe(1); + }); + + it('holds the socket until every concurrent opener has released', async () => { + // Two invocations for one run in one instance: the first to finish must + // not close the channel the second is still writing over. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const releaseA = openWsChannel('wrun_1', directConfig); + const releaseB = openWsChannel('wrun_1', directConfig); + await tick(); + latest().open(); + await tick(); + + releaseA?.(); + await tick(); + expect(latest().readyState).toBe(1); + + releaseB?.(); + await tick(); + expect(latest().readyState).toBe(3); + }); + + it('does not release a channel a later invocation opened', async () => { + // A refused upgrade evicts the instance, so the next opener for the same + // run is registered under the same URL as a *different* channel. A + // release resolved from the URL would find that one and drop a socket + // still in use; this one holds the instance it incremented. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const releaseA = openWsChannel('wrun_1', directConfig); + await tick(); + sockets[0].failHandshake(); + await tick(); + expect(resolveWsTransport('wrun_1', directConfig)).toBeNull(); + + openWsChannel('wrun_1', directConfig); + await tick(); + sockets[1].open(); + await tick(); + expect(sockets[1].readyState).toBe(1); + + releaseA?.(); + await tick(); + + expect(sockets[1].readyState).toBe(1); + expect(resolveWsTransport('wrun_1', directConfig)).not.toBeNull(); + }); + + it('is inert when called twice', async () => { + // Double-releasing must not consume the other holder's claim, which is + // the same over-decrement by another route. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const releaseA = openWsChannel('wrun_1', directConfig); + openWsChannel('wrun_1', directConfig); + await tick(); + latest().open(); + await tick(); + + releaseA?.(); + releaseA?.(); + await tick(); + + expect(latest().readyState).toBe(1); + }); + + it('hands the next invocation a fresh channel when its open lands after the release', async () => { + // The other interleaving at refcount 1. Both halves run start to finish + // with no await, so a release either sees the next opener's increment or + // it doesn't — and when it doesn't, that opener must end up with a live + // channel of its own rather than the instance just de-registered. + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + const release = openWsChannel('wrun_1', directConfig); + await tick(); + latest().open(); + await tick(); + const first = latest(); + + release?.(); + openWsChannel('wrun_1', directConfig); + await tick(); + latest().open(); + await tick(); + + expect(first.readyState).toBe(3); + expect(latest()).not.toBe(first); + expect(latest().readyState).toBe(1); + // Registered, so this invocation's writes take the socket rather than + // silently spending the whole invocation on the HTTP fallback. + expect(resolveWsTransport('wrun_1', directConfig)).not.toBeNull(); + }); + + it('is undefined when the gate is off', () => { + // Nothing was claimed, so there is nothing for the flow route to release. + expect(openWsChannel('wrun_1', directConfig)).toBeUndefined(); + expect(sockets).toHaveLength(0); + }); + }); +}); diff --git a/packages/world-vercel/src/ws-transport.ts b/packages/world-vercel/src/ws-transport.ts new file mode 100644 index 0000000000..d612e98dbf --- /dev/null +++ b/packages/world-vercel/src/ws-transport.ts @@ -0,0 +1,840 @@ +/** + * Client half of the events WebSocket protocol. The normative wire-format and + * lifecycle spec is the server's `docs/ws-protocol.md`. + * + * One socket per `wsUrl` — which embeds the runId — is shared across concurrent + * `createWorkflowRunEventV4` calls and multiplexed by `reqId`. Frame meta is + * `{ reqId, type, ... }`, with the type-specific payload nested under a field + * named after `type` (only `event` is sent today). The caller builds the meta; + * this file owns the socket and the reqId<->promise bookkeeping. + * + * Every failure mode has to reach the caller waiting on it — a swallowed error + * here is a hung invocation, not a lost log line. Whether a failed write is + * re-sent is `event-retry.ts`'s decision, not this file's: `WsTransportError` + * is mapped to the same `code: 'TRANSPORT'` shape a failed `fetch` produces. + * + * Uses the `ws` package rather than the WHATWG global `WebSocket`, which cannot + * set headers on the upgrade request — auth rides the handshake, once per + * connection instead of once per message. + * + * Transport *selection* lives at the bottom of the file — the opt-in flag, + * which Worlds can hold a socket at all, and the pre-warm entry point — so + * `events-v4.ts` consumes one seam instead of assembling the transport. + */ + +import { getVercelOidcToken } from '@vercel/oidc'; +import { WebSocket } from 'ws'; +import { type DecodedFrame, decodeFrames } from './frames.js'; +import { getRequestTimeoutMs, headersToRecord } from './http-core.js'; +import { injectTraceContextIntoHeaders } from './telemetry.js'; +import { type APIConfig, getHttpConfig, getHttpUrl } from './utils.js'; +import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; + +export interface WsFrameReply { + meta: Record; + body: Uint8Array; +} + +/** + * A transport-level failure: the socket closed, the frame could not be handed + * to it, or a reply arrived that cannot be correlated to a caller — always + * before the frame was acked. Distinct from an application-level error (a reply + * carrying a non-2xx status), which the events adapter raises as a typed + * `@workflow/errors` error. Carries no retry policy of its own; + * `postEventFrameOverWs` maps it to `code: 'TRANSPORT'`. + */ +export class WsTransportError extends Error { + constructor(message: string, opts?: { cause?: unknown }) { + super(message, { cause: opts?.cause }); + this.name = 'WsTransportError'; + } +} + +interface PendingRequest { + resolve: (reply: WsFrameReply) => void; + reject: (err: unknown) => void; +} + +/** + * Everything scoped to one socket. `nextReqId` and `pending` live here rather + * than on the transport because the protocol defines `reqId` as a + * per-connection counter: a reconnected socket restarts at 1 and would + * otherwise collide with the previous socket's still-registered waiters. + */ +interface Connection { + ws: WebSocket; + nextReqId: number; + pending: Map; +} + +/** Reserved reqId the server replies under when a frame was too malformed to + * recover a real one from. Client-issued ids start at 1 and only increase. */ +const MALFORMED_FRAME_REQ_ID = -1; + +// Bounded so a server that is down, or an upgrade being rejected outright, +// can't become a hot reconnect loop. Once exhausted the transport goes quiet +// and the next `request()` reconnects lazily instead. +const RECONNECT_MAX_ATTEMPTS = 5; +const RECONNECT_BASE_DELAY_MS = 100; +const RECONNECT_MAX_DELAY_MS = 5_000; + +/** Absent on servers predating the field, which only ever drained for + * maxDuration — so absent reads as `max_duration`. */ +type DrainReason = 'max_duration' | 'auth_expiry'; + +/** `getHeaders` is a caller-supplied thunk and shouldn't have to guarantee + * lowercased keys the way `Headers.forEach` does. */ +function readAuthorization(headers: Record): string | null { + for (const [key, value] of Object.entries(headers)) { + if (key.toLowerCase() === 'authorization') return value; + } + return null; +} + +async function decodeOneFrame(raw: Uint8Array): Promise { + const source = (async function* () { + yield raw; + })(); + for await (const frame of decodeFrames(source)) { + return frame; + } + throw new Error('ws-transport: received an empty/unframed message'); +} + +/** Pull the `{ "message": string }` JSON an `error` frame carries as its body, + * falling back to the raw text if it isn't shaped that way. */ +function errorFrameMessage(body: Uint8Array): string { + if (body.byteLength === 0) return '(no message)'; + const text = new TextDecoder().decode(body); + try { + const parsed = JSON.parse(text) as { message?: unknown }; + return typeof parsed.message === 'string' ? parsed.message : text; + } catch { + return text; + } +} + +function describeError(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +/** + * One multiplexed connection to the events WS endpoint. Not exported — callers + * go through `getWsEventsTransport`, which caches one instance per `wsUrl`. + */ +class WsEventsTransport { + private connection: Connection | null = null; + private connecting: Promise | null = null; + private reconnectAttempts = 0; + private reconnectTimer: ReturnType | null = null; + /** Number of `open()` calls not yet matched by a `release()`. The socket goes + * away at zero. A refcount rather than a flag because concurrent invocations + * for one run share this instance — see `open`. */ + private openCount = 0; + /** Set by `close()`. Suppresses reconnects so an intentional teardown can't + * be undone by the close handler it triggers. */ + private closed = false; + /** Reason from the most recent `drain`, consumed by the close that follows. */ + private lastDrainReason: DrainReason | null = null; + /** A drain said the *token* expired, not that the socket aged out — the next + * connect must not reuse the same bearer. */ + private needsFreshToken = false; + /** Authorization the current socket was opened with, so a forced refresh can + * tell whether it actually produced a new one. */ + private lastAuthorization: string | null = null; + + constructor( + private readonly wsUrl: string, + private readonly getHeaders: (opts: { + forceRefresh: boolean; + }) => Promise> + ) {} + + /** Send one request frame and wait for its matching reply. `buildFrame` + * receives the reqId to embed in the meta before framing. */ + async request( + buildFrame: (reqId: number) => Uint8Array + ): Promise { + if (this.closed) { + // Unreachable through `resolveWsTransport`, which only hands back a + // registered channel and `close()` de-registers synchronously. Reviving + // here instead would recreate the socket nobody owns that the explicit + // open/close pair exists to rule out. + throw new WsTransportError( + `workflow-server events WS channel for ${this.wsUrl} is closed` + ); + } + const conn = await this.ensureConnected(); + + const reqId = conn.nextReqId++; + const frame = buildFrame(reqId); + const timeoutMs = getRequestTimeoutMs(); + let deadline: ReturnType | undefined; + try { + return await new Promise((resolve, reject) => { + conn.pending.set(reqId, { resolve, reject }); + // Deliberately the same knob the HTTP path uses. Without it, a reply + // that never arrives for a reason the error/close handling doesn't + // cover blocks the caller until the platform's `maxDuration` + // SIGTERM — including a server that accepts a frame and never + // answers it. + deadline = setTimeout(() => { + if (!conn.pending.delete(reqId)) return; + reject( + new WsTransportError( + `workflow-server events WS request ${reqId} to ${this.wsUrl} ` + + `timed out after ${timeoutMs}ms with no reply` + ) + ); + // A socket that accepted a frame and never answered it is not one + // the next write should be handed. + this.failConnection( + conn, + `workflow-server events WS connection to ${this.wsUrl} ` + + `abandoned after request ${reqId} timed out` + ); + }, timeoutMs); + deadline.unref?.(); + conn.ws.send(frame, (err) => { + if (!err) return; + // `ws.send()` does not throw when the socket isn't OPEN — it + // reports here instead, so without this callback the request would + // wait for a reply that is never coming. `delete` doubles as the + // already-settled guard. + if (conn.pending.delete(reqId)) { + reject( + new WsTransportError( + `workflow-server events WS send failed: ${describeError(err)}`, + { cause: err } + ) + ); + } + }); + }); + } finally { + if (deadline !== undefined) clearTimeout(deadline); + } + } + + /** + * Claim the channel and start connecting. Called once per invocation that + * intends to write, at the point the run id is known — connecting lazily on + * the first write instead bills the handshake (plus the OIDC mint riding it) + * to whichever event happens to be written first; when that's a `step_started` + * issued as the step body already runs, its server-recorded timestamp lands + * after the work it timestamps and the step reads as shorter than it was. + * + * Fire-and-forget: the connect is unawaited, and `request` awaits the same + * promise, so a write issued while the handshake is still in flight joins it + * rather than racing it. + * + * A failed connect closes the channel rather than leaving it registered for + * `request` to retry, which is what keeps this path from being *worse* than + * HTTP: writes resolve no channel and take the pooled HTTP route instead of + * paying another handshake attempt each. + * + * That de-opt covers the handshake only. A channel that connects and then + * fails every write stays registered and keeps taking the WS path, so the + * fallback is per-invocation rather than a circuit breaker. Tolerable while + * the transport is opt-in; defaulting it on needs a write-failure de-opt too. + */ + open(): void { + if (this.closed) return; + this.openCount++; + if (this.connection !== null || this.connecting !== null) return; + void this.ensureConnected().catch(() => { + // Already logged by `connect`. + this.close('connect failed'); + }); + } + + /** + * Give up this invocation's claim, dropping the socket once the last holder + * does. Concurrent invocations for one run share this instance (inline step + * executions ride the flow topic on per-step topics, so a run's steps can be + * separate invocations of the same container), and the first to finish must + * not tear the socket out from under the others. + * + * Unbalanced calls are inert: without a matching `open` there is nothing to + * release, and the count floors at zero rather than going negative and + * pinning the socket for the life of the process. + */ + release(reason: string): void { + if (this.openCount === 0) return; + this.openCount--; + if (this.openCount === 0) this.close(reason); + } + + /** + * Drop the socket and evict this transport from the cache. Terminal, not a + * pause: the instance stays closed and a later `openWsChannel` for the same + * run constructs a fresh one. Idempotent. Safe with work in flight — those + * requests fail through the socket's own `close` handler, same as any other + * close. + */ + close(reason: string): void { + if (this.closed) return; + this.closed = true; + this.openCount = 0; + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + if (transports.get(this.wsUrl) === this) transports.delete(this.wsUrl); + const conn = this.connection; + this.connection = null; + // Normal closure: a clean client-side release, not an aborted run. + conn?.ws.close(1000, reason); + } + + private ensureConnected(): Promise { + const conn = this.connection; + if (conn && conn.ws.readyState === WebSocket.OPEN) { + return Promise.resolve(conn); + } + // Clearing the slot here rather than from the socket handlers keeps the + // bookkeeping in one place, and stops a stale socket's late `close` from + // nulling out a newer connect that has since taken the slot. + this.connecting ??= this.connect().finally(() => { + this.connecting = null; + }); + return this.connecting; + } + + /** + * Resolve the headers for one upgrade request — once per socket, since the + * bearer only rides the upgrade. Re-resolving per socket is what lets a + * reconnect pick up a fresh token rather than replaying an expired one. + * + * Also the only place this transport can carry W3C trace context, which + * `CLAUDE.md` requires of every outgoing world-vercel request: frames are not + * HTTP requests and carry no headers of their own, so the server parents its + * event spans to whichever invocation opened the socket rather than to the + * individual write. + */ + private async resolveUpgradeHeaders(): Promise> { + const forceRefresh = this.needsFreshToken; + const headers = await this.getHeaders({ forceRefresh }); + const authorization = readAuthorization(headers); + + if (forceRefresh && authorization === this.lastAuthorization) { + // The server says the token is expiring and we can't produce a different + // one. Inside a Vercel function the bearer comes from the invocation's own + // `x-vercel-oidc-token` and is fixed for that invocation, so reconnecting + // now just earns a 401 and burns the attempt budget. Give up on the eager + // reconnect; queue redelivery recovers this by landing in a new + // invocation with a new token. + throw new WsTransportError( + `workflow-server events WS connection to ${this.wsUrl} drained for ` + + `auth expiry, but re-resolving the bearer produced the same token. ` + + `Not reconnecting with a token the server has already rejected; ` + + `the next event write will retry.` + ); + } + + this.lastAuthorization = authorization; + this.needsFreshToken = false; + + // `injectTraceContextIntoHeaders` writes into a `Headers`; the upgrade takes + // a plain record, so carry the propagator's output across. No-op when no + // OTEL SDK is registered. + const carrier = new Headers(); + await injectTraceContextIntoHeaders(carrier); + carrier.forEach((value, key) => { + headers[key] = value; + }); + + return headers; + } + + private consumeDrainReason(): void { + const reason = this.lastDrainReason; + this.lastDrainReason = null; + if (reason === 'auth_expiry') this.needsFreshToken = true; + } + + private connect(): Promise { + return new Promise((resolve, reject) => { + void (async () => { + let conn: Connection; + try { + const headers = await this.resolveUpgradeHeaders(); + const ws = new WebSocket(this.wsUrl, { headers }); + ws.binaryType = 'nodebuffer'; + conn = { ws, nextReqId: 1, pending: new Map() }; + } catch (err) { + console.error( + `world-vercel: ws events transport could not open a connection ` + + `to ${this.wsUrl}: ${describeError(err)}` + ); + // Nothing was sent, so re-sending is safe wherever the shared policy + // allows it. Already-typed errors pass through untouched. + reject( + err instanceof WsTransportError + ? err + : new WsTransportError( + `workflow-server events WS connection to ${this.wsUrl} ` + + `could not be opened: ${describeError(err)}`, + { cause: err } + ) + ); + return; + } + + const { ws } = conn; + let opened = false; + + ws.on('open', () => { + opened = true; + if (this.closed) { + // Released while this handshake was in flight — `close()` could + // only null out the connection it could see. Adopting this one now + // would leave a live socket on a transport nothing will ever close. + ws.close(1000, 'released while connecting'); + reject( + new WsTransportError( + `workflow-server events WS connection to ${this.wsUrl} was ` + + `released while connecting` + ) + ); + return; + } + this.connection = conn; + this.reconnectAttempts = 0; + resolve(conn); + }); + + ws.on('message', (raw: Buffer) => { + void this.handleMessage(conn, new Uint8Array(raw)); + }); + + ws.on('error', (err) => { + console.error( + `world-vercel: ws events transport socket error ` + + `(${this.wsUrl}): ${describeError(err)}` + ); + if (!opened) { + reject(err); + return; + } + // `ws` normally follows `'error'` with `'close'`, which tears down and + // schedules the reconnect. Close explicitly in case it doesn't. + ws.close(); + }); + + ws.on('close', (code: number) => { + const wasActive = this.connection === conn; + if (wasActive) this.connection = null; + + this.consumeDrainReason(); + + if (!opened) { + // A close with no preceding `'error'` would leave `ensureConnected` + // waiting forever. Rejecting an already-settled promise is a no-op. + reject( + new WsTransportError( + `workflow-server events WS connection to ${this.wsUrl} ` + + `closed before opening (code ${code})` + ) + ); + } + + // Unconditional because `pending` is per-connection: a superseded + // socket's late close can only reach its own waiters. The frame was + // in flight when the socket died, so the server either never saw it + // or never acked it — re-sending is safe, createEvent writes are + // conditional on the entity server-side. + this.failAllPending( + conn, + new WsTransportError( + `workflow-server events WS connection closed (code ${code})` + ) + ); + + if (opened || this.reconnectAttempts > 0) { + this.scheduleReconnect(code); + } + }); + })(); + }); + } + + /** + * Reconnect eagerly rather than waiting for the next write: an unexpected + * close mid-run (broken pipe, or the server's drain → 1001) almost always + * means more writes are coming, and paying the handshake now keeps it off the + * next write's critical path. + */ + private scheduleReconnect(closeCode: number): void { + // An intentional teardown lands here via the socket's own `close` event; + // without this guard the transport would reconnect what it just released. + if (this.closed) return; + // A newer socket is already live. Deliberately not also checking + // `this.connecting`: `ws` can emit `'error'` and `'close'` in one tick, so + // that slot may not be cleared yet and bailing would break the retry + // chain. `ensureConnected` dedups, so at worst the timer joins a connect + // already in flight. + if (this.connection !== null) return; + if (this.reconnectTimer !== null) return; + + if (this.reconnectAttempts >= RECONNECT_MAX_ATTEMPTS) { + console.error( + `world-vercel: ws events transport giving up eager reconnect to ` + + `${this.wsUrl} after ${RECONNECT_MAX_ATTEMPTS} attempts ` + + `(last close code ${closeCode}); the next event write will retry.` + ); + return; + } + + const delayMs = Math.min( + RECONNECT_BASE_DELAY_MS * 2 ** this.reconnectAttempts, + RECONNECT_MAX_DELAY_MS + ); + this.reconnectAttempts++; + + const timer = setTimeout(() => { + this.reconnectTimer = null; + void this.ensureConnected().catch(() => {}); + }, delayMs); + timer.unref?.(); + this.reconnectTimer = timer; + } + + private async handleMessage( + conn: Connection, + raw: Uint8Array + ): Promise { + let decoded: DecodedFrame; + try { + decoded = await decodeOneFrame(raw); + } catch (err) { + // Uncorrelatable, and it says the framing on this socket is no longer + // trustworthy — so the connection goes rather than leaving its waiters + // unanswerable. + const detail = + `could not decode a ${raw.byteLength}-byte reply frame from ` + + `${this.wsUrl}: ${describeError(err)}`; + console.error(`world-vercel: ws events transport ${detail}`); + this.failConnection( + conn, + `workflow-server events WS transport ${detail}` + ); + return; + } + + if (decoded.meta.type === 'drain') { + // Unsolicited server push, no reqId. Informational on its own — the + // `close` that follows is what triggers the reconnect and consumes the + // reason recorded here. + const reason: DrainReason = + decoded.meta.reason === 'auth_expiry' ? 'auth_expiry' : 'max_duration'; + this.lastDrainReason = reason; + console.log( + `world-vercel: ws events transport received a drain notice ` + + `(reason: ${reason}, graceMs: ${decoded.meta.graceMs ?? 'unspecified'}); ` + + `connection will close soon.` + ); + return; + } + + const reqId = decoded.meta.reqId; + + if (reqId === MALFORMED_FRAME_REQ_ID) { + // Loud: the server's schema rejected a frame this client sent, which is a + // protocol bug on our side, not a transient failure. Nothing to correlate + // it to by construction. + const detail = + `received a malformed-frame error from ${this.wsUrl} (reserved reqId ` + + `${MALFORMED_FRAME_REQ_ID}, status ` + + `${String(decoded.meta.status ?? 'unknown')}): ` + + `${errorFrameMessage(decoded.body)}. The server could not parse a ` + + `frame this client sent, so the originating write cannot be identified`; + console.error(`world-vercel: ws events transport ${detail}.`); + this.failConnection( + conn, + `workflow-server events WS transport ${detail}` + ); + return; + } + + if (typeof reqId !== 'number') { + const detail = + `received a reply frame with no numeric reqId from ${this.wsUrl} ` + + `(type: ${String(decoded.meta.type ?? 'absent')})`; + console.error(`world-vercel: ws events transport ${detail}.`); + this.failConnection( + conn, + `workflow-server events WS transport ${detail}` + ); + return; + } + + const pending = conn.pending.get(reqId); + if (!pending) { + // The one uncorrelatable case that does NOT tear the connection down: an + // id nobody is waiting on means that request already settled (send error, + // or its own deadline with the reply arriving late). Nothing is orphaned, + // and the frame decoded with a real reqId, so the stream is intact. + console.error( + `world-vercel: ws events transport received a reply for unknown ` + + `reqId ${reqId} from ${this.wsUrl} (already settled, or the request ` + + `was failed by a send error or its deadline); dropping it.` + ); + return; + } + conn.pending.delete(reqId); + pending.resolve({ meta: decoded.meta, body: decoded.body }); + } + + private failAllPending(conn: Connection, err: unknown): void { + for (const pending of conn.pending.values()) { + pending.reject(err); + } + conn.pending.clear(); + } + + /** + * Fail every waiter on `conn` and drop the socket — for a reply that can't be + * correlated to a caller, and for a request that outlived its deadline. The + * whole connection goes because each case says the stream itself is no longer + * understood; logging and moving on would leave the originating request in + * `pending` with nothing able to answer it until the server drains (~680s), + * well past the waiting invocation's `maxDuration`. The socket's own `close` + * handler does the teardown and schedules the reconnect — failing the waiters + * here is what makes callers see this diagnosis, not a bare close code. + */ + private failConnection(conn: Connection, message: string): void { + this.failAllPending(conn, new WsTransportError(message)); + if (this.connection === conn) this.connection = null; + conn.ws.close(); + } +} + +const transports = new Map(); + +/** + * Get (or lazily create) the shared WS transport for `wsUrl`. `getHeaders` runs + * once per socket, at connect time, with `forceRefresh: true` when the previous + * socket drained on an expiring token, so a caching token source knows not to + * serve the stale entry. Only the first caller's thunk for a `wsUrl` is kept — + * fine, since `wsUrl` embeds the runId and one run has one client. An entry + * exists only between `openWsChannel` and the matching `closeWsChannel`, so + * membership *is* the answer to "does this run have a channel" — which is what + * the write path asks, via `resolveWsTransport`. + */ +export function getWsEventsTransport( + wsUrl: string, + getHeaders: (opts: { + forceRefresh: boolean; + }) => Promise> +): WsEventsTransport { + let transport = transports.get(wsUrl); + if (!transport) { + transport = new WsEventsTransport(wsUrl, getHeaders); + transports.set(wsUrl, transport); + } + return transport; +} + +/** + * Test seam: close and drop every cached transport, and re-arm the + * once-per-process log latches below so a test asserting on either message + * isn't silenced by an earlier one having already logged it. + */ +export function resetWsEventsTransportsForTest(): void { + for (const transport of [...transports.values()]) { + transport.close('test reset'); + } + transports.clear(); + loggedWsProxyFallback = false; + loggedWsInUse = false; +} + +/** + * Derive the WS endpoint URL for one run from the http(s) base URL used for v4 + * REST calls. `/websockets/v1/runs/:runId` is versioned independently of the + * `v4` REST API these frames are forwarded into (server `docs/ws-protocol.md`). + * Scoped to one run because one client instance only ever drives one run, so + * `runId` belongs on the connection rather than on every frame. + */ +export function toEventsWsUrl(baseUrl: string, runId: string): string { + const url = new URL(baseUrl); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + url.pathname = `${url.pathname.replace(/\/$/, '')}/websockets/v1/runs/${encodeURIComponent(runId)}`; + return url.toString(); +} + +// ============================================================================= +// Transport selection +// ============================================================================= + +/** + * Re-exported so this module stays the single seam for everything socket-shaped. + * Import it from `./ws-transport-enabled.js` instead when the point is to *not* + * load this module. + */ +export { isWsEventsTransportEnabled }; + +/** + * Open this run's events channel for the duration of one invocation, start + * connecting, and return the release for that claim. Nothing else creates a + * channel, so a code path that doesn't call this writes its events over HTTP + * even with the transport enabled. + * + * Call it as early in an invocation as the run id is known — the flow route does, + * before dispatching to the runtime. Only worth it for a caller that will write + * several events: a single write does not repay a handshake, which is why + * `start()` deliberately doesn't open one for `run_created`. + * + * Silent and synchronous by contract: `undefined` on the HTTP default and on a + * World that can't hold a socket, and it neither awaits nor reports the + * connect. Callers must be able to treat it as free. + * + * An open socket is not `unref`'d, so a caller that drops the release stops the + * process exiting — and keeps a server invocation pinned, one per connection — + * until the platform kills it. The socket drops once every concurrent holder + * has released; see `WsEventsTransport.release`. + * + * **The release closes over the instance, never the URL.** A channel is evicted + * from `transports` the moment it closes, and a refused upgrade does that on + * the connect path, so the next opener for the same run gets a *different* + * instance under the same URL. A release that re-resolved the URL would + * decrement whichever instance is registered by then rather than the one it + * claimed, dropping a socket a live invocation is still writing over — and for + * the event types `EVENT_RETRY_ELIGIBILITY` marks non-retryable there is no + * second attempt to carry that write over HTTP. Idempotent for the same reason: + * a doubled release must not consume another holder's claim. + * + * Synchronous matters beyond cost. Two invocations for one run share a channel, + * so an open races a release whenever the refcount stands at 1 — and because + * neither yields between its map lookup and its refcount write, the release + * either observes the open's increment, and the socket survives for the new + * holder, or lands first, and the opener misses the de-registered instance and + * builds a fresh channel. Adding an await ahead of the refcount write — minting + * a token to derive the URL, say — reopens that window, and an open landing + * inside it hits an already-closed instance, returns early, and leaves that + * invocation on HTTP for its whole duration with nothing to signal it. + */ +export function openWsChannel( + runId: string, + config?: APIConfig +): (() => void) | undefined { + if (!isWsEventsTransportEnabled()) return undefined; + const resolved = resolveChannelUrl(runId, config); + if (!resolved) return undefined; + if (!loggedWsInUse) { + loggedWsInUse = true; + console.log(`world-vercel: using ws events transport (${resolved}).`); + } + // Cheap: a URL plus a map lookup, no token mint and no I/O. The socket work + // happens inside `open`, unawaited. + const transport = getWsEventsTransport(resolved, async ({ forceRefresh }) => { + // `forceRefresh` means the previous socket was drained by the server + // because its *bearer* was expiring, not because the socket aged out. + if (forceRefresh) await refreshOidcTokenBestEffort(); + const { headers } = await getHttpConfig(config); + return headersToRecord(headers); + }); + transport.open(); + + let released = false; + return () => { + if (released) return; + released = true; + transport.release('invocation complete'); + }; +} + +/** + * A buffer wider than any OIDC token's lifetime, so `isExpired()` reports + * true and `@vercel/oidc` takes its refresh path instead of handing back the + * cached entry. 24h comfortably exceeds the 60min TTL. + */ +const OIDC_FORCE_REFRESH_BUFFER_MS = 24 * 60 * 60 * 1000; + +/** + * Ask `@vercel/oidc` for a new token before the next `getHttpConfig()` reads + * one, in response to a drain with `reason: 'auth_expiry'`. Swallows failures: + * an unavailable refresh must not fail a write that `getHttpConfig()` can still + * produce a usable — if soon-to-expire — bearer for. + * + * Only effective outside a Vercel function. `getVercelOidcToken()` prefers + * `getContext().headers['x-vercel-oidc-token']` over + * `process.env.VERCEL_OIDC_TOKEN`, and the refresh only writes the env var, so + * inside a function the invocation's own header token shadows anything minted + * here and there is genuinely no fresher bearer mid-invocation. The transport's + * stale-token guard covers that case instead. In a CLI, local dev or a + * long-lived server there is no context header and the reconnect picks up the + * new token. + */ +async function refreshOidcTokenBestEffort(): Promise { + try { + await getVercelOidcToken({ + expirationBufferMs: OIDC_FORCE_REFRESH_BUFFER_MS, + }); + } catch { + // No refresh path available here. getHttpConfig() below still resolves + // whatever token it can, and the transport decides whether reconnecting + // with it is worth attempting. + } +} + +// Each logged at most once per process — both branches below are expected +// to repeat (every event), and a per-request log would just be noise. +let loggedWsProxyFallback = false; +let loggedWsInUse = false; + +/** + * Resolve this run's channel URL, or `null` when this World can't hold a socket + * at all and every caller must use HTTP. Says nothing about whether a channel is + * *open* — that's `resolveWsTransport`. + */ +function resolveChannelUrl( + runId: string, + config: APIConfig | undefined +): string | null { + // `getHttpUrl` is the cheap, synchronous half of `getHttpConfig`: it resolves + // the route without minting a token. The bearer only rides the upgrade, so + // the transport resolves it once per socket instead of once per write. + const { baseUrl, usingProxy } = getHttpUrl(config); + if (usingProxy) { + // `usingProxy` resolves `baseUrl` to `api.vercel.com/v1/workflow`, an + // HTTP-only REST gateway that does not forward a raw WebSocket upgrade to + // the workflow-server target — it either rejects the upgrade or hands the + // route a plain forwarded request that never went through Vercel's + // platform-level upgrade path, which is what surfaces as + // "experimental_upgradeWebSocket is not available in the current runtime + // environment". Fall back rather than fail a connection it can't serve. + if (!loggedWsProxyFallback) { + loggedWsProxyFallback = true; + console.warn( + `world-vercel: ws events transport requested but a World with projectConfig ` + + `(api-workflow proxy, resolved baseUrl: ${baseUrl}) is active — falling back.` + ); + } + return null; + } + return toEventsWsUrl(baseUrl, runId); +} + +/** + * The write path's question: is there an open channel for this run? `null` means + * write over HTTP — because the transport is disabled, because this World can't + * hold a socket, or because nothing opened a channel for this invocation. + * + * A lookup, never a create. Lazily connecting here is what made the socket's + * lifetime a property of the *last write* rather than of the caller, and left a + * timer as the only thing able to end it. + */ +export function resolveWsTransport( + runId: string, + config: APIConfig | undefined +): { + transport: WsEventsTransport; + wsUrl: string; +} | null { + const wsUrl = resolveChannelUrl(runId, config); + if (!wsUrl) return null; + const transport = transports.get(wsUrl); + return transport ? { transport, wsUrl } : null; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 72197c9c64..93004071e2 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1548,6 +1548,9 @@ importers: undici: specifier: 'catalog:' version: 7.29.0 + ws: + specifier: ^8.20.0 + version: 8.20.0 zod: specifier: 'catalog:' version: 4.3.6 @@ -1567,6 +1570,9 @@ importers: '@types/node': specifier: 'catalog:' version: 22.19.0 + '@types/ws': + specifier: ^8.18.1 + version: 8.18.1 '@workflow/tsconfig': specifier: workspace:* version: link:../tsconfig @@ -9583,6 +9589,9 @@ packages: '@types/unist@3.0.3': resolution: {integrity: sha512-ko/gIFJRv177XgZsZcBwnqJN5x/Gien8qNOn0D5bQU/zAzVf9Zt3BlcUiLqhV9y4ARk0GbT3tnUiPNgnTXzc/Q==} + '@types/ws@8.18.1': + resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} + '@typescript-eslint/types@8.46.4': resolution: {integrity: sha512-USjyxm3gQEePdUwJBFjjGNG18xY9A2grDVGuk7/9AkjIF1L+ZrVnwR5VAU5JXtUnBL/Nwt3H31KlRDaksnM7/w==} engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0} @@ -26026,6 +26035,10 @@ snapshots: '@types/unist@3.0.3': {} + '@types/ws@8.18.1': + dependencies: + '@types/node': 22.19.0 + '@typescript-eslint/types@8.46.4': optional: true @@ -26246,7 +26259,7 @@ snapshots: '@vercel/queue@0.3.1': dependencies: - '@vercel/oidc': 3.8.1 + '@vercel/oidc': 3.8.2 minimatch: 10.2.5 mixpart: 0.0.6 picocolors: 1.1.1 diff --git a/workbench/tanstack-start/vite.config.ts b/workbench/tanstack-start/vite.config.ts index 6b11059419..4b96c58246 100644 --- a/workbench/tanstack-start/vite.config.ts +++ b/workbench/tanstack-start/vite.config.ts @@ -8,5 +8,10 @@ export default defineConfig({ plugins: [workflow(), tanstackStart(), nitro(), viteReact()], nitro: { plugins: ['./plugins/start-pg-world.ts'], + // NOTE: `ws`'s optional native accelerators (`bufferutil`, + // `utf-8-validate`) used to be externalized here. That now happens in + // `@workflow/rollup`'s `workflowTransformPlugin`, which every Nitro-based + // integration already installs — so a real user gets the fix without + // copying this block into their own config. }, }); diff --git a/workbench/vite/vite.config.ts b/workbench/vite/vite.config.ts index 8bd8e7590c..bf8e700ceb 100644 --- a/workbench/vite/vite.config.ts +++ b/workbench/vite/vite.config.ts @@ -19,5 +19,10 @@ export default defineConfig({ nitro: { serverDir: './', plugins: ['plugins/start-pg-world.ts'], + // NOTE: `ws`'s optional native accelerators (`bufferutil`, + // `utf-8-validate`) used to be externalized here. That now happens in + // `@workflow/rollup`'s `workflowTransformPlugin`, which every Nitro-based + // integration already installs — so a real user gets the fix without + // copying this block into their own config. }, });