diff --git a/.changeset/fail-cross-deployment-runs.md b/.changeset/fail-cross-deployment-runs.md new file mode 100644 index 0000000000..37e2570aea --- /dev/null +++ b/.changeset/fail-cross-deployment-runs.md @@ -0,0 +1,6 @@ +--- +"@workflow/core": patch +"@workflow/errors": patch +--- + +Fail a run with the new `DEPLOYMENT_MISMATCH` error code when queue delivery reaches a deployment other than the one that created it, instead of exhausting retries on an undecryptable step. The failure is recorded without the origin deployment's key so it works even after that deployment is gone. diff --git a/packages/core/src/classify-error.test.ts b/packages/core/src/classify-error.test.ts index a2f202c845..bfb3cae683 100644 --- a/packages/core/src/classify-error.test.ts +++ b/packages/core/src/classify-error.test.ts @@ -6,6 +6,7 @@ import { RuntimeDecryptionError, ThrottleError, TooEarlyError, + WorkflowDeploymentMismatchError, WorkflowNotRegisteredError, WorkflowRuntimeError, WorkflowWorldError, @@ -42,6 +43,18 @@ describe('classifyRunError', () => { ); }); + it('classifies WorkflowDeploymentMismatchError as DEPLOYMENT_MISMATCH', () => { + expect( + classifyRunError( + new WorkflowDeploymentMismatchError( + 'wrun_test', + 'dpl_expected', + 'dpl_actual' + ) + ) + ).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH); + }); + it('classifies plain Error as USER_ERROR', () => { expect(classifyRunError(new Error('user code broke'))).toBe( RUN_ERROR_CODES.USER_ERROR diff --git a/packages/core/src/classify-error.ts b/packages/core/src/classify-error.ts index cc21416d02..838d1ff92a 100644 --- a/packages/core/src/classify-error.ts +++ b/packages/core/src/classify-error.ts @@ -6,6 +6,7 @@ import { RuntimeDecryptionError, StepNotRegisteredError, ThrottleError, + WorkflowDeploymentMismatchError, WorkflowNotRegisteredError, WorkflowRuntimeError, WorkflowWorldError, @@ -116,6 +117,12 @@ export function classifyRunError(err: unknown): RunErrorCode { return RUN_ERROR_CODES.CORRUPTED_EVENT_LOG; } + // A run delivered to the wrong deployment gets its own code so dashboards + // and the UI can distinguish code-skew misrouting from generic runtime bugs. + if (WorkflowDeploymentMismatchError.is(err)) { + return RUN_ERROR_CODES.DEPLOYMENT_MISMATCH; + } + // World-layer faults — both a malformed response (contract violation) and a // transient infrastructure failure (throttle / 5xx / transport / timeout, // e.g. a firewall challenge) — are the backend's fault, not the user's. diff --git a/packages/core/src/describe-error.test.ts b/packages/core/src/describe-error.test.ts index 6dac492d01..7878da0d0e 100644 --- a/packages/core/src/describe-error.test.ts +++ b/packages/core/src/describe-error.test.ts @@ -3,6 +3,7 @@ import { RUN_ERROR_CODES, SerializationError, StepNotRegisteredError, + WorkflowDeploymentMismatchError, WorkflowRuntimeError, } from '@workflow/errors'; import { describe, expect, test } from 'vitest'; @@ -108,6 +109,17 @@ describe('describeError', () => { expect(result.hint).toContain('SDK contract'); }); + test('WorkflowDeploymentMismatchError is attributed to the SDK with the deployment hint', () => { + const result = describeError( + new WorkflowDeploymentMismatchError('wrun', 'dpl_a', 'dpl_b') + ); + expect(result.attribution).toBe('sdk'); + expect(result.errorCode).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH); + expect(result.hint).toContain( + 'deployment other than the one that created it' + ); + }); + test('precomputed errorCode wins over classifyRunError when both are provided', () => { // A plain Error would classify as USER_ERROR, but passing REPLAY_TIMEOUT // explicitly overrides that — useful for callers that know the failure @@ -201,6 +213,18 @@ describe('describeRunError', () => { expect(result.hint).toContain('SDK contract'); }); + test('DEPLOYMENT_MISMATCH errorCode is attributed to the SDK', () => { + const result = describeRunError({ + errorCode: RUN_ERROR_CODES.DEPLOYMENT_MISMATCH, + errorName: 'WorkflowDeploymentMismatchError', + }); + expect(result.attribution).toBe('sdk'); + expect(result.errorCode).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH); + expect(result.hint).toContain( + 'deployment other than the one that created it' + ); + }); + test('RUNTIME_ERROR code without errorName still lands as SDK', () => { const result = describeRunError({ errorCode: RUN_ERROR_CODES.RUNTIME_ERROR, diff --git a/packages/core/src/describe-error.ts b/packages/core/src/describe-error.ts index 74a4b5a9f1..bfce3c509f 100644 --- a/packages/core/src/describe-error.ts +++ b/packages/core/src/describe-error.ts @@ -85,6 +85,8 @@ const MAX_DELIVERIES_HINT = 'The workflow queue exceeded its max-delivery budget. This usually indicates a persistent runtime failure — check the most recent stack traces for the underlying cause.'; const WORLD_CONTRACT_HINT = 'The workflow backend returned data that violated the SDK contract. This is not retryable; please report it with the stack trace and runId.'; +const DEPLOYMENT_MISMATCH_HINT = + 'The run was delivered to a deployment other than the one that created it, and was stopped to protect against code-skew errors. This usually happens on preview branches after a new deployment supersedes the one a run started on. Verify that the original deployment is still available and that queue callbacks route to it.'; function normalizeErrorCode(code: string | undefined): RunErrorCode { // Values read back from persisted events are `string | undefined` — we @@ -144,6 +146,9 @@ export function describeRunError( hint: WORLD_CONTRACT_HINT, }; } + if (errorCode === RUN_ERROR_CODES.DEPLOYMENT_MISMATCH) { + return { attribution: 'sdk', errorCode, hint: DEPLOYMENT_MISMATCH_HINT }; + } if (name === 'WorkflowRuntimeError' || name === 'StepNotRegisteredError') { return { attribution: 'sdk', errorCode, hint: RUNTIME_ERROR_HINT }; } @@ -204,6 +209,16 @@ export function describeError( }; } + // Check DEPLOYMENT_MISMATCH before the generic WorkflowRuntimeError branch — + // WorkflowDeploymentMismatchError subclasses it, but has its own code + hint. + if (effectiveCode === RUN_ERROR_CODES.DEPLOYMENT_MISMATCH) { + return { + attribution: 'sdk', + errorCode: effectiveCode, + hint: DEPLOYMENT_MISMATCH_HINT, + }; + } + if (err instanceof WorkflowRuntimeError) { return { attribution: 'sdk', diff --git a/packages/core/src/runtime-trace-mode.test.ts b/packages/core/src/runtime-trace-mode.test.ts index fbf5928b61..402c616645 100644 --- a/packages/core/src/runtime-trace-mode.test.ts +++ b/packages/core/src/runtime-trace-mode.test.ts @@ -125,6 +125,7 @@ async function driveHandler(opts: { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -352,6 +353,7 @@ describe('workflowEntrypoint trace modes', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 6f46637ec4..5aa3e5e283 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -17,6 +17,7 @@ import { workflowEntrypoint } from './runtime.js'; import { dehydrateStepReturnValue, dehydrateWorkflowArguments, + hydrateRunError, } from './serialization.js'; // Capture every promise handed to `waitUntil` so tests can assert that @@ -62,6 +63,7 @@ async function runWorkflowHandlerWithEvents( * pin the log's contents keep full control. */ dynamicEventLog?: boolean; + currentDeploymentId?: string; } = {} ) { const createdEvents = options.createdEvents ?? []; @@ -89,6 +91,9 @@ async function runWorkflowHandlerWithEvents( setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn( + async () => options.currentDeploymentId ?? workflowRun.deploymentId + ), createQueueHandler: vi.fn( ( _prefix: string, @@ -146,6 +151,57 @@ describe('workflowEntrypoint replay guards', () => { `;globalThis.__private_workflows = new Map(); globalThis.__private_workflows.set(${JSON.stringify(workflowName)}, ${workflowName});`; + it('fails a run delivered to a different deployment before executing it', async () => { + // A run may only execute on the deployment that created it. On delivery + // elsewhere the run is failed before any workflow code or step runs, and + // the failure is recorded with the DEPLOYMENT_MISMATCH code without + // resolving the (possibly-gone) origin deployment's encryption key. + const workflowRun: WorkflowRun = { + runId: 'wrun_wrong_deployment', + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments( + [], + 'wrun_wrong_deployment', + undefined, + [] + ), + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'dpl_origin', + }; + + const createdEvents = await runWorkflowHandlerWithEvents( + `async function workflow() { + throw new Error('workflow code must not execute'); + }${getWorkflowTransformCode('workflow')}`, + workflowRun, + [], + { currentDeploymentId: 'dpl_current' } + ); + + const failedEvent = createdEvents.find( + (event: any) => event.eventType === 'run_failed' + ) as any; + expect(failedEvent).toBeDefined(); + expect(failedEvent.eventData.errorCode).toBe( + RUN_ERROR_CODES.DEPLOYMENT_MISMATCH + ); + const error = await hydrateRunError( + failedEvent.eventData.error, + workflowRun.runId, + undefined + ); + expect(error).toMatchObject({ name: 'WorkflowDeploymentMismatchError' }); + expect((error as Error).message).toContain( + 'belongs to deployment "dpl_origin", but was received by deployment "dpl_current"' + ); + expect(createdEvents).not.toContainEqual( + expect.objectContaining({ eventType: 'run_completed' }) + ); + }); + it('records run_failed when run_started response schema validation fails', async () => { const createdEvents: unknown[] = []; const schemaError = new WorkflowWorldError( @@ -173,6 +229,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => 'test-deployment'), createQueueHandler: vi.fn( ( _prefix: string, @@ -271,6 +328,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => 'test-deployment'), createQueueHandler: vi.fn( ( _prefix: string, @@ -368,6 +426,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -447,6 +506,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -808,6 +868,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -917,6 +978,7 @@ describe('workflowEntrypoint replay guards', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -1138,6 +1200,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( ( _prefix: string, @@ -1383,6 +1446,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { }); setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => workflowRun.deploymentId), createQueueHandler: vi.fn( (_p: string, handler: (m: unknown, md: unknown) => Promise) => async () => { @@ -1558,6 +1622,7 @@ describe('workflowEntrypoint turbo mode', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => 'test-deployment'), createQueueHandler: vi.fn( (_p: string, handler: (m: unknown, md: unknown) => Promise) => async () => { @@ -1826,6 +1891,7 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => { setWorld({ specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(async () => 'test-deployment'), createQueueHandler: vi.fn( (_p: string, handler: (m: unknown, md: unknown) => Promise) => async () => { diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 4fc3e8e5f6..ee5f964cf7 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -39,6 +39,7 @@ import { isInlineOwnershipEnabled, isTurboEnabled, } from './runtime/constants.js'; +import { failRunIfDeploymentMismatch } from './runtime/deployment-guard.js'; import { getQueueOverhead, getWorkflowQueueName, @@ -702,6 +703,15 @@ export function workflowEntrypoint( ); return; } + if ( + await failRunIfDeploymentMismatch({ + world, + run: bgRun, + requestId, + }) + ) { + return; + } const bgStartedAt = bgRun.startedAt ? +bgRun.startedAt : Date.now(); @@ -1065,6 +1075,23 @@ export function workflowEntrypoint( } // end else (non-turbo run_started) } // end if (!workflowRun) + // Stop before replaying (or executing inline steps) if this + // run reached a deployment other than the one that owns it. + // Continuing risks code skew, and any step it dispatches would + // resolve the wrong encryption key. `awaitRunReady` ensures a + // turbo-backgrounded run_started has landed before we record + // the failure. + if ( + await failRunIfDeploymentMismatch({ + world, + run: workflowRun, + requestId, + beforeFail: awaitRunReady, + }) + ) { + return; + } + // Resolve the encryption key for this run's deployment. // Used eagerly here since both runWorkflow (input // hydration / hook payload decryption) and the run_failed diff --git a/packages/core/src/runtime/deployment-guard.test.ts b/packages/core/src/runtime/deployment-guard.test.ts new file mode 100644 index 0000000000..d81971b567 --- /dev/null +++ b/packages/core/src/runtime/deployment-guard.test.ts @@ -0,0 +1,81 @@ +import { + RUN_ERROR_CODES, + WorkflowDeploymentMismatchError, +} from '@workflow/errors'; +import type { WorkflowRun, World } from '@workflow/world'; +import { describe, expect, it, vi } from 'vitest'; +import { hydrateRunError } from '../serialization.js'; +import { failRunIfDeploymentMismatch } from './deployment-guard.js'; + +const run = { + runId: 'wrun_test', + deploymentId: 'dpl_origin', + specVersion: 5, +} satisfies Pick; + +function createWorld(currentDeploymentId: string) { + const eventsCreate = vi.fn().mockResolvedValue({ event: {} }); + const getEncryptionKeyForRun = vi.fn().mockResolvedValue(undefined); + const world = { + getDeploymentId: vi.fn().mockResolvedValue(currentDeploymentId), + getEncryptionKeyForRun, + events: { create: eventsCreate }, + } as unknown as World; + return { world, eventsCreate, getEncryptionKeyForRun }; +} + +describe('failRunIfDeploymentMismatch', () => { + it('continues when the run belongs to the current deployment', async () => { + const { world, eventsCreate, getEncryptionKeyForRun } = + createWorld('dpl_origin'); + + await expect(failRunIfDeploymentMismatch({ world, run })).resolves.toBe( + false + ); + expect(eventsCreate).not.toHaveBeenCalled(); + expect(getEncryptionKeyForRun).not.toHaveBeenCalled(); + }); + + it('fails the run with an actionable error on another deployment', async () => { + const { world, eventsCreate, getEncryptionKeyForRun } = + createWorld('dpl_current'); + + const stopped = await failRunIfDeploymentMismatch({ + world, + run, + requestId: 'req_test', + }); + expect(stopped).toBe(true); + + // The failure must be recordable without the origin deployment's key, + // which is often gone once a run outlives its deployment. + expect(getEncryptionKeyForRun).not.toHaveBeenCalled(); + expect(eventsCreate).toHaveBeenCalledWith( + 'wrun_test', + expect.objectContaining({ + eventType: 'run_failed', + eventData: expect.objectContaining({ + errorCode: RUN_ERROR_CODES.DEPLOYMENT_MISMATCH, + }), + }), + { requestId: 'req_test' } + ); + + const failedEvent = eventsCreate.mock.calls[0][1]; + const error = await hydrateRunError( + failedEvent.eventData.error, + 'wrun_test', + undefined + ); + expect(WorkflowDeploymentMismatchError.is(error)).toBe(true); + expect((error as Error).message).toContain( + 'belongs to deployment "dpl_origin", but was received by deployment "dpl_current"' + ); + expect((error as Error).message).toContain( + 'stopped to protect against code-skew errors' + ); + expect((error as Error).message).toContain( + 'Verify that the original deployment is still available' + ); + }); +}); diff --git a/packages/core/src/runtime/deployment-guard.ts b/packages/core/src/runtime/deployment-guard.ts new file mode 100644 index 0000000000..fbfd84a1c7 --- /dev/null +++ b/packages/core/src/runtime/deployment-guard.ts @@ -0,0 +1,90 @@ +import { + EntityConflictError, + RUN_ERROR_CODES, + RunExpiredError, + WorkflowDeploymentMismatchError, +} from '@workflow/errors'; +import { + SPEC_VERSION_CURRENT, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { runtimeLogger } from '../logger.js'; +import { dehydrateRunError } from '../serialization.js'; + +/** + * Fails a run that reached a deployment other than its persisted owner by + * recording a `run_failed` carrying the `DEPLOYMENT_MISMATCH` error code. + * + * Returns `true` when the run was stopped — the caller must ack and stop — and + * `false` when execution is safe to continue. Mirrors `recordFatalRunError`: + * it records and returns rather than throwing, so the caller acks the message + * instead of letting the queue redeliver a run that is now terminal. Only an + * unexpected persistence failure is thrown, so the queue redelivers and retries + * the write. + * + * The failure is recorded **without resolving the run's encryption key**. This + * guard fires precisely when a run outlived its origin deployment, whose key is + * fetched from that deployment's API — often unavailable once it's past its + * retention window. Depending on it would throw here and drop us back into the + * silent retry-exhaustion this guard exists to replace. The error payload is + * written unencrypted (it holds only deployment ids, nothing sensitive), and + * the plaintext `errorCode` is the signal observability and the UI key off. + */ +export async function failRunIfDeploymentMismatch({ + world, + run, + requestId, + beforeFail, +}: { + world: World; + run: Pick; + requestId?: string; + beforeFail?: () => Promise; +}): Promise { + const currentDeploymentId = await world.getDeploymentId(); + if (run.deploymentId === currentDeploymentId) { + return false; + } + + const error = new WorkflowDeploymentMismatchError( + run.runId, + run.deploymentId, + currentDeploymentId + ); + runtimeLogger.error('Workflow run received by the wrong deployment', { + workflowRunId: run.runId, + expectedDeploymentId: run.deploymentId, + actualDeploymentId: currentDeploymentId, + errorMessage: error.message, + }); + + try { + await beforeFail?.(); + await world.events.create( + run.runId, + { + eventType: 'run_failed', + specVersion: SPEC_VERSION_CURRENT, + eventData: { + // Unencrypted (undefined key): see the note on this function — the + // origin deployment's key may be gone. dehydrate/hydrate are + // self-describing, so this reads back without a key. + error: await dehydrateRunError(error, run.runId, undefined), + errorCode: RUN_ERROR_CODES.DEPLOYMENT_MISMATCH, + }, + }, + { requestId } + ); + } catch (failError) { + // Run already reached a terminal state (a concurrent writer failed it, or + // it was cancelled/expired) — still stop. Anything else is a transient + // persistence failure: rethrow so the queue redelivers and we try again. + if (EntityConflictError.is(failError) || RunExpiredError.is(failError)) { + return true; + } + throw failError; + } + + return true; +} diff --git a/packages/core/src/runtime/precondition-guard-replay.test.ts b/packages/core/src/runtime/precondition-guard-replay.test.ts index 63fdfc0bed..8096124aa2 100644 --- a/packages/core/src/runtime/precondition-guard-replay.test.ts +++ b/packages/core/src/runtime/precondition-guard-replay.test.ts @@ -313,6 +313,7 @@ async function runPreconditionScenario(options: { const queue = vi.fn().mockResolvedValue({ messageId: 'msg_step' }); const fakeWorld = { specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn().mockResolvedValue(deploymentId), createQueueHandler: vi.fn((_prefix, handler) => { capturedHandler = handler; return vi.fn(); @@ -488,6 +489,7 @@ async function runCompletedRejectionScenario() { const fakeWorld = { specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn().mockResolvedValue(deploymentId), createQueueHandler: vi.fn((_prefix, handler) => { capturedHandler = handler; return vi.fn(); diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index 8e5f95d61c..5839f626db 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -1,6 +1,7 @@ import { EntityConflictError, FatalError, + RUN_ERROR_CODES, RunExpiredError, ThrottleError, WorkflowWorldError, @@ -23,6 +24,8 @@ const { mockRuntimeLogger, mockStepLogger, mockQueueMessage, + mockRunsGet, + mockGetDeploymentId, mockStepFn, } = vi.hoisted(() => { const mockStepFn = Object.assign(vi.fn().mockResolvedValue('step-result'), { @@ -54,6 +57,12 @@ const { error: vi.fn(), }, mockQueueMessage: vi.fn().mockResolvedValue(undefined), + mockRunsGet: vi.fn().mockResolvedValue({ + runId: 'wrun_test123', + deploymentId: 'test-deployment', + specVersion: 5, + }), + mockGetDeploymentId: vi.fn().mockResolvedValue('test-deployment'), mockStepFn, }; }); @@ -70,7 +79,9 @@ vi.mock('@vercel/functions', () => ({ vi.mock('./world.js', () => ({ getWorld: vi.fn(async () => ({ events: { create: mockEventsCreate }, + runs: { get: mockRunsGet }, queue: mockQueue, + getDeploymentId: mockGetDeploymentId, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), })), getWorldHandlers: vi.fn(async () => ({ @@ -139,6 +150,7 @@ vi.mock('../serialization.js', () => ({ dehydrateStepReturnValue: vi .fn() .mockResolvedValue(new Uint8Array([1, 2, 3])), + dehydrateRunError: vi.fn().mockResolvedValue(new Uint8Array([7, 8, 9])), cancelAbortReaders: vi.fn(), dehydrateStepError: vi.fn().mockResolvedValue(new Uint8Array([4, 5, 6])), })); @@ -198,6 +210,15 @@ import { executeStep } from './step-executor.js'; import { stepEntrypoint } from './step-handler.js'; import { getWorld } from './world.js'; +beforeEach(() => { + mockRunsGet.mockResolvedValue({ + runId: 'wrun_test123', + deploymentId: 'test-deployment', + specVersion: 5, + }); + mockGetDeploymentId.mockResolvedValue('test-deployment'); +}); + function capturedHandler( message: unknown, metadata: { queueName: string; messageId: string; attempt: number } @@ -260,7 +281,9 @@ describe('step-handler 409 handling', () => { // Re-set getWorld mock since clearAllMocks resets it vi.mocked(getWorld).mockResolvedValue({ events: { create: mockEventsCreate }, + runs: { get: mockRunsGet }, queue: mockQueue, + getDeploymentId: mockGetDeploymentId, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as any); // Reset mockEventsCreate fully - mockImplementation persists through clearAllMocks @@ -277,6 +300,31 @@ describe('step-handler 409 handling', () => { mockStepFn.mockResolvedValue('step-result'); }); + it('fails the run before executing a step on another deployment', async () => { + mockGetDeploymentId.mockResolvedValue('dpl_current'); + mockRunsGet.mockResolvedValue({ + runId: 'wrun_test123', + deploymentId: 'dpl_origin', + specVersion: 5, + }); + + await capturedHandler(createMessage(), createMetadata('myStep')); + + expect(mockStepFn).not.toHaveBeenCalled(); + expect(mockEventsCreate).toHaveBeenCalledTimes(1); + expect(mockEventsCreate).toHaveBeenCalledWith( + 'wrun_test123', + expect.objectContaining({ + eventType: 'run_failed', + eventData: expect.objectContaining({ + error: new Uint8Array([7, 8, 9]), + errorCode: RUN_ERROR_CODES.DEPLOYMENT_MISMATCH, + }), + }), + expect.anything() + ); + }); + afterEach(() => { vi.restoreAllMocks(); }); @@ -581,7 +629,9 @@ describe('step-handler max deliveries', () => { mockQueueMessage.mockResolvedValue(undefined); vi.mocked(getWorld).mockResolvedValue({ events: { create: mockEventsCreate }, + runs: { get: mockRunsGet }, queue: mockQueue, + getDeploymentId: mockGetDeploymentId, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as any); mockEventsCreate.mockReset().mockResolvedValue({ @@ -664,7 +714,9 @@ describe('step-handler step not found', () => { mockQueueMessage.mockResolvedValue(undefined); vi.mocked(getWorld).mockResolvedValue({ events: { create: mockEventsCreate }, + runs: { get: mockRunsGet }, queue: mockQueue, + getDeploymentId: mockGetDeploymentId, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as any); mockEventsCreate.mockReset().mockResolvedValue({ @@ -837,7 +889,9 @@ describe('step-handler fatal vs retryable behavior', () => { mockQueueMessage.mockResolvedValue(undefined); vi.mocked(getWorld).mockResolvedValue({ events: { create: mockEventsCreate }, + runs: { get: mockRunsGet }, queue: mockQueue, + getDeploymentId: mockGetDeploymentId, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as any); mockEventsCreate.mockReset().mockResolvedValue({ diff --git a/packages/core/src/runtime/step-handler.ts b/packages/core/src/runtime/step-handler.ts index adbcd6e495..c7e1bfdf18 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -53,6 +53,7 @@ import { } from '../types.js'; import { getMaxQueueDeliveries } from './constants.js'; +import { failRunIfDeploymentMismatch } from './deployment-guard.js'; import { getQueueOverhead, getWorkflowQueueName, @@ -213,6 +214,19 @@ function createStepHandler(namespace?: string) { const world = await getWorld(); const isVercel = process.env.VERCEL_URL !== undefined; + const workflowRun = await world.runs.get(workflowRunId, { + resolveData: 'none', + }); + if ( + await failRunIfDeploymentMismatch({ + world, + run: workflowRun, + requestId, + }) + ) { + return; + } + // Memoized accessor for the per-run AES-256 encryption key. The first // caller (typically `hydrateStepArguments` for input deserialization, // or one of the early-return dehydrateStepError paths if step_started diff --git a/packages/core/src/runtime/wait-completion-replay.test.ts b/packages/core/src/runtime/wait-completion-replay.test.ts index c4ca46982a..fe25d3628b 100644 --- a/packages/core/src/runtime/wait-completion-replay.test.ts +++ b/packages/core/src/runtime/wait-completion-replay.test.ts @@ -329,6 +329,7 @@ async function runStaleWaitReplayScenario(options: { const queue = vi.fn().mockResolvedValue({ messageId: 'msg_step' }); const fakeWorld = { specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn().mockResolvedValue(deploymentId), createQueueHandler: vi.fn((_prefix, handler) => { capturedHandler = handler; return vi.fn(); diff --git a/packages/errors/src/error-codes.ts b/packages/errors/src/error-codes.ts index 656f479496..8e1dd92f1b 100644 --- a/packages/errors/src/error-codes.ts +++ b/packages/errors/src/error-codes.ts @@ -18,6 +18,8 @@ export const RUN_ERROR_CODES = { REPLAY_TIMEOUT: 'REPLAY_TIMEOUT', /** World response violated the SDK contract and cannot be retried safely */ WORLD_CONTRACT_ERROR: 'WORLD_CONTRACT_ERROR', + /** Run was delivered to a deployment other than the one that created it */ + DEPLOYMENT_MISMATCH: 'DEPLOYMENT_MISMATCH', } as const; export type RunErrorCode = diff --git a/packages/errors/src/index.ts b/packages/errors/src/index.ts index 573e1d7218..652d79dffc 100644 --- a/packages/errors/src/index.ts +++ b/packages/errors/src/index.ts @@ -566,6 +566,35 @@ export class WorkflowNotRegisteredError extends WorkflowRuntimeError { } } +/** + * Thrown when a workflow run is delivered to a deployment other than the one + * that created it. Continuing on the receiving deployment is unsafe because + * its workflow and step bundles may not match the run's persisted history. + */ +export class WorkflowDeploymentMismatchError extends WorkflowRuntimeError { + readonly runId: string; + readonly expectedDeploymentId: string; + readonly actualDeploymentId: string; + + constructor( + runId: string, + expectedDeploymentId: string, + actualDeploymentId: string + ) { + super( + `Workflow run "${runId}" belongs to deployment "${expectedDeploymentId}", but was received by deployment "${actualDeploymentId}". The run was stopped to protect against code-skew errors. Verify that the original deployment is still available and that queue callbacks are routed to it.` + ); + this.name = 'WorkflowDeploymentMismatchError'; + this.runId = runId; + this.expectedDeploymentId = expectedDeploymentId; + this.actualDeploymentId = actualDeploymentId; + } + + static is(value: unknown): value is WorkflowDeploymentMismatchError { + return isError(value) && value.name === 'WorkflowDeploymentMismatchError'; + } +} + /** * Thrown when performing operations on a workflow run that does not exist. *