diff --git a/.changeset/lazy-resume-deployment-affinity.md b/.changeset/lazy-resume-deployment-affinity.md new file mode 100644 index 0000000000..03dc47b129 --- /dev/null +++ b/.changeset/lazy-resume-deployment-affinity.md @@ -0,0 +1,6 @@ +--- +'@workflow/world': minor +'@workflow/core': patch +--- + +Carry the run's pinned deployment on lazy resume messages so misrouted deliveries re-route before the `hook_received` write. diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index a68b35b4ad..322b4000d8 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -297,10 +297,12 @@ describe('workflowEntrypoint replay guards', () => { it('re-routes a misrouted lazy hook resume with its payload intact', async () => { // The lazy-resume producer parallelizes the `hook_received` write with this - // queue publish, so `hookInput` may be the only copy of the payload. The - // guard runs ahead of the re-ensure, so nothing is written here — which - // means the re-routed message has to carry `hookInput` or the resume is - // lost when the producer's direct write had not landed. + // queue publish, so `hookInput` may be the only copy of the payload. A + // modern message carries `hookInput.deploymentId`, so the cheap pre-check + // detects the mismatch BEFORE the fast path's hook_received write — + // zero event writes before re-routing — and the re-routed message has to + // carry the complete `hookInput` or the resume is lost when the + // producer's direct write had not landed. const workflowRun = await misroutedRun(); const queueCalls: QueueCall[] = []; const hookInput = { @@ -309,6 +311,7 @@ describe('workflowEntrypoint replay guards', () => { token: 'tok_1', payload: { serialized: true }, payloadDigest: 'sha256:abc', + deploymentId: 'dpl_origin', }; const createdEvents = await runWorkflowHandlerWithEvents( @@ -320,13 +323,19 @@ describe('workflowEntrypoint replay guards', () => { expect(queueCalls).toHaveLength(1); expect(queueCalls[0].opts).toMatchObject({ deploymentId: 'dpl_origin' }); + // The complete hookInput — payload included — survives re-routing. expect(queueCalls[0].message).toMatchObject({ deploymentMismatchRetryCount: 1, hookInput, }); + // Zero event writes before re-routing: neither the fast path's + // hook_received nor the generic setup's run_started ran. expect(createdEvents).not.toContainEqual( expect.objectContaining({ eventType: 'hook_received' }) ); + expect(createdEvents).not.toContainEqual( + expect.objectContaining({ eventType: 'run_started' }) + ); }); it('fails a misrouted run once the re-route budget is spent', async () => { diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 44edf5bddb..885081b048 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1553,6 +1553,63 @@ export function workflowEntrypoint( } } + // Deployment-affinity pre-check for the lazy hook fast + // path below. New lazy-resume messages carry the run's + // pinned deployment (`hookInput.deploymentId`), so a + // misrouted delivery is detectable with a cheap ambient + // deployment-id comparison BEFORE the fast path's + // hook_received write — the correctly-routed common case + // pays no run fetch and no latency. Only a detected + // mismatch fetches the authoritative run and hands it to + // the existing guard (which owns re-route/fail policy); + // the re-routed message keeps the complete `hookInput`, + // since it may hold the only copy of the resume payload. + // When the pinned or ambient id is unavailable (older + // producer message, a world without deployment affinity, + // or a getDeploymentId failure), behavior is unchanged: + // the authoritative guard after run setup — which remains + // the protection before replay and step execution — still + // covers the delivery; the fast path's write is idempotent + // per (runId, resumeId), so a pre-guard write from an + // older message stays convergent. + if ( + !workflowRun && + hookInput?.deploymentId !== undefined && + // Same eligibility as guardDeploymentAffinity: without + // atomic, immutable deployments the ids legitimately + // differ (e.g. dpl_local@) and comparing them + // would trigger spurious run fetches. + world.capabilities?.deploymentAffinity === true + ) { + const pinnedDeploymentId = hookInput.deploymentId; + let currentDeploymentId: string | undefined; + try { + currentDeploymentId = await world.getDeploymentId(); + } catch { + currentDeploymentId = undefined; + } + if ( + currentDeploymentId !== undefined && + currentDeploymentId !== pinnedDeploymentId + ) { + const run = await world.runs.get(runId, { + resolveData: 'none', + }); + if ( + (await guardDeployment( + run, + async () => ({ + ...(await replayMessage()), + hookInput, + }), + awaitRunReady + )) !== 'continue' + ) { + return; + } + } + } + // --- Lazy hook resume fast path --- // A lazy hook delivery (resumeId + digest on the queue // message) hoists the consumer's idempotent hook_received @@ -1980,17 +2037,24 @@ export function workflowEntrypoint( // Covers every flow replay — initial start, step completions, // hook resumptions, wait completions — and stops before any - // workflow code, inline step, or event write happens when the - // run is not pinned here (see `guardDeploymentAffinity`). - // `awaitRunReady` orders turbo's `run_started` before either - // stopping action. + // workflow code or inline step executes when the run is not + // pinned here (see `guardDeploymentAffinity`). This remains + // the AUTHORITATIVE protection before replay and step + // execution. `awaitRunReady` orders turbo's `run_started` + // before either stopping action. // - // Runs ahead of the lazy-hook re-ensure below, so a misrouted - // resume writes nothing here — which means the re-routed - // message must carry `hookInput`, or a resume whose producer - // write had not yet landed would be lost (that re-ensure is - // idempotent per `resumeId`, so the pinned deployment still - // converges on one event). + // Lazy hook resumes are additionally pre-checked above the + // fast path: a new message's `hookInput.deploymentId` is + // compared against the ambient deployment id for free, and + // only a detected mismatch fetches the authoritative run — + // so a misrouted modern resume re-routes with zero event + // writes. Older messages (no `hookInput.deploymentId`) reach + // the fast path's idempotent hook_received write before this + // guard; that write converges per (runId, resumeId) on the + // pinned deployment, so it is safe, just not free. Either + // way the re-routed message must carry `hookInput`, or a + // resume whose producer write had not yet landed would be + // lost. if ( (await guardDeployment( workflowRun, diff --git a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts index 7c569f5514..1e8473fd6e 100644 --- a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts +++ b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts @@ -138,6 +138,13 @@ async function runResumeConsumerScenario(options: { * classification (consume the message vs rethrow for redelivery). */ reEnsureRejection?: Error; + /** + * When set, the queue message's hookInput carries the producer-stamped + * pinned deployment id, activating the consumer's cheap pre-write + * affinity check. Omitted by default so most scenarios double as + * older-message fixtures (no deploymentId still parses and runs). + */ + hookDeploymentId?: string; }) { const hookPreload = options.hookPreload ?? 'event-only'; const runId = 'wrun_resume_consumer_preload'; @@ -325,14 +332,20 @@ async function runResumeConsumerScenario(options: { | ((message: unknown, metadata: unknown) => Promise) | undefined; const queue = vi.fn().mockResolvedValue({ messageId: 'msg_resume' }); + const runsGet = vi.fn(async () => workflowRun); setWorld({ specVersion: SPEC_VERSION_CURRENT, + // Atomic, immutable deployments (as world-vercel declares), with the + // ambient id matching the run's pin — so the affinity pre-check and the + // authoritative guard both see a correctly routed delivery. + capabilities: { deploymentAffinity: true }, + getDeploymentId: vi.fn(async () => deploymentId), createQueueHandler: vi.fn((_prefix, handler) => { capturedHandler = handler; return vi.fn(); }), events: { list: listEvents, create: createEvent }, - runs: { get: vi.fn(async () => workflowRun) }, + runs: { get: runsGet }, queue, getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as unknown as World); @@ -356,6 +369,9 @@ async function runResumeConsumerScenario(options: { token: hookToken, payload: payloadBytes, payloadDigest, + ...(options.hookDeploymentId !== undefined + ? { deploymentId: options.hookDeploymentId } + : {}), }, }, { @@ -388,6 +404,7 @@ async function runResumeConsumerScenario(options: { runCompletedCreates, listEvents, createEvent, + runsGet, handlerError, }; } @@ -428,6 +445,33 @@ describe('lazy hook resume consumer preload', () => { expect(runCompletedCreates).toHaveLength(1); }); + it('adds no run fetch for a correctly routed modern message (matching hookInput.deploymentId)', async () => { + // The pre-write affinity check is a pure ambient-id comparison on the + // matching path: the run is never fetched, and the streamed replay path + // is selected exactly as without the field. + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + listEvents, + runsGet, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'complete', + hookDeploymentId: 'dpl_resume_consumer_preload', + }); + + expect(handlerError).toBeUndefined(); + expect(runsGet).not.toHaveBeenCalled(); + // The streamed replay path remains selected: one setup request, no + // run_started, no events.list. + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(0); + expect(listEvents).not.toHaveBeenCalled(); + expect(runCompletedCreates).toHaveLength(1); + }); + it('initializes replay from the producer-won canonical event in the preload', async () => { const { hookReceivedCreates, diff --git a/packages/core/src/runtime/resume-hook.parallel.test.ts b/packages/core/src/runtime/resume-hook.parallel.test.ts index d0c1237d97..4293aabd5d 100644 --- a/packages/core/src/runtime/resume-hook.parallel.test.ts +++ b/packages/core/src/runtime/resume-hook.parallel.test.ts @@ -116,6 +116,9 @@ describe('resumeHook (parallel fast path)', () => { token: hook.token, payload: PAYLOAD_BYTES, payloadDigest: digest, + // The run's pinned deployment from the resume context, for the + // consumer's cheap pre-write affinity check. + deploymentId: 'deployment_par', }); }); diff --git a/packages/core/src/runtime/resume-hook.ts b/packages/core/src/runtime/resume-hook.ts index daf625afc2..7dba457306 100644 --- a/packages/core/src/runtime/resume-hook.ts +++ b/packages/core/src/runtime/resume-hook.ts @@ -593,6 +593,10 @@ async function resumeHookImpl( token: hook.token, payload: dehydratedPayload, payloadDigest, + // Deployment affinity for the consumer's cheap pre-write + // check: lets a misrouted delivery re-route before its + // hoisted hook_received write instead of after. + deploymentId: resumeContext.deploymentId, }, } satisfies WorkflowInvokePayload, queueOptions diff --git a/packages/world/src/queue.ts b/packages/world/src/queue.ts index 5e2479193f..dce7ee844d 100644 --- a/packages/world/src/queue.ts +++ b/packages/world/src/queue.ts @@ -169,6 +169,16 @@ export const HookResumeInputSchema = z.object({ * content-stable server-side. */ payloadDigest: z.string(), + /** + * The deployment the run is pinned to, from the producer's resume context. + * Lets the consumer detect a misrouted delivery with a cheap ambient + * deployment-id comparison BEFORE its hoisted `hook_received` replay-preload + * write — only a detected mismatch pays for the authoritative run fetch and + * the deployment-affinity guard. Optional for queued-message compatibility: + * messages from older producers omit it and simply skip the pre-write + * check (the authoritative guard before replay still protects them). + */ + deploymentId: z.string().optional(), }); export type HookResumeInput = z.infer;