From 3a6dc5bab805175441169eb306b20b6fe4da83f0 Mon Sep 17 00:00:00 2001 From: Karthik Kalyanaraman Date: Thu, 6 Aug 2026 12:38:24 -0700 Subject: [PATCH] fix(core): pre-check deployment affinity before the lazy resume write MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lazy hook fast path (#3345) hoisted the consumer's hook_received write above the deployment-affinity guard (#2960), so a misrouted lazy resume wrote its event before the guard could re-route the delivery. Stamp the run's pinned deployment on the resume message (hookInput.deploymentId, from the producer's resume context) and, on the consumer, compare it against the ambient deployment id immediately before the fast path: a match continues with no run fetch, a mismatch fetches the authoritative run and hands it to the existing guard — which keeps sole ownership of re-route/fail policy and remains the authoritative protection before replay and step execution. The re-routed message preserves the complete hookInput (it may hold the only copy of the resume payload). Older messages without the field, and worlds without deployment affinity, are unchanged: they skip the pre-check and rely on the authoritative guard, the pre-guard write staying convergent per (runId, resumeId). Fixes the misrouted-lazy-resume unit test broken by the #2960/#3345 ordering: a modern misrouted resume now re-routes with zero event writes, asserted for both hook_received and run_started. Co-Authored-By: Claude Fable 5 --- .changeset/lazy-resume-deployment-affinity.md | 6 ++ packages/core/src/runtime.test.ts | 17 +++- packages/core/src/runtime.ts | 84 ++++++++++++++++--- .../resume-hook.consumer-preload.test.ts | 46 +++++++++- .../src/runtime/resume-hook.parallel.test.ts | 3 + packages/core/src/runtime/resume-hook.ts | 4 + packages/world/src/queue.ts | 10 +++ 7 files changed, 155 insertions(+), 15 deletions(-) create mode 100644 .changeset/lazy-resume-deployment-affinity.md 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;