diff --git a/.changeset/global-singleton-helper.md b/.changeset/global-singleton-helper.md new file mode 100644 index 0000000000..8da5f89aea --- /dev/null +++ b/.changeset/global-singleton-helper.md @@ -0,0 +1,5 @@ +--- +'@workflow/utils': minor +--- + +Add `globalSingleton()`, which parks a package's process-wide state on `globalThis` so bundled copies of a module in one process share it. diff --git a/.changeset/module-scope-lint-hardening.md b/.changeset/module-scope-lint-hardening.md new file mode 100644 index 0000000000..29968dc95f --- /dev/null +++ b/.changeset/module-scope-lint-hardening.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-testing': patch +--- + +Annotate the test server's per-run invocation counter as deliberately per-copy, so it passes the module-scope state rule. diff --git a/.changeset/module-scope-state-all-bundled-packages.md b/.changeset/module-scope-state-all-bundled-packages.md new file mode 100644 index 0000000000..e73a33cb4a --- /dev/null +++ b/.changeset/module-scope-state-all-bundled-packages.md @@ -0,0 +1,8 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +'@workflow/ai': patch +'@workflow/nest': patch +--- + +Hold process-wide state on `globalThis` rather than at module scope in the packages that get bundled into the host application's server build, where a bundler compiles one copy of each module per layer. Covers warn-once latches, lazy caches, the VM script and QuickJS asset caches, the dev-server port cache, and step single-flight, whose per-copy map was not actually single-flight. State that is deliberately per-copy is annotated with the reason. diff --git a/.changeset/reuse-runtime-world-for-route-handlers.md b/.changeset/reuse-runtime-world-for-route-handlers.md new file mode 100644 index 0000000000..bb5a8d8e61 --- /dev/null +++ b/.changeset/reuse-runtime-world-for-route-handlers.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Build the workflow entrypoint's queue handler from the runtime World (`getWorld()`) instead of `getWorldHandlers()`, so a process creates one World rather than two. A stateful World no longer gets duplicate connection pools or queue workers. diff --git a/.changeset/utils-side-effects-free.md b/.changeset/utils-side-effects-free.md new file mode 100644 index 0000000000..eae52bf9ba --- /dev/null +++ b/.changeset/utils-side-effects-free.md @@ -0,0 +1,5 @@ +--- +'@workflow/utils': patch +--- + +Declare `sideEffects: false` so bundlers can drop the unused parts of the barrel from a host application's build. diff --git a/.changeset/world-module-scope-state.md b/.changeset/world-module-scope-state.md new file mode 100644 index 0000000000..107e243983 --- /dev/null +++ b/.changeset/world-module-scope-state.md @@ -0,0 +1,6 @@ +--- +'@workflow/world-vercel': patch +'@workflow/world-local': patch +--- + +Hold process-wide state (the WebSocket transport registry, HTTP connection pools, ULID factories, caches, log-once latches) on `globalThis` instead of at module scope. This de-duplicates state across bundled packages. Fixes WebSocket transport, which was registered in one module state but looked up in another. diff --git a/AGENTS.md b/AGENTS.md index 573d600925..7c4773dccc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -478,6 +478,45 @@ The `executionContext` field on workflow runs is a flexible JSONB/CBOR object th ### Observability data hydration `packages/core/src/observability.ts` contains `hydrateResourceIO` which strips certain fields (like `executionContext`) before UI display. If you need to display data from stripped fields, extract it before the stripping occurs. +### World packages must not hold mutable module state + +`@workflow/world-local` and `@workflow/world-vercel` are bundled into the host +application's server build (see `VERCEL_WORLD_DEPENDENCY_PACKAGES` in +`packages/next/src/index.ts`). Bundlers key module identity on +`(resource, layer)`, and Next.js alone compiles `instrument`, app-route, `ssr` +and `edge` as separate module graphs, so one process holds one copy of every +module in these packages **per bundler layer**. A top-level `let`, or a `const` +holding a `Map`, is per-copy state, not the process singleton it reads as. A +duplicated mutex stops mutually excluding; a duplicated registry is a +deterministic miss; duplicated ID generators can fork a sequence. + +Hold such state on the World instance where it is per-World, or on `globalThis` +via `globalSingleton()` from `@workflow/utils` where it is genuinely +process-wide. State that is deliberately per-copy needs a +`// per-copy-ok: ` annotation. `scripts/lint/module-scope-state.mjs` +enforces this across every published `packages/world-*`, run from +`@workflow/utils`'s test suite (with a local mirror in each world package), so +adding a new world package is covered automatically. + +Custom worlds loaded through `WORKFLOW_TARGET_WORLD` are deduped by Node's +module cache and are safe today, but that is a property of how they are loaded, +not of how they are written, and it changed for world-vercel in #3493. Keep them +clean too. The author-facing version of this rule is in +`docs/content/worlds/{v4,v5}/building-a-world.mdx`; keep both versions in sync. + +The sweep covers every package that ends up inside the host application's +server build: all published `packages/world-*` (discovered at runtime, so a new +world is covered the day it is added) plus `core`, `world`, `ai` and `nest`, +which are named in `BUNDLED_RUNTIME_PACKAGES` in +`packages/utils/src/module-scope-state.test.ts`. Adding a package that runs in +the host server means adding it to that list: "does this run inside the host's +server bundle" is a judgement, not something to infer from a directory name. + +Deliberately outside the sweep, because a single module graph makes the hazard +impossible: `next`, `builders` and `sveltekit` (build-time code), `cli` (its own +process), `web` and `web-shared` (the observability UI), `vitest` (the test +runner's process), and private packages such as `world-sim`. + ### Trace context propagation (world-vercel HTTP requests) Every outgoing HTTP request from `@workflow/world-vercel` to workflow-server (or the queue) MUST explicitly inject W3C trace context so the server can parent its spans to the caller and traces stay correlated end to end. Call `injectTraceContextIntoHeaders(headers)` (from `packages/world-vercel/src/telemetry.ts`) on the outgoing headers, inside the client span when one exists. `makeRequest` in `utils.ts` is the reference implementation. It is a no-op when no OpenTelemetry SDK is registered. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/get-world-handlers.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/get-world-handlers.mdx index 8eeef050c2..aa2594ffb4 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/get-world-handlers.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/get-world-handlers.mdx @@ -35,10 +35,10 @@ type WorldHandlers = Pick; ``` - This is SDK infrastructure used by framework adapters and the workflow entrypoint. Application code should use [`getWorld()`](/docs/api-reference/workflow-runtime/get-world) instead. + This is SDK infrastructure used by framework adapters at build time. Runtime routes and application code should use [`getWorld()`](/docs/api-reference/workflow-runtime/get-world) instead. ## Related functions - [`getWorld()`](/docs/api-reference/workflow-runtime/get-world): Resolve the full World instance at runtime. -- [`workflowEntrypoint()`](/docs/api-reference/workflow-runtime/workflow-entrypoint): The route handler factory built on these handlers. +- [`workflowEntrypoint()`](/docs/api-reference/workflow-runtime/workflow-entrypoint): Create the runtime route handler that shares the full World instance. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/workflow-entrypoint.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/workflow-entrypoint.mdx index d1d776b51e..bb0aa328e0 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/workflow-entrypoint.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/workflow-entrypoint.mdx @@ -38,5 +38,6 @@ Returns a fetch-style request handler: `(req: Request) => Promise`. ## Related functions -- [`getWorldHandlers()`](/docs/api-reference/workflow-runtime/get-world-handlers): The build-time World access this handler is built on. +- [`getWorld()`](/docs/api-reference/workflow-runtime/get-world): Resolve the runtime World instance this handler shares with workflow execution. +- [`getWorldHandlers()`](/docs/api-reference/workflow-runtime/get-world-handlers): Access build-time-safe World handlers for framework tooling. - [`healthCheck()`](/docs/api-reference/workflow-runtime/health-check): Verify the entrypoint processes queue messages end-to-end. diff --git a/docs/content/worlds/v4/building-a-world.mdx b/docs/content/worlds/v4/building-a-world.mdx index ceee349678..a5078235dc 100644 --- a/docs/content/worlds/v4/building-a-world.mdx +++ b/docs/content/worlds/v4/building-a-world.mdx @@ -217,6 +217,65 @@ Streams are identified by a combination of `runId` and `name`. Each workflow run `getChunks` returns a paginated snapshot of currently available chunks (unlike `get` which returns a live `ReadableStream` that waits for new chunks). `getInfo` returns the tail index (last chunk index, 0-based, or `-1` when empty) and whether the stream is complete, which is useful for resolving negative `startIndex` values into absolute positions. +## Process-wide state + +Hold state that must be process-wide on `globalThis`, not at module scope. + +A World is loaded in one of two ways, and only one of them gives your package a +single module instance: + +- **Loaded at runtime.** `WORKFLOW_TARGET_WORLD=@your-org/world-foo` is resolved + with `require()` at runtime, so Node's module cache dedupes it and one process + holds one copy. +- **Bundled.** The host application's bundler compiles your package into its + server build. Bundlers key module identity on `(resource, layer)`, and a + framework routinely builds several server layers. Next.js compiles + `instrument`, app-route, `ssr` and `edge` as separate module graphs. Your + package is then compiled into each one, so a single process holds several + copies of every one of your modules, each with its own module scope. + +The two built-in worlds are bundled. A custom world is not today, but that is a +property of how it is loaded rather than of how it is written, and it can change +under you. `@workflow/world-vercel` was external until it wasn't, and every +module-scope variable in it silently became per-copy state. + +So a top-level `let` or a `const` holding a `Map` is not the singleton it looks +like: + +```typescript +// Wrong: one Map per copy. Writes from one part of the app are invisible to +// another, and a mutex like this simply stops mutually excluding. +const locks = new Map>(); +``` + +Reach for `globalThis` under a `Symbol.for()` key instead, so every copy shares +one object: + +```typescript +type WorldState = { locks: Map> }; + +const StateKey = Symbol.for('@your-org/world-foo//locks/v1'); +const store = globalThis as typeof globalThis & + Record; + +const state: WorldState = (store[StateKey] ??= { locks: new Map() }); +``` + +Version the key. Two releases of your package can end up in one process, and a +key without a version lets an older copy read a state object it does not +understand. + +Inside this repository, `globalSingleton()` from `@workflow/utils` does exactly +this and is what the first-party worlds use; the hand-rolled form above is +written out so a world published outside this repository does not need the +dependency. `scripts/lint/module-scope-state.mjs` accepts either. + +Better still, keep the state on the World instance your `createWorld()` returns. +Connection pools, caches, and open channels are usually per-World rather than +per-process, and instance state cannot be duplicated by a bundler. Reserve the +global for the few things that are genuinely process-wide: ID generators whose +sequence must not fork, and log-once latches. + ## Reference implementations Study these implementations for guidance: diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index 3b853c6c18..72023dc014 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -373,6 +373,65 @@ If you implement this namespace, observe the following requirements: See the [Analytics API reference](/docs/api-reference/workflow-runtime/world/analytics) for per-method parameters, row shapes, and `pageInfo` semantics. +## Process-wide state + +Hold state that must be process-wide on `globalThis`, not at module scope. + +A World is loaded in one of two ways, and only one of them gives your package a +single module instance: + +- **Loaded at runtime.** `WORKFLOW_TARGET_WORLD=@your-org/world-foo` is resolved + with `require()` at runtime, so Node's module cache dedupes it and one process + holds one copy. +- **Bundled.** The host application's bundler compiles your package into its + server build. Bundlers key module identity on `(resource, layer)`, and a + framework routinely builds several server layers. Next.js compiles + `instrument`, app-route, `ssr` and `edge` as separate module graphs. Your + package is then compiled into each one, so a single process holds several + copies of every one of your modules, each with its own module scope. + +The two built-in worlds are bundled. A custom world is not today, but that is a +property of how it is loaded rather than of how it is written, and it can change +under you. `@workflow/world-vercel` was external until it wasn't, and every +module-scope variable in it silently became per-copy state. + +So a top-level `let` or a `const` holding a `Map` is not the singleton it looks +like: + +```typescript +// Wrong: one Map per copy. Writes from one part of the app are invisible to +// another, and a mutex like this simply stops mutually excluding. +const locks = new Map>(); +``` + +Reach for `globalThis` under a `Symbol.for()` key instead, so every copy shares +one object: + +```typescript +type WorldState = { locks: Map> }; + +const StateKey = Symbol.for('@your-org/world-foo//locks/v1'); +const store = globalThis as typeof globalThis & + Record; + +const state: WorldState = (store[StateKey] ??= { locks: new Map() }); +``` + +Version the key. Two releases of your package can end up in one process, and a +key without a version lets an older copy read a state object it does not +understand. + +Inside this repository, `globalSingleton()` from `@workflow/utils` does exactly +this and is what the first-party worlds use; the hand-rolled form above is +written out so a world published outside this repository does not need the +dependency. `scripts/lint/module-scope-state.mjs` accepts either. + +Better still, keep the state on the World instance your `createWorld()` returns. +Connection pools, caches, and open channels are usually per-World rather than +per-process, and instance state cannot be duplicated by a bundler. Reserve the +global for the few things that are genuinely process-wide: ID generators whose +sequence must not fork, and log-once latches. + ## Reference implementations Study these implementations for guidance: diff --git a/packages/ai/package.json b/packages/ai/package.json index 5ce72283c4..3837c100be 100644 --- a/packages/ai/package.json +++ b/packages/ai/package.json @@ -85,6 +85,7 @@ "dependencies": { "@ai-sdk/provider": "^3.0.0", "@workflow/serde": "workspace:^", + "@workflow/utils": "workspace:*", "zod": "catalog:" }, "optionalDependencies": { diff --git a/packages/ai/src/agent/telemetry.ts b/packages/ai/src/agent/telemetry.ts index 5e4b4d1cd7..a725951ac2 100644 --- a/packages/ai/src/agent/telemetry.ts +++ b/packages/ai/src/agent/telemetry.ts @@ -1,3 +1,4 @@ +import { globalSingleton } from '@workflow/utils'; import type { TelemetrySettings } from './durable-agent.js'; // Minimal OTel type shims so we don't depend on @opentelemetry/api at compile time. @@ -38,24 +39,30 @@ interface OtelApi { SpanStatusCode: { ERROR: number }; } -// Lazy-loaded OTel API: self-initializes on first use (item 5) -let otelApi: OtelApi | null = null; -let otelLoadAttempted = false; +// Lazy-loaded OTel API: self-initializes on first use (item 5). +// +// On `globalThis` rather than at module scope because this package is bundled +// into the host application's server build, which gives one copy of this module +// per bundler layer; per-copy state would re-attempt the import once per layer. +const otel = globalSingleton('@workflow/ai//agentTelemetry', 1, () => ({ + api: null as OtelApi | null, + loadAttempted: false, +})); async function ensureOtelApi(): Promise { - if (otelLoadAttempted) return otelApi; - otelLoadAttempted = true; + if (otel.loadAttempted) return otel.api; + otel.loadAttempted = true; try { // Dynamic import, since @opentelemetry/api is an optional peer dependency. // Use Function() to hide the import from bundlers that would fail at // compile time when the package is absent. - otelApi = await (Function( + otel.api = await (Function( 'return import("@opentelemetry/api")' )() as Promise); } catch { - otelApi = null; + otel.api = null; } - return otelApi; + return otel.api; } /** @@ -64,9 +71,9 @@ async function ensureOtelApi(): Promise { * don't need a separate init step. */ function getTracer(telemetry?: TelemetrySettings): Tracer | null { - if (!telemetry?.isEnabled || !otelApi) return null; + if (!telemetry?.isEnabled || !otel.api) return null; if (telemetry.tracer) return telemetry.tracer as Tracer; - return otelApi.trace.getTracer('ai'); + return otel.api.trace.getTracer('ai'); } // ── Attribute helpers ────────────────────────────────────────────────── @@ -130,11 +137,11 @@ function recordErrorOnSpan(span: Span, error: unknown): void { stack: error.stack, }); span.setStatus({ - code: otelApi?.SpanStatusCode.ERROR ?? 2, + code: otel.api?.SpanStatusCode.ERROR ?? 2, message: error.message, }); } else { - span.setStatus({ code: otelApi?.SpanStatusCode.ERROR ?? 2 }); + span.setStatus({ code: otel.api?.SpanStatusCode.ERROR ?? 2 }); } } @@ -172,12 +179,12 @@ export async function recordSpan(options: { fn: (span?: Span) => PromiseLike | T; }): Promise { // Self-initialize on first call (item 5) - if (!otelLoadAttempted) { + if (!otel.loadAttempted) { await ensureOtelApi(); } const tracer = getTracer(options.telemetry); - if (!tracer || !otelApi) { + if (!tracer || !otel.api) { return options.fn(undefined); } @@ -192,11 +199,13 @@ export async function recordSpan(options: { { attributes: attrs }, async (span) => { // Capture current context so nested spans parent correctly (item 4). - // otelApi is guaranteed non-null here (checked before startActiveSpan). - const ctx = otelApi!.context.active(); + // otel.api is guaranteed non-null here (checked before startActiveSpan). + const ctx = otel.api!.context.active(); try { - const result = await otelApi!.context.with(ctx, () => options.fn(span)); + const result = await otel.api!.context.with(ctx, () => + options.fn(span) + ); span.end(); return result; } catch (error) { @@ -228,12 +237,12 @@ export async function createSpan(options: { telemetry?: TelemetrySettings; attributes?: Attributes; }): Promise { - if (!otelLoadAttempted) { + if (!otel.loadAttempted) { await ensureOtelApi(); } const tracer = getTracer(options.telemetry); - if (!tracer || !otelApi) return undefined; + if (!tracer || !otel.api) return undefined; const attrs = buildAttributes( options.name, @@ -243,9 +252,9 @@ export async function createSpan(options: { // Capture the active context so the span parents under the caller's // current span, matching how recordSpan uses context.with(). - const parentCtx = otelApi.context.active(); + const parentCtx = otel.api.context.active(); const span = tracer.startSpan(options.name, { attributes: attrs }, parentCtx); - const context = otelApi.trace.setSpan(parentCtx, span); + const context = otel.api.trace.setSpan(parentCtx, span); return { span, context }; } @@ -263,8 +272,8 @@ export function runInContext( handle: SpanHandle | undefined, fn: () => T ): T { - if (!handle || !otelApi) return fn(); - return otelApi.context.with(handle.context, fn); + if (!handle || !otel.api) return fn(); + return otel.api.context.with(handle.context, fn); } /** diff --git a/packages/ai/src/module-scope-state.test.ts b/packages/ai/src/module-scope-state.test.ts new file mode 100644 index 0000000000..8cb839d79b --- /dev/null +++ b/packages/ai/src/module-scope-state.test.ts @@ -0,0 +1,26 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests. Repeated here so the signal arrives when you run just this + * package's tests. + * + * This package is compiled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/ai', () => { + const findings = scanPackage(path.join(repoRoot, 'packages/ai'), repoRoot); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/core/src/module-scope-state.test.ts b/packages/core/src/module-scope-state.test.ts new file mode 100644 index 0000000000..b2a74b3c37 --- /dev/null +++ b/packages/core/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests. Repeated here so the signal arrives when you run just this + * package's tests. + * + * This package is compiled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/core', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/core'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 2cd9b28bc8..dc6b01457e 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -703,6 +703,9 @@ export function registerDeliveryBarrier( * so the context interface (constructed literally by many test harnesses) * needs no new field; entries drop with the context. */ +// per-copy-ok: keyed on the orchestrator context object, and a context is armed +// and observed by the same copy that created it. Entries can never be looked up +// from another copy because that copy has no reference to the key. const activeBarrierSafetyNets = new WeakSet(); /** diff --git a/packages/core/src/runtime-world-singleton.test.ts b/packages/core/src/runtime-world-singleton.test.ts new file mode 100644 index 0000000000..d789bafbbf --- /dev/null +++ b/packages/core/src/runtime-world-singleton.test.ts @@ -0,0 +1,40 @@ +import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { getWorld, setWorld } from './runtime/world.js'; +import { workflowEntrypoint } from './runtime.js'; + +const createLocalWorld = vi.hoisted(() => vi.fn()); + +vi.mock('@workflow/world-local', () => ({ + createWorld: createLocalWorld, +})); + +describe('workflowEntrypoint world initialization', () => { + const world = { + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn( + () => async () => new Response(null, { status: 204 }) + ), + } as unknown as World; + + beforeEach(() => { + setWorld(undefined); + createLocalWorld.mockReset(); + createLocalWorld.mockResolvedValue(world); + }); + + afterEach(() => { + setWorld(undefined); + vi.clearAllMocks(); + }); + + it('reuses the runtime World after initializing the route handler', async () => { + const handler = workflowEntrypoint(''); + + const response = await handler(new Request('https://example.test')); + + expect(response.status).toBe(204); + await expect(getWorld()).resolves.toBe(world); + expect(createLocalWorld).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 5962cc4473..cada587c47 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -114,11 +114,7 @@ import { runStepSingleFlight } from './runtime/step-single-flight.js'; import { handleSuspension } from './runtime/suspension-handler.js'; import { useQuickJSVm } from './runtime/vm-mode.js'; import { getWaitContinuationDispatch } from './runtime/wait-continuation.js'; -import { - getWorld, - getWorldHandlers, - type WorldHandlers, -} from './runtime/world.js'; +import { getWorld, type WorldHandlers } from './runtime/world.js'; import { dehydrateRunError } from './serialization.js'; import { remapErrorStack } from './source-map.js'; import * as Attribute from './telemetry/semantic-conventions.js'; @@ -4683,9 +4679,19 @@ export function workflowEntrypoint( async (span) => { if (!cachedHandler) { cachedHandler = await trace('workflow.route.init', async () => { + // The full runtime World, not `getWorldHandlers()`. That accessor + // owns a second, build-time-safe cache, so calling it here built a + // second World in the same process: duplicate connection pools and + // queue workers for a stateful World, plus a second copy of that + // world package's modules once it is bundled, which is what + // silently demoted the events WebSocket transport to HTTP. #3665. + // + // The span keeps its original name. It is a distinct span from the + // per-request `workflow.route.get_world` at the top of the flow + // route, and renaming it would collide with that one. const worldHandlers = await trace( 'workflow.route.get_world_handlers', - async () => getWorldHandlers() + async () => getWorld() ); return handler(worldHandlers); }); diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 2f7bb669d0..4f09975b21 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -1,3 +1,4 @@ +import { globalSingleton } from '@workflow/utils'; import { envNumber } from '@workflow/world'; import { runtimeLogger } from '../logger.js'; @@ -73,15 +74,23 @@ export const MAX_REPLAY_TIMEOUT_MS = 780_000; // Track which raw env var values we've already warned about so the warning // log only fires once per process (the function may be called many times). -const warnedReplayTimeoutValues = new Set(); +// +// On `globalThis` rather than at module scope so "once per process" survives +// bundling: this package is compiled into the host application's server build +// once per bundler layer, and per-copy sets warn once per layer instead. +const warned = globalSingleton('@workflow/core//envWarnings', 1, () => ({ + replayTimeoutValues: new Set(), + maxInlineStepsValues: new Set(), + maxEventsValues: new Set(), +})); function warnOnce( raw: string, message: string, data: Record ): void { - if (warnedReplayTimeoutValues.has(raw)) return; - warnedReplayTimeoutValues.add(raw); + if (warned.replayTimeoutValues.has(raw)) return; + warned.replayTimeoutValues.add(raw); runtimeLogger.warn(message, data); } @@ -132,7 +141,7 @@ export function getReplayTimeoutMs(): number { * @internal */ export function _resetReplayTimeoutWarnCacheForTests(): void { - warnedReplayTimeoutValues.clear(); + warned.replayTimeoutValues.clear(); } // Number of queue delivery attempts to allow before permanently failing a run @@ -175,9 +184,6 @@ export const MIN_MAX_INLINE_STEPS = 1; */ export const MAX_MAX_INLINE_STEPS = 16; -// Warn-once cache for WORKFLOW_MAX_INLINE_STEPS, keyed by raw env value. -const warnedMaxInlineStepsValues = new Set(); - /** * Resolve the effective max number of inline steps for the current process. * @@ -191,8 +197,8 @@ export function getMaxInlineSteps(): number { if (!raw) return MAX_INLINE_STEPS; const parsed = Number(raw); if (!Number.isInteger(parsed) || parsed <= 0) { - if (!warnedMaxInlineStepsValues.has(raw)) { - warnedMaxInlineStepsValues.add(raw); + if (!warned.maxInlineStepsValues.has(raw)) { + warned.maxInlineStepsValues.add(raw); runtimeLogger.warn( 'Ignoring WORKFLOW_MAX_INLINE_STEPS: not a positive integer; using default', { raw, defaultValue: MAX_INLINE_STEPS } @@ -202,8 +208,8 @@ export function getMaxInlineSteps(): number { } if (parsed < MIN_MAX_INLINE_STEPS) return MIN_MAX_INLINE_STEPS; if (parsed > MAX_MAX_INLINE_STEPS) { - if (!warnedMaxInlineStepsValues.has(raw)) { - warnedMaxInlineStepsValues.add(raw); + if (!warned.maxInlineStepsValues.has(raw)) { + warned.maxInlineStepsValues.add(raw); runtimeLogger.warn('WORKFLOW_MAX_INLINE_STEPS above maximum; clamped', { raw, clampedValue: MAX_MAX_INLINE_STEPS, @@ -287,8 +293,6 @@ export function isBatchTransitionsEnabled(): boolean { */ export const MAX_BATCH_FANOUT_EVENTS = 32; -const warnedMaxEventsValues = new Set(); - /** * Optional client-side override for the server-supplied per-run event ceiling. * When set to a positive integer, the runtime clamps the server's limit *down* @@ -304,8 +308,8 @@ export function getMaxEventsOverride(): number | undefined { if (!raw) return undefined; const parsed = Number(raw); if (!Number.isInteger(parsed) || parsed <= 0) { - if (!warnedMaxEventsValues.has(raw)) { - warnedMaxEventsValues.add(raw); + if (!warned.maxEventsValues.has(raw)) { + warned.maxEventsValues.add(raw); runtimeLogger.warn( 'Ignoring WORKFLOW_MAX_EVENTS_OVERRIDE: not a positive integer; using server limit', { raw } diff --git a/packages/core/src/runtime/get-port-lazy.ts b/packages/core/src/runtime/get-port-lazy.ts index 8b337b1fcc..c6b779ebaf 100644 --- a/packages/core/src/runtime/get-port-lazy.ts +++ b/packages/core/src/runtime/get-port-lazy.ts @@ -9,15 +9,24 @@ import { createRequire } from 'node:module'; import { pathToFileURL } from 'node:url'; +import { globalSingleton } from '@workflow/utils'; -let _getPort: (() => Promise) | undefined; +// All three live on `globalThis` (see `globalSingleton`) rather than at module +// scope. The resolved port is a fact about the process, not about one copy of +// this module, and per-copy caches would each pay the ~60ms `lsof` discovery +// and could each pin their own answer. +const portState = globalSingleton('@workflow/core//devServerPort', 1, () => ({ + getPort: undefined as (() => Promise) | undefined, + cachedPort: undefined as number | undefined, + inFlight: undefined as Promise | undefined, +})); // Per-process cache of the resolved port. The workflow server listens on a // stable port for the lifetime of the process, but `getPort()` rediscovers it // on every call by querying the OS for the process's listening sockets. On // macOS that shells out to `lsof` (~60ms), which the runtime pays on EVERY // workflow replay or step invocation. Since the port does not change within a -// process, resolve it once and reuse it. `_inFlight` +// process, resolve it once and reuse it. `portState.inFlight` // dedupes concurrent first calls so discovery never runs more than once. // // The first concrete port is pinned for the lifetime of the process: there is @@ -25,20 +34,18 @@ let _getPort: (() => Promise) | undefined; // the already-listening dev-server process, and `getPort()` -> `getAllPorts()` // returns a deterministic order, so repeated calls would resolve the same port // anyway. -let _cachedPort: number | undefined; -let _inFlight: Promise | undefined; export async function getPortLazy(): Promise { // Fast path: already resolved a concrete port for this process. - if (_cachedPort !== undefined) { - return _cachedPort; + if (portState.cachedPort !== undefined) { + return portState.cachedPort; } // A discovery is already running, so share it rather than starting a second. - if (_inFlight) { - return _inFlight; + if (portState.inFlight) { + return portState.inFlight; } - if (!_getPort) { + if (!portState.getPort) { try { // Construct specifier at runtime to defeat bundler static analysis. const spec = ['@workflow/utils', 'get-port'].join('/'); @@ -48,43 +55,43 @@ export async function getPortLazy(): Promise { pathToFileURL(process.cwd() + '/package.json').href ); const mod = _require(spec); - _getPort = mod.getPort; + portState.getPort = mod.getPort; } catch { // Module not available (e.g., in a browser or minimal bundle) - _getPort = async () => undefined; + portState.getPort = async () => undefined; } } - // `_getPort` is always assigned by the block above; the fallback keeps the + // `portState.getPort` is always assigned by the block above; the fallback keeps the // type non-nullable without a non-null assertion. - const resolver = _getPort ?? (async () => undefined); - _inFlight = resolver() + const resolver = portState.getPort ?? (async () => undefined); + portState.inFlight = resolver() .then((port) => { // Only cache a concrete port. A transient `undefined` (e.g. the server is // not listening yet on the first replay) must not poison the cache: // leaving it unset lets the next call retry discovery. if (typeof port === 'number') { - _cachedPort = port; + portState.cachedPort = port; } return port; }) .finally(() => { - _inFlight = undefined; + portState.inFlight = undefined; }); - return _inFlight; + return portState.inFlight; } /** * Resets the per-process port cache. Intended for tests; not used on the hot * path. Callers must let any in-flight lookup settle (await the pending - * `getPortLazy()` call) before resetting: clearing `_inFlight` here does not + * `getPortLazy()` call) before resetting: clearing `portState.inFlight` here does not * cancel an already-scheduled resolution, so a late `.then` could otherwise - * repopulate `_cachedPort` after the reset and bleed into the next test. + * repopulate `portState.cachedPort` after the reset and bleed into the next test. */ export function resetPortCacheForTesting(): void { - _getPort = undefined; - _cachedPort = undefined; - _inFlight = undefined; + portState.getPort = undefined; + portState.cachedPort = undefined; + portState.inFlight = undefined; } /** @@ -96,7 +103,7 @@ export function resetPortCacheForTesting(): void { export function setPortResolverForTesting( fn: () => Promise ): void { - _getPort = fn; - _cachedPort = undefined; - _inFlight = undefined; + portState.getPort = fn; + portState.cachedPort = undefined; + portState.inFlight = undefined; } diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index 18e04a93f5..d81f8d00a9 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -30,6 +30,7 @@ */ import { SerializationError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import { type Event, isSealedNoopEvent, @@ -1048,16 +1049,25 @@ type CompiledExtension = Omit & { * only needs to happen once per process. The promise is cached (not the * result) so concurrent first invocations share a single compilation. */ -let compiledAssetsPromise: - | Promise<{ - wasm: object; - extensions: CompiledExtension[]; - }> - | undefined; +// On `globalThis` (see `globalSingleton`): the comment above says once per +// process, and module scope would make it once per bundler layer, recompiling +// the ~600 KB runtime binary for each. +const quickjsAssets = globalSingleton( + '@workflow/core//quickjsCompiledAssets', + 1, + () => ({ + promise: undefined as + | Promise<{ + wasm: object; + extensions: CompiledExtension[]; + }> + | undefined, + }) +); function getCompiledAssets() { - if (!compiledAssetsPromise) { - compiledAssetsPromise = (async () => { + if (!quickjsAssets.promise) { + quickjsAssets.promise = (async () => { const [wasm, ...extensionModules] = await Promise.all([ WebAssemblyGlobal.compile(quickjsWasm), ...quickjsExtensions.map((ext) => @@ -1074,11 +1084,11 @@ function getCompiledAssets() { })(); // On failure, clear the cache so a later invocation can retry rather // than being stuck with a rejected promise forever. - compiledAssetsPromise.catch(() => { - compiledAssetsPromise = undefined; + quickjsAssets.promise.catch(() => { + quickjsAssets.promise = undefined; }); } - return compiledAssetsPromise; + return quickjsAssets.promise; } /** @@ -1177,19 +1187,28 @@ type BaselineEntry = } | { state: 'ineligible'; reason: string }; -const baselineCache = new Map>(); +// On `globalThis` (see `globalSingleton`): a snapshot is expensive to build and +// is keyed by bundle, so per-copy caches would build the same baseline once per +// bundler layer while each enforcing its own bound. +const baselines = globalSingleton( + '@workflow/core//quickjsBaselines', + 1, + () => ({ + byKey: new Map>(), + }) +); const BASELINE_CACHE_MAX_ENTRIES = 4; /** Test-only: reset the baseline cache between test cases. */ export function __clearBaselineSnapshotCacheForTests(): void { - baselineCache.clear(); + baselines.byKey.clear(); } /** Test-only: observe how a bundle was classified. */ export async function __peekBaselineEntryForTests( workflowCode: string ): Promise { - return baselineCache.get(workflowCode); + return baselines.byKey.get(workflowCode); } /** @@ -1313,15 +1332,15 @@ function getBaselineEntry( workflowCode: string, workflowId: string ): Promise { - let entry = baselineCache.get(workflowCode); + let entry = baselines.byKey.get(workflowCode); if (!entry) { - if (baselineCache.size >= BASELINE_CACHE_MAX_ENTRIES) { - const oldest = baselineCache.keys().next().value; - if (oldest !== undefined) baselineCache.delete(oldest); + if (baselines.byKey.size >= BASELINE_CACHE_MAX_ENTRIES) { + const oldest = baselines.byKey.keys().next().value; + if (oldest !== undefined) baselines.byKey.delete(oldest); } entry = prepareBaselineSnapshot(workflowCode, workflowId); - baselineCache.set(workflowCode, entry); - entry.catch(() => baselineCache.delete(workflowCode)); + baselines.byKey.set(workflowCode, entry); + entry.catch(() => baselines.byKey.delete(workflowCode)); } return entry; } @@ -2453,6 +2472,9 @@ function markCreated(vm: QuickJS, cidJs: string, opType?: string): void { * its bytes are computed once even though the op is re-collected on every * suspension it stays pending through. */ +// per-copy-ok: keyed on the VM instance, and a VM is created and driven by one +// copy. Another copy holds no reference to the key, so a shared map could never +// be read from it. const pendingByteCache = new WeakMap>(); function ensurePendingByteCache(vm: QuickJS): Map { diff --git a/packages/core/src/runtime/start.ts b/packages/core/src/runtime/start.ts index 693ca6adfa..fab61ad943 100644 --- a/packages/core/src/runtime/start.ts +++ b/packages/core/src/runtime/start.ts @@ -1,4 +1,5 @@ import { EntityConflictError, WorkflowRuntimeError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import { workflowDisplayName } from '@workflow/utils/parse-name'; import type { WorkflowInvokePayload, World } from '@workflow/world'; import { @@ -80,7 +81,13 @@ function resolveLineageAttributes(): Record | undefined { // The warning that explains this only needs to fire once per process: a // workflow that hardcodes 'latest' for its Vercel deployment would otherwise // log it on every local/Postgres run, flooding tight dev loops. -let hasWarnedLatestNoOp = false; +// On `globalThis` (see `globalSingleton`) so "once per process" is not once +// per bundler layer. +const latestNoOpWarning = globalSingleton( + '@workflow/core//latestNoOpWarning', + 1, + () => ({ warned: false }) +); /** * Reset the `deploymentId: 'latest'` no-op warn-once guard. Test-only, @@ -89,7 +96,7 @@ let hasWarnedLatestNoOp = false; * @internal */ export function _resetLatestNoOpWarnForTests(): void { - hasWarnedLatestNoOp = false; + latestNoOpWarning.warned = false; } export interface StartOptionsBase { @@ -300,9 +307,9 @@ export async function start( if (world.resolveLatestDeploymentId) { deploymentId = await world.resolveLatestDeploymentId(); } else { - // Warn once per process; see hasWarnedLatestNoOp above. - if (!hasWarnedLatestNoOp) { - hasWarnedLatestNoOp = true; + // Warn once per process; see latestNoOpWarning.warned above. + if (!latestNoOpWarning.warned) { + latestNoOpWarning.warned = true; runtimeLogger.warn( "deploymentId: 'latest' has no effect in this world and was ignored. " + 'It is only supported by worlds with atomic deployments, such as Vercel. ' + diff --git a/packages/core/src/runtime/step-single-flight.ts b/packages/core/src/runtime/step-single-flight.ts index e96582b2bf..b886d4627c 100644 --- a/packages/core/src/runtime/step-single-flight.ts +++ b/packages/core/src/runtime/step-single-flight.ts @@ -1,3 +1,4 @@ +import { globalSingleton } from '@workflow/utils'; import { runtimeLogger } from '../logger.js'; import type { StepExecutionResult } from './step-executor.js'; @@ -29,7 +30,16 @@ import type { StepExecutionResult } from './step-executor.js'; * multi-instance self-hosted worlds (mitigate by raising * `WORKFLOW_INLINE_OWNERSHIP_LEASE_SECONDS`). */ -const inFlightSteps = new Map>(); +// On `globalThis` (see `globalSingleton`), not module scope: a per-copy map is +// not single-flight. Two invocations reaching this module through different +// bundler layers would each believe they were the only one in the process and +// both run the step body, degrading in-process dedup to the cross-process +// residual the doc above scopes out. +const singleFlight = globalSingleton( + '@workflow/core//stepSingleFlight', + 1, + () => ({ inFlight: new Map>() }) +); /** * Run `execute` unless an execution for the same run + step correlation ID is @@ -46,7 +56,7 @@ export async function runStepSingleFlight( execute: () => Promise ): Promise { const key = `${runId}:${correlationId}`; - const existing = inFlightSteps.get(key); + const existing = singleFlight.inFlight.get(key); if (existing) { // warn (always printed, unlike debug/info): the single-flight is // absorbing what would have been a duplicate execution, typically a @@ -68,10 +78,10 @@ export async function runStepSingleFlight( } const promise = execute(); - inFlightSteps.set(key, promise); + singleFlight.inFlight.set(key, promise); try { return await promise; } finally { - inFlightSteps.delete(key); + singleFlight.inFlight.delete(key); } } diff --git a/packages/core/src/runtime/world.ts b/packages/core/src/runtime/world.ts index a39943a12f..ca4ae08629 100644 --- a/packages/core/src/runtime/world.ts +++ b/packages/core/src/runtime/world.ts @@ -5,6 +5,16 @@ import { resolveWorkflowTargetWorld, } from '@workflow/utils'; import type { World } from '@workflow/world'; +// Static imports, so these two are compiled into the host application's server +// build. A bundler keys module identity on (resource, layer) and Next.js alone +// builds `instrument`, app-route, `ssr` and `edge` layers, so one process holds +// one copy of each of their modules *per layer*. Custom worlds below load +// through `getRuntimeRequire()` and are deduped by Node's module cache instead. +// +// Neither package may therefore keep mutable state at module scope; both hold +// it on `globalThis` via `globalSingleton()` from `@workflow/utils`, enforced +// by `scripts/lint/module-scope-state.mjs`. See that helper's doc comment, and +// `docs/content/worlds/*/building-a-world.mdx` for the rule world authors get. import { createWorld as createLocalWorld } from '@workflow/world-local'; import { createWorld as createVercelWorld } from '@workflow/world-vercel'; import { assertWorldSupportsRuntimeProtocol } from './world-compatibility.js'; diff --git a/packages/core/src/serialization-format.ts b/packages/core/src/serialization-format.ts index 887efa740d..5da401b4f0 100644 --- a/packages/core/src/serialization-format.ts +++ b/packages/core/src/serialization-format.ts @@ -6,6 +6,7 @@ * o11y, CLI o11y). It has NO Node.js dependencies. */ +import { globalSingleton } from '@workflow/utils'; import { getEventDataRefFields } from '@workflow/world'; import { parse, unflatten } from 'devalue'; @@ -280,9 +281,14 @@ function decompressSyncIfAvailable( * Web `DecompressionStream` has no zstd support. Node decodes via `node:zlib` * and never needs this. See `registerZstdDecoder`. */ -let zstdBrowserDecoder: - | ((payload: Uint8Array) => Promise) - | undefined; +// On `globalThis` (see `globalSingleton`): the o11y host registers the decoder +// once, and a per-copy slot would leave every other copy of this module without +// one. +const zstd = globalSingleton('@workflow/core//zstd.decoder', 1, () => ({ + decoder: undefined as + | ((payload: Uint8Array) => Promise) + | undefined, +})); /** * Register a browser zstd decoder (e.g. a WASM-backed one). The web o11y UI @@ -292,7 +298,7 @@ let zstdBrowserDecoder: export function registerZstdDecoder( decoder: (payload: Uint8Array) => Promise ): void { - zstdBrowserDecoder = decoder; + zstd.decoder = decoder; } /** @@ -307,7 +313,7 @@ async function decompressAsync( if (format === SerializationFormat.ZSTD) { const sync = decompressSyncIfAvailable(format, payload); if (sync) return sync; - if (zstdBrowserDecoder) return zstdBrowserDecoder(payload); + if (zstd.decoder) return zstd.decoder(payload); throw new Error( 'zstd-compressed workflow data encountered but no zstd decoder is ' + 'available. Node.js 22.15+ decodes natively; in the browser register ' + diff --git a/packages/core/src/serialization/hardened.ts b/packages/core/src/serialization/hardened.ts index da3b18612f..18c04a0316 100644 --- a/packages/core/src/serialization/hardened.ts +++ b/packages/core/src/serialization/hardened.ts @@ -44,6 +44,7 @@ */ import { types } from 'node:util'; +import { globalSingleton } from '@workflow/utils'; import type { StringifyOperations } from 'devalue'; import { defaultStringifyOperations } from 'devalue'; @@ -83,7 +84,12 @@ export interface GuestCodeStats { executions: GuestCodeExecution[]; } +// per-copy-ok: both are set and cleared by `withGuestCodeStats` around a single +// synchronous call, so the sink is only ever read by the same copy that armed +// it. A shared slot would let two copies recording concurrently clobber each +// other's sink. let activeStats: GuestCodeStats | null = null; +// per-copy-ok: same scope as `activeStats` above, armed and cleared together. let reportedProxies: WeakSet | null = null; /** @@ -121,17 +127,24 @@ export function withGuestCodeStats( * report entry, never incorrect output; closing it means branding at the * compiler, which does not belong here. */ -const useStepClosureFns = new WeakSet(); +// On `globalThis` (see `globalSingleton`): functions cross module copies freely, +// so a closure marked by one copy would not be recognized by another, costing a +// report entry for no reason. +const useStepClosures = globalSingleton( + '@workflow/core//useStepClosures.fns', + 1, + () => ({ fns: new WeakSet() }) +); /** Marks a function as having been passed to `useStep`. */ export function markUseStepClosureFn(fn: T): T { - useStepClosureFns.add(fn); + useStepClosures.fns.add(fn); return fn; } /** Whether `fn` was marked by {@link markUseStepClosureFn}. */ export function isUseStepClosureFn(fn: unknown): boolean { - return typeof fn === 'function' && useStepClosureFns.has(fn as object); + return typeof fn === 'function' && useStepClosures.fns.has(fn as object); } export function recordGuestCode( diff --git a/packages/core/src/serialization/workflow-vm.ts b/packages/core/src/serialization/workflow-vm.ts index 9d92d6d36a..0b7700ae00 100644 --- a/packages/core/src/serialization/workflow-vm.ts +++ b/packages/core/src/serialization/workflow-vm.ts @@ -13,19 +13,24 @@ * format-prefixed devalue data ("devl" + devalue.stringify output). */ +import { globalSingleton } from '@workflow/utils'; import { devalueVmCodec } from './codec-devalue-vm.js'; import { isFormatPrefix, SerializationFormat } from './types.js'; const FORMAT_PREFIX_LENGTH = 4; -let _encoder: { encode(s: string): Uint8Array }; -let _decoder: { decode(d: Uint8Array): string }; -function getEncoder() { - if (!_encoder) _encoder = new (globalThis as any).TextEncoder(); - return _encoder; +// On `globalThis` (see `globalSingleton`) so one process builds one pair, +// rather than one per bundler layer this module is compiled into. +const codecs = globalSingleton('@workflow/core//vmTextCodecs', 1, () => ({ + encoder: undefined as { encode(s: string): Uint8Array } | undefined, + decoder: undefined as { decode(d: Uint8Array): string } | undefined, +})); +function getEncoder(): { encode(s: string): Uint8Array } { + codecs.encoder ??= new (globalThis as any).TextEncoder(); + return codecs.encoder as { encode(s: string): Uint8Array }; } -function getDecoder() { - if (!_decoder) _decoder = new (globalThis as any).TextDecoder(); - return _decoder; +function getDecoder(): { decode(d: Uint8Array): string } { + codecs.decoder ??= new (globalThis as any).TextDecoder(); + return codecs.decoder as { decode(d: Uint8Array): string }; } /** diff --git a/packages/core/src/source-map.ts b/packages/core/src/source-map.ts index 83dec355c8..b5bf4085aa 100644 --- a/packages/core/src/source-map.ts +++ b/packages/core/src/source-map.ts @@ -1,4 +1,5 @@ import { originalPositionFor, TraceMap } from '@jridgewell/trace-mapping'; +import { globalSingleton } from '@workflow/utils'; /** Marker prefix of an inline source map comment emitted by bundlers. */ const INLINE_SOURCE_MAP_MARKER = @@ -110,15 +111,20 @@ function extractInlineSourceMapBase64(source: string): string | undefined { * string per edit; the bound keeps the few most-recent ones and evicts the * rest instead of pinning every historical version. */ -const tracerCache = new Map(); +// On `globalThis` (see `globalSingleton`): the cache exists to avoid re-parsing +// one build-time bundle for the life of the process, which per-copy state would +// do once per bundler layer. +const tracers = globalSingleton('@workflow/core//sourceMapTracers', 1, () => ({ + byCode: new Map(), +})); const MAX_TRACERS = 8; function getTraceMapForCode(workflowCode: string): TraceMap | null { - const cached = tracerCache.get(workflowCode); + const cached = tracers.byCode.get(workflowCode); if (cached !== undefined) { // Move to most-recently-used position (end of insertion order). - tracerCache.delete(workflowCode); - tracerCache.set(workflowCode, cached); + tracers.byCode.delete(workflowCode); + tracers.byCode.set(workflowCode, cached); return cached; } @@ -136,13 +142,13 @@ function getTraceMapForCode(workflowCode: string): TraceMap | null { } } - tracerCache.set(workflowCode, tracer); + tracers.byCode.set(workflowCode, tracer); // Evict the least-recently-used entries when over the cap. New entries are // appended at the end, so the oldest live at the front. - while (tracerCache.size > MAX_TRACERS) { - const oldest = tracerCache.keys().next().value; + while (tracers.byCode.size > MAX_TRACERS) { + const oldest = tracers.byCode.keys().next().value; if (oldest === undefined) break; - tracerCache.delete(oldest); + tracers.byCode.delete(oldest); } return tracer; } diff --git a/packages/core/src/telemetry.ts b/packages/core/src/telemetry.ts index b2a1102724..e23ef6fc45 100644 --- a/packages/core/src/telemetry.ts +++ b/packages/core/src/telemetry.ts @@ -1,6 +1,6 @@ import type * as api from '@opentelemetry/api'; import type { Span, SpanKind, SpanOptions } from '@opentelemetry/api'; -import { once } from '@workflow/utils'; +import { globalSingleton, once } from '@workflow/utils'; import { WorkflowSuspension } from './global.js'; import { runtimeLogger } from './logger.js'; import * as Attr from './telemetry/semantic-conventions.js'; @@ -23,8 +23,16 @@ import * as Attr from './telemetry/semantic-conventions.js'; */ export type WorkflowTraceMode = 'linked' | 'continuous'; -/** Unrecognized `WORKFLOW_TRACE_MODE` values we already warned about. */ -const warnedUnrecognizedTraceModes = new Set(); +/** + * Unrecognized `WORKFLOW_TRACE_MODE` values we already warned about. On + * `globalThis` (see `globalSingleton`) so the warning stays once per process + * rather than once per bundler layer. + */ +const traceModeWarnings = globalSingleton( + '@workflow/core//traceModeWarnings', + 1, + () => ({ unrecognized: new Set() }) +); /** * Resolves the active trace mode from the `WORKFLOW_TRACE_MODE` env var. @@ -35,8 +43,12 @@ const warnedUnrecognizedTraceModes = new Set(); export function getWorkflowTraceMode(): WorkflowTraceMode { const value = process.env.WORKFLOW_TRACE_MODE; if (value === 'continuous') return 'continuous'; - if (value && value !== 'linked' && !warnedUnrecognizedTraceModes.has(value)) { - warnedUnrecognizedTraceModes.add(value); + if ( + value && + value !== 'linked' && + !traceModeWarnings.unrecognized.has(value) + ) { + traceModeWarnings.unrecognized.add(value); runtimeLogger.warn( `Unrecognized WORKFLOW_TRACE_MODE value "${value}"; expected "linked" or "continuous". Falling back to "linked".` ); @@ -198,6 +210,9 @@ const StepExecutionDurationHistogram = once(async () => { * of `@opentelemetry/api` sees the global registration, so a deployment's * logs show the two packages' views side by side. */ +// per-copy-ok: this diagnostic reports how THIS module instance sees the global +// OTel registration, which is the whole point of the log. With several copies +// in a process, each one's view is what is worth seeing. let otelDiagLogged = false; function logOtelDiagnosticOnce(otel: typeof api, tracer: api.Tracer): void { const debugEnabled = diff --git a/packages/core/src/vm/script-cache.ts b/packages/core/src/vm/script-cache.ts index a0f994da2b..fe69884523 100644 --- a/packages/core/src/vm/script-cache.ts +++ b/packages/core/src/vm/script-cache.ts @@ -1,4 +1,5 @@ import { type Context, Script } from 'node:vm'; +import { globalSingleton } from '@workflow/utils'; /** * Module-level cache of compiled workflow-bundle `vm.Script` objects. @@ -60,7 +61,12 @@ import { type Context, Script } from 'node:vm'; * source files in a bundle and is dropped wholesale when its parent `code` * entry is evicted. */ -const scriptCache = new Map>(); +// On `globalThis` (see `globalSingleton`): compiling a bundle is the expensive +// part this cache exists to skip, and per-copy caches would pay it once per +// bundler layer that compiles a workflow. +const scripts = globalSingleton('@workflow/core//vmScriptCache', 1, () => ({ + byCode: new Map>(), +})); /** * Max number of distinct bundle (`code`) versions to retain. One is enough for @@ -78,13 +84,13 @@ const MAX_BUNDLES = 8; * least-recently-used eviction candidate. */ function touchBundle(code: string): Map | undefined { - const byFilename = scriptCache.get(code); + const byFilename = scripts.byCode.get(code); if (byFilename === undefined) { return undefined; } // Move to the most-recently-used position (end of insertion order). - scriptCache.delete(code); - scriptCache.set(code, byFilename); + scripts.byCode.delete(code); + scripts.byCode.set(code, byFilename); return byFilename; } @@ -105,15 +111,15 @@ export function getCachedWorkflowScript( let byFilename = touchBundle(code); if (byFilename === undefined) { byFilename = new Map(); - scriptCache.set(code, byFilename); + scripts.byCode.set(code, byFilename); // Evict the least-recently-used bundle(s) when over the cap. New bundles // are appended at the end, so the oldest live at the front. - while (scriptCache.size > MAX_BUNDLES) { - const oldest = scriptCache.keys().next().value; + while (scripts.byCode.size > MAX_BUNDLES) { + const oldest = scripts.byCode.keys().next().value; if (oldest === undefined) { break; } - scriptCache.delete(oldest); + scripts.byCode.delete(oldest); } } let script = byFilename.get(filename); @@ -141,7 +147,7 @@ export function runCachedWorkflowScript( * compile-vs-cache behaviour in isolation; not used on the hot path. */ export function clearWorkflowScriptCache(): void { - scriptCache.clear(); + scripts.byCode.clear(); } /** @@ -149,5 +155,5 @@ export function clearWorkflowScriptCache(): void { * tests asserting the LRU bound; not used on the hot path. */ export function workflowScriptCacheSize(): number { - return scriptCache.size; + return scripts.byCode.size; } diff --git a/packages/errors/src/module-scope-state.test.ts b/packages/errors/src/module-scope-state.test.ts new file mode 100644 index 0000000000..918d1747f7 --- /dev/null +++ b/packages/errors/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests. Repeated here so the signal arrives when you run just this + * package's tests. + * + * This package is compiled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/errors', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/errors'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/nest/package.json b/packages/nest/package.json index f69f61b8e1..8047628ea1 100644 --- a/packages/nest/package.json +++ b/packages/nest/package.json @@ -40,6 +40,7 @@ "@swc/core": "catalog:", "@workflow/builders": "workspace:*", "@workflow/swc-plugin": "workspace:*", + "@workflow/utils": "workspace:*", "esbuild": "catalog:", "pathe": "2.0.3" }, diff --git a/packages/nest/src/module-scope-state.test.ts b/packages/nest/src/module-scope-state.test.ts new file mode 100644 index 0000000000..8ca2b544a0 --- /dev/null +++ b/packages/nest/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests. Repeated here so the signal arrives when you run just this + * package's tests. + * + * This package is compiled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/nest', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/nest'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/nest/src/workflow.controller.ts b/packages/nest/src/workflow.controller.ts index a040bace25..ab5bac9a91 100644 --- a/packages/nest/src/workflow.controller.ts +++ b/packages/nest/src/workflow.controller.ts @@ -1,16 +1,27 @@ import { readFileSync } from 'node:fs'; import { pathToFileURL } from 'node:url'; import { All, Controller, Get, Post, Req, Res } from '@nestjs/common'; +import { globalSingleton } from '@workflow/utils'; import { join } from 'pathe'; -// Module-level state for configuration -let configuredOutDir: string | null = null; +// Configuration, set once at bootstrap and read on every request. +// +// On `globalThis` rather than at module scope because a bundler can compile +// this module into the host application's build more than once (see +// `globalSingleton`), and the copy that `configureWorkflowController()` writes +// would then not be the copy the request path reads, leaving the controller +// unconfigured for the life of the process. +const controllerConfig = globalSingleton( + '@workflow/nest//controllerConfig', + 1, + () => ({ outDir: null as string | null }) +); /** * Configure the workflow controller with the output directory */ export function configureWorkflowController(outDir: string): void { - configuredOutDir = outDir; + controllerConfig.outDir = outDir; } /** @@ -72,12 +83,12 @@ async function sendWebResponse( } function getOutDir(): string { - if (!configuredOutDir) { + if (!controllerConfig.outDir) { throw new Error( 'WorkflowController not configured. Call configureWorkflowController first.' ); } - return configuredOutDir; + return controllerConfig.outDir; } /** diff --git a/packages/nest/src/workflow.module.ts b/packages/nest/src/workflow.module.ts index 34d13abb0b..88aa43a22e 100644 --- a/packages/nest/src/workflow.module.ts +++ b/packages/nest/src/workflow.module.ts @@ -4,6 +4,7 @@ import { type OnModuleDestroy, type OnModuleInit, } from '@nestjs/common'; +import { globalSingleton } from '@workflow/utils'; import { join } from 'pathe'; import type { NestBuilderOptions } from './builder.js'; import { @@ -34,8 +35,19 @@ const DEFAULT_OUT_DIR = '.nestjs/workflow'; */ @Module({}) export class WorkflowModule implements OnModuleInit, OnModuleDestroy { - private static options: WorkflowModuleOptions | null = null; - private static outDir: string | null = null; + // On `globalThis` rather than in static fields: a bundler can compile this + // module into the host build more than once (see `globalSingleton`), and + // `forRoot()` would then configure one copy while the module lifecycle hooks + // read another. Static fields are module-scope state with a class for a + // namespace, and duplicate exactly the same way. + private static readonly state = globalSingleton( + '@workflow/nest//moduleConfig', + 1, + () => ({ + options: null as WorkflowModuleOptions | null, + outDir: null as string | null, + }) + ); /** * Configure the WorkflowModule with options. @@ -56,8 +68,8 @@ export class WorkflowModule implements OnModuleInit, OnModuleDestroy { // Configure the controller with the output directory configureWorkflowController(outDir); - WorkflowModule.options = options; - WorkflowModule.outDir = outDir; + WorkflowModule.state.options = options; + WorkflowModule.state.outDir = outDir; return { module: WorkflowModule, @@ -73,7 +85,7 @@ export class WorkflowModule implements OnModuleInit, OnModuleDestroy { } async onModuleInit() { - const options = WorkflowModule.options; + const options = WorkflowModule.state.options; if (!options || options.skipBuild) { return; } @@ -84,13 +96,13 @@ export class WorkflowModule implements OnModuleInit, OnModuleDestroy { ]); const builder = new NestLocalBuilder({ ...options, - outDir: WorkflowModule.outDir ?? undefined, + outDir: WorkflowModule.state.outDir ?? undefined, }); await createBuildQueue()(() => builder.build()); } async onModuleDestroy() { // Cleanup if needed - WorkflowModule.options = null; + WorkflowModule.state.options = null; } } diff --git a/packages/nest/tsconfig.json b/packages/nest/tsconfig.json index 715a46d4b1..fcfdb375b3 100644 --- a/packages/nest/tsconfig.json +++ b/packages/nest/tsconfig.json @@ -9,5 +9,5 @@ "experimentalDecorators": true }, "include": ["src/**/*.ts"], - "exclude": ["node_modules", "dist"] + "exclude": ["node_modules", "dist", "**/*.test.ts"] } diff --git a/packages/utils/package.json b/packages/utils/package.json index f7db6b5ac4..1105eb5061 100644 --- a/packages/utils/package.json +++ b/packages/utils/package.json @@ -3,6 +3,7 @@ "description": "Utility functions for Workflow SDK", "version": "5.0.0-beta.8", "type": "module", + "sideEffects": false, "main": "dist/index.js", "files": [ "dist" diff --git a/packages/utils/src/global-singleton.test.ts b/packages/utils/src/global-singleton.test.ts new file mode 100644 index 0000000000..660a72c604 --- /dev/null +++ b/packages/utils/src/global-singleton.test.ts @@ -0,0 +1,109 @@ +import { afterEach, describe, expect, test } from 'vitest'; +import { + globalSingleton, + resetGlobalSingletonForTest, +} from './global-singleton'; + +const NAME = '@workflow/utils//globalSingletonTest'; + +afterEach(() => { + for (const version of [1, 2]) { + resetGlobalSingletonForTest(NAME, version); + } +}); + +describe('globalSingleton', () => { + test('returns the same object for repeated calls', () => { + const first = globalSingleton(NAME, 1, () => ({ calls: 0 })); + const second = globalSingleton(NAME, 1, () => ({ calls: 0 })); + + expect(second).toBe(first); + }); + + test('runs the factory exactly once', () => { + let factoryRuns = 0; + const create = () => { + factoryRuns++; + return { value: factoryRuns }; + }; + + globalSingleton(NAME, 1, create); + globalSingleton(NAME, 1, create); + globalSingleton(NAME, 1, create); + + expect(factoryRuns).toBe(1); + }); + + test('mutations are visible to every holder', () => { + // The point of the helper: two module copies each call globalSingleton and + // then write through their own reference. A second `const` per copy (the + // bug this replaces) would make these two objects independent. + const copyA = globalSingleton(NAME, 1, () => ({ + transports: new Map(), + })); + const copyB = globalSingleton(NAME, 1, () => ({ + transports: new Map(), + })); + + copyA.transports.set('run_1', 'ws'); + + expect(copyB.transports.get('run_1')).toBe('ws'); + }); + + test('reaches across module instances via globalThis, not module scope', () => { + const created = globalSingleton(NAME, 1, () => ({ marker: 'shared' })); + + // A second copy of a bundled module has its own module scope but the same + // realm, so the only thing it shares is the global. Read it the way that + // copy would: off globalThis, by well-known symbol. + const key = Symbol.for(`${NAME}/v1`); + const fromGlobal = (globalThis as Record)[key]; + + expect(fromGlobal).toBe(created); + }); + + test('different shape versions do not share state', () => { + const v1 = globalSingleton(NAME, 1, () => ({ shape: 'old' })); + const v2 = globalSingleton(NAME, 2, () => ({ shape: 'new' })); + + expect(v2).not.toBe(v1); + expect(v1.shape).toBe('old'); + expect(v2.shape).toBe('new'); + }); + + test('different names do not share state', () => { + const a = globalSingleton(`${NAME}/a`, 1, () => ({ which: 'a' })); + const b = globalSingleton(`${NAME}/b`, 1, () => ({ which: 'b' })); + + expect(b).not.toBe(a); + + resetGlobalSingletonForTest(`${NAME}/a`, 1); + resetGlobalSingletonForTest(`${NAME}/b`, 1); + }); +}); + +describe('resetGlobalSingletonForTest', () => { + test('makes the next call build a fresh object', () => { + const before = globalSingleton(NAME, 1, () => ({ id: 'first' })); + + resetGlobalSingletonForTest(NAME, 1); + const after = globalSingleton(NAME, 1, () => ({ id: 'second' })); + + expect(after).not.toBe(before); + expect(after.id).toBe('second'); + }); + + test('only clears the version it names', () => { + const v1 = globalSingleton(NAME, 1, () => ({ shape: 'old' })); + const v2 = globalSingleton(NAME, 2, () => ({ shape: 'new' })); + + resetGlobalSingletonForTest(NAME, 1); + + expect(globalSingleton(NAME, 1, () => ({ shape: 'rebuilt' }))).not.toBe(v1); + expect(globalSingleton(NAME, 2, () => ({ shape: 'unused' }))).toBe(v2); + }); + + test('is a no-op when nothing was created', () => { + expect(() => resetGlobalSingletonForTest(NAME, 1)).not.toThrow(); + }); +}); diff --git a/packages/utils/src/global-singleton.ts b/packages/utils/src/global-singleton.ts new file mode 100644 index 0000000000..534e8af57e --- /dev/null +++ b/packages/utils/src/global-singleton.ts @@ -0,0 +1,101 @@ +/** + * Process-wide state for packages a bundler may duplicate. + * + * # Why this exists + * + * A module's top-level `const`/`let` is a singleton per *module instance*, not + * per process, and a Next.js server routinely holds several instances of the + * same file. Next compiles its server output into independent module graphs + * (`instrument`, app-route, `ssr`, `edge`), and a bundled module is compiled + * into each one separately, with its own module-scope bindings. Only a package + * left in `serverExternalPackages` collapses to one instance, because that + * emits a runtime `require()` and Node's module cache dedupes it. + * + * `@workflow/core` has always been bundled, hence the `Symbol.for` World cache + * in `runtime/world.ts`. `@workflow/world-vercel` was external until + * vercel/workflow#3493 bundled it, and every module-scope singleton in it + * quietly became one-per-layer. The visible casualty was the WS events + * transport: the queue consumer registered its channel in the route copy's + * registry and the write path looked it up in the instrumentation copy's empty + * one, so every event fell back to HTTP for the life of the process. + * + * The combination that makes this bite rather than merely waste memory is that + * core caches the *World object* on `globalThis` while the module state that + * World closes over stays layer-local. Anything a World reaches at request time + * therefore has to be process-wide too. + * + * # Using it + * + * Hold the state in one object and read through it, rather than reaching for a + * top-level `let`: + * + * ```ts + * const state = globalSingleton('@workflow/world-vercel//wsEventsTransports', 1, () => ({ + * transports: new Map(), + * loggedWsInUse: false, + * })); + * + * state.transports.set(url, transport); + * state.loggedWsInUse = true; + * ``` + * + * A `let` cannot be shared by reference, so log-once latches and lazy caches + * become fields on the state object. That is the whole migration. + * + * # Shape versions + * + * `shapeVersion` is part of the key. Two *different releases* of a package can + * share one process (a transitive dependency pinning an older copy), and they + * would otherwise meet on the same key with different expectations of the + * object. Bump it whenever the state's shape changes incompatibly; an older + * copy then keeps its own state instead of misreading yours. + */ + +/** + * Get the process-wide state for `name`, creating it on first use. + * + * Every copy of the calling module in the process gets the same object back, + * because the object hangs off a `Symbol.for` key on `globalThis` rather than + * off the module. + * + * @param name - Stable identifier, conventionally `//` (e.g. + * `@workflow/world-vercel//httpDispatchers`). It is global to the process, so + * qualify it with the package name. + * @param shapeVersion - Version of the state object's shape. Bump on an + * incompatible change so copies expecting the old shape do not read the new + * one. See "Shape versions" above. + * @param create - Builds the initial state. Runs at most once per process: + * whichever copy asks first wins, so it must not close over anything + * copy-specific. + */ +export function globalSingleton( + name: string, + shapeVersion: number, + create: () => T +): T { + const key = Symbol.for(`${name}/v${shapeVersion}`); + const store = globalThis as typeof globalThis & Record; + const existing = store[key]; + if (existing !== undefined) { + return existing; + } + const created = create(); + store[key] = created; + return created; +} + +/** + * Drop the process-wide state for `name`, so the next {@link globalSingleton} + * call rebuilds it. + * + * @internal A test seam. Production code should reset fields on the state object instead: + * other copies of the module hold a reference to the object this discards, and + * would keep writing to the orphan. + */ +export function resetGlobalSingletonForTest( + name: string, + shapeVersion: number +): void { + const key = Symbol.for(`${name}/v${shapeVersion}`); + delete (globalThis as typeof globalThis & Record)[key]; +} diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index 9748032089..5285f045a9 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -1,3 +1,7 @@ +export { + globalSingleton, + resetGlobalSingletonForTest, +} from './global-singleton.js'; export { formatStepName, formatWorkflowName, diff --git a/packages/utils/src/module-scope-state.test.ts b/packages/utils/src/module-scope-state.test.ts new file mode 100644 index 0000000000..004383eb49 --- /dev/null +++ b/packages/utils/src/module-scope-state.test.ts @@ -0,0 +1,364 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { afterEach, describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** A throwaway package directory holding the given `src/` files. */ +const tempPackages: string[] = []; +function packageWithFiles(files: Record): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'module-scope-state-')); + tempPackages.push(dir); + fs.mkdirSync(path.join(dir, 'src')); + for (const [name, source] of Object.entries(files)) { + fs.writeFileSync(path.join(dir, 'src', name), source); + } + return dir; +} + +/** The common case: one `src/state.ts`. */ +function packageWith(source: string): string { + return packageWithFiles({ 'state.ts': source }); +} + +afterEach(() => { + for (const dir of tempPackages.splice(0)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +/** + * Packages that end up inside the host application's server build, where a + * bundler compiles one copy of every module per layer. + * + * Every published `world-*` is discovered rather than listed, so a new world is + * covered the day it is added; private ones (`@workflow/world-sim`) are out of + * scope because nothing bundles them into an application. The rest are named, + * because "does this package run inside the host's server bundle" is a + * judgement rather than something to infer from the directory name. + * + * Deliberately absent, and why: + * - `next`, `builders`, `sveltekit`: build-time code. The build is one + * process with one module graph. + * - `cli`: its own process. + * - `web`, `web-shared`: the observability UI, not the host's server. + * - `vitest`: the test runner's process. + * + * Adding a package that runs in the host server means adding it here. + */ +const BUNDLED_RUNTIME_PACKAGES = [ + 'core', + 'workflow', + 'world', + 'utils', + 'errors', + 'serde', + 'ai', + 'nest', +]; + +function bundledPackages(): string[] { + const packages = path.join(repoRoot, 'packages'); + const published = (dir: string) => { + const manifest = path.join(dir, 'package.json'); + if (!fs.existsSync(manifest)) return false; + return !JSON.parse(fs.readFileSync(manifest, 'utf8')).private; + }; + const worlds = fs + .readdirSync(packages) + .filter((name) => name.startsWith('world-')) + .map((name) => path.join(packages, name)) + .filter(published); + const named = BUNDLED_RUNTIME_PACKAGES.map((name) => + path.join(packages, name) + ); + return [...worlds, ...named]; +} + +describe('module-scope state rule', () => { + const bundled = bundledPackages(); + + it('finds the packages to check', () => { + // Guards the sweep below against silently checking nothing. + expect(bundled.map((dir) => path.basename(dir))).toEqual( + expect.arrayContaining(['world-local', 'world-vercel', 'core', 'utils']) + ); + }); + + it.each(bundled)('reports nothing for %s', (dir) => { + const findings = scanPackage(dir, repoRoot); + expect(findings, formatFindings(findings)).toEqual([]); + }); + + it('flags a module-scope Map that is written to', () => { + const dir = packageWith( + [ + 'const transports = new Map();', + 'export function open(id: string) {', + ' transports.set(id, 1);', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'transports', keyword: 'const', reason: '`.set()`' }, + ]); + }); + + it('flags a module-scope `let` that is reassigned', () => { + const dir = packageWith( + [ + 'let started = false;', + 'export function start() {', + ' started = true;', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'started', keyword: 'let', reason: 'reassigned' }, + ]); + }); + + it('flags a field written through a member chain', () => { + const dir = packageWith( + [ + 'const state = { count: 0 };', + 'export function bump() {', + ' state.count += 1;', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'state', reason: 'field written' }, + ]); + }); + + it('ignores module-scope state that never changes', () => { + const dir = packageWith( + [ + 'const LIMIT = 10;', + 'const NAMES = new Set(["a"]);', + 'export const total = () => LIMIT + NAMES.size;', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('accepts state parked on globalThis by globalSingleton()', () => { + const dir = packageWith( + [ + "import { globalSingleton } from '@workflow/utils';", + "const state = globalSingleton('pkg//transports', 1, () => ({", + ' transports: new Map(),', + '}));', + 'export function open(id: string) {', + ' state.transports.set(id, 1);', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('accepts a declaration annotated `per-copy-ok:`', () => { + const dir = packageWith( + [ + '// per-copy-ok: reports what THIS copy sees, so once-per-copy is the point.', + 'let logged = false;', + 'export function warnOnce() {', + ' logged = true;', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('ignores a table filled once at module evaluation', () => { + // Every copy computes the same bytes at init, so per-copy costs memory and + // nothing else. Only a write that can happen later, per request, diverges. + const dir = packageWith( + [ + 'const BASE64_LOOKUP = new Uint8Array(256);', + 'for (let i = 0; i < 64; i++) BASE64_LOOKUP[i] = i;', + 'export function decode(i: number) {', + ' return BASE64_LOOKUP[i];', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('accepts state hand-rolled onto globalThis, through an alias', () => { + // The shape `docs/content/worlds/*/building-a-world.mdx` documents for + // custom world authors, and the one `packages/core` already uses. + const dir = packageWith( + [ + 'type WorldState = { locks: Map> };', + "const StateKey = Symbol.for('@your-org/world-foo//locks/v1');", + 'const store = globalThis as typeof globalThis &', + ' Record;', + 'const state: WorldState = (store[StateKey] ??= { locks: new Map() });', + 'export function open(id: string) {', + ' state.locks.set(id, Promise.resolve());', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('flags a static class field, which is module state with a namespace', () => { + const dir = packageWith( + [ + 'export class Registry {', + ' static transports = new Map();', + ' static open(id: string) {', + ' Registry.transports.set(id, 1);', + ' }', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'Registry.transports', keyword: 'static' }, + ]); + }); + + it('reports each static field on a class separately', () => { + // Keyed `Class.field`, not by the class: keying on the bare class name let + // the second static overwrite the first, so one of the two went unreported + // and the survivor was labelled with the other one's mutation. + const dir = packageWith( + [ + 'export class Registry {', + ' static transports = new Map();', + ' static latch = false;', + ' static open(id: string) {', + ' Registry.transports.set(id, 1);', + ' }', + ' static mark() {', + ' Registry.latch = true;', + ' }', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'Registry.transports', keyword: 'static', reason: '`.set()`' }, + { name: 'Registry.latch', keyword: 'static', reason: 'field written' }, + ]); + }); + + it('resolves `this` to the class inside a static member', () => { + const dir = packageWith( + [ + 'export class Counters {', + ' static hits = new Map();', + ' static bump(id: string) {', + ' this.hits.set(id, 1);', + ' }', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'Counters.hits', keyword: 'static' }, + ]); + }); + + it('ignores an instance field, which is per-instance not per-copy', () => { + const dir = packageWith( + [ + 'export class Session {', + ' seen = new Map();', + ' mark(id: string) {', + ' this.seen.set(id, 1);', + ' }', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('flags a field incremented with `++`, like one written with `+=`', () => { + const dir = packageWith( + [ + 'const state = { count: 0 };', + 'export function bump() {', + ' state.count++;', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'state', reason: 'field written' }, + ]); + }); + + it('flags an exported empty collection filled from another file', () => { + // The shipped bug's exact shape, with the registry and its mutators split + // across files. A single-file walk cannot see the write, so the export plus + // the empty initializer is the signal. + const dir = packageWithFiles({ + 'registry.ts': 'export const transports = new Map();\n', + 'consumer.ts': [ + "import { transports } from './registry.js';", + 'export function open(id: string) {', + ' transports.set(id, 1);', + '}', + '', + ].join('\n'), + }); + expect(scanPackage(dir, dir)).toMatchObject([ + { name: 'transports', reason: 'exported empty collection' }, + ]); + }); + + it('leaves a non-empty exported lookup table alone', () => { + const dir = packageWith("export const LIMITS = new Map([['a', 1]]);\n"); + expect(scanPackage(dir, dir)).toEqual([]); + }); + + it('scans `.mts` sources', () => { + // `@workflow/world-testing` is authored in `.mts`; while the walk was + // `.ts`-only its entry in the sweep below passed vacuously. + const dir = packageWithFiles({ + 'state.mts': [ + 'const counts = new Map();', + 'export function bump(id: string) {', + ' counts.set(id, 1);', + '}', + '', + ].join('\n'), + }); + expect(scanPackage(dir, dir)).toMatchObject([{ name: 'counts' }]); + }); + + it('does not accept a bare `per-copy-ok` with no reason', () => { + const dir = packageWith( + [ + '// per-copy-ok:', + 'let logged = false;', + 'export function warnOnce() {', + ' logged = true;', + '}', + '', + ].join('\n') + ); + expect(scanPackage(dir, dir)).toMatchObject([{ name: 'logged' }]); + }); +}); diff --git a/packages/workflow/src/module-scope-state.test.ts b/packages/workflow/src/module-scope-state.test.ts new file mode 100644 index 0000000000..aedf7b8d8a --- /dev/null +++ b/packages/workflow/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests. Repeated here so the signal arrives when you run just this + * package's tests. + * + * This package is compiled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for workflow', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/workflow'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/world-local/src/build-target-mismatch.ts b/packages/world-local/src/build-target-mismatch.ts index 4fbf7ed4ab..cb57e8a255 100644 --- a/packages/world-local/src/build-target-mismatch.ts +++ b/packages/world-local/src/build-target-mismatch.ts @@ -1,6 +1,7 @@ import { tmpdir } from 'node:os'; import path from 'node:path'; import { WorkflowWorldError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; /** * How to get out of a deployment running against the wrong world. The world is @@ -51,11 +52,18 @@ export function isUnwritableDirCode(code: string | undefined): boolean { return code !== undefined && UNWRITABLE_DIR_CODES.has(code); } -let warnedAboutVercelDeployment = false; +// Warned at most once per process; a field rather than a module-level `let` +// because a bundler can put several copies of this file in one process and +// "once" should not become once per copy (see `globalSingleton`). +const warnings = globalSingleton( + '@workflow/world-local//buildTargetWarnings', + 1, + () => ({ warnedAboutVercelDeployment: false }) +); /** Test seam: the warning is emitted once per process. */ export function resetVercelDeploymentWarning(): void { - warnedAboutVercelDeployment = false; + warnings.warnedAboutVercelDeployment = false; } /** @@ -75,7 +83,10 @@ export function resetVercelDeploymentWarning(): void { * deliberate choice rather than a misconfiguration. */ export function warnIfRunningInVercelDeployment(dataDir: string): void { - if (warnedAboutVercelDeployment || !process.env.VERCEL_DEPLOYMENT_ID) { + if ( + warnings.warnedAboutVercelDeployment || + !process.env.VERCEL_DEPLOYMENT_ID + ) { return; } const resolvedDataDir = path.resolve(dataDir); @@ -86,7 +97,7 @@ export function warnIfRunningInVercelDeployment(dataDir: string): void { ) { return; } - warnedAboutVercelDeployment = true; + warnings.warnedAboutVercelDeployment = true; console.warn( `[workflow] Warning: the local (filesystem) world is running inside a Vercel deployment, writing to ${resolvedDataDir}. ` + 'That filesystem is read-only, so workflow runs will fail before their first step. ' + diff --git a/packages/world-local/src/fs.ts b/packages/world-local/src/fs.ts index cf35e49f91..b0cbf7413c 100644 --- a/packages/world-local/src/fs.ts +++ b/packages/world-local/src/fs.ts @@ -1,6 +1,7 @@ import { promises as fs } from 'node:fs'; import path from 'node:path'; import { EntityConflictError, WorkflowWorldError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import type { PaginatedResponse } from '@workflow/world'; import { monotonicFactory } from 'ulid'; import { z } from 'zod'; @@ -9,7 +10,24 @@ import { UnwritableDataDirError, } from './build-target-mismatch.js'; -const ulid = monotonicFactory(() => Math.random()); +/** + * Temp-file suffixes for atomic writes, and the write-path caches below. + * + * On `globalThis` rather than at module scope because a bundler can put several + * copies of this file in one process (see `globalSingleton`): per-copy monotonic + * factories can hand two writers the same suffix in the same millisecond, and + * per-copy caches make the syscalls they exist to skip happen once per copy. + */ +const fsState = globalSingleton('@workflow/world-local//fs', 1, () => ({ + ulid: monotonicFactory(() => Math.random()), + // In-memory cache of created files to avoid expensive fs.access() calls. + // Safe because we only write once per file path (no overwrites without an + // explicit flag). + createdFilesCache: new Set(), + // Writes repeatedly target a small fixed set of entity directories. Once one + // exists in this process, avoid another recursive mkdir syscall per event. + createdDirectoriesCache: new Set(), +})); /** * Truncate a possibly-untrusted value for inclusion in an error message. @@ -129,19 +147,12 @@ export async function withWindowsRetry( throw new Error('Retry loop exited unexpectedly'); } -// In-memory cache of created files to avoid expensive fs.access() calls -// This is safe because we only write once per file path (no overwrites without explicit flag) -const createdFilesCache = new Set(); -// Writes repeatedly target a small fixed set of entity directories. Once one -// exists in this process, avoid another recursive mkdir syscall per event. -const createdDirectoriesCache = new Set(); - /** * Clear write-path caches. Useful for testing or when files are deleted externally. */ export function clearCreatedFilesCache(): void { - createdFilesCache.clear(); - createdDirectoriesCache.clear(); + fsState.createdFilesCache.clear(); + fsState.createdDirectoriesCache.clear(); } export { ulidToDate } from '@workflow/world'; @@ -274,12 +285,12 @@ export async function listTaggedFilesByExtension( export async function ensureDir(dirPath: string): Promise { const resolvedPath = path.resolve(dirPath); - if (createdDirectoriesCache.has(resolvedPath)) { + if (fsState.createdDirectoriesCache.has(resolvedPath)) { return; } try { await fs.mkdir(resolvedPath, { recursive: true }); - createdDirectoriesCache.add(resolvedPath); + fsState.createdDirectoriesCache.add(resolvedPath); } catch (error) { // A filesystem that refuses the directory outright will refuse every write // into it too, and the caller's write would surface as a confusing ENOENT @@ -320,7 +331,7 @@ async function withEnsuredDirectory( // A dev server may outlive an external cleanup of its data directory. // Forget the cached directory and retry once after recreating it. - createdDirectoriesCache.delete(path.resolve(dirPath)); + fsState.createdDirectoriesCache.delete(path.resolve(dirPath)); await ensureDir(dirPath); return operation(); } @@ -384,7 +395,7 @@ export async function write( if (!opts?.overwrite) { // Fast path: check in-memory cache first to avoid expensive fs.access() calls // This provides significant performance improvement when creating many files - if (createdFilesCache.has(filePath)) { + if (fsState.createdFilesCache.has(filePath)) { throw new EntityConflictError( `File ${filePath} already exists and 'overwrite' is false` ); @@ -394,7 +405,7 @@ export async function write( try { await fs.access(filePath); // File exists on disk, add to cache for future checks - createdFilesCache.add(filePath); + fsState.createdFilesCache.add(filePath); throw new EntityConflictError( `File ${filePath} already exists and 'overwrite' is false` ); @@ -406,7 +417,7 @@ export async function write( } } - const tempPath = `${filePath}.tmp.${ulid()}`; + const tempPath = `${filePath}.tmp.${fsState.ulid()}`; let tempFileCreated = false; try { await withEnsuredDirectory(path.dirname(filePath), async () => { @@ -415,7 +426,7 @@ export async function write( await withWindowsRetry(() => fs.rename(tempPath, filePath)); }); // Track this file in cache so future writes know it exists - createdFilesCache.add(filePath); + fsState.createdFilesCache.add(filePath); } catch (error) { // Only try to clean up temp file if it was actually created if (tempFileCreated) { @@ -482,7 +493,7 @@ export async function writeExclusive( filePath: string, data: string ): Promise { - const tempPath = `${filePath}.tmp.${ulid()}`; + const tempPath = `${filePath}.tmp.${fsState.ulid()}`; let tempFileCreated = false; try { diff --git a/packages/world-local/src/init.ts b/packages/world-local/src/init.ts index 1190664fdd..b4a35e65e8 100644 --- a/packages/world-local/src/init.ts +++ b/packages/world-local/src/init.ts @@ -8,6 +8,7 @@ import { } from 'node:fs/promises'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { globalSingleton } from '@workflow/utils'; /** Package name - hardcoded since it doesn't change */ const PACKAGE_NAME = '@workflow/world-local'; @@ -17,7 +18,13 @@ interface PackageInfo { version: string; } -let cachedPackageInfo: PackageInfo | null = null; +// On `globalThis` rather than at module scope so several copies of this file +// in one process share the resolved manifest (see `globalSingleton`). +const packageInfo = globalSingleton( + '@workflow/world-local//packageInfo', + 1, + () => ({ cached: null as PackageInfo | null }) +); /** * Get the directory path for this module. @@ -39,8 +46,8 @@ function getModuleDir(): string | null { * returns 'bundled' as the version. */ export async function getPackageInfo(): Promise { - if (cachedPackageInfo) { - return cachedPackageInfo; + if (packageInfo.cached) { + return packageInfo.cached; } const moduleDir = getModuleDir(); @@ -50,19 +57,19 @@ export async function getPackageInfo(): Promise { path.join(moduleDir, '../package.json'), 'utf-8' ); - cachedPackageInfo = JSON.parse(content) as PackageInfo; - return cachedPackageInfo; + packageInfo.cached = JSON.parse(content) as PackageInfo; + return packageInfo.cached; } catch { // Fall through to bundled fallback } } // Bundled context - package.json not accessible - cachedPackageInfo = { + packageInfo.cached = { name: PACKAGE_NAME, version: 'bundled', }; - return cachedPackageInfo; + return packageInfo.cached; } /** Filename for storing version information in the data directory */ diff --git a/packages/world-local/src/module-scope-state.test.ts b/packages/world-local/src/module-scope-state.test.ts new file mode 100644 index 0000000000..2b4277ef38 --- /dev/null +++ b/packages/world-local/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests and checks every published world package. Repeated here so the + * signal arrives when you run just this package's tests. + * + * This package is bundled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/world-local', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/world-local'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/world-local/src/storage/helpers.ts b/packages/world-local/src/storage/helpers.ts index e748d4b55b..a4fb238636 100644 --- a/packages/world-local/src/storage/helpers.ts +++ b/packages/world-local/src/storage/helpers.ts @@ -2,6 +2,7 @@ import { createHash } from 'node:crypto'; import fs from 'node:fs/promises'; import path from 'node:path'; import { WorkflowWorldError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import { eventIdToSlot } from '@workflow/world'; import { lock } from 'proper-lockfile'; import { decodeTime, monotonicFactory } from 'ulid'; @@ -502,10 +503,23 @@ export function hookRecoveryMarkerPath( } /** - * Create a monotonic ULID factory that ensures ULIDs are always increasing - * even when generated within the same millisecond. + * Monotonic ULID source: IDs are always increasing even when generated within + * the same millisecond. + * + * The factory lives on `globalThis` rather than at module scope because a + * bundler can put several copies of this file in one process (see + * `globalSingleton`). These IDs name events (`evnt_…`), so two copies each + * advancing their own sequence could mint the same event ID twice in one + * millisecond, or mint them out of order. */ -export const monotonicUlid = monotonicFactory(() => Math.random()); +const ulids = globalSingleton( + '@workflow/world-local//storageMonotonicUlid', + 1, + () => ({ next: monotonicFactory(() => Math.random()) }) +); + +export const monotonicUlid = (seedTime?: number): string => + ulids.next(seedTime); /** * Creates a function to extract createdAt date from a filename based on ULID. diff --git a/packages/world-local/src/storage/hook-index.ts b/packages/world-local/src/storage/hook-index.ts index c85ebf718f..b22f23d120 100644 --- a/packages/world-local/src/storage/hook-index.ts +++ b/packages/world-local/src/storage/hook-index.ts @@ -1,5 +1,6 @@ import fs from 'node:fs/promises'; import path from 'node:path'; +import { globalSingleton } from '@workflow/utils'; import type { Event } from '@workflow/world'; import { EventSchema, HookSchema } from '@workflow/world'; import { z } from 'zod'; @@ -196,12 +197,19 @@ export async function deleteHookByRunMarkerFile( await deleteJSON(path.join(byRunDir(basedir), `${fileId}.json`)); } -// Per-process ensure cache; only successful backfills are cached. -const ensuredBasedirs = new Map>(); +// Per-process ensure cache; only successful backfills are cached. On +// `globalThis` rather than at module scope so "per-process" stays true when a +// bundler puts several copies of this file in one process (see +// `globalSingleton`), otherwise each copy runs the full scan again. +const hookIndex = globalSingleton( + '@workflow/world-local//hookIndexEnsureCache', + 1, + () => ({ ensuredBasedirs: new Map>() }) +); /** Forget completed backfills (data-dir reset / tests). */ export function resetHookIndexEnsureCache(): void { - ensuredBasedirs.clear(); + hookIndex.ensuredBasedirs.clear(); } /** @@ -212,13 +220,13 @@ export function resetHookIndexEnsureCache(): void { */ export async function ensureHookIndexes(basedir: string): Promise { const key = path.resolve(basedir); - let pending = ensuredBasedirs.get(key); + let pending = hookIndex.ensuredBasedirs.get(key); if (!pending) { pending = ensureHookIndexesImpl(key).catch((error) => { - ensuredBasedirs.delete(key); + hookIndex.ensuredBasedirs.delete(key); throw error; }); - ensuredBasedirs.set(key, pending); + hookIndex.ensuredBasedirs.set(key, pending); } return pending; } diff --git a/packages/world-local/src/storage/runs-storage.ts b/packages/world-local/src/storage/runs-storage.ts index ac092918e0..4f1a7443e9 100644 --- a/packages/world-local/src/storage/runs-storage.ts +++ b/packages/world-local/src/storage/runs-storage.ts @@ -1,5 +1,6 @@ import path from 'node:path'; import { WorkflowRunNotFoundError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import type { AttributeChange, ExperimentalSetAttributesResult, @@ -75,27 +76,36 @@ export interface LocalRunsStorage { * Lifecycle writers acquire the lock and re-read the run file inside * the critical section to pick up any attributes that landed since * their pre-validation read. + * + * Held on `globalThis` rather than at module scope: a bundler can put several + * copies of this file in one process (see `globalSingleton`), and a per-copy + * lock table is not a lock: two copies would each believe they held the key + * and interleave exactly the read-modify-write this exists to serialize. */ -const runFileLocks = new Map>(); +const runLocks = globalSingleton( + '@workflow/world-local//runFileLocks', + 1, + () => ({ byKey: new Map>() }) +); export function withRunFileLock( key: string, fn: () => Promise ): Promise { - const prev = runFileLocks.get(key); + const prev = runLocks.byKey.get(key); const taskBox: { task?: Promise } = {}; const task = (async () => { if (prev) await prev.catch(() => undefined); try { return await fn(); } finally { - if (runFileLocks.get(key) === taskBox.task) { - runFileLocks.delete(key); + if (runLocks.byKey.get(key) === taskBox.task) { + runLocks.byKey.delete(key); } } })(); taskBox.task = task; - runFileLocks.set(key, task); + runLocks.byKey.set(key, task); return task; } diff --git a/packages/world-local/src/streamer.ts b/packages/world-local/src/streamer.ts index 2d514b26f2..a4379de59a 100644 --- a/packages/world-local/src/streamer.ts +++ b/packages/world-local/src/streamer.ts @@ -1,6 +1,7 @@ import { EventEmitter } from 'node:events'; import fs from 'node:fs/promises'; import path from 'node:path'; +import { globalSingleton } from '@workflow/utils'; import type { GetChunksOptions, StreamChunksResponse, @@ -19,9 +20,16 @@ import { writeJSON, } from './fs.js'; -// Create a monotonic ULID factory that ensures ULIDs are always increasing -// even when generated within the same millisecond -const monotonicUlid = monotonicFactory(() => Math.random()); +// Monotonic ULID source for chunk IDs: always increasing even within one +// millisecond. On `globalThis` rather than at module scope because a bundler +// can put several copies of this file in one process (see `globalSingleton`), +// and two copies advancing their own sequences can mint the same `chnk_` ID. +const chunkIds = globalSingleton( + '@workflow/world-local//streamerMonotonicUlid', + 1, + () => ({ next: monotonicFactory(() => Math.random()) }) +); +const monotonicUlid = (seedTime?: number): string => chunkIds.next(seedTime); // Schema for the run-to-streams mapping file const RunStreamsSchema = z.object({ diff --git a/packages/world-local/src/telemetry.ts b/packages/world-local/src/telemetry.ts index 9d78d8e6e9..fda6d7d262 100644 --- a/packages/world-local/src/telemetry.ts +++ b/packages/world-local/src/telemetry.ts @@ -10,12 +10,23 @@ */ import type * as api from '@opentelemetry/api'; import type { Span, SpanKind, SpanOptions } from '@opentelemetry/api'; +import { globalSingleton } from '@workflow/utils'; -// Lazy load OpenTelemetry API to make it optional -let otelApiPromise: Promise | null = null; +/** + * This module's process-wide state: the OpenTelemetry API, imported lazily so + * it stays optional, and the tracer built from it. + * + * On `globalThis` rather than at module scope because a bundler can put several + * copies of this file in one process (see `globalSingleton`), which would + * import the API and build a tracer once per copy. + */ +const otel = globalSingleton('@workflow/world-local//telemetry', 1, () => ({ + apiPromise: null as Promise | null, + tracerPromise: null as Promise | null, +})); async function getOtelApi(): Promise { - if (!otelApiPromise) { + if (!otel.apiPromise) { // Static specifier is intentional: esbuild-bundled targets (the CLI's // `vercel-build-output-api` build, Nitro, Astro) ship a self-contained // bundle with no node_modules, so `@opentelemetry/api` (an optional peer) @@ -23,20 +34,18 @@ async function getOtelApi(): Promise { // esbuild and would silently disable tracing there. Bundlers that reject // an unresolvable static `import()` when the peer is absent (Rollup/Vite, // e.g. SvelteKit) externalize it in the framework integration instead. - otelApiPromise = import('@opentelemetry/api').catch(() => null); + otel.apiPromise = import('@opentelemetry/api').catch(() => null); } - return otelApiPromise; + return otel.apiPromise; } -let tracerPromise: Promise | null = null; - async function getTracer(): Promise { - if (!tracerPromise) { - tracerPromise = getOtelApi().then((otel) => - otel ? otel.trace.getTracer('workflow') : null + if (!otel.tracerPromise) { + otel.tracerPromise = getOtelApi().then((otelApi) => + otelApi ? otelApi.trace.getTracer('workflow') : null ); } - return tracerPromise; + return otel.tracerPromise; } /** diff --git a/packages/world-postgres/src/module-scope-state.test.ts b/packages/world-postgres/src/module-scope-state.test.ts new file mode 100644 index 0000000000..88e9cbc502 --- /dev/null +++ b/packages/world-postgres/src/module-scope-state.test.ts @@ -0,0 +1,30 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests and checks every published world package. Repeated here so the + * signal arrives when you run just this package's tests. + * + * This package is deduped today only because `getRuntimeRequire()` loads it: + * a property of how it is loaded, not of how it is written, and exactly what + * changed for world-vercel in vercel/workflow#3493. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/world-postgres', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/world-postgres'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/world-testing/src/server.mts b/packages/world-testing/src/server.mts index b5f24ce602..069d1e1588 100644 --- a/packages/world-testing/src/server.mts +++ b/packages/world-testing/src/server.mts @@ -42,6 +42,9 @@ const Invoke = z }); // Track flow handler invocations per run for testing inline execution +// per-copy-ok: this file is a standalone test server entry (it calls `serve()` +// below), so it runs as its own process with one module instance. There is no +// host bundler to compile it into several layers. const flowInvocationCounts = new Map(); const app = new Hono() diff --git a/packages/world-vercel/package.json b/packages/world-vercel/package.json index 9a03eb2a43..c7517276fd 100644 --- a/packages/world-vercel/package.json +++ b/packages/world-vercel/package.json @@ -38,6 +38,7 @@ "@vercel/oidc": "catalog:", "@vercel/queue": "catalog:", "@workflow/errors": "workspace:*", + "@workflow/utils": "workspace:*", "@workflow/world": "workspace:*", "cbor-x": "1.6.0", "ulid": "catalog:", diff --git a/packages/world-vercel/src/create-run-id.ts b/packages/world-vercel/src/create-run-id.ts index 012c33c343..536114babc 100644 --- a/packages/world-vercel/src/create-run-id.ts +++ b/packages/world-vercel/src/create-run-id.ts @@ -1,3 +1,4 @@ +import { globalSingleton } from '@workflow/utils'; import { monotonicFactory } from 'ulid'; import { bytesToUlid, ulidToBytes } from './run-id/codec.js'; import { decode, encode } from './run-id/index.js'; @@ -8,21 +9,33 @@ import { } from './run-id/regions.js'; /** - * Underlying monotonic ULID factory. {@link encode} overwrites only the - * top 11 bits of the randomness section, so the factory's same-millisecond - * bottom-bit increments survive encoding and consecutive IDs with the same - * region/version metadata are naturally monotonic. The per-process check in - * {@link createRunId} exists for the remaining edge case: the metadata - * changing (e.g. a different `region`) within a single millisecond. - */ -const ulid = monotonicFactory(); - -/** - * Last emitted run ID (the encoded/tagged form), used to enforce strict - * lexicographic monotonicity across calls within a single process even - * when the region/version metadata changes between same-millisecond calls. + * This module's process-wide state: the monotonic ULID factory and the last + * emitted run ID (the encoded/tagged form), which together enforce strict + * lexicographic monotonicity across calls within a single process even when + * the region/version metadata changes between same-millisecond calls. + * + * {@link encode} overwrites only the top 11 bits of the randomness section, so + * the factory's same-millisecond bottom-bit increments survive encoding and + * consecutive IDs with the same region/version metadata are naturally monotonic + * on their own. The `lastRunId` comparison covers the remaining edge case: the + * metadata changing (e.g. a different `region`) within a single millisecond. + * + * On `globalThis` rather than at module scope because a bundler can put several + * copies of this file in one process (see `globalSingleton`), and both halves of + * the monotonicity guarantee are per-copy state. Two copies minting IDs in the + * same millisecond (a page in the `ssr` graph and a route handler in the + * app-route one both calling `start()`) would each advance their own factory + * and compare against their own `lastRunId`, so the process could emit the same + * ID twice, or emit them out of order. */ -let lastRunId: string | undefined; +const runIds = globalSingleton( + '@workflow/world-vercel//runIdFactory', + 1, + () => ({ + ulid: monotonicFactory(), + lastRunId: undefined as string | undefined, + }) +); /** * Increment the bit immediately above the 11-bit metadata window of a @@ -109,13 +122,13 @@ export function createRunId( ): string { const region = resolveRegion(options); const regionId = REGION_IDS[region]; - let candidate = encode(ulid(), regionId); - if (lastRunId !== undefined) { - while (candidate <= lastRunId) { - candidate = encode(bumpAboveMetadata(lastRunId), regionId); + let candidate = encode(runIds.ulid(), regionId); + if (runIds.lastRunId !== undefined) { + while (candidate <= runIds.lastRunId) { + candidate = encode(bumpAboveMetadata(runIds.lastRunId), regionId); } } - lastRunId = candidate; + runIds.lastRunId = candidate; return candidate; } diff --git a/packages/world-vercel/src/http-client.ts b/packages/world-vercel/src/http-client.ts index 2b4799f8d6..6157ad8c88 100644 --- a/packages/world-vercel/src/http-client.ts +++ b/packages/world-vercel/src/http-client.ts @@ -1,3 +1,4 @@ +import { globalSingleton } from '@workflow/utils'; import { isNodeHttpEnabled } from '@workflow/world'; import { createNodeHttpAgents, @@ -6,11 +7,25 @@ import { } from '@workflow/world/node-http.js'; import { Agent, type Dispatcher, RetryAgent, type RetryHandler } from 'undici'; import type { APIConfig } from './utils.js'; +import { version } from './version.js'; -let _dispatcher: RetryAgent | undefined; -let _streamDispatcher: RetryAgent | undefined; -let _streamCloseDispatcher: RetryAgent | undefined; -let _nodeHttpAgents: NodeHttpAgents | undefined; +/** + * This module's process-wide state: the shared connection pools. + * + * On `globalThis` rather than at module scope because a bundler can put several + * copies of this file in one process (see `globalSingleton`). Per-copy pools + * would mean per-copy keep-alive connections: a `register()` that warms the + * world would warm a pool no route ever dispatches on, and every layer would + * pay its own TCP and TLS handshake on its first request. The recycler's + * failure accounting would be split the same way, so a wedged origin would have + * to be detected once per copy. + */ +const pools = globalSingleton('@workflow/world-vercel//httpPools', 1, () => ({ + dispatcher: undefined as RetryAgent | undefined, + streamDispatcher: undefined as RetryAgent | undefined, + streamCloseDispatcher: undefined as RetryAgent | undefined, + nodeHttpAgents: undefined as NodeHttpAgents | undefined, +})); /** * Shared between all agents: connection pooling only. `pipelining` is @@ -523,9 +538,17 @@ export function createDispatcherRecycler( * black-holed HTTP/2 session self-healing. See createDispatcherRecycler and * EVENTS_RECYCLE_AFTER_CONSECUTIVE_FAILURES. */ -const eventsRecycler = createDispatcherRecycler( - () => createEventsDispatcher(), - 'events transport' +const eventsRecycler = globalSingleton( + // Version-keyed for the same reason as the WS registry in `ws-transport.ts`: + // this holds a recycler closed over *this* copy's `createEventsDispatcher`, + // so an unversioned key would silently apply one published version's undici + // and HTTP/2 options, and its failure accounting, to another's requests. The + // plain connection pools above stay unversioned: sharing a keep-alive pool + // across copies is the point, and they hold no module-local behavior. + `@workflow/world-vercel//eventsDispatcherRecycler@${version}`, + 1, + () => + createDispatcherRecycler(() => createEventsDispatcher(), 'events transport') ); /** @@ -583,17 +606,17 @@ export function getNodeHttpAgents( ): NodeHttpAgents | undefined { if (config?.dispatcher) return undefined; if (!isNodeHttpEnabled()) return undefined; - _nodeHttpAgents ??= createNodeHttpAgents({ + pools.nodeHttpAgents ??= createNodeHttpAgents({ maxSockets: BASE_AGENT_OPTIONS.connections, keepAliveMs: BASE_AGENT_OPTIONS.keepAliveTimeout, }); - return _nodeHttpAgents; + return pools.nodeHttpAgents; } /** Drop the shared node:http pool. Exported for tests; production keeps it. */ export function _resetNodeHttpAgentsForTests(): void { - if (_nodeHttpAgents) destroyNodeHttpAgents(_nodeHttpAgents); - _nodeHttpAgents = undefined; + if (pools.nodeHttpAgents) destroyNodeHttpAgents(pools.nodeHttpAgents); + pools.nodeHttpAgents = undefined; } /** @@ -752,11 +775,11 @@ export function createStreamDispatcher( * the `Retry-After` header when present. */ function getDefaultDispatcher(): RetryAgent { - _dispatcher ??= makeRetryDispatcher( + pools.dispatcher ??= makeRetryDispatcher( DEFAULT_AGENT_OPTIONS, RETRY_AGENT_OPTIONS ); - return _dispatcher; + return pools.dispatcher; } /** @@ -774,12 +797,14 @@ function getDefaultDispatcher(): RetryAgent { * at once. */ function getDefaultStreamDispatcher(): RetryAgent { - _streamDispatcher ??= createStreamDispatcher(STREAM_RETRY_OPTIONS); - return _streamDispatcher; + pools.streamDispatcher ??= createStreamDispatcher(STREAM_RETRY_OPTIONS); + return pools.streamDispatcher; } /** Shared agent for the idempotent stream close (5xx retriable). */ function getDefaultStreamCloseDispatcher(): RetryAgent { - _streamCloseDispatcher ??= createStreamDispatcher(STREAM_CLOSE_RETRY_OPTIONS); - return _streamCloseDispatcher; + pools.streamCloseDispatcher ??= createStreamDispatcher( + STREAM_CLOSE_RETRY_OPTIONS + ); + return pools.streamCloseDispatcher; } diff --git a/packages/world-vercel/src/module-scope-state.test.ts b/packages/world-vercel/src/module-scope-state.test.ts new file mode 100644 index 0000000000..3fa4e78ede --- /dev/null +++ b/packages/world-vercel/src/module-scope-state.test.ts @@ -0,0 +1,29 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +// @ts-expect-error -- plain JS lint rule, no type declarations +import { + formatFindings, + scanPackage, +} from '../../../scripts/lint/module-scope-state.mjs'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repoRoot = path.resolve(here, '../../..'); + +/** + * A local mirror of the sweep in `@workflow/utils`, which owns this rule and + * its own tests and checks every published world package. Repeated here so the + * signal arrives when you run just this package's tests. + * + * This package is bundled into the host application's server build, so one + * process holds one copy of each of its modules per bundler layer. + */ +describe('module-scope state rule', () => { + it('reports nothing for @workflow/world-vercel', () => { + const findings = scanPackage( + path.join(repoRoot, 'packages/world-vercel'), + repoRoot + ); + expect(findings, formatFindings(findings)).toEqual([]); + }); +}); diff --git a/packages/world-vercel/src/queue.ts b/packages/world-vercel/src/queue.ts index 655c1906dd..0f1eec291e 100644 --- a/packages/world-vercel/src/queue.ts +++ b/packages/world-vercel/src/queue.ts @@ -1,6 +1,7 @@ import { AsyncLocalStorage } from 'node:async_hooks'; import type { Transport } from '@vercel/queue'; import { ConsumerDiscoveryError, QueueClient } from '@vercel/queue'; +import { globalSingleton } from '@workflow/utils'; import { MessageId, type Queue, @@ -99,6 +100,9 @@ class DualTransport implements Transport { } } +// per-copy-ok: both ends of this store live in the same `createQueueHandler` +// closure: the `run()` wrapper and the `getStore()` read always come from the +// same module copy, so the context never has to cross a copy boundary. const requestIdStorage = new AsyncLocalStorage(); const MessageWrapper = z.object({ @@ -338,7 +342,14 @@ function getHeadersFromPayload( */ const FLOW_TOPIC_PATTERN = /^__([a-z][a-z0-9]*_)?wkf_workflow_/; -let loggedSequentialReplays = false; +// Logged at most once per process; a field rather than a module-level `let` +// because a bundler can put several copies of this file in one process and +// "once" should not become once per copy (see `globalSingleton`). +const queueLogs = globalSingleton( + '@workflow/world-vercel//queueLogLatches', + 1, + () => ({ loggedSequentialReplays: false }) +); /** * Whether sequential replays are enabled (`WORKFLOW_SEQUENTIAL_REPLAYS=1`). @@ -356,8 +367,8 @@ function getPhysicalQueueName( if (!isSequentialReplaysEnabled() || !FLOW_TOPIC_PATTERN.test(queueName)) { return queueName; } - if (!loggedSequentialReplays) { - loggedSequentialReplays = true; + if (!queueLogs.loggedSequentialReplays) { + queueLogs.loggedSequentialReplays = true; // One-time breadcrumb so a half-applied configuration (env var set without // a maxConcurrency-bearing flow trigger, or vice versa) is diagnosable // from function logs. Must go to stderr: this code also runs inside CLI diff --git a/packages/world-vercel/src/runs.ts b/packages/world-vercel/src/runs.ts index d74f7421ab..28df1297cf 100644 --- a/packages/world-vercel/src/runs.ts +++ b/packages/world-vercel/src/runs.ts @@ -1,4 +1,5 @@ import { WorkflowRunNotFoundError, WorkflowWorldError } from '@workflow/errors'; +import { globalSingleton } from '@workflow/utils'; import { type AttributeChange, type BulkCancelWorkflowRunsRequest, @@ -283,11 +284,15 @@ const LONG_POLL_UNSUPPORTED_TTL_MS = 5 * 60 * 1000; * fast path for the other. Bounded by construction: the key is the resolved * base URL, of which a process has a handful at most. */ -const longPollUnsupportedUntilByBaseUrl = new Map(); +const longPoll = globalSingleton( + '@workflow/world-vercel//runStatusLongPollSupport', + 1, + () => ({ unsupportedUntilByBaseUrl: new Map() }) +); /** Test-only: forget that the long-poll route was unavailable. @internal */ export function _resetRunStatusLongPollSupportForTests(): void { - longPollUnsupportedUntilByBaseUrl.clear(); + longPoll.unsupportedUntilByBaseUrl.clear(); } /** @@ -379,7 +384,7 @@ export async function waitForWorkflowRunTerminalStatus( ); const { baseUrl } = getHttpUrl(config); - const unsupportedUntil = longPollUnsupportedUntilByBaseUrl.get(baseUrl) ?? 0; + const unsupportedUntil = longPoll.unsupportedUntilByBaseUrl.get(baseUrl) ?? 0; if (waitMs === 0 || Date.now() < unsupportedUntil) { return getWorkflowRun(id, { resolveData }, config); @@ -410,7 +415,7 @@ export async function waitForWorkflowRunTerminalStatus( // Throws WorkflowRunNotFoundError when the run is what was missing. const run = await getWorkflowRun(id, { resolveData }, config); - longPollUnsupportedUntilByBaseUrl.set( + longPoll.unsupportedUntilByBaseUrl.set( baseUrl, Date.now() + LONG_POLL_UNSUPPORTED_TTL_MS ); diff --git a/packages/world-vercel/src/telemetry.ts b/packages/world-vercel/src/telemetry.ts index 438c12faff..95b4490d5e 100644 --- a/packages/world-vercel/src/telemetry.ts +++ b/packages/world-vercel/src/telemetry.ts @@ -15,12 +15,24 @@ */ import type * as api from '@opentelemetry/api'; import type { Span, SpanKind, SpanOptions } from '@opentelemetry/api'; +import { globalSingleton } from '@workflow/utils'; -// Lazy load OpenTelemetry API to make it optional -let otelApiPromise: Promise | null = null; +/** + * This module's process-wide state: the lazily-imported OpenTelemetry API and + * the tracer built from it. + * + * On `globalThis` rather than at module scope because a bundler can put several + * copies of this file in one process (see `globalSingleton`); per-copy caches + * would import `@opentelemetry/api` and build a tracer once per copy. + */ +const otel = globalSingleton('@workflow/world-vercel//telemetry', 1, () => ({ + // Lazy load OpenTelemetry API to make it optional + apiPromise: null as Promise | null, + tracerPromise: null as Promise | null, +})); async function getOtelApi(): Promise { - if (!otelApiPromise) { + if (!otel.apiPromise) { // Static specifier is intentional: esbuild-bundled targets (the CLI's // `vercel-build-output-api` build, Nitro, Astro) ship a self-contained // bundle with no node_modules, so `@opentelemetry/api` (an optional peer) @@ -28,7 +40,7 @@ async function getOtelApi(): Promise { // esbuild and would silently disable tracing there. Bundlers that reject // an unresolvable static `import()` when the peer is absent (Rollup/Vite, // e.g. SvelteKit) externalize it in the framework integration instead. - otelApiPromise = import('@opentelemetry/api').catch((error) => { + otel.apiPromise = import('@opentelemetry/api').catch((error) => { // A missing module is expected for apps without OTEL, but the same // silent null also swallows bundler/resolution failures in apps that // DO register a tracer, which then lose every world-vercel span. @@ -46,11 +58,9 @@ async function getOtelApi(): Promise { return null; }); } - return otelApiPromise; + return otel.apiPromise; } -let tracerPromise: Promise | null = null; - function workflowDebugEnabled(): boolean { return ( typeof process !== 'undefined' && @@ -59,6 +69,9 @@ function workflowDebugEnabled(): boolean { ); } +// per-copy-ok: this diagnostic reports how THIS module instance sees the +// global OTel registration, so "once" is deliberately once per copy. With +// several copies in a process, each one's view is the thing worth seeing. let otelDiagLogged = false; /** @@ -100,15 +113,15 @@ function logOtelDiagnosticOnce(otel: typeof api, tracer: api.Tracer): void { } async function getTracer(): Promise { - if (!tracerPromise) { - tracerPromise = getOtelApi().then((otel) => { - if (!otel) return null; - const tracer = otel.trace.getTracer('workflow'); - logOtelDiagnosticOnce(otel, tracer); + if (!otel.tracerPromise) { + otel.tracerPromise = getOtelApi().then((otelApi) => { + if (!otelApi) return null; + const tracer = otelApi.trace.getTracer('workflow'); + logOtelDiagnosticOnce(otelApi, tracer); return tracer; }); } - return tracerPromise; + return otel.tracerPromise; } /** diff --git a/packages/world-vercel/src/ws-transport-module-copies.test.ts b/packages/world-vercel/src/ws-transport-module-copies.test.ts new file mode 100644 index 0000000000..cd3d882882 --- /dev/null +++ b/packages/world-vercel/src/ws-transport-module-copies.test.ts @@ -0,0 +1,47 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import * as first from './ws-transport.js'; +// A second, independent instance of the same module. Vite keys its module +// registry on the specifier, so the query suffix buys what a bundler layer +// buys in a Next.js server build: the same file, compiled and evaluated twice +// in one process. +// @ts-expect-error -- same module, distinct instance; no declaration for the query form +import * as second from './ws-transport.js?copy=2'; + +const WS_URL = 'wss://vercel-workflow.test/websockets/v1/runs/wrun_copies'; +const headers = async () => ({ authorization: 'Bearer test' }); + +afterEach(() => { + first.resetWsEventsTransportsForTest(); +}); + +describe('ws transport registry across module copies', () => { + /** + * Guards the test against becoming vacuous: if the two specifiers ever + * collapsed to one module instance, every assertion below would pass for the + * wrong reason. Class identity is module-scope state, so distinct classes + * means distinct instances, and is itself the thing that used to make the + * registry diverge. + */ + it('imports two genuinely distinct instances of the module', () => { + expect(second.WsTransportError).not.toBe(first.WsTransportError); + }); + + /** + * The vercel/workflow#3493 regression, in miniature. `@workflow/world-vercel` + * became bundled rather than external, so one process holds one copy of this + * module per bundler layer. The queue consumer opened its channel from the + * `instrument` copy and the events write path looked it up from the route + * copy's own, empty `Map`. Every event then silently fell back to HTTP for + * the life of the process. + */ + it('finds a transport registered by the other copy', () => { + const registered = first.getWsEventsTransport(WS_URL, headers); + expect(second.getWsEventsTransport(WS_URL, headers)).toBe(registered); + }); + + it('drops it for both copies when either one resets', () => { + const registered = first.getWsEventsTransport(WS_URL, headers); + second.resetWsEventsTransportsForTest(); + expect(first.getWsEventsTransport(WS_URL, headers)).not.toBe(registered); + }); +}); diff --git a/packages/world-vercel/src/ws-transport.ts b/packages/world-vercel/src/ws-transport.ts index 78e780ebad..2b0e7397d3 100644 --- a/packages/world-vercel/src/ws-transport.ts +++ b/packages/world-vercel/src/ws-transport.ts @@ -23,6 +23,7 @@ */ import { getVercelOidcToken } from '@vercel/oidc'; +import { globalSingleton } from '@workflow/utils'; import { WebSocket } from 'ws'; import { type DecodedFrame, decodeFrames } from './frames.js'; import { @@ -38,6 +39,7 @@ import { WorkflowWsReconnectAttempt, } from './telemetry.js'; import { type APIConfig, getHttpConfig, getHttpUrl } from './utils.js'; +import { version } from './version.js'; import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; export interface WsFrameReply { @@ -290,7 +292,9 @@ class WsEventsTransport { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } - if (transports.get(this.wsUrl) === this) transports.delete(this.wsUrl); + if (wsState.transports.get(this.wsUrl) === this) { + wsState.transports.delete(this.wsUrl); + } const conn = this.connection; this.connection = null; // Normal closure: a clean client-side release, not an aborted run. @@ -660,7 +664,40 @@ class WsEventsTransport { } } -const transports = new Map(); +/** + * Process-wide, not module-scope: `@workflow/world-vercel` is bundled into the + * Next.js server output, so a plain `const` here would be one Map per bundler + * layer. The queue consumer registers a channel from the `instrument` layer + * copy and the write path looks it up from the route layer copy: a + * deterministic miss that silently demotes every event to HTTP. See + * `globalSingleton`'s doc comment. + * + * `loggedWsProxyFallback` / `loggedWsInUse` live here for the same reason: + * they are once-*per-process* latches, and a `let` cannot be shared by + * reference. + * Both branches they guard repeat on every event, so a per-copy log would be + * the same noise the latch exists to prevent. + */ +const wsState = globalSingleton( + // Keyed by package version, unlike the state that holds only plain data. + // Two different published versions of this package can share one process (a + // transitive dependency pinning an older `@workflow/core`, which depends on + // this package by exact version), and this Map holds `WsEventsTransport` + // instances. An unversioned key would hand one version's write path an object + // built by the other version's class, and `events-v4.ts` would then frame and + // parse against a protocol the other copy may not share. There is no version + // negotiation on this socket to catch that. `shapeVersion` cannot express it: + // the container shape is stable, the hazard is in the contents. Two versions + // therefore keep separate registries, which costs a second socket and is what + // happened before this package was bundled anyway. + `@workflow/world-vercel//wsEventsTransports@${version}`, + 1, + () => ({ + transports: new Map(), + loggedWsProxyFallback: false, + loggedWsInUse: false, + }) +); /** * Get (or lazily create) the shared WS transport for `wsUrl`. `getHeaders` runs @@ -678,10 +715,10 @@ export function getWsEventsTransport( forceRefresh: boolean; }) => Promise> ): WsEventsTransport { - let transport = transports.get(wsUrl); + let transport = wsState.transports.get(wsUrl); if (!transport) { transport = new WsEventsTransport(wsUrl, getHeaders); - transports.set(wsUrl, transport); + wsState.transports.set(wsUrl, transport); } return transport; } @@ -692,12 +729,12 @@ export function getWsEventsTransport( * isn't silenced by an earlier one having already logged it. */ export function resetWsEventsTransportsForTest(): void { - for (const transport of [...transports.values()]) { + for (const transport of [...wsState.transports.values()]) { transport.close('test reset'); } - transports.clear(); - loggedWsProxyFallback = false; - loggedWsInUse = false; + wsState.transports.clear(); + wsState.loggedWsProxyFallback = false; + wsState.loggedWsInUse = false; } /** @@ -773,8 +810,8 @@ export function openWsChannel( if (!isWsEventsTransportEnabled()) return undefined; const resolved = resolveChannelUrl(runId, config); if (!resolved) return undefined; - if (!loggedWsInUse) { - loggedWsInUse = true; + if (!wsState.loggedWsInUse) { + wsState.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 @@ -830,11 +867,6 @@ async function refreshOidcTokenBestEffort(): Promise { } } -// Each logged at most once per process: both branches below are expected -// to repeat (every event), and a per-request log would add 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 @@ -856,8 +888,8 @@ function resolveChannelUrl( // 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; + if (!wsState.loggedWsProxyFallback) { + wsState.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.` @@ -886,6 +918,6 @@ export function resolveWsTransport( } | null { const wsUrl = resolveChannelUrl(runId, config); if (!wsUrl) return null; - const transport = transports.get(wsUrl); + const transport = wsState.transports.get(wsUrl); return transport ? { transport, wsUrl } : null; } diff --git a/packages/world/README.md b/packages/world/README.md index c5f28b767e..e0fab7a2fe 100644 --- a/packages/world/README.md +++ b/packages/world/README.md @@ -5,3 +5,24 @@ Core interfaces and types for Workflow SDK storage backends. This package defines the `World` interface that abstracts workflow storage, queuing, authentication, and streaming operations. Implementation packages like `@workflow/world-local` and `@workflow/world-vercel` provide concrete implementations. Used internally by `@workflow/core` and world implementations. Should not be used directly in application code. + +## Implementation constraint: no mutable module state + +A World implementation must not keep mutable state at module scope. Hold it on +the World instance, or, when it is genuinely process-wide (an ID generator +whose sequence must not fork, a log-once latch), on `globalThis` via +`globalSingleton()` from `@workflow/utils`. + +`@workflow/world-local` and `@workflow/world-vercel` are bundled into the host +application's server build, and a bundler keys module identity on +`(resource, layer)`: Next.js alone compiles `instrument`, app-route, `ssr` and +`edge` as separate module graphs, so one process holds one copy of every module +in these packages *per layer*. A top-level `let`, or a `const` holding a `Map`, +is therefore per-copy state rather than the singleton it reads as. + +A world loaded at runtime through `WORKFLOW_TARGET_WORLD` is deduped by Node's +module cache and does not have this problem today, but that is a property of +how it is loaded, not of how it is written, and it has changed before +(vercel/workflow#3493). `scripts/lint/module-scope-state.mjs` enforces the rule +across every published world package; see +`docs/content/worlds/*/building-a-world.mdx` for the author-facing version. diff --git a/packages/world/src/env-config.ts b/packages/world/src/env-config.ts index ff75ed8b2b..f59711c9e3 100644 --- a/packages/world/src/env-config.ts +++ b/packages/world/src/env-config.ts @@ -27,7 +27,21 @@ export interface EnvNumberOptions { // Raw "name=value" pairs already warned about, so a bad env var warns once // per process rather than on every (lazy) read. -const warnedEnvValues = new Set(); +// +// On `globalThis` rather than at module scope so "per process" survives +// bundling: this package is compiled into the host application's server build, +// which gives one copy of this module per bundler layer, and a per-copy Set +// would warn once per layer. Hand-rolled rather than `globalSingleton()` from +// `@workflow/utils` because this package deliberately carries no dependencies; +// the two are equivalent and the rule accepts both. +const WarnedEnvValuesKey = Symbol.for('@workflow/world//warnedEnvValues/v1'); +const globalStore = globalThis as typeof globalThis & + Record | undefined>; +// The same globalThis-backed idiom as `packages/core/src/private.ts`. Keeping +// the initializer an expression is also what +// `scripts/lint/module-scope-state.mjs` recognizes as off-module state. +// biome-ignore lint/suspicious/noAssignInExpressions: off-module state idiom +const warnedEnvValues = (globalStore[WarnedEnvValuesKey] ??= new Set()); function warnOnce(key: string, message: string): void { if (warnedEnvValues.has(key)) return; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2a9741f39b..e03357768b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -293,6 +293,9 @@ importers: '@workflow/serde': specifier: workspace:^ version: link:../serde + '@workflow/utils': + specifier: workspace:* + version: link:../utils zod: specifier: 'catalog:' version: 4.3.6 @@ -709,6 +712,9 @@ importers: '@workflow/swc-plugin': specifier: workspace:* version: link:../swc-plugin-workflow + '@workflow/utils': + specifier: workspace:* + version: link:../utils esbuild: specifier: 'catalog:' version: 0.28.1 @@ -1589,6 +1595,9 @@ importers: '@workflow/errors': specifier: workspace:* version: link:../errors + '@workflow/utils': + specifier: workspace:* + version: link:../utils '@workflow/world': specifier: workspace:* version: link:../world diff --git a/scripts/lint/module-scope-state.mjs b/scripts/lint/module-scope-state.mjs new file mode 100644 index 0000000000..60a4badd7e --- /dev/null +++ b/scripts/lint/module-scope-state.mjs @@ -0,0 +1,544 @@ +/** + * Finds module-scope state that changes at runtime. + * + * `@workflow/world-vercel` and `@workflow/world-local` are *bundled* into the + * host application's server build (see `VERCEL_WORLD_DEPENDENCY_PACKAGES` in + * `packages/next/src/index.ts`). A bundler keys module identity on + * (resource, layer), so one process holds one copy of each of these modules + * *per layer*. Next.js alone builds `instrument`, app-route, `ssr` and `edge` + * layers, and code registered from `instrumentation.ts` therefore does not + * share module scope with code that runs in a route handler. + * + * That makes every mutable module-scope binding a per-copy variable rather + * than the process-wide singleton its author assumed. vercel/workflow#3493 + * turned these packages from external into bundled and the WebSocket events + * transport silently regressed to HTTP for exactly this reason: the queue + * consumer registered its channel in the `instrument` copy's `Map` and the + * write path looked it up in the route copy's empty one. + * + * The fix is `globalSingleton()` from `@workflow/utils`, which parks the state + * on `globalThis` under a `Symbol.for()` key so every copy shares one object. + * This rule fails the build on anything that reintroduces the pattern. + * + * Two escapes: + * - initialize the binding from `globalSingleton(...)` or from `globalThis` + * directly, the fix itself; + * - annotate it `// per-copy-ok: ` when the + * state is deliberately per module instance (a diagnostic describing what + * *this* copy sees, for example). + * + * What it sees: `const`/`let` statements and `static` class fields, mutated + * from inside a function body. Writes in top-level statements are ignored, + * because they run identically in every copy at module evaluation, so a + * precomputed lookup table is not a finding. An *exported* binding initialized + * to an empty collection is a finding on its own, since the code that fills it + * is often in another file. + * + * What it does not see: a write to an imported binding, resolved across files. + * That needs whole-package resolution. The exported-empty-collection rule above + * is the cheap approximation, and it is why exporting a mutable registry is + * reported even when this file never writes to it. + * + * Usage: node scripts/lint/module-scope-state.mjs [...] + */ +import fs from 'node:fs'; +import path from 'node:path'; +import ts from 'typescript'; + +/** Methods that mutate the receiver in place. */ +const MUTATORS = new Set([ + 'set', + 'delete', + 'clear', + 'add', + 'push', + 'pop', + 'shift', + 'unshift', + 'splice', + 'sort', + 'reverse', + 'fill', + 'copyWithin', +]); + +const SINGLETON_HELPER = 'globalSingleton'; +const PRAGMA = /(?:^|\s)per-copy-ok:\s*(\S.*)$/; + +function walkSourceFiles(dir, out = []) { + if (!fs.existsSync(dir)) return out; + for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + walkSourceFiles(full, out); + continue; + } + // `.mts`/`.cts` as well as `.ts`: `@workflow/world-testing` is authored in + // `.mts`, and skipping those extensions made its sweep pass vacuously. + if (!/\.(ts|mts|cts)$/.test(entry.name)) continue; + if (/\.(test|spec)\.(ts|mts|cts)$/.test(entry.name)) continue; + if (/\.d\.(ts|mts|cts)$/.test(entry.name)) continue; + out.push(full); + } + return out; +} + +/** `globalSingleton(...)`, including a namespaced `utils.globalSingleton(...)`. */ +function isGlobalSingletonCall(node) { + if (!node) return false; + if (ts.isAsExpression(node) || ts.isTypeAssertionExpression(node)) { + return isGlobalSingletonCall(node.expression); + } + if (!ts.isCallExpression(node)) return false; + const callee = node.expression; + if (ts.isIdentifier(callee)) return callee.text === SINGLETON_HELPER; + if (ts.isPropertyAccessExpression(callee)) { + return callee.name.text === SINGLETON_HELPER; + } + return false; +} + +/** + * Whether an initializer reaches `globalThis`, so the hand-rolled + * `const store = globalThis as …` / `const x = (globalThis[Key] ??= …)` shape is + * accepted alongside `globalSingleton()`. Both park the state off-module, which + * is the property this rule is actually checking for; `packages/core`'s step + * registry and the pattern documented for custom world authors in + * `docs/content/worlds/*\/building-a-world.mdx` are both written this way. + */ +function isGlobalThisBacked(node, aliases = new Set()) { + if (!node) return false; + if (ts.isIdentifier(node)) { + return node.text === 'globalThis' || aliases.has(node.text); + } + if ( + ts.isAsExpression(node) || + ts.isTypeAssertionExpression(node) || + ts.isNonNullExpression(node) || + ts.isParenthesizedExpression(node) || + ts.isPropertyAccessExpression(node) || + ts.isElementAccessExpression(node) + ) { + return isGlobalThisBacked(node.expression, aliases); + } + if (ts.isBinaryExpression(node)) { + // `globalThis[Key] ??= {…}` and friends. + return ( + isGlobalThisBacked(node.left, aliases) || + isGlobalThisBacked(node.right, aliases) + ); + } + return false; +} + +/** + * Names in this file that are themselves globalThis-backed, so a binding + * derived from one is too. The documented pattern takes two statements: an + * alias for `globalThis`, then the state read off it. + */ +function globalThisAliases(declared) { + const aliases = new Set(); + for (const binding of declared.values()) { + if (isGlobalThisBacked(binding.declaration.initializer, aliases)) { + aliases.add(binding.name); + } + } + return aliases; +} + +/** + * The identifier a member chain is rooted at, so `state.pools.set(…)` is + * recognized as a mutation of `state`. + */ +function rootIdentifier(node) { + let current = node; + while ( + ts.isPropertyAccessExpression(current) || + ts.isElementAccessExpression(current) || + ts.isNonNullExpression(current) || + ts.isParenthesizedExpression(current) + ) { + current = current.expression; + } + return ts.isIdentifier(current) ? current.text : undefined; +} + +/** A `// per-copy-ok: ` comment directly above the declaration. */ +function perCopyReason(statement, text) { + const ranges = ts.getLeadingCommentRanges(text, statement.getFullStart()); + if (!ranges) return undefined; + for (const range of ranges) { + const match = PRAGMA.exec(text.slice(range.pos, range.end).trim()); + if (match) return match[1].trim(); + } + return undefined; +} + +/** + * Module-scope bindings in `source`, keyed by the name a mutation would be + * attributed to. + * + * Covers `const`/`let` statements and `static` class fields. A static field is + * module-scope state wearing a class as its namespace: `Registry.transports` + * duplicates per copy exactly like a top-level `const` would. Static fields are + * keyed `Class.field`, so a class with several of them yields one entry each and + * every finding names the field that is actually written. + */ +function collectDeclarations(source) { + const declared = new Map(); + for (const statement of source.statements) { + if (ts.isVariableStatement(statement)) { + const isConst = + (statement.declarationList.flags & ts.NodeFlags.Const) !== 0; + for (const declaration of statement.declarationList.declarations) { + if (!ts.isIdentifier(declaration.name)) continue; + declared.set(declaration.name.text, { + name: declaration.name.text, + isConst, + declaration, + statement, + }); + } + continue; + } + if (!ts.isClassDeclaration(statement) || !statement.name) continue; + for (const member of statement.members) { + if (!ts.isPropertyDeclaration(member) || !ts.isIdentifier(member.name)) { + continue; + } + const isStatic = member.modifiers?.some( + (m) => m.kind === ts.SyntaxKind.StaticKeyword + ); + if (!isStatic) continue; + // Keyed `Class.field`, which is what `memberPath` reads off a write like + // `Registry.transports.set(…)`. Keying on the bare class name would let a + // second static field overwrite the first, and would then attach one + // field's mutation to the other field's declaration. + const key = `${statement.name.text}.${member.name.text}`; + declared.set(key, { + name: key, + isConst: false, + declaration: member, + statement, + keyword: 'static', + }); + } + } + return declared; +} + +/** + * The class `this` refers to, when `this` *is* the class: inside a `static` + * member. Undefined inside an instance member, where `this` is an instance and + * the state it holds is per-instance rather than per-copy, and undefined inside + * a nested `function`, which rebinds `this`. + */ +function staticClassOf(node) { + for (let n = node.parent; n; n = n.parent) { + if (ts.isFunctionDeclaration(n) || ts.isFunctionExpression(n)) { + return undefined; + } + const isMember = + ts.isMethodDeclaration(n) || + ts.isPropertyDeclaration(n) || + ts.isGetAccessorDeclaration(n) || + ts.isSetAccessorDeclaration(n) || + ts.isClassStaticBlockDeclaration(n); + if (!isMember) continue; + const isStatic = + ts.isClassStaticBlockDeclaration(n) || + n.modifiers?.some((m) => m.kind === ts.SyntaxKind.StaticKeyword); + if (!isStatic) return undefined; + return ts.isClassDeclaration(n.parent) && n.parent.name + ? n.parent.name.text + : undefined; + } + return undefined; +} + +/** + * The `Root.field` prefix of a member chain, or undefined when there is no named + * first property. Lets a write to `Registry.transports.set(…)` be attributed to + * the static field `Registry.transports`, which `rootIdentifier` alone cannot + * distinguish from a write to any other static on the same class. `this.field` + * inside a static member resolves to the class, where `this` is the class. + */ +function memberPath(node) { + const segments = []; + let current = node; + while ( + ts.isPropertyAccessExpression(current) || + ts.isElementAccessExpression(current) || + ts.isNonNullExpression(current) || + ts.isParenthesizedExpression(current) + ) { + if (ts.isPropertyAccessExpression(current)) { + segments.unshift(current.name.text); + } else if (ts.isElementAccessExpression(current)) { + // A computed key names no field, so the chain stops being addressable. + segments.unshift(undefined); + } + current = current.expression; + } + const root = ts.isIdentifier(current) + ? current.text + : current.kind === ts.SyntaxKind.ThisKeyword + ? staticClassOf(current) + : undefined; + if (!root || segments[0] === undefined) return undefined; + return `${root}.${segments[0]}`; +} + +/** `x = …`, `x.field = …`, `x += …`. */ +function assignment(node) { + if ( + !ts.isBinaryExpression(node) || + node.operatorToken.kind < ts.SyntaxKind.FirstAssignment || + node.operatorToken.kind > ts.SyntaxKind.LastAssignment + ) { + return undefined; + } + if (ts.isIdentifier(node.left)) { + return { name: node.left.text, reason: 'reassigned' }; + } + if ( + ts.isPropertyAccessExpression(node.left) || + ts.isElementAccessExpression(node.left) + ) { + return { + name: rootIdentifier(node.left), + target: node.left, + reason: 'field written', + }; + } + return undefined; +} + +/** `x++`, `--x`, `x.field++`. */ +function increment(node) { + if (!ts.isPrefixUnaryExpression(node) && !ts.isPostfixUnaryExpression(node)) { + return undefined; + } + if ( + node.operator !== ts.SyntaxKind.PlusPlusToken && + node.operator !== ts.SyntaxKind.MinusMinusToken + ) { + return undefined; + } + if (ts.isIdentifier(node.operand)) { + return { name: node.operand.text, reason: 'reassigned' }; + } + // `state.count++` mutates just as much as `state.count += 1`, which + // `assignment` already reports. + if ( + ts.isPropertyAccessExpression(node.operand) || + ts.isElementAccessExpression(node.operand) + ) { + return { + name: rootIdentifier(node.operand), + target: node.operand, + reason: 'field written', + }; + } + return undefined; +} + +/** `x.set(…)`, `x.items.push(…)`: a call that mutates its receiver. */ +function mutatingCall(node) { + if ( + !ts.isCallExpression(node) || + !ts.isPropertyAccessExpression(node.expression) || + !MUTATORS.has(node.expression.name.text) + ) { + return undefined; + } + return { + name: rootIdentifier(node.expression.expression), + target: node.expression.expression, + reason: `\`.${node.expression.name.text}()\``, + }; +} + +/** `delete x.field`. */ +function deletion(node) { + if ( + !ts.isDeleteExpression(node) || + (!ts.isPropertyAccessExpression(node.expression) && + !ts.isElementAccessExpression(node.expression)) + ) { + return undefined; + } + return { + name: rootIdentifier(node.expression), + target: node.expression, + reason: 'field deleted', + }; +} + +/** How `node` changes a binding, if it changes one at all. */ +function mutationIn(node) { + return ( + assignment(node) ?? increment(node) ?? mutatingCall(node) ?? deletion(node) + ); +} + +/** + * An empty collection literal: `new Map()`, `new Set()`, `[]`. A module-scope + * binding initialized to one and *exported* is a registry something fills, and + * the filling is often in another file, which this single-file walk cannot see. + * That is the shipped bug's exact shape, so the emptiness plus the export is + * treated as the signal. A non-empty initializer is a lookup table and is left + * alone. + */ +function isEmptyCollection(node) { + if (!node) return false; + if (ts.isArrayLiteralExpression(node)) return node.elements.length === 0; + if (!ts.isNewExpression(node) || !ts.isIdentifier(node.expression)) { + return false; + } + const collections = new Set(['Map', 'Set', 'WeakMap', 'WeakSet']); + if (!collections.has(node.expression.text)) return false; + return !node.arguments || node.arguments.length === 0; +} + +function isExported(statement) { + return Boolean( + statement.modifiers?.some((m) => m.kind === ts.SyntaxKind.ExportKeyword) + ); +} + +const FUNCTION_LIKE = new Set([ + ts.SyntaxKind.FunctionDeclaration, + ts.SyntaxKind.FunctionExpression, + ts.SyntaxKind.ArrowFunction, + ts.SyntaxKind.MethodDeclaration, + ts.SyntaxKind.Constructor, + ts.SyntaxKind.GetAccessor, + ts.SyntaxKind.SetAccessor, +]); + +/** + * Record how `mutation` changes a declared binding, most specific key first: + * `Registry.transports` before `Registry`, so a class carrying several static + * fields attributes each write to the field that actually took it. Only the + * first sighting of a binding is kept, which is the one the finding cites. + */ +function recordMutation(mutation, declared, mutations) { + if (!mutation) return; + const path = mutation.target ? memberPath(mutation.target) : undefined; + for (const key of [path, mutation.name]) { + if (!key || !declared.has(key)) continue; + if (!mutations.has(key)) mutations.set(key, mutation.reason); + return; + } +} + +function scanFile(file, repoRoot) { + const text = fs.readFileSync(file, 'utf8'); + const source = ts.createSourceFile( + file, + text, + ts.ScriptTarget.ESNext, + /* setParentNodes */ true + ); + + const declared = collectDeclarations(source); + if (declared.size === 0) return []; + const aliases = globalThisAliases(declared); + + /** key -> how it was first seen changing. */ + const mutations = new Map(); + // Only mutations inside a function body count. A write in a top-level + // statement runs once per copy at module evaluation and produces the same + // value in each, so a precomputed lookup table is not the hazard this rule + // is looking for; divergence needs a write that happens later, per request. + const visit = (node, inFunction) => { + if (inFunction) recordMutation(mutationIn(node), declared, mutations); + const nowInFunction = inFunction || FUNCTION_LIKE.has(node.kind); + ts.forEachChild(node, (child) => visit(child, nowInFunction)); + }; + visit(source, false); + + const findings = []; + for (const [key, binding] of declared) { + const initializer = binding.declaration.initializer; + let how = mutations.get(key); + if ( + !how && + isExported(binding.statement) && + isEmptyCollection(initializer) + ) { + how = 'exported empty collection'; + } + if (!how) continue; // never changes: one copy per layer is harmless + if (isGlobalSingletonCall(initializer)) continue; + if (isGlobalThisBacked(initializer, aliases)) continue; + if (perCopyReason(binding.statement, text)) continue; + + const { line } = source.getLineAndCharacterOfPosition( + binding.declaration.getStart(source) + ); + findings.push({ + file: path.relative(repoRoot, file), + line: line + 1, + name: binding.name, + keyword: binding.keyword ?? (binding.isConst ? 'const' : 'let'), + reason: how, + }); + } + return findings; +} + +/** Scan one package directory (the one holding its `package.json`). */ +export function scanPackage(packageDir, repoRoot = process.cwd()) { + const findings = []; + for (const file of walkSourceFiles(path.join(packageDir, 'src'))) { + findings.push(...scanFile(file, repoRoot)); + } + return findings.sort((a, b) => + a.file === b.file ? a.line - b.line : a.file.localeCompare(b.file) + ); +} + +export function formatFindings(findings) { + return findings + .map( + (f) => + `${f.file}:${f.line} ${f.keyword} ${f.name} (${f.reason})\n` + + ' A bundler keys module identity on (resource, layer), so once this\n' + + ' package is bundled one process holds one copy of this module per\n' + + ' layer. Next.js alone builds instrument, app-route, ssr and edge.\n' + + ' This binding is therefore per-copy state, not a process singleton.\n' + + '\n' + + ' Hold it on the World instance if it is per-World, or on globalThis\n' + + ' via globalSingleton() from @workflow/utils if it is process-wide.\n' + + ' If per-copy is what you want, say why:\n' + + ' // per-copy-ok: \n' + + '\n' + + ' Background: packages/utils/src/global-singleton.ts, and\n' + + ' docs/content/worlds/v5/building-a-world.mdx#process-wide-state.' + ) + .join('\n\n'); +} + +const invokedDirectly = + process.argv[1] && import.meta.url === `file://${process.argv[1]}`; + +if (invokedDirectly) { + const packages = process.argv.slice(2); + if (packages.length === 0) { + console.error( + 'usage: node scripts/lint/module-scope-state.mjs [...]' + ); + process.exit(2); + } + let total = 0; + for (const pkg of packages) { + const findings = scanPackage(pkg); + total += findings.length; + console.log(`\n${pkg}: ${findings.length}`); + if (findings.length > 0) console.log(formatFindings(findings)); + } + console.log(`\nTOTAL ${total}`); + process.exit(total === 0 ? 0 : 1); +}