diff --git a/.changeset/retained-serialization-taint.md b/.changeset/retained-serialization-taint.md new file mode 100644 index 0000000000..c7a79c81ea --- /dev/null +++ b/.changeset/retained-serialization-taint.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': minor +'workflow': minor +--- + +Retained-VM boundaries now accept plain data and standard built-ins (`Map`, `Set`, `Date`, `Error`, typed arrays, `URL`, `Headers`, …) as step inputs. Serialization reads through captured intrinsics and reports when it had to execute workflow code (getters, proxies, custom serializers); only those boundaries fall back to ordinary replay. diff --git a/biome.json b/biome.json index 222f843041..009f58f0fd 100644 --- a/biome.json +++ b/biome.json @@ -27,7 +27,8 @@ "!**/.nitro", "!**/.output", "!**/.tanstack", - "!**/routeTree.gen.ts" + "!**/routeTree.gen.ts", + "!packages/core/src/vendor" ], "ignoreUnknown": true }, diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 9007fb7c2a..4209134853 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -81,7 +81,7 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: enabled - Keeps the suspended workflow VM alive across inline steps within one invocation, so each iteration of the inline loop appends only the newly written events instead of replaying the whole event log in a fresh VM. - Suspensions involving hooks, waits, or attributes — and any replay divergence — always fall back to a full replay. -- Step inputs of primitive values remain retainable; anything whose serialization could execute workflow code falls back to replay for that boundary. (Support for plain objects, arrays, and standard built-ins lands in a follow-up.) +- Step inputs made of plain data (objects, arrays, primitives) and standard built-ins (`Map`, `Set`, `Date`, `RegExp`, `Error`, typed arrays, `ArrayBuffer`, `URL`, `Headers`) remain retainable: serialization reads them through captured intrinsics, so patching or polyfilling built-in prototypes does not affect retention. Anything whose serialization executes workflow code — getters, proxies, custom class serializers, step-function closures — falls back to replay for that boundary. - Set `0` or `false` to replay from scratch in a fresh VM on every iteration. ### `WORKFLOW_INLINE_OWNERSHIP` diff --git a/packages/core/e2e/local-build.test.ts b/packages/core/e2e/local-build.test.ts index 880a56b8cc..cf415c10be 100644 --- a/packages/core/e2e/local-build.test.ts +++ b/packages/core/e2e/local-build.test.ts @@ -79,8 +79,10 @@ async function readFileIfExists(filePath: string): Promise { * Projects that use the VercelBuildOutputAPIBuilder and produce ESM step bundles. */ const ESM_STEP_BUNDLE_PROJECTS: Record = { + // Steps are bundled into the combined flow function (there is no separate + // step.func in the Build Output API layout). example: - '.vercel/output/functions/.well-known/workflow/v1/step.func/index.mjs', + '.vercel/output/functions/.well-known/workflow/v1/flow.func/index.mjs', }; const DIAGNOSTICS_MANIFEST_PATHS: Record = { diff --git a/packages/core/package.json b/packages/core/package.json index 42ba989369..09fea2b15d 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -85,7 +85,7 @@ "./_workflow": "./dist/workflow/index.js" }, "scripts": { - "build": "genversion --es6 src/version.ts && tsc", + "build": "genversion --es6 src/version.ts && tsc && cp src/vendor/devalue/LICENSE src/vendor/devalue/README.md dist/vendor/devalue/", "dev": "genversion --es6 src/version.ts && tsc --watch", "clean": "tsc --build --clean && rm -rf dist src/version.ts docs ||:", "test": "cross-env WORKFLOW_TARGET_WORLD=local vitest run src", @@ -107,7 +107,6 @@ "@workflow/world-local": "workspace:*", "@workflow/world-vercel": "workspace:*", "debug": "4.4.3", - "devalue": "5.8.1", "ms": "2.1.3", "nanoid": "5.1.6", "seedrandom": "3.0.5", diff --git a/packages/core/src/retained-vm-loop.test.ts b/packages/core/src/retained-vm-loop.test.ts index d24b59bcc2..9ef6630529 100644 --- a/packages/core/src/retained-vm-loop.test.ts +++ b/packages/core/src/retained-vm-loop.test.ts @@ -100,6 +100,56 @@ const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP" } globalThis.__private_workflows = new Map([["workflow", workflow]]);`; +// Map/Date/typed-array arguments serialize through captured host intrinsics +// (see serialization/operations.ts), so these boundaries stay retainable. +const builtinArgsWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2"); + async function workflow() { + const a = await s1({ index: new Map([["k", 1]]), when: new Date(1234) }); + const b = await s2(new Uint8Array([1, 2, 3])); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + +// The Temporal / core-js pattern: polyfills add new data-valued methods to +// built-in prototypes and constructor statics. Serialization never reads +// them, so retention is unaffected. +const polyfillArgsWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2"); + Date.prototype.toTemporalInstant = function () { return "instant"; }; + Set.prototype.union = function (other) { return new Set([...this, ...other]); }; + Object.groupBy = function () { return {}; }; + async function workflow() { + const a = await s1(new Date(1234)); + const b = await s2(new Set([1, 2])); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + +// Replacing a serialization-relevant member (Date.prototype.toISOString) +// does not affect retention: the Date reducer reads through captured host +// intrinsics (see serialization/operations.ts), so the patched member never +// executes and the serialized bytes stay pristine in both modes. +const patchedDateArgWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2"); + Date.prototype.toISOString = function () { return "patched"; }; + async function workflow() { + const a = await s1(new Date(1234)); + const b = await s2(); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + +const patchedDateStringArgWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2"); + Date.prototype.toISOString = function () { return "patched"; }; + async function workflow() { + const a = await s1(new Date(1234).toISOString()); + const b = await s2(); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + // `crypto.subtle.digest` computes synchronously via node:crypto, so a // digest-using VM stays quiescent at suspension and remains retainable. const digestWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); @@ -319,6 +369,42 @@ describe('retained VM through the inline replay loop', () => { expect(vmBuilds).toBeGreaterThan(1); }); + it('retains boundaries whose args are supported built-ins', async () => { + const { vmBuilds, output } = await drive( + 'wrun_retained_builtins', + builtinArgsWorkflow + ); + expect(output).toBeInstanceOf(Uint8Array); + expect(vmBuilds).toBe(1); + }); + + it('retains boundaries when prototypes carry polyfilled data methods', async () => { + const { vmBuilds, output } = await drive( + 'wrun_retained_polyfill', + polyfillArgsWorkflow + ); + expect(output).toBeInstanceOf(Uint8Array); + expect(vmBuilds).toBe(1); + }); + + it('retains a Date arg even when a serialization member is replaced', async () => { + const { vmBuilds, output } = await drive( + 'wrun_retained_patched_date', + patchedDateArgWorkflow + ); + expect(output).toBeInstanceOf(Uint8Array); + expect(vmBuilds).toBe(1); + }); + + it('retains when the patched type is converted to a string first', async () => { + const { vmBuilds, output } = await drive( + 'wrun_retained_patched_date_string', + patchedDateStringArgWorkflow + ); + expect(output).toBeInstanceOf(Uint8Array); + expect(vmBuilds).toBe(1); + }); + it('retains a VM that used the synchronous crypto.subtle.digest', async () => { const { vmBuilds, output } = await drive( 'wrun_retained_digest', diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 404caea4d2..7352b01ef8 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -397,3 +397,81 @@ describe('handleSuspension', () => { ).toBe(false); }); }); + +describe('retainedStepInputsSafe (serialization passivity gate)', () => { + function stepPending(args: unknown[]) { + return new Map([ + [ + 'step_1', + { + type: 'step' as const, + correlationId: 'step_1', + stepName: 'someStep', + args, + }, + ], + ]); + } + + async function runSuspension(args: unknown[]) { + const eventsCreate = vi + .fn() + .mockImplementation(async (_runId, event) => ({ event })); + const world = createWorld(eventsCreate); + return handleSuspension({ + suspension: new WorkflowSuspension(stepPending(args), globalThis), + world, + run, + }); + } + + it('reports safe for plain data and supported built-ins', async () => { + const result = await runSuspension([ + { nested: [{ ok: true }, 'text', 42n], flag: false }, + new Map([['k', new Set([1])]]), + new Date(1700000000000), + new Uint8Array([1, 2, 3]), + /pattern/gi, + new URL('https://example.com/'), + new Error('recorded failure'), + ]); + expect(result.retainedStepInputsSafe).toBe(true); + }); + + it('reports unsafe when serializing an argument executes a getter', async () => { + const value: Record = {}; + Object.defineProperty(value, 'lazy', { + enumerable: true, + get: () => 'computed', + }); + const result = await runSuspension([{ deep: [value] }]); + expect(result.retainedStepInputsSafe).toBe(false); + }); + + it('reports unsafe when an argument is a proxy', async () => { + const result = await runSuspension([new Proxy({ a: 1 }, {})]); + expect(result.retainedStepInputsSafe).toBe(false); + }); + + it('still serializes tainted inputs successfully (bytes are unaffected)', async () => { + const value: Record = {}; + Object.defineProperty(value, 'lazy', { + enumerable: true, + get: () => 'computed', + }); + const eventsCreate = vi + .fn() + .mockImplementation(async (_runId, event) => ({ event })); + const world = createWorld(eventsCreate); + const result = await handleSuspension({ + suspension: new WorkflowSuspension(stepPending([value]), globalThis), + world, + run, + }); + expect(result.retainedStepInputsSafe).toBe(false); + // The step is still prepared for execution as usual. + expect( + result.lazyInlineSteps.length + result.createdStepCorrelationIds.size + ).toBeGreaterThan(0); + }); +}); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 72f5411f3c..bfbe76ece2 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -26,25 +26,13 @@ import type { WorkflowSuspension, } from '../global.js'; import { runtimeLogger } from '../logger.js'; +import type { SerializationPassivityReport } from '../serialization/operations.js'; import { dehydrateStepArguments } from '../serialization.js'; import * as Attribute from '../telemetry/semantic-conventions.js'; import { getAbortStreamIdFromToken } from '../util.js'; import { getMaxInlineSteps } from './constants.js'; import { type MutableEventLog, withPreconditionRetry } from './helpers.js'; -// Serializing a primitive executes no code of any kind. BigInt is excluded: -// its encoding calls a prototype method. Widened to plain data and standard -// built-ins by the retained-input walker in a follow-up. -function isPrimitiveStepArgument(value: unknown): boolean { - return ( - value === null || - value === undefined || - typeof value === 'boolean' || - typeof value === 'number' || - typeof value === 'string' - ); -} - export interface SuspensionHandlerParams { suspension: WorkflowSuspension; world: World; @@ -134,7 +122,12 @@ export interface SuspensionHandlerResult { * durably creating the user's hooks doesn't count as runtime overhead. */ hookCreationMs: number; - /** Whether every newly serialized step input was passive retained-VM data. */ + /** + * Whether serializing this suspension's new step inputs was passive (did + * not execute workflow-owned code such as getters, proxy traps, or custom + * serializers). `false` means the retained VM may have diverged from what + * a cold replay would compute, so the caller must demote to replay. + */ retainedStepInputsSafe: boolean; } @@ -506,20 +499,18 @@ export async function handleSuspension({ // Serialization always runs through the one ordinary path below, so the // durable bytes cannot depend on retention. What retention needs to know is - // whether that serialization will execute workflow code (getters, hooks, - // patched prototype members) — side effects a cold replay would not repeat. - // For now only primitive arguments are provably passive (serializing them - // executes no code at all); a follow-up widens this to plain data and the - // standard built-ins. If any input in the batch is not provably passive, - // the caller demotes the session so the side effects land in a VM that is - // about to be discarded, exactly like the pre-retention runtime. - const retainedStepInputsSafe = stepItems.every( - (item) => - !stepsNeedingCreation.has(item.correlationId) || - (item.thisVal === undefined && - item.closureVars === undefined && - item.args.every(isPrimitiveStepArgument)) - ); + // whether that serialization *executed* workflow code (getters, proxy + // traps, custom serializers) — side effects a cold replay would not + // repeat, since a replay skips dehydration for already-recorded steps. + // The hardened stringify operations record exactly that into this report + // (see ../serialization/operations.ts); when any input in the batch + // taints it, the caller demotes the session so the side effects land in a + // VM that is about to be discarded, exactly like the pre-retention + // runtime. + const passivityReport: SerializationPassivityReport = { + tainted: false, + reasons: [], + }; // Lazy inline start: defer the step_created write for up to // `getMaxInlineSteps()` steps the caller will run inline (in parallel). Each @@ -564,7 +555,8 @@ export async function handleSuspension({ encryptionKey, suspension.globalThis, false, - compression + compression, + passivityReport ); // Deferred (lazy) inline step: skip the step_created write — the // caller's inline executeStep will send a lazy step_started carrying @@ -706,6 +698,15 @@ export async function handleSuspension({ // step_created and re-dispatches, and recovers the run instead of orphaning it. await Promise.all(ops); + // The step-input dehydrations above have settled, so the report is final. + const retainedStepInputsSafe = !passivityReport.tainted; + if (passivityReport.tainted) { + runtimeLogger.debug( + 'Serializing step inputs executed workflow code; falling back to replay instead of retaining the VM', + { workflowRunId: runId, reasons: passivityReport.reasons } + ); + } + // Rebuild the inline batch in deterministic order. `lazyInlineCorrelationIds` // is a Set seeded from the ordered first-N slice, so iterating it preserves // stepItems order; every id in it was set by the lazy branch above. diff --git a/packages/core/src/serialization-format.test.ts b/packages/core/src/serialization-format.test.ts index e81d45e02f..7f456ec7c4 100644 --- a/packages/core/src/serialization-format.test.ts +++ b/packages/core/src/serialization-format.test.ts @@ -1,4 +1,4 @@ -import { stringify } from 'devalue'; +import { stringify } from './vendor/devalue/index.js'; import { describe, expect, it } from 'vitest'; import { ClassInstanceRef, diff --git a/packages/core/src/serialization-format.ts b/packages/core/src/serialization-format.ts index c2415961b3..52352d29af 100644 --- a/packages/core/src/serialization-format.ts +++ b/packages/core/src/serialization-format.ts @@ -7,7 +7,7 @@ */ import { getEventDataRefFields } from '@workflow/world'; -import { parse, unflatten } from 'devalue'; +import { parse, unflatten } from './vendor/devalue/index.js'; // --------------------------------------------------------------------------- // Format prefix constants and encoding/decoding diff --git a/packages/core/src/serialization.test.ts b/packages/core/src/serialization.test.ts index 5c21c55c3c..75a5d21f71 100644 --- a/packages/core/src/serialization.test.ts +++ b/packages/core/src/serialization.test.ts @@ -4767,7 +4767,7 @@ describe('getDeserializeStream legacy fallback', () => { } it('should parse legacy newline-delimited devalue text', async () => { - const { stringify } = await import('devalue'); + const { stringify } = await import('./vendor/devalue/index.js'); const encoder = new TextEncoder(); const line1 = `${stringify({ hello: 'world' })}\n`; @@ -4781,7 +4781,7 @@ describe('getDeserializeStream legacy fallback', () => { }); it('should parse legacy single-line chunks', async () => { - const { stringify } = await import('devalue'); + const { stringify } = await import('./vendor/devalue/index.js'); const encoder = new TextEncoder(); const chunk1 = encoder.encode(`${stringify('hello')}\n`); diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index ffd5815d01..4e0f3a100d 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -4,7 +4,6 @@ import { WorkflowRuntimeError, } from '@workflow/errors'; import { envNumber } from '@workflow/world'; -import { parse, stringify, unflatten } from 'devalue'; import { monotonicFactory } from 'ulid'; import { decrypt as aesGcmDecrypt, @@ -48,6 +47,17 @@ import { isEncrypted, peekFormatPrefix, } from './serialization/format.js'; +// All devalue stringify calls go through the hardened wrapper so the +// side-effect-free operations (and passivity tainting) apply uniformly — +// see ./serialization/operations.ts. +import { + capturedIntrinsics, + isInstanceOfPrototype, + passiveGet, + type SerializationPassivityReport, + hardenedStringify as stringify, + withPassivityReport, +} from './serialization/operations.js'; import { getClassReducers, getClassRevivers, @@ -87,6 +97,7 @@ import { import * as Attr from './telemetry/semantic-conventions.js'; import { getActiveSpan, getSpanKind, recordElapsedSpan } from './telemetry.js'; import { getAbortStreamId } from './util.js'; +import { parse, unflatten } from './vendor/devalue/index.js'; import { WorkflowAbortSignal } from './workflow/abort-controller.js'; // Re-export types and utilities from the modular serialization modules @@ -1286,9 +1297,31 @@ import type { * Base reducers shared across all serialization boundaries. * Composes: class + step-function + common reducers from the modular modules. */ +/** + * Passively resolve `global[name].prototype` for prototype-chain brand + * checks (see `isInstanceOfPrototype`). Returns undefined when the binding + * is missing or not a function (e.g. the workflow VM's `AbortSignal`, which + * is a plain object) — matching how the old `instanceof` predicates guarded + * on `typeof global[name] === 'function'`. + */ +function resolvePrototype( + global: Record, + name: string +): object | undefined { + const ctor = passiveGet(global, name); + if (typeof ctor !== 'function') return undefined; + return passiveGet(ctor, 'prototype') as object | undefined; +} + function getAllBaseReducers( global: Record = globalThis ): Partial { + // Resolve the realm's constructors once per reducer set. In the workflow + // VM these are custom classes installed by workflow.ts (not host classes), + // so brand checks must go through the passed `global`. + const RequestPrototype = resolvePrototype(global, 'Request'); + const ResponsePrototype = resolvePrototype(global, 'Response'); + // Class/Instance MUST come before Error so that custom Error subclasses // with WORKFLOW_SERIALIZE take precedence (devalue uses first-match-wins). return { @@ -1298,17 +1331,20 @@ function getAllBaseReducers( // Request and Response reducers are mode-specific and added by // getExternalReducers / getWorkflowReducers / getStepReducers below. Request: (value) => { - if (!(value instanceof global.Request)) return false; + if (!isInstanceOfPrototype(value, RequestPrototype)) return false; const data: SerializableSpecial['Request'] = { - method: value.method, - url: value.url, - headers: value.headers, - body: value.body, - duplex: value.duplex, + method: passiveGet(value, 'method') as string, + url: passiveGet(value, 'url') as string, + headers: passiveGet(value, 'headers') as Headers, + body: passiveGet(value, 'body') as ReadableStream | null, + duplex: passiveGet( + value, + 'duplex' + ) as SerializableSpecial['Request']['duplex'], }; - const responseWritable = value[WEBHOOK_RESPONSE_WRITABLE]; + const responseWritable = passiveGet(value, WEBHOOK_RESPONSE_WRITABLE); if (responseWritable) { - data.responseWritable = responseWritable; + data.responseWritable = responseWritable as WritableStream; } // Forward the signal in two cases: // 1. Already aborted — preserve aborted=true/reason so the hydrated @@ -1319,25 +1355,30 @@ function getAllBaseReducers( // Plain non-aborted native signals are intentionally dropped (would // mint stream infra for every Request, including the auto-generated // signal on `new Request(url)`). + const signal = passiveGet(value, 'signal') as + | (AbortSignal & AbortInternals) + | undefined; if ( - value.signal && - (value.signal.aborted || - (value.signal as AbortInternals)[ABORT_STREAM_NAME]) + signal && + (passiveGet(signal, 'aborted') || passiveGet(signal, ABORT_STREAM_NAME)) ) { - data.signal = value.signal; + data.signal = signal; } return data; }, Response: (value) => { - if (!(value instanceof global.Response)) return false; + if (!isInstanceOfPrototype(value, ResponsePrototype)) return false; return { - type: value.type, - url: value.url, - status: value.status, - statusText: value.statusText, - headers: value.headers, - body: value.body, - redirected: value.redirected, + type: passiveGet( + value, + 'type' + ) as SerializableSpecial['Response']['type'], + url: passiveGet(value, 'url') as string, + status: passiveGet(value, 'status') as number, + statusText: passiveGet(value, 'statusText') as string, + headers: passiveGet(value, 'headers') as Headers, + body: passiveGet(value, 'body') as ReadableStream | null, + redirected: passiveGet(value, 'redirected') as boolean, }; }, }; @@ -1426,18 +1467,35 @@ function reduceAbortBySymbol( signal: { aborted: boolean; reason?: unknown }, holder: AbortHolder ): AbortSerializedData | false { - const streamName = - holder[ABORT_STREAM_NAME] ?? holder.signal?.[ABORT_STREAM_NAME]; - const hookToken = - holder[ABORT_HOOK_TOKEN] ?? holder.signal?.[ABORT_HOOK_TOKEN]; + // All reads are passive: the symbols are own data properties on both the + // VM's Workflow* classes and reduced natives, and `aborted`/`reason` are + // either data properties (VM) or the captured native prototype accessors. + // `signal` is the holder itself (AbortSignal reducer) or was already read + // passively from the holder (AbortController reducer), so symbol probes on + // both cover holder[SYM] ?? holder.signal?.[SYM]. + const streamName = (passiveGet(holder, ABORT_STREAM_NAME) ?? + passiveGet(signal as object, ABORT_STREAM_NAME)) as string | undefined; + const hookToken = (passiveGet(holder, ABORT_HOOK_TOKEN) ?? + passiveGet(signal as object, ABORT_HOOK_TOKEN)) as string | undefined; if (!streamName) { throw new Error('AbortController/AbortSignal stream name is not set'); } + const aborted = passiveGet( + signal as object, + 'aborted', + capturedIntrinsics.abortSignalAborted + ) as boolean; return { streamName, hookToken: hookToken!, - aborted: signal.aborted, - reason: signal.aborted ? signal.reason : undefined, + aborted, + reason: aborted + ? passiveGet( + signal as object, + 'reason', + capturedIntrinsics.abortSignalReason + ) + : undefined, }; } @@ -1664,37 +1722,45 @@ export function getExternalReducers( export function getWorkflowReducers( global: Record = globalThis ): Partial { + // In the workflow VM these are custom classes installed by workflow.ts, + // so resolve their prototypes from the passed `global` (passively, once). + const readableStreamPrototype = resolvePrototype(global, 'ReadableStream'); + const writableStreamPrototype = resolvePrototype(global, 'WritableStream'); + const abortControllerPrototype = resolvePrototype(global, 'AbortController'); + const abortSignalPrototype = resolvePrototype(global, 'AbortSignal'); + return { ...getAllBaseReducers(global), // Readable/Writable streams from within the workflow execution environment // are simply "handles" that can be passed around to other steps. ReadableStream: (value) => { - if (!(value instanceof global.ReadableStream)) return false; + if (!isInstanceOfPrototype(value, readableStreamPrototype)) return false; // Check if this is a fake stream storing BodyInit from Request/Response constructor - const bodyInit = value[BODY_INIT_SYMBOL]; + const bodyInit = passiveGet(value, BODY_INIT_SYMBOL); if (bodyInit !== undefined) { // This is a fake stream - serialize the BodyInit directly // devalue will handle serializing strings, Uint8Array, etc. return { bodyInit }; } - const name = value[STREAM_NAME_SYMBOL]; + const name = passiveGet(value, STREAM_NAME_SYMBOL) as string | undefined; if (!name) { throw new WorkflowRuntimeError('ReadableStream `name` is not set'); } const s: SerializableSpecial['ReadableStream'] = { name }; - const type = value[STREAM_TYPE_SYMBOL]; + const type = passiveGet(value, STREAM_TYPE_SYMBOL) as 'bytes' | undefined; if (type) s.type = type; - const framing: ByteStreamFraming | undefined = - value[STREAM_FRAMING_SYMBOL]; + const framing = passiveGet(value, STREAM_FRAMING_SYMBOL) as + | ByteStreamFraming + | undefined; if (framing) s.framing = framing; return s; }, WritableStream: (value) => { - if (!(value instanceof global.WritableStream)) return false; - const name = value[STREAM_NAME_SYMBOL]; + if (!isInstanceOfPrototype(value, writableStreamPrototype)) return false; + const name = passiveGet(value, STREAM_NAME_SYMBOL) as string | undefined; if (!name) { throw new WorkflowRuntimeError('WritableStream `name` is not set'); } @@ -1702,9 +1768,12 @@ export function getWorkflowReducers( // When the handle was forwarded from another run (parent → child // via `start()`), preserve the foreign runId so the step-side // reviver opens the writable against the original stream. - const foreignRunId = value[STREAM_SERVER_RUN_ID_SYMBOL]; + const foreignRunId = passiveGet(value, STREAM_SERVER_RUN_ID_SYMBOL); if (typeof foreignRunId === 'string') s.runId = foreignRunId; - const foreignDeploymentId = value[STREAM_SERVER_DEPLOYMENT_ID_SYMBOL]; + const foreignDeploymentId = passiveGet( + value, + STREAM_SERVER_DEPLOYMENT_ID_SYMBOL + ); if (typeof foreignDeploymentId === 'string') { s.deploymentId = foreignDeploymentId; } @@ -1716,26 +1785,44 @@ export function getWorkflowReducers( // is a plain object (not a class), so instanceof checks won't work for signals. // Detect instances by the presence of the ABORT_STREAM_NAME symbol instead. AbortController: (value) => { - if (!value || !value.signal) return false; + if ( + value === null || + (typeof value !== 'object' && typeof value !== 'function') + ) { + return false; + } + const signal = passiveGet( + value, + 'signal', + capturedIntrinsics.abortControllerSignal + ) as (AbortSignal & AbortInternals) | undefined; + if (!signal) return false; const holder = value as AbortController & AbortHolder; const hasAbortSymbol = - holder[ABORT_STREAM_NAME] ?? holder.signal?.[ABORT_STREAM_NAME]; - const isNativeAbortController = - global.AbortController && - typeof global.AbortController === 'function' && - value instanceof global.AbortController; + passiveGet(holder, ABORT_STREAM_NAME) ?? + passiveGet(signal, ABORT_STREAM_NAME); + const isNativeAbortController = isInstanceOfPrototype( + value, + abortControllerPrototype + ); if (!hasAbortSymbol && !isNativeAbortController) return false; - return reduceAbortBySymbol(value.signal, holder); + return reduceAbortBySymbol(signal, holder); }, AbortSignal: (value) => { - const signal = value as (AbortSignal & AbortInternals) | undefined; - const hasAbortSymbol = signal?.[ABORT_STREAM_NAME]; - const isNativeAbortSignal = - global.AbortSignal && - typeof global.AbortSignal === 'function' && - value instanceof global.AbortSignal; + if ( + value === null || + (typeof value !== 'object' && typeof value !== 'function') + ) { + return false; + } + const signal = value as AbortSignal & AbortInternals; + const hasAbortSymbol = passiveGet(signal, ABORT_STREAM_NAME); + const isNativeAbortSignal = isInstanceOfPrototype( + value, + abortSignalPrototype + ); if (!hasAbortSymbol && !isNativeAbortSignal) return false; - return reduceAbortBySymbol(value, value as AbortHolder); + return reduceAbortBySymbol(signal, value as AbortHolder); }, }; } @@ -3111,19 +3198,31 @@ export async function dehydrateStepArguments( key: CryptoKey | undefined, global: Record = globalThis, v1Compat = false, - compression = false + compression = false, + passivityReport?: SerializationPassivityReport ): Promise { + // Reducer construction resolves prototypes off the workflow global (see + // resolvePrototype) before stringify activates the report, so it must run + // inside the report scope too — a getter or proxy planted on a global + // constructor is value-owned code like any other. if (v1Compat) { - const str = stringify(value, getWorkflowReducers(global)); + const reducers = withPassivityReport(passivityReport, () => + getWorkflowReducers(global) + ); + const str = stringify(value, reducers, passivityReport); return revive(str); } try { const compressionStats: CompressionStats = {}; + const extraReducers = withPassivityReport(passivityReport, () => + getStreamAndRequestReducers(getWorkflowReducers(global)) + ); const result = await stepModule.serialize(value, key, { global, - extraReducers: getStreamAndRequestReducers(getWorkflowReducers(global)), + extraReducers, compression, compressionStats, + passivityReport, }); await recordCompression(compressionStats, 'serialize'); return result; diff --git a/packages/core/src/serialization/byte-corpus.snapshot.json b/packages/core/src/serialization/byte-corpus.snapshot.json new file mode 100644 index 0000000000..e0ddc38fc2 --- /dev/null +++ b/packages/core/src/serialization/byte-corpus.snapshot.json @@ -0,0 +1,60 @@ +{ + "primitives": "[[1,2,3,4,5,6],null,true,false,\"text\",42,-1.5]", + "special numbers": "[[-3,-4,-5,-6]]", + "undefined": "-1", + "bigint": "[{\"big\":1,\"negative\":3},[\"BigInt\",2],\"42\",[\"BigInt\",4],\"-7\"]", + "string escapes": "[[1,2,3,4],\"\\u003Cscript>\",\"line\\nbreak\",\"tab\\there\",\"u2028\\u2028u2029\\u2029\"]", + "nested plain object": "[{\"a\":1,\"nested\":2},1,{\"b\":3,\"c\":6},[4,5],2,3,{\"deep\":7},\"yes\"]", + "null prototype object": "[[\"null\",\"x\",1],1]", + "sparse array": "[[1,-2,2],1,3]", + "very sparse array": "[[-7,100,0,1,99,2],1,2]", + "repeated references": "[{\"first\":1,\"second\":1},{\"shared\":2},true]", + "cyclic object": "[{\"name\":1,\"self\":0},\"cycle\"]", + "date": "[[\"Date\",1],\"2023-11-14T22:13:20.000Z\"]", + "invalid date": "[[\"Date\",1],\".\"]", + "map": "[[\"Map\",1],[2,6],[3,4],\"k\",{\"v\":5},1,[7,8],2,\"two\"]", + "set": "[[\"Set\",1],[2,3,4],1,\"two\",{\"three\":5},3]", + "nested map in set": "[[\"Set\",1],[2],[\"Map\",3],[4],[5,6],\"inner\",1]", + "regexp": "[[\"RegExp\",1],{\"source\":2,\"flags\":3},\"ab+c\",\"gi\"]", + "regexp no flags": "[[\"RegExp\",1],{\"source\":2,\"flags\":3},\"plain\",\"\"]", + "url": "[[\"URL\",1],\"https://example.com/path?q=1#frag\"]", + "url search params": "[[\"URLSearchParams\",1],\"a=1&b=2\"]", + "empty url search params": "[[\"URLSearchParams\",1],\".\"]", + "headers": "[[\"Headers\",1],[2],[3,4],\"content-type\",\"text/plain\"]", + "array buffer": "[[\"ArrayBuffer\",1],\"AQIDBAUGBwg=\"]", + "empty array buffer": "[[\"ArrayBuffer\",1],\".\"]", + "uint8 array": "[[\"Uint8Array\",1],\"AQID\"]", + "uint8 subarray": "[[\"Uint8Array\",1],\"AwQFBg==\"]", + "int16 array": "[[\"Int16Array\",1],\"//8CAP3/\"]", + "float32 array": "[[\"Float32Array\",1],\"AADAPwAAIMA=\"]", + "float64 array": "[[\"Float64Array\",1],\"GC1EVPshCUA=\"]", + "bigint64 array": "[[\"BigInt64Array\",1],\"AQAAAAAAAAD+/////////w==\"]", + "uint8 clamped array": "[[\"Uint8ClampedArray\",1],\"AP8=\"]", + "data view": "[[\"DataView\",1],[\"ArrayBuffer\",2],\"CQgH\"]", + "data view subview": "[[\"DataView\",1,1,undefined],[\"ArrayBuffer\",2],\"AQIDBAUGBwg=\"]", + "node buffer": "[[\"Uint8Array\",1],\"aGk=\"]", + "boxed number": "[[\"Object\",1],42]", + "boxed string": "[[\"Object\",1],\"boxed\"]", + "boxed boolean": "[[\"Object\",1],false]", + "boxed bigint": "[[\"Object\",1],[\"BigInt\",2],\"123\"]", + "plain error": "[[\"Error\",1],{\"name\":2,\"message\":3,\"stack\":4},\"Error\",\"boom\",\"Error: boom\\n at \\u003Ccorpus>\"]", + "error subclasses": "[[1,5,9],[\"TypeError\",2],{\"message\":3,\"stack\":4},\"t\",\"TypeError: t\\n at \\u003Ccorpus>\",[\"RangeError\",6],{\"message\":7,\"stack\":8},\"r\",\"RangeError: r\\n at \\u003Ccorpus>\",[\"SyntaxError\",10],{\"message\":11,\"stack\":12},\"s\",\"SyntaxError: s\\n at \\u003Ccorpus>\"]", + "error with cause chain": "[[\"Error\",1],{\"name\":2,\"message\":3,\"stack\":4,\"cause\":5},\"Error\",\"outer\",\"Error: outer\\n at \\u003Ccorpus>\",[\"TypeError\",6],{\"message\":7,\"stack\":8},\"inner\",\"TypeError: inner\\n at \\u003Ccorpus>\"]", + "error with undefined cause": "[[\"Error\",1],{\"name\":2,\"message\":3,\"stack\":4,\"cause\":-1},\"Error\",\"has undefined cause\",\"Error: has undefined cause\\n at \\u003Ccorpus>\"]", + "custom named error": "[[\"Error\",1],{\"name\":2,\"message\":3,\"stack\":4},\"MyCustomError\",\"custom\",\"MyCustomError: custom\\n at \\u003Ccorpus>\"]", + "fatal error": "[[\"FatalError\",1],{\"message\":2,\"stack\":3},\"fatal\",\"FatalError: fatal\\n at \\u003Ccorpus>\"]", + "retryable error": "[[\"RetryableError\",1],{\"message\":2,\"stack\":3,\"retryAfter\":4},\"try again\",\"RetryableError: try again\\n at \\u003Ccorpus>\",1700000060000]", + "aggregate error": "[[\"AggregateError\",1],{\"message\":2,\"stack\":3,\"errors\":4},\"many\",\"AggregateError: many\\n at \\u003Ccorpus>\",[5],[\"Error\",6],{\"name\":7,\"message\":8,\"stack\":9},\"Error\",\"a\",\"Error: a\\n at \\u003Ccorpus>\"]", + "workflow function reference": "[[\"WorkflowFunction\",1],{\"workflowId\":2},\"workflow//corpus\"]", + "class reference": "[[\"Class\",1],{\"classId\":2},\"corpus/Class\"]", + "custom serialized instance": "[[\"Instance\",1],{\"classId\":2,\"data\":3},\"corpus/Thing\",{\"label\":4},\"hi\"]", + "object with getter": "[{\"plain\":1,\"computed\":2},1,\"from getter\"]", + "proxy over plain object": "[{\"wrapped\":1,\"n\":2},true,2]", + "step argument envelope": "[{\"args\":1,\"closureVars\":-1,\"thisVal\":-1},[2,6],{\"userId\":3,\"when\":4},\"u_1\",[\"Date\",5],\"2023-11-14T22:13:20.000Z\",3]", + "throws: bare function": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization", + "throws: symbol": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization", + "throws: pojo with symbol keys": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization", + "throws: class instance without serializer": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization", + "throws: promise (sync stringify)": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization", + "throws: proto key": "Failed to serialize step arguments\n╰▶ hint: Ensure you're passing workflow serializable types. Check the serialization docs to see what's serializable: https://workflow-sdk.dev/docs/foundations/serialization" +} diff --git a/packages/core/src/serialization/byte-corpus.ts b/packages/core/src/serialization/byte-corpus.ts new file mode 100644 index 0000000000..9e4162e8b4 --- /dev/null +++ b/packages/core/src/serialization/byte-corpus.ts @@ -0,0 +1,233 @@ +/** + * Shared corpus for the serialization byte-stability test. + * + * Each entry builds a value whose serialized devalue string is pinned in + * `byte-corpus.snapshot.json`. The snapshot was recorded from the + * serialization code as it existed BEFORE the hardened stringify operations + * were introduced (see operations.ts), so the accompanying test proves the + * hardening did not change a single byte of the wire format. + * + * Regenerating the snapshot is an explicit act: it means "I intend to change + * the wire format". See byte-stability.test.ts for how. + */ + +import { FatalError, RetryableError } from '@workflow/errors'; +import { WORKFLOW_SERIALIZE } from '@workflow/serde'; + +/** An error with a deterministic stack, safe to pin in a snapshot. */ +function fixedError(error: T, stack: string): T { + error.stack = stack; + return error; +} + +export class CorpusSerializableThing { + static classId = 'corpus/Thing'; + static [WORKFLOW_SERIALIZE](instance: CorpusSerializableThing) { + return { label: instance.label }; + } + constructor(public label: string) {} +} + +export function buildByteCorpus(): Array<{ name: string; value: unknown }> { + const shared = { shared: true }; + const cyclic: Record = { name: 'cycle' }; + cyclic.self = cyclic; + + const buffer = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]).buffer; + + const causeChain = fixedError( + new Error('outer', { + cause: fixedError( + new TypeError('inner'), + 'TypeError: inner\n at ' + ), + }), + 'Error: outer\n at ' + ); + + const retryable = fixedError( + new RetryableError('try again', { + retryAfter: new Date(1_700_000_060_000), + }), + 'RetryableError: try again\n at ' + ); + + const aggregate = fixedError( + new AggregateError( + [fixedError(new Error('a'), 'Error: a\n at ')], + 'many' + ), + 'AggregateError: many\n at ' + ); + + return [ + // ---- primitives & special numbers ---- + { name: 'primitives', value: [null, true, false, 'text', 42, -1.5] }, + { name: 'special numbers', value: [Number.NaN, Infinity, -Infinity, -0] }, + { name: 'undefined', value: undefined }, + { name: 'bigint', value: { big: 42n, negative: -7n } }, + { + name: 'string escapes', + value: ['