diff --git a/.changeset/retain-vms-across-hooks.md b/.changeset/retain-vms-across-hooks.md new file mode 100644 index 0000000000..7f0e5e3c50 --- /dev/null +++ b/.changeset/retain-vms-across-hooks.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Retain workflow VMs across safe step boundaries that also contain open hooks. diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index d4fa637695..19b0c2fc0e 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -5,7 +5,7 @@ import { withResolvers } from '@workflow/utils'; import type { WorldCapabilities } from '@workflow/world'; import type { EventsConsumer } from './events-consumer.js'; -import type { QueueItem } from './global.js'; +import { type QueueItem, WorkflowSuspension } from './global.js'; import type { ReplayPayloadCache } from './replay-payload-cache.js'; import type { Serializable } from './schemas.js'; import type { DecryptionKey } from './serialization/encryption.js'; @@ -137,12 +137,7 @@ export interface WorkflowOrchestratorContext { globalThis: typeof globalThis; /** * Increments when a suspension is accepted and on every retained-session - * resume. STEP suspension signals capture it when scheduled and no-op if - * it moved (see step.ts) — this drops same-boundary sibling signals and - * timers queued at boundary N that would fire after the session resumed - * into boundary N+1. Sleep/hook/attribute signals are intentionally - * unguarded: their presence makes the boundary unretainable, so a late - * signal correctly demotes the session (workflow.ts `onWorkflowError`). + * resume. Step and hook signals capture it so stale signals no-op. */ suspensionGeneration: number; eventsConsumer: EventsConsumer; @@ -793,3 +788,16 @@ export function scheduleWhenIdle( }; setTimeout(check, 0); } + +/** Schedule a generation-guarded suspension after deliveries settle. */ +export function scheduleWorkflowSuspension( + ctx: WorkflowOrchestratorContext +): void { + const generation = ctx.suspensionGeneration; + scheduleWhenIdle(ctx, () => { + if (generation !== ctx.suspensionGeneration) return; + ctx.onWorkflowError( + new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) + ); + }); +} diff --git a/packages/core/src/retained-vm-loop.test.ts b/packages/core/src/retained-vm-loop.test.ts index b7a88a979f..e1be1c5f86 100644 --- a/packages/core/src/retained-vm-loop.test.ts +++ b/packages/core/src/retained-vm-loop.test.ts @@ -5,7 +5,15 @@ import { slotToEventId, type WorkflowRun, } from '@workflow/world'; -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + afterEach, + assert, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; // Spy on VM-context construction while preserving the real implementation, so // we can prove the retained path builds ONE VM for a whole run instead of one @@ -21,9 +29,11 @@ const { registerSerializationClass } = await import('./class-serialization.js'); const { registerStepFunction } = await import('./private.js'); const { setWorld } = await import('./runtime/world.js'); const { workflowEntrypoint } = await import('./runtime.js'); -const { dehydrateWorkflowArguments, hydrateWorkflowReturnValue } = await import( - './serialization.js' -); +const { + dehydrateStepReturnValue, + dehydrateWorkflowArguments, + hydrateWorkflowReturnValue, +} = await import('./serialization.js'); vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn((p: Promise) => { @@ -101,6 +111,39 @@ const parallelBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_ST } globalThis.__private_workflows = new Map([["workflow", workflow]]);`; +// An open hook and a step both signal the first suspension. The step wins the +// Promise.race, then a second step advances the same inline replay loop. A +// retained session must absorb the losing hook's same-generation suspension +// signal and keep the one VM alive across both step completions. +const openHookRaceWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2"); + async function workflow() { + const hook = createHook({ token: "retained-hook" }); + const a = await Promise.race([hook.then(() => 999), s1()]); + const b = await s2(); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + +// Hook metadata is serialized at the same suspension boundary as step input. +// A getter mutating workflow state must demote the retained VM just like an +// unsafe step argument, because a cold replay skips serialization after the +// hook_created event exists. +const impureHookMetadataWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + const echo = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_echo"); + async function workflow() { + let counter = 0; + createHook({ + token: "unsafe-retained-hook", + metadata: { get value() { counter++; return 1; } }, + }); + await s1(); + return await echo(counter); + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + // A parallel batch where one sibling's input is unsafe must serialize the // WHOLE batch through the ordinary VM path (all-or-nothing) and demote. const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); @@ -196,12 +239,17 @@ registerStepFunction('r_echo', async (value) => value); // Drive the full workflow handler over a stateful (dynamic) event log so the // inline loop makes real progress across its own writes, exactly like a World. // Non-turbo (no runInput, attempt 2) to keep the path simple and deterministic. +type DriveMode = + | { type: 'normal' } + | { type: 'fail-event'; eventType: Event['eventType'] } + | { type: 'inject-hook' }; + async function drive( runId: string, workflowCode = twoStepWorkflow, - options: { failEventTypeOnce?: string } = {} + initialMode: DriveMode = { type: 'normal' } ) { - let { failEventTypeOnce } = options; + let mode = initialMode; const run: WorkflowRun = { runId, workflowName: 'workflow', @@ -217,8 +265,8 @@ async function drive( let seq = 0; const eventsCreate = vi.fn(async (_runId: string, data: any) => { - if (data.eventType === failEventTypeOnce) { - failEventTypeOnce = undefined; + if (mode.type === 'fail-event' && data.eventType === mode.eventType) { + mode = { type: 'normal' }; throw new PreconditionFailedError('stale snapshot (test-injected)'); } createdEvents.push(data); @@ -232,6 +280,29 @@ async function drive( ...data, } as Event; events.push(event); + if (data.eventType === 'step_started' && mode.type === 'inject-hook') { + mode = { type: 'normal' }; + const hookCreated = events.find( + (candidate) => candidate.eventType === 'hook_created' + ); + assert(hookCreated, 'expected hook_created before step'); + events.push({ + eventId: slotToEventId(++seq), + runId, + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hookCreated.correlationId, + eventData: { + token: hookCreated.eventData.token, + payload: await dehydrateStepReturnValue( + { source: 'external-hook' }, + runId, + undefined + ), + }, + createdAt: new Date(), + }); + } // step_started returns a running step entity so executeStep proceeds to // run the body and write step_completed. if (data.eventType === 'step_started') { @@ -358,6 +429,36 @@ describe('retained VM through the inline replay loop', () => { expect(vmBuilds).toBe(1); }); + it('retains when hook_received extends the log between step_started and step_completed', async () => { + const { vmBuilds, result } = await drive( + 'wrun_retained_hook_between_step_events', + openHookRaceWorkflow, + { type: 'inject-hook' } + ); + // The hook event precedes step_completed in the durable log, so it wins + // the race on resume while the already-started step still completes. + expect(result).toBe(1019); + expect(vmBuilds).toBe(1); + }); + + it('matches cold replay when hook metadata serialization mutates workflow state', async () => { + process.env.WORKFLOW_RETAINED_VM = '0'; + const off = await drive( + 'wrun_impure_hook_metadata_off', + impureHookMetadataWorkflow + ); + createContextSpy.mockClear(); + delete process.env.WORKFLOW_RETAINED_VM; + + const on = await drive( + 'wrun_impure_hook_metadata_on', + impureHookMetadataWorkflow + ); + expect(off.result).toBe(0); + expect(on.result).toBe(0); + expect(on.vmBuilds).toBeGreaterThan(1); + }); + it('demotes retention when any input in a parallel batch is unsafe', async () => { const { vmBuilds, result } = await drive( 'wrun_retained_mixed_batch', @@ -376,7 +477,7 @@ describe('retained VM through the inline replay loop', () => { const { vmBuilds, result } = await drive( 'wrun_retained_412_restart', twoStepWorkflow, - { failEventTypeOnce: 'run_completed' } + { type: 'fail-event', eventType: 'run_completed' } ); expect(result).toBe(30); expect(vmBuilds).toBeGreaterThan(1); diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 24ec515471..cb8ab79467 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -486,11 +486,8 @@ function rootRunIdFrom( * `wait_completed`, which the wait timer can resolve with * `wait_completed`). * - * This gates VM retention, the inline-delta fast path, and turbo's forced - * optimistic start. A terminal-step delta can omit an event appended - * concurrently after that write. With no open hook or wait, only cancellation - * can do so, and observing it one replay late is safe because the next entity - * write is rejected. + * Open waits block VM retention and inline deltas. Open hooks and waits block + * turbo's optimistic start; hooks also require a `step_started` claim. * * Step-body `attr_set` writes are NOT a concern: they land before the * step's terminal write and are therefore already inside the returned @@ -567,47 +564,6 @@ function replayEventDeploymentId(event: Event): string | undefined { return undefined; } -/** - * The whole retention predicate: keep the session only for a pure step - * boundary (every suspension item is a step — any other item type, present - * or future, is unretainable by default) whose new step inputs serialized - * without executing workflow code, with no out-of-band continuation source: - * attributes require replay; hooks and waits can wake another invocation. - * `WORKFLOW_RETAINED_VM=0` disables retention entirely. - * - * The open hook/wait scan is O(events), so it is taken through a lazy getter - * and consulted last, after every cheap check has passed. - * - * INVARIANT this predicate leans on: every suspension signaler that does NOT - * carry the step-consumer generation guard (sleep, hook, attribute — see - * `suspensionGeneration` in private.ts) must be unretainable here, either via - * a non-step queue item or the open hook/wait scan. A new signaler that - * satisfies neither would let a stale signal be accepted as a fresh - * suspension on a resumed session. - * - * Quiescence assumes workflow code stays inside the sandbox's determinism - * contract. Escaping to the host realm (e.g. recovering the host `Function` - * constructor from an exposed host class to schedule real timers) makes a - * workflow nondeterministic under ordinary replay too, and is not defended - * here. - */ -function canRetainWorkflowSession( - suspension: WorkflowSuspension, - stepInputsSafe: boolean, - openHookWait: { value: ReturnType } -): boolean { - if ( - !isVmRetentionEnabled() || - !stepInputsSafe || - suspension.steps.length === 0 || - !suspension.steps.every((item) => item.type === 'step') - ) { - return false; - } - const { openHook, openWait } = openHookWait.value; - return !openHook && !openWait; -} - /** * Maximum inline-execution duration for a single handler invocation. * @@ -3341,32 +3297,37 @@ export function workflowEntrypoint( } // Open hooks/waits in the log as loaded for this - // replay. This suspension's own hook/wait writes are - // NOT in it — they never reach retention anyway, - // because a suspension containing a non-step item - // fails canRetainWorkflowSession's type check before - // the scan is consulted. Computed - // lazily, at most once, and shared between the - // retention decision here and the delta/turbo gates - // below — the attr-detour and hook-conflict paths - // return/continue before the gates and usually - // short-circuit before ever scanning the log. + // replay. Computed lazily, at most once, and shared + // between the retention decision here and the + // delta/turbo gates below — the attr-detour and + // hook-conflict paths return/continue before the + // gates and usually short-circuit before scanning. const openHookWait = once(() => { assert(eventLog.type === 'ready'); return openHookAndWaitState(eventLog.events); }); - // The single retention decision: keep the parked - // session only across a pure step boundary with no - // out-of-band continuation source and provably - // passive step inputs. if ( retainedSession && - !canRetainWorkflowSession( - err, - suspensionResult.retainedStepInputsSafe, - openHookWait - ) + (!isVmRetentionEnabled() || + !suspensionResult.serializationWasPassive || + err.stepCount === 0 || + !err.steps.every((item) => { + switch (item.type) { + case 'step': + case 'hook': + return true; + case 'wait': + case 'attribute': + return false; + default: + item satisfies never; + throw new Error( + 'Unknown workflow suspension item' + ); + } + }) || + openHookWait.value.openWait) ) { retainedSession = null; } diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index aee668d278..ae644cb369 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -13,7 +13,7 @@ import { type World, } from '@workflow/world'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { WorkflowSuspension } from '../global.js'; +import { type QueueItem, WorkflowSuspension } from '../global.js'; import { maxEventSlot, stepDispatchIdempotencyKey } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; import { handleSuspension } from './suspension-handler.js'; @@ -832,21 +832,7 @@ describe('resilient step dispatch', () => { }); }); -describe('retainedStepInputsSafe (serialization passivity gate)', () => { - function stepPending(args: unknown[]) { - return new Map([ - [ - 'step_1', - { - type: 'step' as const, - correlationId: 'step_1', - stepName: 'someStep', - args, - }, - ], - ]); - } - +describe('serializationWasPassive', () => { /** An object with a VM-realm getter — exactly what the sink records. */ function vmGetterObject() { return runInNewContext( @@ -859,20 +845,32 @@ describe('retainedStepInputsSafe (serialization passivity gate)', () => { ); } - async function runSuspension(args: unknown[]) { + async function runSuspension(item: QueueItem) { const eventsCreate = vi .fn() .mockImplementation(async (_runId, event) => ({ event })); const world = createWorld(eventsCreate); return handleSuspension({ - suspension: new WorkflowSuspension(stepPending(args), globalThis), + suspension: new WorkflowSuspension( + new Map([[item.correlationId, item]]), + globalThis + ), world, run, }); } + function runStep(args: Extract['args']) { + return runSuspension({ + type: 'step', + correlationId: 'step_1', + stepName: 'someStep', + args, + }); + } + it('reports safe for plain data and supported built-ins', async () => { - const result = await runSuspension([ + const result = await runStep([ { nested: [{ ok: true }, 'text', 42n], flag: false }, new Map([['k', new Set([1])]]), new Date(1700000000000), @@ -880,7 +878,7 @@ describe('retainedStepInputsSafe (serialization passivity gate)', () => { /pattern/gi, new URL('https://example.com/'), ]); - expect(result.retainedStepInputsSafe).toBe(true); + expect(result.serializationWasPassive).toBe(true); }); it('reports unsafe for an Error argument (stack materialization)', async () => { @@ -888,25 +886,53 @@ describe('retainedStepInputsSafe (serialization passivity gate)', () => { // invocation formats-and-caches the trace and runs any // `Error.prepareStackTrace` — neither is repeated by a cold replay, so // the boundary must demote. - const result = await runSuspension([new Error('lazy stack')]); - expect(result.retainedStepInputsSafe).toBe(false); + const result = await runStep([new Error('lazy stack')]); + expect(result.serializationWasPassive).toBe(false); }); it('reports unsafe when serializing an argument executes a getter', async () => { const value = vmGetterObject(); - const result = await runSuspension([{ deep: [value] }]); - expect(result.retainedStepInputsSafe).toBe(false); + const result = await runStep([{ deep: [value] }]); + expect(result.serializationWasPassive).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); + const result = await runStep([new Proxy({ a: 1 }, {})]); + expect(result.serializationWasPassive).toBe(false); + }); + + it.each([ + [ + 'hook metadata', + { + type: 'hook', + correlationId: 'hook_unsafe_metadata', + token: 'unsafe-metadata', + metadata: vmGetterObject(), + }, + ], + [ + 'a hook abort reason', + { + type: 'hook', + correlationId: 'hook_unsafe_abort', + token: 'unsafe-abort', + hasCreatedEvent: true, + abortRequested: true, + abortReason: vmGetterObject(), + }, + ], + ] satisfies [ + string, + QueueItem, + ][])('reports unsafe for %s', async (_, item) => { + expect((await runSuspension(item)).serializationWasPassive).toBe(false); }); it('still serializes recorded inputs successfully (bytes are unaffected)', async () => { const value = vmGetterObject(); - const result = await runSuspension([value]); - expect(result.retainedStepInputsSafe).toBe(false); + const result = await runStep([value]); + expect(result.serializationWasPassive).toBe(false); // The step is still prepared for execution as usual (a single uncreated // step always lands in the lazy inline slice). expect(result.lazyInlineSteps).toHaveLength(1); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 0e50ac4e09..54b4260a03 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -185,13 +185,8 @@ export interface SuspensionHandlerResult { * durably creating the user's hooks doesn't count as runtime overhead. */ hookCreationMs: number; - /** - * 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; + /** Whether serializing new step and hook data was passive. */ + serializationWasPassive: boolean; } async function createHookEvent({ @@ -433,6 +428,22 @@ export async function handleSuspension({ const compression = (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION; + let serializationWasPassive = true; + async function dehydrateInput(value: unknown): Promise { + const stats: GuestCodeStats = { executions: [] }; + const dehydrated = await dehydrateStepArguments( + value, + runId, + encryptionKey, + suspension.globalThis, + false, + compression, + stats + ); + if (stats.executions.length > 0) serializationWasPassive = false; + return dehydrated as SerializedData; + } + async function disposeHook( queueItem: HookInvocationQueueItem ): Promise { @@ -495,17 +506,10 @@ export async function handleSuspension({ let creationConflicted = false; if (!queueItem.hasCreatedEvent) { - const hookMetadata: SerializedData | undefined = + const hookMetadata = typeof queueItem.metadata === 'undefined' ? undefined - : ((await dehydrateStepArguments( - queueItem.metadata, - runId, - encryptionKey, - suspension.globalThis, - false, - compression - )) as SerializedData); + : await dehydrateInput(queueItem.metadata); const hookEvent: CreateEventRequest = { eventType: 'hook_created' as const, specVersion: SPEC_VERSION_CURRENT, @@ -553,14 +557,10 @@ export async function handleSuspension({ hooksNeedingAbort.map(async (queueItem) => { try { // Dehydrate the abort payload for storage - const abortPayload = await dehydrateStepArguments( - { aborted: true, reason: queueItem.abortReason }, - runId, - encryptionKey, - suspension.globalThis, - false, - compression - ); + const abortPayload = await dehydrateInput({ + aborted: true, + reason: queueItem.abortReason, + }); // Create hook_received event with abort payload await createGuarded({ @@ -632,18 +632,6 @@ export async function handleSuspension({ // racing with concurrent handlers on step execution. const createdStepCorrelationIds = new Set(); - // 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 *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 serializer records exactly that into this sink (see - // ../serialization/hardened.ts); when any input in the batch records an - // execution, 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 guestCodeStats: GuestCodeStats = { executions: [] }; - // Lazy inline start: defer the step_created write for up to // `getMaxInlineSteps()` steps the caller will run inline (in parallel). Each // step is created on the fly by the lazy `step_started` executeStep sends @@ -753,24 +741,11 @@ export async function handleSuspension({ // order, before the concurrent dehydration runs). const stepOrder = batchOrderCounter++; const stepOp = (async () => { - // Per-step sink, merged below: the dehydrate wrapper emits span - // attributes from the sink it is handed, so sharing one across - // steps would re-emit (and misattribute) earlier steps' entries. - const stepGuestCode: GuestCodeStats = { executions: [] }; - const dehydratedInput = await dehydrateStepArguments( - { - args: queueItem.args, - closureVars: queueItem.closureVars, - thisVal: queueItem.thisVal, - }, - runId, - encryptionKey, - suspension.globalThis, - false, - compression, - stepGuestCode - ); - guestCodeStats.executions.push(...stepGuestCode.executions); + const dehydratedInput = await dehydrateInput({ + args: queueItem.args, + closureVars: queueItem.closureVars, + thisVal: queueItem.thisVal, + }); // Deferred (lazy) inline step: skip the step_created write — the // caller's inline executeStep will send a lazy step_started carrying // this input, and the world creates the step (entity + synthetic @@ -781,7 +756,7 @@ export async function handleSuspension({ lazyInlineByCorrelationId.set(queueItem.correlationId, { correlationId: queueItem.correlationId, stepName: queueItem.stepName, - dehydratedInput: dehydratedInput as SerializedData, + dehydratedInput, }); return; } @@ -1176,20 +1151,6 @@ export async function handleSuspension({ // step_created and re-dispatches, and recovers the run instead of orphaning it. await settlePhase(ops); - // The step-input dehydrations above have settled, so the sink is final. - const retainedStepInputsSafe = guestCodeStats.executions.length === 0; - if (!retainedStepInputsSafe) { - runtimeLogger.debug( - 'Serializing step inputs executed workflow code; falling back to replay instead of retaining the VM', - { - workflowRunId: runId, - executions: guestCodeStats.executions - .slice(0, 5) - .map((e) => (e.detail ? `${e.kind}(${e.detail})` : e.kind)), - } - ); - } - // 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. @@ -1237,7 +1198,7 @@ export async function handleSuspension({ hasAttributeEvents: attributeItems.length > 0, hasHookEvents: hooksNeedingCreation.length > 0, hookCreationMs, - retainedStepInputsSafe, + serializationWasPassive, reportedEventCount: reportedEvents, }; } diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts index 78a9379cee..71c2d34fc2 100644 --- a/packages/core/src/step.ts +++ b/packages/core/src/step.ts @@ -1,12 +1,12 @@ import { FatalError, ReplayDivergenceError } from '@workflow/errors'; import { withResolvers } from '@workflow/utils'; import { EventConsumerResult } from './events-consumer.js'; -import { type StepInvocationQueueItem, WorkflowSuspension } from './global.js'; +import type { StepInvocationQueueItem } from './global.js'; import { stepLogger } from './logger.js'; import { awaitEarlierDeliveries, registerDeliveryBarrier, - scheduleWhenIdle, + scheduleWorkflowSuspension, type WorkflowOrchestratorContext, } from './private.js'; import type { Serializable } from './schemas.js'; @@ -59,15 +59,7 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) { // Crucially, if we got here, then this step Promise does // not resolve so that the user workflow code does not proceed any further. // Notify the workflow handler that this step has not been run / has not completed yet. - const generation = ctx.suspensionGeneration; - scheduleWhenIdle(ctx, () => { - // A retained session may have resumed past this boundary while - // the timer was queued; a stale signal must not fire. - if (generation !== ctx.suspensionGeneration) return; - ctx.onWorkflowError( - new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) - ); - }); + scheduleWorkflowSuspension(ctx); return EventConsumerResult.NotConsumed; } diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 9ce4a39883..295d313847 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -386,16 +386,15 @@ async function createWorkflowSessionInner( state = WorkflowSuspension.is(error) ? { type: 'suspended', suspension: error } : { type: 'replay' }; - // Each parked step consumer schedules its own (identical) suspension - // signal; the first one lands here, and bumping the generation makes - // the step-consumer guard drop the rest at fire time. + // Step and hook consumers can schedule the same suspension. The first + // signal advances the generation so the rest no-op. workflowContext.suspensionGeneration++; interruption.reject(error); return; } case 'suspended': // Same-boundary duplicates were staled by the generation bump above, - // so anything landing here is out-of-band — an unguarded sleep/hook/ + // so anything landing here is out-of-band — an unguarded sleep/ // attribute signal or a divergence. Those boundaries are unretainable // (the runtime demotes them too), so fall back to replay. state = { type: 'replay' }; diff --git a/packages/core/src/workflow/hook.ts b/packages/core/src/workflow/hook.ts index 16e14ccedc..c475f627cd 100644 --- a/packages/core/src/workflow/hook.ts +++ b/packages/core/src/workflow/hook.ts @@ -13,12 +13,11 @@ import type { HookConflictEvent } from '@workflow/world'; import { getSerializationClass, RUN_CLASS_ID } from '../class-serialization.js'; import type { Hook, HookOptions } from '../create-hook.js'; import { EventConsumerResult } from '../events-consumer.js'; -import { WorkflowSuspension } from '../global.js'; import { webhookLogger } from '../logger.js'; import { awaitEarlierDeliveries, registerDeliveryBarrier, - scheduleWhenIdle, + scheduleWorkflowSuspension, type WorkflowOrchestratorContext, } from '../private.js'; import type { Run } from '../runtime/run.js'; @@ -178,11 +177,7 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { (promises.length > 0 && payloadsQueue.length === 0) || (getConflictPromises.length > 0 && !hasCreated && !hasConflict) ) { - scheduleWhenIdle(ctx, () => { - ctx.onWorkflowError( - new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) - ); - }); + scheduleWorkflowSuspension(ctx); } return EventConsumerResult.NotConsumed; } @@ -502,11 +497,7 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { } if (eventLogEmpty) { - scheduleWhenIdle(ctx, () => { - ctx.onWorkflowError( - new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) - ); - }); + scheduleWorkflowSuspension(ctx); } promises.push(resolvers); @@ -546,11 +537,7 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { } if (eventLogEmpty) { - scheduleWhenIdle(ctx, () => { - ctx.onWorkflowError( - new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) - ); - }); + scheduleWorkflowSuspension(ctx); } getConflictPromises.push(resolvers); @@ -581,11 +568,7 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { // never deliver another hook_received after disposal. if (promises.length > 0) { promises.length = 0; - scheduleWhenIdle(ctx, () => { - ctx.onWorkflowError( - new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis) - ); - }); + scheduleWorkflowSuspension(ctx); } webhookLogger.debug('Hook disposed', { correlationId, token });