diff --git a/.changeset/catchable-step-arg-serialization-errors.md b/.changeset/catchable-step-arg-serialization-errors.md new file mode 100644 index 0000000000..4b4cdb6b6c --- /dev/null +++ b/.changeset/catchable-step-arg-serialization-errors.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Step-argument serialization failures now fail the step with a catchable error (via a `step_failed` event, like a step-body failure) instead of failing the run from outside the workflow, and when uncaught they fail the run immediately as a `USER_ERROR` rather than retrying until max queue deliveries. diff --git a/docs/content/docs/errors/serialization-failed.mdx b/docs/content/docs/errors/serialization-failed.mdx index 5d3d97c1f3..14fb82baa5 100644 --- a/docs/content/docs/errors/serialization-failed.mdx +++ b/docs/content/docs/errors/serialization-failed.mdx @@ -25,6 +25,34 @@ This error can appear when: - Serializing step arguments - Serializing step return values +## Where the Error Surfaces + +Where you observe the failure depends on which boundary it crosses: + +- **Workflow arguments** — `start()` throws synchronously in your application code. +- **Step arguments** — the *step* fails, exactly like a step whose body threw a `FatalError`: no retries (the failure is deterministic), and a `try/catch` around the step call in your workflow code observes it. The step's recorded input shows `[input unavailable: step argument serialization failed]`, since the real arguments are precisely what refused to serialize. +- **Workflow return values** — the workflow body has already returned, so nothing can catch it; the run fails. + +```typescript lineNumbers +async function stepWithBadArguments(value: unknown) { + "use step"; + return value; +} + +export async function processWorkflow(someValue: unknown) { + "use workflow"; + + try { + await stepWithBadArguments(someValue); + } catch (err) { + // (err as Error).message starts with + // "Failed to serialize step arguments" + } +} +``` + +Uncaught, the error propagates out of the workflow body and the run fails immediately with the error code `USER_ERROR` — it does not retry. + ## Why This Happens Workflows persist their state using an event log. Every value that crosses execution boundaries must be: diff --git a/docs/content/docs/foundations/errors-and-retries.mdx b/docs/content/docs/foundations/errors-and-retries.mdx index 2355fc640c..fc6c309765 100644 --- a/docs/content/docs/foundations/errors-and-retries.mdx +++ b/docs/content/docs/foundations/errors-and-retries.mdx @@ -139,6 +139,31 @@ callApi.maxRetries = 5; // Retry up to 5 times on failure (6 total attempts) step can run up to 4 times total (1 initial attempt + 3 retries). +## Serialization Failures + +A step whose arguments cannot be [serialized](/docs/foundations/serialization) fails like a step whose body threw a `FatalError`: the failure is deterministic, so it skips the retry loop, and a `try/catch` around the step call observes it: + +```typescript lineNumbers +async function someStep(input: unknown) { + "use step"; + return input; +} + +export async function myWorkflow(input: unknown) { + "use workflow"; + + try { + await someStep(input); + } catch (err) { + if ((err as Error).message.startsWith("Failed to serialize step arguments")) { + // e.g. `Failed to serialize step arguments at path "..."` + } + } +} +``` + +Uncaught, the run fails immediately with the `USER_ERROR` code — without retrying. See [serialization-failed](/docs/errors/serialization-failed) for common causes and fixes. + ## Error Codes When a workflow run fails, the error may include a `code` that classifies the failure. You can access it programmatically via the `Run` class: diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index c40e3300bc..8f4c8dda6a 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1472,6 +1472,68 @@ describe('e2e', () => { ); }); + describe('serialization failures', () => { + test( + 'step-argument serialization failure is catchable in workflow code', + { timeout: 60_000 }, + async () => { + // Passing an unserializable value (a class instance with no serde + // model) to a step must fail THAT STEP — step_created + + // step_failed — not the whole run, so a try/catch around the step + // call observes the serialization error. + const run = await start( + await e2e('serializationErrorStepArgsCaught'), + [] + ); + const result = await run.returnValue; + + expect(result.caught).toBe(true); + expect(result.messageIncludesStepArguments).toBe(true); + + // The workflow completed (the error was caught) … + const { json: runData } = await cliInspectJson(`runs ${run.runId}`); + expect(runData.status).toBe('completed'); + + // … and the step itself is recorded as failed. + const { json: steps } = await cliInspectJson( + `steps --runId ${run.runId}` + ); + const step = steps.find((s: any) => + s.stepName.includes('acceptAnyValue') + ); + expect(step.status).toBe('failed'); + } + ); + + test( + 'uncaught step-argument serialization failure fails the run as USER_ERROR without redelivery retries', + { timeout: 60_000 }, + async () => { + // Regression coverage for the production failure mode where a + // step-argument serialization error caused the run to redeliver + // until "exceeded max deliveries (49/48)". The run must fail + // promptly (well within this test's timeout — 48 redeliveries + // with backoff would take many minutes) and classify as + // USER_ERROR, not MAX_DELIVERIES_EXCEEDED. + const run = await start( + await e2e('serializationErrorStepArgsUncaught'), + [] + ); + const error = await run.returnValue.catch((e: unknown) => e); + + expect(WorkflowRunFailedError.is(error)).toBe(true); + assert(WorkflowRunFailedError.is(error)); + expect(error.cause.code).toBe('USER_ERROR'); + expect(error.cause.message).toContain( + 'Failed to serialize step arguments' + ); + + const { json: runData } = await cliInspectJson(`runs ${run.runId}`); + expect(runData.status).toBe('failed'); + } + ); + }); + describe('not registered', () => { test( 'WorkflowNotRegisteredError fails the run when workflow does not exist', diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index 13476ab0be..2b0830d60a 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -109,6 +109,9 @@ vi.mock('../serialization.js', () => ({ dehydrateStepReturnValue: vi .fn() .mockResolvedValue(new Uint8Array([1, 2, 3])), + formatSerializationError: vi.fn( + (context: string) => `Failed to serialize ${context}.` + ), })); // Mock context storage @@ -138,6 +141,7 @@ vi.mock('../private.js', () => ({ // which populates capturedHandlerRef import './step-handler.js'; import { getStepFunction } from '../private.js'; +import { hydrateStepArguments } from '../serialization.js'; import { getErrorName, getErrorStack, @@ -148,6 +152,10 @@ import { resetPortCacheForTesting, setPortResolverForTesting, } from './get-port-lazy.js'; +import { + UNSERIALIZABLE_STEP_INPUT_MARKER, + unserializableStepInputPlaceholder, +} from './unserializable-step.js'; import { getWorld } from './world.js'; const mockPortResolver = vi.fn(async () => 3000); @@ -729,3 +737,94 @@ describe('step-handler step not found', () => { expect(mockQueueMessage).not.toHaveBeenCalled(); }); }); + +describe('step-handler unserializable-argument placeholder recovery', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getStepFunction).mockReturnValue(mockStepFn); + vi.mocked(normalizeUnknownError).mockImplementation( + async (err: unknown) => ({ + message: err instanceof Error ? err.message : String(err), + name: err instanceof Error ? err.name : 'Error', + stack: err instanceof Error ? err.stack : undefined, + }) + ); + vi.mocked(getErrorName).mockReturnValue('FatalError'); + vi.mocked(getErrorStack).mockReturnValue(''); + mockStepFn.mockReset().mockResolvedValue('step-result'); + mockStepFn.maxRetries = 3; + mockQueueMessage.mockResolvedValue(undefined); + vi.mocked(getWorld).mockReturnValue({ + events: { create: mockEventsCreate }, + queue: mockQueue, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as any); + mockEventsCreate.mockReset().mockResolvedValue({ + step: { + stepId: 'step_abc', + status: 'running', + attempt: 1, + startedAt: new Date(), + input: [], + }, + event: {}, + }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + vi.mocked(hydrateStepArguments).mockResolvedValue({ + args: [], + thisVal: null, + closureVars: undefined, + } as any); + }); + + it('fails a placeholder-input step without running the step body', async () => { + // A crash between finalization's two durable writes leaves a lone + // step_created carrying the placeholder; crash recovery dispatches it + // here. The body must not run — the intended failure completes instead. + vi.mocked(hydrateStepArguments).mockResolvedValue( + unserializableStepInputPlaceholder() as any + ); + + const result = await capturedHandler( + createMessage(), + createMetadata('acceptAnyValue') + ); + + expect(result).toBeUndefined(); + expect(mockStepFn).not.toHaveBeenCalled(); + expect(mockEventsCreate).toHaveBeenCalledWith( + 'wrun_test123', + expect.objectContaining({ + eventType: 'step_failed', + eventData: expect.objectContaining({ + error: expect.stringContaining('Failed to serialize step arguments'), + }), + }), + expect.anything() + ); + // Fatal: no step_retrying, so no queue attempts are burned. + expect(mockEventsCreate).not.toHaveBeenCalledWith( + 'wrun_test123', + expect.objectContaining({ eventType: 'step_retrying' }), + expect.anything() + ); + }); + + it('does not trip on a genuine argument equal to the display marker', async () => { + // The structural flag — not the human-readable marker string — is what + // identifies the placeholder, so a step legitimately called with the + // marker text still runs. + vi.mocked(hydrateStepArguments).mockResolvedValue({ + args: [UNSERIALIZABLE_STEP_INPUT_MARKER], + thisVal: null, + closureVars: undefined, + } as any); + + await capturedHandler(createMessage(), createMetadata('acceptAnyValue')); + + expect(mockStepFn).toHaveBeenCalledWith(UNSERIALIZABLE_STEP_INPUT_MARKER); + }); +}); diff --git a/packages/core/src/runtime/step-handler.ts b/packages/core/src/runtime/step-handler.ts index c0efefd524..db13fcfa01 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -22,6 +22,7 @@ import { runtimeLogger, stepLogger } from '../logger.js'; import { getStepFunction } from '../private.js'; import { dehydrateStepReturnValue, + formatSerializationError, hydrateStepArguments, } from '../serialization.js'; import { contextStorage } from '../step/context-storage.js'; @@ -49,6 +50,7 @@ import { queueMessage, withHealthCheck, } from './helpers.js'; +import { isUnserializableStepInputPlaceholder } from './unserializable-step.js'; import { safeWaitUntil } from './wait-until.js'; import { getWorld, getWorldHandlers } from './world.js'; @@ -533,6 +535,21 @@ const stepHandler = createQueueHandler( const executionStartTime = Date.now(); try { + // Finalization of an unserializable-argument step writes + // step_created (placeholder input) and step_failed as two + // separate durable writes (see finalizeUnserializableStep in + // suspension-handler.ts). A crash or transient failure between + // them leaves this step pending with the placeholder stored as + // its input, and normal crash recovery then dispatches it here. + // NEVER run user code with placeholder arguments — complete the + // intended failure instead. FatalError skips the retry loop + // below and writes step_failed, exactly what the interrupted + // finalization was about to do. + if (isUnserializableStepInputPlaceholder(hydratedInput)) { + throw new FatalError( + formatSerializationError('step arguments', undefined) + ); + } result = await trace('step.execute', {}, async () => { return await contextStorage.run( { diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 2c391479d5..3bc9c49525 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -1,7 +1,13 @@ +import { EntityConflictError, RunExpiredError } from '@workflow/errors'; import type { WorkflowRun, World } from '@workflow/world'; import { describe, expect, it, vi } from 'vitest'; import { WorkflowSuspension } from '../global.js'; +import { hydrateStepArguments } from '../serialization.js'; import { handleSuspension } from './suspension-handler.js'; +import { + isUnserializableStepInputPlaceholder, + UNSERIALIZABLE_STEP_INPUT_MARKER, +} from './unserializable-step.js'; vi.mock('../version.js', () => ({ version: '0.0.0-test' })); @@ -144,3 +150,209 @@ describe('handleSuspension', () => { expect(result.timeoutSeconds).toBeUndefined(); }); }); + +describe('step-argument serialization failure', () => { + // A value the workflow serializer cannot dehydrate: a class instance with + // no registered serde model. Mirrors serialization.test.ts's unsupported + // type coverage — dehydrateStepArguments throws. + class Unserializable { + secret = 'not-a-pojo'; + } + + function stepItem(id: string, args: unknown[] = []) { + return { + type: 'step' as const, + correlationId: id, + stepName: id, + args, + }; + } + + it('finalizes the step as step_created + step_failed instead of rejecting the suspension', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + // The suspension itself resolves — the failure is scoped to the step. + expect(eventsCreate).toHaveBeenCalledTimes(2); + const [createdCall, failedCall] = eventsCreate.mock.calls; + expect(createdCall[1]).toMatchObject({ + eventType: 'step_created', + correlationId: 's_bad', + eventData: expect.objectContaining({ + stepName: 's_bad', + workflowName: run.workflowName, + }), + }); + expect(failedCall[1]).toMatchObject({ + eventType: 'step_failed', + correlationId: 's_bad', + eventData: expect.objectContaining({ + stepName: 's_bad', + error: expect.stringContaining('Failed to serialize step arguments'), + }), + }); + expect([...(result.failedStepCorrelationIds ?? [])]).toEqual(['s_bad']); + // The step never runs, so no execution message is dispatched — the + // handler instead schedules the immediate replay that observes the + // step_failed event. + expect(world.queue).not.toHaveBeenCalled(); + expect(result.timeoutSeconds).toBe(0); + }); + + it('writes a recoverable placeholder input, not a genuine-looking empty input', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ]); + + await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + const createdEvent = eventsCreate.mock.calls[0][1]; + expect(createdEvent.eventType).toBe('step_created'); + const hydrated = await hydrateStepArguments( + createdEvent.eventData.input, + run.runId, + undefined, + [] + ); + expect(isUnserializableStepInputPlaceholder(hydrated)).toBe(true); + expect((hydrated as { args: unknown[] }).args).toEqual([ + UNSERIALIZABLE_STEP_INPUT_MARKER, + ]); + }); + + it('finalizes the bad step while healthy siblings are still queued', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ['s_good', stepItem('s_good', ['fine'])], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + expect([...(result.failedStepCorrelationIds ?? [])]).toEqual(['s_bad']); + expect( + eventsCreate.mock.calls.map(([, event]) => [ + event.eventType, + event.correlationId, + ]) + ).toEqual( + expect.arrayContaining([ + ['step_created', 's_bad'], + ['step_failed', 's_bad'], + ['step_created', 's_good'], + ]) + ); + expect(world.queue).toHaveBeenCalledTimes(1); + expect(world.queue).toHaveBeenCalledWith( + '__wkf_step_s_good', + expect.objectContaining({ stepId: 's_good' }), + expect.anything() + ); + }); + + it('tolerates a concurrent handler having already finalized the step', async () => { + // Both writes conflict: a concurrent replay hit the same deterministic + // serialization failure and wrote step_created + step_failed first. + const eventsCreate = vi + .fn() + .mockRejectedValue(new EntityConflictError('already exists')); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + expect([...(result.failedStepCorrelationIds ?? [])]).toEqual(['s_bad']); + }); + + it('skips finalization when the run has already finished', async () => { + const eventsCreate = vi + .fn() + .mockRejectedValue(new RunExpiredError('run is gone')); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + // Nothing to observe the failure — no replay is forced. + expect(result.failedStepCorrelationIds?.size ?? 0).toBe(0); + expect(result.timeoutSeconds).toBeUndefined(); + }); + + it('rejects the suspension when step_failed cannot be written after step_created landed', async () => { + // The two finalization writes are separate durable writes. If the second + // fails transiently, the suspension must reject so the message + // redelivers — leaving a lone placeholder step_created behind. Recovery + // for that window lives in the step handler: the placeholder carries a + // structural flag (see unserializable-step.ts) that the handler + // completes as the intended step_failed instead of running user code + // with placeholder arguments. + const writeError = new Error('storage unavailable'); + const eventsCreate = vi + .fn() + .mockImplementationOnce(async (_runId, event) => ({ event })) + .mockRejectedValueOnce(writeError); + const world = createWorld(eventsCreate); + const pending = new Map([ + ['s_bad', stepItem('s_bad', [new Unserializable()])], + ]); + + await expect( + handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }) + ).rejects.toBe(writeError); + + // The lone step_created that redelivery will find carries the + // recoverable placeholder. + expect(eventsCreate).toHaveBeenCalledTimes(2); + const createdEvent = eventsCreate.mock.calls[0][1]; + expect(createdEvent.eventType).toBe('step_created'); + const hydrated = await hydrateStepArguments( + createdEvent.eventData.input, + run.runId, + undefined, + [] + ); + expect(isUnserializableStepInputPlaceholder(hydrated)).toBe(true); + }); +}); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 20f4a32eae..aa9b9e9a6c 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -3,6 +3,7 @@ import { EntityConflictError, HookNotFoundError, RunExpiredError, + RuntimeDecryptionError, } from '@workflow/errors'; import { type CreateEventParams, @@ -29,6 +30,7 @@ import { queueMessage, withPreconditionRetry, } from './helpers.js'; +import { unserializableStepInputPlaceholder } from './unserializable-step.js'; /** * Extracts W3C trace context headers from a trace carrier for HTTP propagation. @@ -65,6 +67,18 @@ export interface SuspensionHandlerParams { export interface SuspensionHandlerResult { timeoutSeconds?: number; + /** + * Correlation IDs of steps whose arguments failed to serialize. Each was + * finalized here as `step_created` (with a placeholder input — the real + * input is precisely what refused to serialize) followed by `step_failed` + * carrying the serialization error, so the next replay rejects the step's + * promise and a try/catch around the step call observes the error — + * exactly like a step-body failure. No step-execution message is + * dispatched for these, so an immediate re-invocation is scheduled + * (`timeoutSeconds: 0`): when the failed step was the only pending work, + * nothing else would ever wake the run to observe the terminal event. + */ + failedStepCorrelationIds?: Set; } async function createHookEvent({ @@ -300,6 +314,113 @@ export async function handleSuspension({ .map((queueItem) => queueItem.correlationId) ); + // Correlation IDs of steps finalized as failed because their arguments + // refused to serialize — see finalizeUnserializableStep below. + const failedStepCorrelationIds = new Set(); + + /** + * A step whose arguments fail to serialize is deterministic: every replay + * re-derives the same unserializable value, so redelivering the + * orchestrator message can never succeed. Instead of rejecting the whole + * suspension (which fails the run from the outside, where no user code can + * observe it), treat it exactly like a step-body failure: write + * `step_created` with a placeholder input (every World requires the step + * entity to exist before a terminal step event, and the real input is + * precisely what refused to serialize) followed by `step_failed` carrying + * the serialization error. The next replay rejects the step's promise with + * it, so a try/catch around the step call observes the error; uncaught, it + * propagates out of the workflow body and fails the run as a USER_ERROR — + * without burning queue redeliveries either way. + */ + const finalizeUnserializableStep = async ( + queueItem: StepInvocationQueueItem, + error: Error + ): Promise => { + runtimeLogger.warn( + 'Step arguments failed to serialize; failing the step so the ' + + 'workflow can observe the error', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + stepName: queueItem.stepName, + error: error.message, + } + ); + // Marker placeholder (not empty args): byte-identical-to-zero-args would + // make `workflow inspect steps` show "no arguments" for the one step + // whose entire problem was its arguments. + const placeholderInput = (await dehydrateStepArguments( + unserializableStepInputPlaceholder(), + runId, + encryptionKey, + suspension.globalThis + )) as SerializedData; + try { + await createGuarded( + { + eventType: 'step_created' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: queueItem.correlationId, + eventData: { + stepName: queueItem.stepName, + workflowName: run.workflowName, + input: placeholderInput, + }, + }, + { requestId } + ); + } catch (createErr) { + if (EntityConflictError.is(createErr)) { + // A concurrent handler already created the step — the failure is + // deterministic, so it is racing toward the same step_failed below. + runtimeLogger.info('Step already exists, continuing', { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: createErr.message, + }); + } else if (RunExpiredError.is(createErr)) { + // Run already finished — nothing to observe the failure. + return; + } else { + throw createErr; + } + } + try { + await createGuarded( + { + eventType: 'step_failed' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: queueItem.correlationId, + eventData: { + stepName: queueItem.stepName, + // The message carries the framed serialization hint; the step + // consumer rejects the step's promise with a FatalError built + // from it, exactly like any other step failure. + error: error.message, + stack: error.stack, + }, + }, + { requestId } + ); + } catch (failErr) { + if (EntityConflictError.is(failErr) || RunExpiredError.is(failErr)) { + // Step already terminal (a concurrent handler wrote the same + // deterministic failure) or the run already finished. + runtimeLogger.info( + 'Tried failing step, but step or run has already finished.', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: failErr.message, + } + ); + } else { + throw failErr; + } + } + failedStepCorrelationIds.add(queueItem.correlationId); + }; + // Process steps and waits in parallel // Each step: create event (if needed) -> queue message // Each wait: create event (if needed) @@ -311,16 +432,32 @@ export async function handleSuspension({ (async () => { // Create step event if not already created if (stepsNeedingCreation.has(queueItem.correlationId)) { - const dehydratedInput = await dehydrateStepArguments( - { - args: queueItem.args, - closureVars: queueItem.closureVars, - thisVal: queueItem.thisVal, - }, - runId, - encryptionKey, - suspension.globalThis - ); + let dehydratedInput: Uint8Array | unknown; + try { + dehydratedInput = await dehydrateStepArguments( + { + args: queueItem.args, + closureVars: queueItem.closureVars, + thisVal: queueItem.thisVal, + }, + runId, + encryptionKey, + suspension.globalThis + ); + } catch (err) { + if (RuntimeDecryptionError.is(err)) { + // An SDK fault, not a user value problem. Keep its identity + // (RUNTIME_ERROR) and current fail-the-suspension behavior. + throw err; + } + // Every other failure out of dehydrateStepArguments is a + // serialization failure on the step's own arguments. + await finalizeUnserializableStep( + queueItem, + err instanceof Error ? err : new Error(String(err)) + ); + return; + } const stepEvent: CreateEventRequest = { eventType: 'step_created' as const, specVersion: SPEC_VERSION_CURRENT, @@ -432,8 +569,26 @@ export async function handleSuspension({ ...Attribute.WorkflowStepsCreated(stepItems.length), ...Attribute.WorkflowHooksCreated(hooksNeedingCreation.length), ...Attribute.WorkflowWaitsCreated(waitItems.length), + ...(failedStepCorrelationIds.size > 0 + ? Attribute.WorkflowStepsFailedSerialization( + failedStepCorrelationIds.size + ) + : {}), }); + // Steps whose arguments failed to serialize were finalized above as + // step_created + step_failed (see finalizeUnserializableStep). No + // step-execution message is dispatched for them, so when such a step is + // the only pending work nothing would ever wake the run again — schedule + // an immediate re-invocation. The replay rejects the step's promise with + // the serialization error, which a try/catch around the step call + // observes; uncaught, it propagates out of the workflow body and fails + // the run as a USER_ERROR. Healthy sibling steps were already queued + // above, so they execute in parallel invocations regardless. + if (failedStepCorrelationIds.size > 0) { + return { timeoutSeconds: 0, failedStepCorrelationIds }; + } + // If any hook conflicts occurred, re-enqueue the workflow immediately // On the next iteration, the hook consumer will see the hook_conflict event // and reject the promise with a WorkflowRuntimeError diff --git a/packages/core/src/runtime/unserializable-step.ts b/packages/core/src/runtime/unserializable-step.ts new file mode 100644 index 0000000000..c3fc642dc8 --- /dev/null +++ b/packages/core/src/runtime/unserializable-step.ts @@ -0,0 +1,59 @@ +/** + * Shared shape for finalizing a step whose arguments failed to serialize + * (used by both the node:vm suspension handler and the QuickJS entrypoint). + * + * The world requires a `step_created` before any terminal step event, and + * the step's real input is precisely what refused to serialize — so the + * finalization writes a placeholder input. The marker string makes the + * placeholder distinguishable from a genuine zero-argument step in + * `workflow inspect steps` and the observability UI: a reader sees + * "input unavailable" instead of "no arguments". + */ +export const UNSERIALIZABLE_STEP_INPUT_MARKER = + '[input unavailable: step argument serialization failed]'; + +/** + * Structural discriminator on the placeholder's top level. The + * `{ args, closureVars, thisVal }` triple is built by the SDK — user code + * never controls its top-level keys — so this flag cannot false-positive on + * a legitimate input, unlike the display marker inside `args`. + */ +const UNSERIALIZABLE_FLAG = '__workflowUnserializableStepInput'; + +/** + * The placeholder value serialized into the failed step's `step_created` + * input. Matches the `{ args, closureVars, thisVal }` triple + * `dehydrateStepArguments` / the QuickJS bootstrap produce for real steps, + * so every consumer hydrates it uniformly, plus the structural flag the + * step executor checks before running user code (see + * {@link isUnserializableStepInputPlaceholder}). + */ +export function unserializableStepInputPlaceholder(): Record { + return { + args: [UNSERIALIZABLE_STEP_INPUT_MARKER], + closureVars: [], + thisVal: undefined, + [UNSERIALIZABLE_FLAG]: true, + }; +} + +/** + * Whether a hydrated step input is the finalization placeholder. + * + * Finalization writes `step_created` (placeholder) and `step_failed` as two + * separate durable writes; a crash or transient failure between them leaves + * a pending step whose stored input is the placeholder. Redelivery then + * dispatches that step through normal crash recovery — the executor calls + * this before running user code and completes the intended failure (a fatal + * SerializationError → `step_failed`) instead of silently invoking the step + * body with placeholder arguments. + */ +export function isUnserializableStepInputPlaceholder( + hydratedInput: unknown +): boolean { + return ( + typeof hydratedInput === 'object' && + hydratedInput !== null && + (hydratedInput as Record)[UNSERIALIZABLE_FLAG] === true + ); +} diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 12b133bfdc..6a027c7ac2 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -213,7 +213,10 @@ const defaultUlid = monotonicFactory(); * Extracts path, value, and reason from devalue's DevalueError when available. * Logs the problematic value to the console for better debugging. */ -function formatSerializationError(context: string, error: unknown): string { +export function formatSerializationError( + context: string, + error: unknown +): string { // Use "returning" for return values, "passing" for arguments/inputs const verb = context.includes('return value') ? 'returning' : 'passing'; diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index fcc5d0694f..67e27094d3 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -122,6 +122,15 @@ export const WorkflowWaitsCreated = SemanticConvention( 'workflow.waits.created' ); +/** + * Number of steps this suspension finalized as failed because their + * arguments refused to serialize (step_created placeholder + step_failed + * carrying the serialization error — see finalizeUnserializableStep). + */ +export const WorkflowStepsFailedSerialization = SemanticConvention( + 'workflow.steps.failed_serialization' +); + // Step attributes /** Name of the step function being executed */ diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index ac12fcab0f..432c2002a7 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -1209,6 +1209,54 @@ export async function errorFatalCatchable() { } } +// --- + +/** + * A class instance with no registered serde model cannot cross the + * workflow/step boundary — the serializer rejects non-POJO instances. + * Used by the serialization-error tests below. + */ +class UnserializableValue { + secret = 'not-serializable'; +} + +async function acceptAnyValue(value: unknown) { + 'use step'; + return { received: value !== undefined }; +} + +/** + * Test: step ARGUMENTS that cannot be serialized. The suspension handler + * fails the step (step_created + step_failed) instead of failing the run + * from the outside, so a try/catch around the step call observes the + * serialization error — same shape as catching a step-body failure. + */ +export async function serializationErrorStepArgsCaught() { + 'use workflow'; + try { + await acceptAnyValue(new UnserializableValue()); + return { caught: false } as any; + } catch (err: any) { + return { + caught: true, + messageIncludesStepArguments: + typeof err?.message === 'string' && + err.message.includes('Failed to serialize step arguments'), + }; + } +} + +/** + * Test: uncaught step-argument serialization failure fails the run as a + * fatal USER_ERROR immediately — no queue-redelivery retry loop. + */ +export async function serializationErrorStepArgsUncaught() { + 'use workflow'; + // Don't catch — the serialization error propagates and fails the run. + await acceptAnyValue(new UnserializableValue()); + return { caught: false }; +} + // ------------------------------------------------------------ // SECTION 4: NOT REGISTERED ERRORS // Tests for step/workflow not registered in the current deployment