diff --git a/.changeset/async-step-completed.md b/.changeset/async-step-completed.md new file mode 100644 index 0000000000..24fc9e3c20 --- /dev/null +++ b/.changeset/async-step-completed.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': minor +--- + +Add opt-in `WORKFLOW_ASYNC_STEP_COMPLETED=1`: defer a single sequential inline step's `step_completed` write so the next replay iteration (and, under optimistic start, the next step's body) overlaps it, with all subsequent durable writes and the queue ack ordered behind the deferred write. Only engages with no open hook or wait, on turbo's first delivery — the one case where no concurrent orchestrator invocation can exist. diff --git a/.changeset/inline-delta-turbo-unlocks.md b/.changeset/inline-delta-turbo-unlocks.md new file mode 100644 index 0000000000..0ce6c7a760 --- /dev/null +++ b/.changeset/inline-delta-turbo-unlocks.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Keep the inline event-log delta fast path active with open hooks when `WORKFLOW_PRECONDITION_GUARD=1` and the World declares `capabilities.preconditionGuard`. The lazy inline `step_started` claim now carries the guard snapshot so a stale replay's claim is fenced (412 → fresh replay), and guard-enforced batches with an open hook take the await-then-run path so a fenced claim never executes the step body. diff --git a/.changeset/world-capabilities.md b/.changeset/world-capabilities.md new file mode 100644 index 0000000000..cea7332918 --- /dev/null +++ b/.changeset/world-capabilities.md @@ -0,0 +1,6 @@ +--- +'@workflow/world': minor +'@workflow/world-vercel': patch +--- + +Add an optional `capabilities?: WorldCapabilities` field to the World interface so implementations can declare backend feature support (`preconditionGuard`, `maxConcurrency`) instead of the runtime inferring it from environment variables; the Vercel World declares both. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 581be18c07..1e1316d2b1 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -43,7 +43,9 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: disabled - Set `1` to enable an optimistic-concurrency guard for event creation: replay-context event creations send a `stateUpdatedAt` snapshot timestamp, and a backend that supports the guard rejects a creation with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) when a newer out-of-band event (a received hook or a completed step) was recorded after that snapshot. - On rejection the runtime reloads the event log and retries, falling back to a queue re-invocation with a fresh replay if it cannot catch up. -- Backends that do not support the guard ignore the snapshot. +- When enabled — and the World declares that it enforces the guard (`capabilities.preconditionGuard`; the Vercel World does) — the runtime also keeps the per-step event-log delta optimization (consuming the delta returned by a step's terminal write instead of issuing an extra `events.list` per step) active while the run has an open hook. Without an enforced guard, an open hook disables it. +- While a hook is open on a guard-enforcing deployment, inline steps take the await-then-run path even when optimistic inline start is enabled: the step's `step_started` claim carries the snapshot and is awaited before the body runs, so a claim the backend rejects as stale never executes user code. +- Backends that do not support the guard ignore the snapshot; they must not declare the capability, so guard-dependent optimizations stay off against them even when the flag is set. ## Inline execution @@ -63,6 +65,7 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: enabled - Fast path for a run's first delivery. +- Stops forcing optimistic inline step start once the run creates a hook or wait. - Set `0` or `false` to disable. ### `WORKFLOW_OPTIMISTIC_INLINE_START` @@ -72,6 +75,13 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Use only when step side effects are idempotent. - Set `0` or `false` to force it off, including the first-delivery fast path used by `WORKFLOW_TURBO`. +### `WORKFLOW_ASYNC_STEP_COMPLETED` + +- Default: disabled +- Set `1` to let the runtime defer an inline step's `step_completed` write: the next replay iteration runs on locally-synthesized events (and, with optimistic start active, the next step's body starts) while the write settles in the background. Every subsequent durable write is chained behind the deferred write, and the queue message is only acknowledged after it settles. +- Only engages for a single sequential inline step with no open hook or wait, and only on turbo's first delivery — the one case where no concurrent orchestrator invocation can exist. +- Requires idempotent step side effects, more strongly than `WORKFLOW_OPTIMISTIC_INLINE_START`: a crash while the write is in flight re-executes the step on redelivery, which can produce a different result than the one the next step's body already consumed. + ### `WORKFLOW_INLINE_OWNERSHIP` - Default: enabled diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index e687ae44de..fdad3a04d5 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -1,4 +1,6 @@ import { + EntityConflictError, + PreconditionFailedError, RUN_ERROR_CODES, ThrottleError, WorkflowWorldError, @@ -1425,9 +1427,9 @@ describe('workflowEntrypoint turbo mode', () => { const ORIG_TURBO = process.env.WORKFLOW_TURBO; const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; - // Default: turbo ON (unset) and the global optimistic flag OFF (unset). Any - // optimistic behavior observed in these tests therefore comes from turbo - // forcing it — never from WORKFLOW_OPTIMISTIC_INLINE_START. + // Default: turbo ON (unset) and the global optimistic flag OFF (unset). + // Any optimistic behavior observed in these tests therefore comes from + // turbo forcing it — never from WORKFLOW_OPTIMISTIC_INLINE_START. beforeEach(() => { delete process.env.WORKFLOW_TURBO; delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; @@ -1716,6 +1718,652 @@ describe('workflowEntrypoint turbo mode', () => { }); }); +describe('workflowEntrypoint deferred step_completed (WORKFLOW_ASYNC_STEP_COMPLETED)', () => { + const ORIG_TURBO = process.env.WORKFLOW_TURBO; + const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + const ORIG_ASYNC = process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + + // Default: turbo ON (unset), everything else OFF (unset). Individual tests + // opt into WORKFLOW_ASYNC_STEP_COMPLETED. + beforeEach(() => { + delete process.env.WORKFLOW_TURBO; + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + delete process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + asyncOrder = []; + }); + afterEach(() => { + if (ORIG_TURBO === undefined) delete process.env.WORKFLOW_TURBO; + else process.env.WORKFLOW_TURBO = ORIG_TURBO; + if (ORIG_OPT === undefined) { + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + } else { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = ORIG_OPT; + } + if (ORIG_ASYNC === undefined) { + delete process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + } else { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = ORIG_ASYNC; + } + setWorld(undefined); + vi.clearAllMocks(); + waitUntilPromises.length = 0; + }); + + let asyncOrder: string[] = []; + registerStepFunction('asyncStepA', async () => { + asyncOrder.push('bodyA'); + return 'a'; + }); + registerStepFunction('asyncStepB', async () => { + asyncOrder.push('bodyB'); + return 'b'; + }); + + // Two sequential steps: B's input work depends on A having resolved in the + // workflow, so B's body running while A's step_completed create is still + // in flight proves the deferred-completed overlap. + const twoStepWorkflow = `const a = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("asyncStepA"); + const b = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("asyncStepB"); + async function workflow() { await a(); await b(); return 'done'; } + ;globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set("workflow", workflow);`; + + /** + * Drives the handler with a first-invocation message at the given delivery + * attempt. `completedGates` holds the resolution of a step's step_completed + * create until released (keyed by stepName); `completedRejects` makes the + * create reject instead. The mock returns no inline delta on terminal + * writes, so the deferred path exercises its reconcile-by-fetch fallback. + */ + async function driveAsyncCompleted(opts: { + runId: string; + attempt?: number; + completedGates?: Record>; + completedRejects?: Record; + /** World capabilities to declare (absent by default — fail closed). */ + capabilities?: { preconditionGuard?: boolean; maxConcurrency?: boolean }; + /** Workflow source to run (defaults to twoStepWorkflow). */ + source?: string; + }) { + const { runId } = opts; + const attempt = opts.attempt ?? 1; + const order = asyncOrder; + const durable: Event[] = []; + let seq = 0; + const rec = (data: any): Event => { + seq += 1; + const e = { + eventId: `e-${seq}`, + runId, + createdAt: new Date(), + ...data, + } as Event; + durable.push(e); + return e; + }; + const runEntity: WorkflowRun = { + runId, + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments([], runId, 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: 'test-deployment', + }; + + const queueMock = vi.fn(async () => ({ messageId: null })); + const eventsCreate = vi.fn(async (_runId: string, data: any) => { + if (data.eventType === 'run_started') { + return { run: runEntity, events: [] as Event[] }; + } + if (data.eventType === 'step_started') { + const d = data.eventData as { stepName?: string; input?: unknown }; + order.push(`step_started:${d?.stepName}`); + if (d?.input !== undefined) { + rec({ + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: data.correlationId, + eventData: { stepName: d.stepName, input: d.input }, + }); + } + return { + event: rec(data), + step: { + runId, + stepId: data.correlationId, + stepName: d?.stepName, + status: 'running' as const, + attempt: 1, + input: d?.input, + startedAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }, + ...(d?.input !== undefined ? { stepCreated: true } : {}), + }; + } + if (data.eventType === 'step_completed') { + const stepName = (data.eventData as { stepName?: string })?.stepName; + order.push(`step_completed_called:${stepName}`); + const reject = stepName ? opts.completedRejects?.[stepName] : undefined; + if (reject) throw reject; + const gate = stepName ? opts.completedGates?.[stepName] : undefined; + if (gate) await gate; + order.push(`step_completed_resolved:${stepName}`); + return { event: rec(data) }; + } + if (data.eventType === 'run_completed') { + order.push('run_completed'); + return { event: rec(data) }; + } + if (data.eventType === 'run_failed') { + order.push('run_failed'); + return { event: rec(data) }; + } + return { event: rec(data) }; + }); + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn( + (_p: string, handler: (m: unknown, md: unknown) => Promise) => + async () => { + await handler( + { + runId, + requestedAt: new Date('2024-01-01T00:00:00.000Z'), + runInput: { + input: await dehydrateWorkflowArguments( + [], + runId, + undefined, + [] + ), + deploymentId: 'test-deployment', + workflowName: 'workflow', + specVersion: SPEC_VERSION_CURRENT, + executionContext: {}, + }, + }, + { + requestId: 'req_async_completed', + attempt, + queueName: '__wkf_workflow_workflow', + messageId: 'msg_async_completed', + } + ); + return new Response(null, { status: 204 }); + } + ), + events: { + create: eventsCreate, + list: vi.fn(async () => ({ + data: [...durable], + hasMore: false, + cursor: 'cursor_async_completed', + })), + }, + runs: { get: vi.fn(async () => runEntity) }, + queue: queueMock, + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as any); + + const handlerPromise = workflowEntrypoint(opts.source ?? twoStepWorkflow)( + new Request('https://example.test') + ) as Promise; + return { handlerPromise, order, eventsCreate, queueMock }; + } + + it('runs the next step body while the previous step_completed create is in flight, and orders the next step_started behind it', async () => { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = '1'; + let release!: () => void; + const gate = new Promise((r) => { + release = r; + }); + + const { handlerPromise, order } = await driveAsyncCompleted({ + runId: 'wrun_async_completed_overlap', + completedGates: { asyncStepA: gate }, + }); + + // Step B's body runs while step A's terminal write is still held open — + // impossible when the loop awaits the write before replaying. + await vi.waitFor(() => expect(order).toContain('bodyB'), { + timeout: 15_000, + }); + expect(order).not.toContain('step_completed_resolved:asyncStepA'); + // ...but B's own durable write is chained behind A's terminal write, so + // the event log can never record B's start without A's completion. + expect(order).not.toContain('step_started:asyncStepB'); + + release(); + const res = await handlerPromise; + expect(res.status).toBe(204); + expect(order).toContain('run_completed'); + expect(order.indexOf('step_completed_resolved:asyncStepA')).toBeLessThan( + order.indexOf('step_started:asyncStepB') + ); + }); + + it('awaits the step_completed write before the next replay when the flag is off (default)', async () => { + let release!: () => void; + const gate = new Promise((r) => { + release = r; + }); + + const { handlerPromise, order } = await driveAsyncCompleted({ + runId: 'wrun_async_completed_default', + completedGates: { asyncStepA: gate }, + }); + + await vi.waitFor( + () => expect(order).toContain('step_completed_called:asyncStepA'), + { timeout: 15_000 } + ); + // Give the loop a real chance to (incorrectly) run ahead. + await new Promise((r) => setTimeout(r, 150)); + expect(order).not.toContain('bodyB'); + + release(); + const res = await handlerPromise; + expect(res.status).toBe(204); + expect(order.indexOf('step_completed_resolved:asyncStepA')).toBeLessThan( + order.indexOf('bodyB') + ); + }); + + it('fails closed when the deferred write is superseded: no ack, no run_completed, no run_failed', async () => { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = '1'; + + const { handlerPromise, order } = await driveAsyncCompleted({ + runId: 'wrun_async_completed_conflict', + completedRejects: { + asyncStepA: new EntityConflictError('step already in a terminal state'), + }, + }); + + // The conflict is converted to a PreconditionFailedError (this invocation + // speculated on a result another writer superseded) and propagates out of + // the handler, leaving the message unacked so redelivery recovers the + // step. It must not be treated as a benign skip, and it must never + // surface as run_failed (transient concurrency, not a workflow error). + await expect(handlerPromise).rejects.toThrow(/superseded/); + expect(order).not.toContain('run_completed'); + expect(order).not.toContain('run_failed'); + }); + + it('settles a superseded deferred write before the run_failed path: the run re-invokes instead of failing on speculative state', async () => { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = '1'; + + // The workflow consumes step A's (speculative) result and then throws. + // The failure handling path acks the message (recovery re-queue or + // run_failed write), so it must settle the deferred write first: here the + // write was superseded (EntityConflict → PreconditionFailedError), which + // must force a fresh replay — NOT permanently fail the run on a result + // another writer already superseded. + const throwingWorkflow = `const a = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("asyncStepA"); + async function workflow() { await a(); throw new Error('boom'); } + ;globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set("workflow", workflow);`; + + const { handlerPromise, order, queueMock } = await driveAsyncCompleted({ + runId: 'wrun_async_completed_fail_path', + source: throwingWorkflow, + completedRejects: { + asyncStepA: new EntityConflictError('step already in a terminal state'), + }, + }); + + const res = await handlerPromise; + expect(res.status).toBe(204); + // Re-invoked for a fresh replay (turbo reinvoke enqueues an explicit + // continuation) instead of writing run_failed on speculative state. + expect(order).not.toContain('run_failed'); + expect(order).not.toContain('run_completed'); + expect(queueMock).toHaveBeenCalled(); + }); +}); + +describe('workflowEntrypoint inline-delta gate with open hooks', () => { + const ORIG_GUARD = process.env.WORKFLOW_PRECONDITION_GUARD; + const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + + beforeEach(() => { + delete process.env.WORKFLOW_PRECONDITION_GUARD; + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + deltaGateBodyRuns = []; + }); + afterEach(() => { + if (ORIG_GUARD === undefined) { + delete process.env.WORKFLOW_PRECONDITION_GUARD; + } else { + process.env.WORKFLOW_PRECONDITION_GUARD = ORIG_GUARD; + } + if (ORIG_OPT === undefined) { + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + } else { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = ORIG_OPT; + } + setWorld(undefined); + vi.clearAllMocks(); + waitUntilPromises.length = 0; + }); + + registerStepFunction('deltaGateStep', async () => undefined); + let deltaGateBodyRuns: string[] = []; + registerStepFunction('deltaGateStepB', async () => { + deltaGateBodyRuns.push('B'); + return undefined; + }); + + // A fire-and-forget hook alongside a single awaited step: the suspension + // creates a hook AND schedules one lazy inline step, leaving the hook open + // for the rest of the run. + const hookAndStepWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + const s = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("deltaGateStep"); + async function workflow() { + const hook = createHook({ token: 'delta-gate-token' }); + return await s(); + };globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set("workflow", workflow);`; + + // Same open hook, but two sequential steps — used to interleave an + // out-of-band event between step A's completion and step B's claim. + const hookAndTwoStepWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + const a = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("deltaGateStep"); + const b = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("deltaGateStepB"); + async function workflow() { + const hook = createHook({ token: 'delta-gate-token' }); + await a(); + return await b(); + };globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set("workflow", workflow);`; + + /** + * Drives the handler with a continuation message (no runInput, so turbo is + * off and the initial events.list — which supplies the cursor the delta + * diffs against — runs). Returns the events.create mock so tests can + * inspect the step-terminal write's params for `sinceCursor` and the + * step_started claims' params for `stateUpdatedAt`. + */ + async function driveDeltaGate( + runId: string, + opts: { + /** + * World capabilities to declare. Absent by default — capability-gated + * fast paths must fail closed without them. + */ + capabilities?: { preconditionGuard?: boolean; maxConcurrency?: boolean }; + /** Workflow source to run (defaults to hookAndStepWorkflow). */ + source?: string; + /** + * Reject the lazy step_started claim for this stepName with the given + * error (once), simulating a guard-enforcing backend 412-ing a stale + * claim after an out-of-band event bumped the run's marker. + */ + rejectClaimOnce?: { stepName: string; error: Error }; + } = {} + ) { + const workflowRun: WorkflowRun = { + runId, + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments([], runId, 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: 'test-deployment', + }; + + const durableEvents: Event[] = []; + const recordEvent = (data: any): Event => { + // ULID-shaped event IDs so the runtime's stateUpdatedAt snapshot + // (derived from the latest event id's ULID timestamp) is computable. + const created = { + eventId: `evnt_${ulid()}`, + runId, + createdAt: new Date(), + ...data, + } as Event; + durableEvents.push(created); + return created; + }; + + let claimRejected = false; + const eventsCreate = vi.fn( + async (_runId: string, data: any, _params?: any) => { + if (data.eventType === 'run_started') { + return { run: workflowRun, events: [] as Event[] }; + } + if (data.eventType === 'step_started') { + const lazy = data.eventData as { + stepName?: string; + input?: unknown; + }; + if ( + opts.rejectClaimOnce && + !claimRejected && + lazy?.stepName === opts.rejectClaimOnce.stepName + ) { + claimRejected = true; + throw opts.rejectClaimOnce.error; + } + if (lazy?.input !== undefined) { + recordEvent({ + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: data.correlationId, + eventData: { stepName: lazy.stepName, input: lazy.input }, + }); + } + return { + event: recordEvent(data), + step: { + runId, + stepId: data.correlationId, + stepName: lazy?.stepName, + status: 'running' as const, + attempt: 1, + input: lazy?.input, + startedAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }, + ...(lazy?.input !== undefined ? { stepCreated: true } : {}), + }; + } + // The World returns no delta (like a World that doesn't support + // sinceCursor), so the next iteration falls back to events.list — + // these tests only assert whether the delta was REQUESTED. + return { event: recordEvent(data) }; + } + ); + + const queueMock = vi.fn(async () => ({ messageId: null })); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: opts.capabilities, + createQueueHandler: vi.fn( + ( + _prefix: string, + handler: (message: unknown, metadata: unknown) => Promise + ) => { + return async () => { + await handler( + { + runId, + requestedAt: new Date('2024-01-01T00:00:00.000Z'), + }, + { + requestId: 'req_delta_gate', + attempt: 1, + queueName: '__wkf_workflow_workflow', + messageId: 'msg_delta_gate', + } + ); + return new Response(null, { status: 204 }); + }; + } + ), + events: { + create: eventsCreate, + list: vi.fn(async () => ({ + data: [...durableEvents], + hasMore: false, + cursor: 'cursor_delta_gate', + })), + }, + runs: { get: vi.fn(async () => workflowRun) }, + queue: queueMock, + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as any); + + const handler = workflowEntrypoint(opts.source ?? hookAndStepWorkflow); + const res = (await handler( + new Request('https://example.test') + )) as Response; + return { res, eventsCreate, queueMock }; + } + + function stepCompletedParams(eventsCreate: ReturnType) { + const call = eventsCreate.mock.calls.find( + (c) => (c[1] as any).eventType === 'step_completed' + ); + expect(call).toBeDefined(); + return call?.[2] as { sinceCursor?: string } | undefined; + } + + it('requests the inline delta despite the open hook when the precondition guard is enabled and the World enforces it', async () => { + process.env.WORKFLOW_PRECONDITION_GUARD = '1'; + const { res, eventsCreate } = await driveDeltaGate( + 'wrun_delta_gate_guard_on', + { capabilities: { preconditionGuard: true } } + ); + expect(res.status).toBe(204); + // The suspension created a hook (left open) and one lazy inline step — + // with an enforced guard, a hook_received missed by the delta window is + // fenced by the outside-event marker, so the fast path stays active. + expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBe( + 'cursor_delta_gate' + ); + expect(eventsCreate.mock.calls).toContainEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: 'hook_created' }), + ]) + ); + }); + + it('does not request the inline delta with an open hook when the guard is disabled', async () => { + const { res, eventsCreate } = await driveDeltaGate( + 'wrun_delta_gate_guard_off' + ); + expect(res.status).toBe(204); + // Without the guard there is no fence for a hook_received landing in the + // delta window, so the conservative gate keeps the fetch path. + expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBeUndefined(); + }); + + it('does not request the inline delta when the env flag is set but the World does not enforce the guard', async () => { + process.env.WORKFLOW_PRECONDITION_GUARD = '1'; + // No capabilities declared: the env flag only makes the runtime SEND + // snapshots — a World that ignores stateUpdatedAt provides no 412 fence, + // so the relaxation must fail closed to the conservative gate. + const { res, eventsCreate } = await driveDeltaGate( + 'wrun_delta_gate_guard_no_capability' + ); + expect(res.status).toBe(204); + expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBeUndefined(); + }); + + it('abandons the batch and re-invokes when a stale lazy claim is rejected by the guard (interleaved hook_received)', async () => { + process.env.WORKFLOW_PRECONDITION_GUARD = '1'; + // Simulates the interleaving the fence exists for: after step A's + // terminal write, an out-of-band hook_received bumps the run's marker; + // the next replay (working from a view that misses it) schedules step B, + // whose lazy step_started claim the backend rejects as stale (412). + const { res, eventsCreate, queueMock } = await driveDeltaGate( + 'wrun_delta_gate_stale_claim', + { + capabilities: { preconditionGuard: true }, + source: hookAndTwoStepWorkflow, + rejectClaimOnce: { + stepName: 'deltaGateStepB', + error: new PreconditionFailedError( + 'stale stateUpdatedAt: a newer outside event exists' + ), + }, + } + ); + // The handler responds normally: the rejection is mapped to an abandoned + // batch + re-invocation (a `{ timeoutSeconds: 0 }` redelivery outside + // turbo), never a run_failed. + expect(res.status).toBe(204); + // Step B's claim was issued from a loaded (non-empty) log, so it carried + // the guard snapshot — that is what lets the backend fence it. (The very + // first batch of a run loads an empty log and has no snapshot to send; + // the guard is best-effort there, matching the suspension creates.) + const rejectedClaim = eventsCreate.mock.calls.find( + (c) => + (c[1] as any).eventType === 'step_started' && + ((c[1] as any).eventData as { stepName?: string })?.stepName === + 'deltaGateStepB' + ); + expect(typeof (rejectedClaim?.[2] as any)?.stateUpdatedAt).toBe('number'); + // The fenced step never ran its body and never wrote events. + expect(deltaGateBodyRuns).toEqual([]); + expect(eventsCreate.mock.calls).not.toContainEqual( + expect.arrayContaining([ + expect.objectContaining({ + eventType: 'step_completed', + eventData: expect.objectContaining({ stepName: 'deltaGateStepB' }), + }), + ]) + ); + expect(eventsCreate.mock.calls).not.toContainEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: 'run_failed' }), + ]) + ); + // Outside turbo the re-invocation is a redelivery of the current message, + // not an explicit continuation enqueue. + expect(queueMock).not.toHaveBeenCalled(); + }); + + it('suppresses optimistic start on guarded stale-sensitive batches: a 412-fenced step never runs its body even with WORKFLOW_OPTIMISTIC_INLINE_START=1', async () => { + process.env.WORKFLOW_PRECONDITION_GUARD = '1'; + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; + // Same interleaving as above, but with optimistic start enabled globally. + // Without suppression, executeStep would begin step B's body immediately + // (before the claim settles) and only discard the result after the 412 — + // the side effects would already have run. With an open hook and the + // guard in force, the runtime takes the await-then-run path instead, so + // the fence covers user code: the rejected claim means the body never + // begins. + const { res, eventsCreate } = await driveDeltaGate( + 'wrun_delta_gate_stale_claim_optimistic', + { + capabilities: { preconditionGuard: true }, + source: hookAndTwoStepWorkflow, + rejectClaimOnce: { + stepName: 'deltaGateStepB', + error: new PreconditionFailedError( + 'stale stateUpdatedAt: a newer outside event exists' + ), + }, + } + ); + expect(res.status).toBe(204); + expect(deltaGateBodyRuns).toEqual([]); + expect(eventsCreate.mock.calls).not.toContainEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: 'run_failed' }), + ]) + ); + }); +}); + describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => { const ORIG_TURBO = process.env.WORKFLOW_TURBO; diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 5fcd0fd680..120fc9a958 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -37,13 +37,16 @@ import { runtimeLogger } from './logger.js'; import { getMaxQueueDeliveries, getReplayDivergenceMaxRetries, + isAsyncStepCompletedEnabled, isInlineOwnershipEnabled, + isOptimisticInlineStartEnabled, isTurboEnabled, } from './runtime/constants.js'; import { getQueueOverhead, getWorkflowQueueName, handleHealthCheckMessage, + isPreconditionGuardEnabled, loadWorkflowRunEvents, type MutableEventLog, memoizeEncryptionKey, @@ -259,7 +262,7 @@ function rootRunIdFrom( } /** - * Whether the run has any hook or wait that an out-of-band writer could + * Whether the run has a hook and/or wait that an out-of-band writer could * append an event for between an inline step's `step_completed` write and * the next replay — namely an open hook (a `hook_created` not yet * `hook_disposed`, which a webhook receiver can resolve with @@ -267,15 +270,14 @@ function rootRunIdFrom( * `wait_completed`, which the wait timer can resolve with * `wait_completed`). * - * This gates the inline-delta fast path. The delta returned by the + * This gates the inline-delta fast path (per kind — see the gate) and the + * turbo forced-optimistic-start latch. The delta returned by the * step-terminal write is the event log as of that write; it is consumed * on the NEXT loop iteration, so any event a concurrent writer appends in * that window would be present in a real `events.list` fetch but absent - * from the stale delta. Missing such an event for one replay can durably - * commit a scheduling decision (e.g. inline-executing the loser of a - * `Promise.race([hook, step])`) that the eventual full replay cannot - * consume — a `ReplayDivergenceError`. When no hook or wait is open, the - * only out-of-band writer is cancellation, which is benign to observe one + * from the stale delta — the replay observes it one iteration later than + * the fetch path would. When no hook or wait is open, the only + * out-of-band writer is cancellation, which is benign to observe one * iteration late (the next entity write is rejected against the terminal * run and the run is already terminal), so the fast path is safe. * @@ -283,7 +285,10 @@ function rootRunIdFrom( * step's terminal write and are therefore already inside the returned * delta. */ -function hasOpenHookOrWait(events: Event[]): boolean { +function openHookAndWaitState(events: Event[]): { + openHook: boolean; + openWait: boolean; +} { const disposedHookIds = new Set(); const completedWaitIds = new Set(); for (const e of events) { @@ -292,21 +297,23 @@ function hasOpenHookOrWait(events: Event[]): boolean { completedWaitIds.add(e.correlationId); } } + let openHook = false; + let openWait = false; for (const e of events) { if ( e.eventType === 'hook_created' && !disposedHookIds.has(e.correlationId) ) { - return true; - } - if ( + openHook = true; + } else if ( e.eventType === 'wait_created' && !completedWaitIds.has(e.correlationId) ) { - return true; + openWait = true; } + if (openHook && openWait) break; } - return false; + return { openHook, openWait }; } /** @@ -555,6 +562,53 @@ export function workflowEntrypoint( cursor: string | null; } | null = null; + // Deferred step_completed (WORKFLOW_ASYNC_STEP_COMPLETED, + // opt-in): when the previous iteration's single inline step + // deferred its terminal write, `pendingSynthesizedEvents` + // holds local stand-ins for that step's events — consumed by + // exactly ONE subsequent replay pass (appended transiently to + // the replay's event view, never persisted into + // cachedEvents) — and `deferredCompleted.promise` tracks the + // in-flight write chained with its reconcile step (merge the + // canonical events into cachedEvents, advance the cursor). + // Every subsequent durable write is chained behind that + // promise, and every ack path settles it (fail-closed) so + // the queue message is never acknowledged while the terminal + // write is in flight or failed. At most one deferral is + // pending at a time: a new one's write is transitively + // ordered behind the old one's reconcile via the chained + // step_started create. + let pendingSynthesizedEvents: Event[] | null = null; + let deferredCompleted: { + promise: Promise; + } | null = null; + // Await (fail-closed) the pending deferred write, clearing + // both pieces of state. A rejection propagates: converted + // EntityConflict/RunExpired surface as + // PreconditionFailedError (fresh replay via reinvoke where a + // catch routes it) and everything else rethrows so the + // unacked message redelivers and owned recovery re-runs the + // step. The stale synthesized tail is dropped either way — + // after reconcile the canonical events are (or will be, via + // the next fetch) in cachedEvents. + const settleDeferredCompleted = async (): Promise => { + pendingSynthesizedEvents = null; + const settling = deferredCompleted; + if (!settling) return; + deferredCompleted = null; + await settling.promise; + }; + // Compose the turbo run-ready barrier with the deferred + // write for executeStep's pre-write ordering. Rejection of + // either rejects the composition (executeStep's chained + // step_started create then rejects, exactly like a failed + // create on the awaited path). + const composeStepWriteBarrier = ( + a?: Promise, + b?: Promise + ): Promise | undefined => + a && b ? Promise.all([a, b]) : (a ?? b); + // Shared state: set by either the background step path // or the run_started setup below. let workflowRun: WorkflowRun | undefined; @@ -692,6 +746,13 @@ export function workflowEntrypoint( const reinvoke = async ( delaySeconds: number ): Promise<{ timeoutSeconds: number } | undefined> => { + // Both reinvoke shapes ack the current message (the + // explicit continuation acks it; `{ timeoutSeconds }` + // reschedules it as done-for-now), so a deferred + // step_completed must be durable first — a failure here + // rethrows, leaving the message unacked so redelivery's + // owned recovery re-runs the step. + await settleDeferredCompleted(); assertNoInFlightOwnedSteps('reinvoke'); if (!turbo) return { timeoutSeconds: delaySeconds }; await queueMessage( @@ -1110,12 +1171,27 @@ export function workflowEntrypoint( while (true) { loopIteration++; + // Deferred step_completed: unless this iteration is the + // designated synthesized-tail replay (the one iteration + // allowed to run ahead of the in-flight write), settle the + // deferral before doing anything else — every other load + // path fetches from the server and must observe the + // reconciled (canonical) event log. Fail-closed: a + // rejection propagates out of the handler, the message + // stays unacked, and redelivery recovers the step. + if (deferredCompleted && !pendingSynthesizedEvents) { + await settleDeferredCompleted(); + } + // Replay-budget check: bail out (retry or fail) if // non-step time within this invocation has exceeded // the configured budget. Step bodies are excluded // because replayBudget.pause()/resume() bracket every // `executeStep` call. if (replayBudget.isExhausted()) { + // Bail-out acks or fails the run — the deferred write + // must settle first even on the tail-pending fast path. + await settleDeferredCompleted(); await handleReplayBudgetExhausted({ runId, workflowName, @@ -1135,6 +1211,10 @@ export function workflowEntrypoint( Date.now() - invocationStartTime >= NO_INLINE_REPLAY_AFTER_MS ) { + // Queueing the continuation + returning acks this + // message — the deferred write must settle first even + // on the tail-pending fast path. + await settleDeferredCompleted(); runtimeLogger.info( 'V2 timeout reached, re-scheduling workflow', { @@ -1166,7 +1246,27 @@ export function workflowEntrypoint( // The server always returns a cursor when there are events (even on the // final page), so we can reliably use it for incremental loading. let events: Event[]; - if (pendingInlineDelta && cachedEvents) { + // Deferred step_completed: local stand-ins for the + // previous inline step's events, appended to THIS + // replay's event view only (see replayEvents below) and + // never persisted into cachedEvents — the reconcile step + // chained on the in-flight write merges the canonical + // events there instead. + let transientSynthesizedTail: Event[] | null = null; + if (pendingSynthesizedEvents) { + // Fast path: the previous iteration's inline step + // deferred its terminal write; replay over the + // synthesized events without any fetch. No other + // writer can have appended in the window (the + // deferCompletedWrite gate requires no open hook or + // wait and no concurrent orchestrator invocation), so + // the synthesized tail is the complete delta modulo a + // cancellation, which is observed one iteration late — + // the same tolerance as the inline-delta path below. + transientSynthesizedTail = pendingSynthesizedEvents; + pendingSynthesizedEvents = null; + events = cachedEvents ?? []; + } else if (pendingInlineDelta && cachedEvents) { // Fast path: the previous iteration's inline step // terminal write returned the authoritative event-log // delta since the pre-write cursor, so we consume it @@ -1393,6 +1493,17 @@ export function workflowEntrypoint( // Update cache reference (may have been set for first time) cachedEvents = events; + // Deferred step_completed: this replay's event view is + // the durable cache plus the transient synthesized tail. + // The concat isolates the tail — nothing below persists + // it, and the reconcile step merges the canonical + // events into cachedEvents when the in-flight write + // lands (before any subsequent durable write, which all + // chain behind it). + const replayEvents = transientSynthesizedTail + ? [...events, ...transientSynthesizedTail] + : events; + // Latency telemetry: judge TTFS eligibility against the // invocation's first snapshot. Waits completed above // would already disqualify via the event-type check, so @@ -1429,7 +1540,7 @@ export function workflowEntrypoint( const result = await runWorkflow( workflowCode, workflowRun, - events, + replayEvents, encryptionKey, stepHydrationCache, // Turbo: the end-of-run drain inside runWorkflow commits @@ -1452,6 +1563,14 @@ export function workflowEntrypoint( // result. The catch below lets PreconditionFailedError // propagate to the queue for re-invocation. try { + // A deferred step_completed must be durable — and its + // reconcile must have merged the canonical events, so + // the stateUpdatedAt snapshot below reflects the run's + // own latest terminal event — before the run's + // terminal write. Fail-closed: a rejection routes + // through the catch below (PreconditionFailedError → + // fresh replay; transient → redelivery). + await settleDeferredCompleted(); // Turbo: a workflow that finishes with no steps reaches // here before the backgrounded run_started; order the // terminal write after it so the run exists. @@ -1566,6 +1685,14 @@ export function workflowEntrypoint( requestId, eventLog: suspensionLog, runReadyBarrier, + // Deferred step_completed: order every suspension + // write after the in-flight terminal write and its + // reconcile (fail-closed — see + // SuspensionHandlerParams.preWriteBarrier). + // Undefined when nothing is deferred. The pure + // inline hot path writes nothing here, so the + // barrier does not serialize it. + preWriteBarrier: deferredCompleted?.promise, }); } catch (suspensionError) { // A suspension create whose stale (412) rejection @@ -1967,6 +2094,10 @@ export function workflowEntrypoint( if (suspensionResult.hasAwaitedHookCreation) { return await reinvoke(0); } + // Returning acks this message — settle any deferred + // step_completed first (defensive: the suspension's + // chained writes normally settled it already). + await settleDeferredCompleted(); return; } @@ -1977,6 +2108,12 @@ export function workflowEntrypoint( // this the replay-budget check at the top of the // next loop iteration would (incorrectly) charge // the step body against the budget. + // Open hooks/waits in the cumulative log, computed + // once for the two gates below. + const openHookWaitState = openHookAndWaitState( + cachedEvents ?? [] + ); + // Inline-delta fast path gate. We request the delta — // and on the next iteration consume it in place of the // events.list — only when ALL hold: @@ -1986,40 +2123,111 @@ export function workflowEntrypoint( // the initial load). // - This is the clean single-step sequential case: // this suspension produced exactly one step and no - // hooks or waits (`err.{step,hook,wait}Count`), that - // one step is the lone pending step - // (`pendingSteps.length === 1`) and the lone inline - // step (`lazyInlineSteps.length === 1` — no parallel + // waits (`err.{step,wait}Count`), that one step is + // the lone pending step (`pendingSteps.length === 1`) + // and the lone inline step + // (`lazyInlineSteps.length === 1` — no parallel // siblings queued to background handlers, and no other // inline step writing its own events out of band). - // - No pending wait timer from THIS suspension. - // - The run has NO pre-existing open hook or wait. This - // plus the per-suspension counts above is the - // load-bearing safety check: the delta snapshots the - // log at the step_completed write but is consumed on - // the next replay, so an out-of-band hook_received / - // wait_completed landing in that window would be in a - // real fetch yet absent from the stale delta — - // risking a divergent replay. With no hook/wait open - // (neither carried over nor created this suspension), - // the only out-of-band writer is cancellation, which - // is safe to observe one iteration late. See - // hasOpenHookOrWait. + // - No pending wait timer from THIS suspension, and no + // open wait in the cumulative log: a concurrent + // `wait_completed` landing after the delta snapshot + // does not bump the outside-event marker, so nothing + // fences a replay from the stale delta. + // - No open (or this-suspension-created) hook — UNLESS + // the precondition guard is enabled AND the World + // declares it actually enforces the guard + // (`capabilities.preconditionGuard`; the env flag + // alone only makes the runtime SEND snapshots, which + // an unsupporting backend ignores — no fence). The + // delta snapshots the log at the step_completed + // write but is consumed on the next replay, so an + // out-of-band `hook_received` landing in that window + // is absent from the delta and observed one + // iteration later than a real fetch would observe + // it. That staleness is qualitatively the same + // read-to-write race the fetch path already has (an + // event can land right after `events.list` returns + // and before the suspension's writes); with an + // enforced guard it is also fenced: `hook_received` + // bumps the per-run outside-event marker, so every + // durable write the stale replay attempts is + // rejected with 412 — its guarded suspension creates + // (retried over the reloaded log, or exhausted into + // a queue re-invocation), AND the lazy step_started + // claim of its next inline step, which carries the + // snapshot too (threaded below via + // `stateUpdatedAt`; on rejection the batch is + // abandoned and re-invoked for a fresh replay, so a + // stale view can never commit a step). Hooks created + // by THIS suspension are inside the delta (their + // `hook_created` lands before the step-terminal + // write), so only their `hook_received` responses + // are subject to the same fenced window. Without an + // enforced guard there is no fence, so keep the + // conservative gate. + // - With no hook or wait open at all, the only + // out-of-band writer is cancellation, which is safe + // to observe one iteration late. See + // openHookAndWaitState. // // When more than one step runs inline, each writes its // own events and the per-write delta would be partial, so // the delta is not requested (the gate below is false for // multi-step) and the next iteration does a normal fetch. - const requestInlineDelta = - typeof preInlineWriteCursor === 'string' && + // Whether the precondition guard is actually in force: + // enabled by env AND enforced by the World. The env + // flag alone only makes the runtime send snapshots, + // which an unsupporting backend ignores (no fence). + const guardEnforced = + isPreconditionGuardEnabled() && + world.capabilities?.preconditionGuard === true; + + // The clean single-step sequential shape shared by the + // inline-delta and deferred-completed gates: this + // suspension produced exactly one step and no waits, + // that one step is the lone pending step and the lone + // inline step (no parallel siblings queued to + // background handlers, no other inline step writing + // its own events out of band, no owned-recovery + // re-execution), and no wait timer is pending. + const singleCleanInlineStep = err.stepCount === 1 && - err.hookCount === 0 && err.waitCount === 0 && pendingSteps.length === 1 && lazyInlineSteps.length === 1 && ownedRecoverySteps.length === 0 && - !suspensionResult.waitTimeout && - !hasOpenHookOrWait(cachedEvents ?? []); + !suspensionResult.waitTimeout; + + const requestInlineDelta = + typeof preInlineWriteCursor === 'string' && + singleCleanInlineStep && + !openHookWaitState.openWait && + (guardEnforced || + (err.hookCount === 0 && + !openHookWaitState.openHook)); + + // Stale-sensitive batch: a hook is open in the run (or + // was created by this suspension, so its hook_received + // can land any moment) — an out-of-band event can make + // the view this batch was scheduled from stale. With + // the guard in force, the fence rejects a stale + // claim's durable writes — but it cannot un-run a step + // BODY that optimistic start began before the claim + // settled. Suppress optimistic start for these batches + // (take await-then-run) so a 412-fenced step never + // executes user code at all: the fence then covers + // side effects, not just the event log. Costs one + // claim round-trip per step while a hook is open, only + // on guard-enforcing deployments. Without the guard + // nothing 412s, so suppression would buy nothing — + // stale-view exposure there is the pre-existing + // optimistic-start contract (idempotent side effects). + const suppressOptimisticStart = + guardEnforced && + (openHookWaitState.openHook || + err.hookCount > 0 || + suspensionResult.hasHookEvents); // Turbo mode forces optimistic inline start for this // batch — but only while the run is still "clean" (a pure @@ -2043,15 +2251,77 @@ export function workflowEntrypoint( // step suspensions). Once any hook or wait is open in the // cumulative log, resume/parallel invocations are possible // for the rest of the run, so turbo must latch off - // permanently — checked here via `hasOpenHookOrWait` over - // the cumulative `cachedEvents`. + // permanently — checked here via `openHookAndWaitState` + // over the cumulative `cachedEvents`. + // + // NOTE: `WORKFLOW_SEQUENTIAL_REPLAYS=1` (per-run flow + // topics consumed with `maxConcurrency: 1`) would in + // principle waive this latch — serialized orchestrator + // invocations restore the single-handler guarantee for + // the whole delivery. The waiver is intentionally NOT + // taken: the env var is a runtime-process setting that + // cannot prove the BUILT flow trigger actually carries + // `maxConcurrency: 1` (it must be set at build time + // too, and some integrations write their own trigger + // config), and `capabilities.maxConcurrency` only + // declares queue support, not deployed configuration. + // Until the build emits a verifiable signal that the + // deployed trigger is serialized, the conservative + // latch stays. const forceOptimisticStart = turbo && + !suspensionResult.hasAttributeEvents && !suspensionResult.waitTimeout && !suspensionResult.hasHookEvents && - !suspensionResult.hasAttributeEvents && !suspensionResult.hasAwaitedHookCreation && - !hasOpenHookOrWait(cachedEvents ?? []); + !openHookWaitState.openHook && + !openHookWaitState.openWait; + + // Deferred step_completed gate + // (WORKFLOW_ASYNC_STEP_COMPLETED, opt-in). Beyond the + // clean single-step shape, all of these must hold: + // + // - No hook or wait open anywhere in the run and none + // created by this suspension — stricter than the + // delta gate's guard-relaxed arm. An open hook/wait + // admits out-of-band writers into the deferral + // window, and the synthesized-tail replay would not + // see their events even one iteration late (it + // skips the fetch entirely). + // - This is turbo's first delivery (no resume or + // peer invocation exists yet), so no concurrent + // replay can observe (and act on) the durable log + // mid-window. A WORKFLOW_SEQUENTIAL_REPLAYS arm + // could widen this to later deliveries, but the + // runtime env var cannot prove the built flow + // trigger is actually serialized — see the NOTE on + // forceOptimisticStart above; same follow-up. + // - Optimistic inline start is active (forced by + // turbo, or via WORKFLOW_OPTIMISTIC_INLINE_START): + // without it the next step's body waits for its + // step_started create, which chains behind the + // deferred write anyway — deferring would buy + // nothing. + // + // What deferral changes: the next replay iteration + // consumes locally-synthesized events for this step + // (see pendingSynthesizedEvents) and the next step's + // body may run against this step's result before that + // result is durable. A crash inside the window + // re-executes THIS step on redelivery, which can + // produce a different result than the one the next + // body already consumed — step side effects must + // tolerate that (a stronger requirement than + // optimistic start alone, hence the opt-in flag). + const deferCompletedWrite = + isAsyncStepCompletedEnabled() && + singleCleanInlineStep && + err.hookCount === 0 && + !openHookWaitState.openHook && + !openHookWaitState.openWait && + turbo && + (forceOptimisticStart || + isOptimisticInlineStartEnabled()); // Execute the inline steps in parallel. The replay // budget is paused for the whole batch — step duration is @@ -2086,82 +2356,141 @@ export function workflowEntrypoint( turbo, }); + // Precondition-guard snapshot for the inline + // step_started claims: the lazy claim is the first + // durable write of a hot-path step (its step_created + // is deferred), so without a snapshot it would bypass + // the guard entirely and a stale replay could claim — + // and commit — a step scheduled off a view that misses + // an out-of-band event. `stateUpdatedAtForCreate` + // returns undefined when the guard env flag is off, so + // this is a no-op outside guarded deployments; Worlds + // that don't enforce the guard ignore it. + const inlineClaimStateUpdatedAt = + stateUpdatedAtForCreate(cachedEvents ?? []); + replayBudget.pause(); let stepResults: Awaited< ReturnType >[]; + const stepExecutionPromises = inlineExecutions.map( + (s, stepIndex) => { + const run = () => + executeStep({ + world, + workflowRunId: runId, + workflowDeploymentId: workflowRun.deploymentId, + workflowName, + workflowStartedAt, + rootRunId: rootRunIdFrom( + workflowRun.attributes, + runId + ), + stepId: s.correlationId, + stepName: s.stepName, + runSpecVersion: workflowRun.specVersion, + // Lazy inline start: send the deferred step's + // input on step_started so the world creates + // the step on the fly. Absent for + // owned-recovery steps, whose input hydrates + // from the existing step entity. + lazyStepInput: s.lazyStepInput, + // Inline ownership: stamp (or re-stamp) this + // invocation's queue message ID on the + // step_started, so wake replays see the body + // as in flight here and suppress the + // immediate requeue (workflow#2780). + ownerMessageId: metadata.messageId, + // Turbo: force optimistic start and hold the + // lazy step_started until the backgrounded + // run_started lands (the body still runs + // immediately). Both are undefined/false + // outside turbo. The barrier additionally + // composes any deferred step_completed from + // the previous iteration, so this step's + // writes are ordered after (and fail with) + // that write — the body still starts + // immediately under optimistic start. + forceOptimisticStart, + // Guard-enforced batches with an open hook + // await the claim before running the body, so + // a 412-fenced step never executes user code — + // see suppressOptimisticStart above. + suppressOptimisticStart, + // The barrier composes any deferred + // step_completed from the previous iteration, + // so this step's writes are ordered after + // (and fail with) that write. + runReadyBarrier: composeStepWriteBarrier( + runReadyBarrier, + deferredCompleted?.promise + ), + stateUpdatedAt: inlineClaimStateUpdatedAt, + // Deferred terminal write — see the + // deferCompletedWrite gate above. False for + // multi-step batches by construction. + deferCompletedWrite, + ...(stepIndex === 0 && + s.lazyStepInput !== undefined && + latencyTracking + ? { latencyTracking } + : {}), + ...(requestInlineDelta && preInlineWriteCursor + ? { + inlineDeltaSinceCursor: + preInlineWriteCursor, + } + : {}), + }); + // Invariant bookkeeping: this invocation owns + // these bodies until they settle — see + // assertNoInFlightOwnedSteps. + inFlightOwnedSteps.add(s.correlationId); + // Lazy steps are brand-new (their create-claim + // is the exactly-once gate), but an + // owned-recovery step already exists and its + // delayed backstop message may fire mid-body + // in this same process — route those through + // the in-process single-flight. + const executed = + s.lazyStepInput === undefined + ? runStepSingleFlight( + runId, + s.correlationId, + run + ) + : run(); + return executed.finally(() => + inFlightOwnedSteps.delete(s.correlationId) + ); + } + ); try { stepResults = await Promise.all( - inlineExecutions.map((s, stepIndex) => { - const run = () => - executeStep({ - world, - workflowRunId: runId, - workflowDeploymentId: - workflowRun.deploymentId, - workflowName, - workflowStartedAt, - rootRunId: rootRunIdFrom( - workflowRun.attributes, - runId - ), - stepId: s.correlationId, - stepName: s.stepName, - runSpecVersion: workflowRun.specVersion, - // Lazy inline start: send the deferred step's - // input on step_started so the world creates - // the step on the fly. Absent for - // owned-recovery steps, whose input hydrates - // from the existing step entity. - lazyStepInput: s.lazyStepInput, - // Inline ownership: stamp (or re-stamp) this - // invocation's queue message ID on the - // step_started, so wake replays see the body - // as in flight here and suppress the - // immediate requeue (workflow#2780). - ownerMessageId: metadata.messageId, - // Turbo: force optimistic start and hold the - // lazy step_started until the backgrounded - // run_started lands (the body still runs - // immediately). Both are undefined/false - // outside turbo. - forceOptimisticStart, - runReadyBarrier, - ...(stepIndex === 0 && - s.lazyStepInput !== undefined && - latencyTracking - ? { latencyTracking } - : {}), - ...(requestInlineDelta && preInlineWriteCursor - ? { - inlineDeltaSinceCursor: - preInlineWriteCursor, - } - : {}), - }); - // Invariant bookkeeping: this invocation owns - // these bodies until they settle — see - // assertNoInFlightOwnedSteps. - inFlightOwnedSteps.add(s.correlationId); - // Lazy steps are brand-new (their create-claim - // is the exactly-once gate), but an - // owned-recovery step already exists and its - // delayed backstop message may fire mid-body - // in this same process — route those through - // the in-process single-flight. - const executed = - s.lazyStepInput === undefined - ? runStepSingleFlight( - runId, - s.correlationId, - run - ) - : run(); - return executed.finally(() => - inFlightOwnedSteps.delete(s.correlationId) - ); - }) + stepExecutionPromises ); + } catch (stepErr) { + // A stale (412) rejection of an inline step_started + // claim: the loaded view this batch was scheduled + // from is behind an out-of-band event (e.g. a + // received hook), so the claim was fenced by the + // guard and no step events were written. Abandon the + // batch — any optimistic body result is discarded by + // executeStep's reconciliation — and re-invoke for a + // fresh replay that observes the new event. Wait for + // the sibling executions to settle first so no owned + // body is in flight when the ack path runs. + if (PreconditionFailedError.is(stepErr)) { + await Promise.allSettled(stepExecutionPromises); + runtimeLogger.warn( + 'Inline step claim rejected as stale; re-invoking run for a fresh replay', + { workflowRunId: runId, loopIteration } + ); + // The finally below resumes the replay budget + // before this return completes. + return await reinvoke(0); + } + throw stepErr; } finally { replayBudget.resume(); } @@ -2337,6 +2666,85 @@ export function workflowEntrypoint( if (inlineExecutions.length === 1) { const only = stepResults[0]; if ( + only.type === 'completed' && + only.deferredCompleted + ) { + // Deferred step_completed: stash the synthesized + // events for the next iteration's fetch-free + // replay, and chain the reconcile step on the + // in-flight write: merge the canonical events into + // cachedEvents (via the delta the World returned, + // or an incremental fetch when it didn't) and + // advance the cursor. Every subsequent durable + // write chains behind this promise; every ack path + // settles it. An EntityConflict/RunExpired + // rejection means another writer owns the step's + // outcome — since this invocation already + // speculated on OUR result, convert it to + // PreconditionFailedError so the existing routing + // forces a fresh replay instead of treating it as + // a benign skip. + pendingSynthesizedEvents = + only.deferredCompleted.synthesizedEvents; + const deferredStepId = + inlineExecutions[0].correlationId; + // The step stays "owned in flight" until its + // terminal write settles — see + // assertNoInFlightOwnedSteps. + inFlightOwnedSteps.add(deferredStepId); + const reconciled = only.deferredCompleted.write + .then(async (delta) => { + const cache = cachedEvents; + if (!cache) return; + let deltaEvents: Event[]; + if (delta) { + deltaEvents = delta.events; + eventsCursor = delta.cursor ?? eventsCursor; + } else { + const loaded = await loadWorkflowRunEvents( + runId, + eventsCursor ?? undefined + ); + deltaEvents = loaded.events; + eventsCursor = loaded.cursor ?? eventsCursor; + } + if (deltaEvents.length > 0) { + const existingIds = new Set( + cache.map((e) => e.eventId) + ); + for (const e of deltaEvents) { + if (!existingIds.has(e.eventId)) { + existingIds.add(e.eventId); + cache.push(e); + } + } + } + }) + .catch((writeErr) => { + if ( + EntityConflictError.is(writeErr) || + RunExpiredError.is(writeErr) + ) { + throw new PreconditionFailedError( + `Deferred step_completed write for step ${deferredStepId} was superseded (${ + writeErr instanceof Error + ? writeErr.message + : String(writeErr) + }); forcing a fresh replay to observe the durable outcome` + ); + } + throw writeErr; + }) + .finally(() => { + inFlightOwnedSteps.delete(deferredStepId); + }); + deferredCompleted = { promise: reconciled }; + // Rejections are consumed by the chained writes / + // settle sites; this detached catch only prevents + // an unhandledRejection if the process dies before + // any of them attach. + reconciled.catch(() => {}); + } else if ( only.type === 'completed' && only.inlineDelta && !only.inlineDelta.hasMore @@ -2394,6 +2802,38 @@ export function workflowEntrypoint( throw err; } + // Deferred step_completed + // (WORKFLOW_ASYNC_STEP_COMPLETED, opt-in): every path + // below acks the queue message (the ReplayDivergence + // recovery re-queue and the run_failed terminal write + // both return after acknowledging). Per the invariant + // (see the deferredCompleted declaration) the message + // must never be acked while the deferred write is in + // flight or failed, so settle it fail-closed FIRST. A + // settle rejection takes precedence over the current + // error: a PreconditionFailedError (the deferred write + // was superseded) forces a fresh replay instead of + // failing the run on speculative, already-superseded + // state (mirrors the PreconditionFailedError branch + // above and run_completed's settle-then-catch), while + // anything else (transient) rethrows so the message + // stays unacked and redelivery re-runs the step. A + // no-op outside the opt-in (deferredCompleted null). + if (deferredCompleted) { + try { + await settleDeferredCompleted(); + } catch (settleErr) { + if (PreconditionFailedError.is(settleErr)) { + runtimeLogger.warn( + 'Deferred step_completed superseded during failure handling; re-invoking run for a fresh replay instead of failing', + { workflowRunId: runId, loopIteration } + ); + return await reinvoke(0); + } + throw settleErr; + } + } + let terminalError = err; if (ReplayDivergenceError.is(err)) { const divergenceCount = diff --git a/packages/core/src/runtime/constants.test.ts b/packages/core/src/runtime/constants.test.ts index 208250e8e6..9cb5277ca4 100644 --- a/packages/core/src/runtime/constants.test.ts +++ b/packages/core/src/runtime/constants.test.ts @@ -7,6 +7,7 @@ import { getMaxQueueDeliveries, getReplayTimeoutMs, INLINE_OWNERSHIP_LEASE_SECONDS, + isAsyncStepCompletedEnabled, isInlineOwnershipEnabled, isOptimisticInlineStartEnabled, isOptimisticInlineStartExplicitlyDisabled, @@ -289,6 +290,35 @@ describe('isTurboEnabled', () => { }); }); +describe('isAsyncStepCompletedEnabled', () => { + const originalEnv = process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + } else { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = originalEnv; + } + }); + + it('defaults to disabled when unset', () => { + delete process.env.WORKFLOW_ASYNC_STEP_COMPLETED; + expect(isAsyncStepCompletedEnabled()).toBe(false); + }); + + it('is enabled by exactly "1"', () => { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = '1'; + expect(isAsyncStepCompletedEnabled()).toBe(true); + }); + + it('stays disabled for other values', () => { + for (const value of ['', '0', 'true', 'yes']) { + process.env.WORKFLOW_ASYNC_STEP_COMPLETED = value; + expect(isAsyncStepCompletedEnabled()).toBe(false); + } + }); +}); + describe('getMaxQueueDeliveries', () => { const ENV = 'WORKFLOW_MAX_QUEUE_DELIVERIES'; diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 8db7a3f723..71e0519b38 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -275,6 +275,29 @@ export function isTurboEnabled(): boolean { return !(raw === '0' || raw.toLowerCase() === 'false'); } +/** + * Whether `WORKFLOW_ASYNC_STEP_COMPLETED=1` opts the runtime into deferring an + * inline step's `step_completed` write: instead of awaiting the write before + * the next replay iteration, the runtime continues on locally-synthesized + * events (feeding the step's result into the workflow to discover and start + * the next step's body) while the write settles in the background. Every + * subsequent durable write is chained behind the deferred write, so the event + * log can never record downstream events without their upstream terminal. + * + * Only engages on the single clean sequential inline step shape, with no open + * hook or wait, and only on turbo's first delivery — the one case where no + * concurrent orchestrator invocation can exist (see the `deferCompletedWrite` + * gate in runtime.ts). + * + * Off by default: a crash while the write is in flight re-executes the step + * on redelivery, which can produce a different result than the one the next + * step's body already consumed — a stronger idempotency requirement on step + * side effects than optimistic inline start alone. + */ +export function isAsyncStepCompletedEnabled(): boolean { + return process.env.WORKFLOW_ASYNC_STEP_COMPLETED === '1'; +} + /** * Whether inline step ownership is enabled (default ON). When on, the lazy * `step_started` that creates an inline step records the owning queue diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index db0477c64a..e3fa1c6d46 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -127,6 +127,49 @@ export interface StepExecutorParams { * handler is the sole inline writer for the run on this iteration. */ inlineDeltaSinceCursor?: string; + /** + * Precondition-guard snapshot (epoch ms of the latest event the caller's + * replay loaded) to attach to this step's `step_started` claim. On the lazy + * inline path the claim is the step's FIRST durable write (its + * `step_created` is deferred), so without this the claim would bypass the + * optimistic-concurrency guard entirely: a replay working from a stale view + * could claim — and then commit — a step scheduled without observing an + * out-of-band event. A guard-enforcing World rejects a stale claim with + * `PreconditionFailedError` (412); executeStep does NOT translate that + * rejection (re-claiming in place would still commit the stale schedule), + * so it propagates for the caller to abandon the batch and force a fresh + * replay. Undefined when the guard is disabled or the caller has no + * snapshot; Worlds that don't enforce the guard ignore it. + */ + stateUpdatedAt?: number; + /** + * Suppress optimistic inline start for this step regardless of + * `WORKFLOW_OPTIMISTIC_INLINE_START` / `forceOptimisticStart`: take the + * await-then-run path so the body only runs after the `step_started` claim + * succeeds. Set by the runtime for guard-enforced batches that are + * stale-sensitive (an open hook means an out-of-band event can make the + * scheduling view stale): the guard's 412 fence can reject a stale claim's + * durable writes, but it cannot un-run a body that optimistic start began + * before the claim settled — awaiting the claim extends the fence to user + * code. Wins over `forceOptimisticStart` and the env flag. + */ + suppressOptimisticStart?: boolean; + /** + * Defer the `step_completed` write (opt-in via + * `WORKFLOW_ASYNC_STEP_COMPLETED=1`; see the `deferCompletedWrite` gate in + * runtime.ts). When set and the step completes cleanly with no pending ops, + * the terminal `events.create` is initiated but NOT awaited: the result + * carries the in-flight write plus locally-synthesized + * `step_created`/`step_started`/`step_completed` events so the caller can + * feed the step's outcome into the next replay iteration while the write + * settles. The caller owns the ordering invariant: no subsequent durable + * write may be issued until the deferred write resolves, and no queue ack + * may happen until it settles. Only meaningful together with + * `lazyStepInput` (the synthesis needs the step input); ignored otherwise, + * and ignored when the step has pending background ops, failed, retried, or + * lost its claim — all of those await their writes as usual. + */ + deferCompletedWrite?: boolean; /** * Force optimistic inline start regardless of * `WORKFLOW_OPTIMISTIC_INLINE_START`. Set by turbo mode on the first delivery @@ -170,6 +213,24 @@ export interface InlineEventDelta { hasMore: boolean; } +/** + * A deferred `step_completed` write (see + * {@link StepExecutorParams.deferCompletedWrite}). `write` settles when the + * terminal `events.create` does, resolving to the inline delta the World + * returned (or undefined — the caller then reconciles via an incremental + * fetch). `synthesizedEvents` are local stand-ins for the canonical + * `step_created`/`step_started`/`step_completed` this step will produce: + * content-equal to the durable events (same correlationId, stepName, input, + * result bytes) but with locally-minted event IDs and timestamps. They are + * only valid for feeding ONE subsequent replay pass and must never be + * persisted into the durable event cache — the canonical events replace them + * at reconcile time. + */ +export interface DeferredCompletedWrite { + write: Promise; + synthesizedEvents: Event[]; +} + /** * Result of a step execution attempt. The caller decides what to do * based on the result type (e.g., queue workflow continuation, replay inline, etc.). @@ -185,6 +246,12 @@ export type StepExecutionResult = type: 'completed'; hasPendingOps?: boolean; inlineDelta?: InlineEventDelta; + /** + * Present when the caller requested {@link StepExecutorParams.deferCompletedWrite} + * and this completion's terminal write was deferred. Mutually exclusive + * with `inlineDelta` (the delta, if any, arrives on `deferredCompleted.write`). + */ + deferredCompleted?: DeferredCompletedWrite; } | { type: 'failed' } | { type: 'retry'; timeoutSeconds: number } @@ -390,6 +457,10 @@ export async function executeStep( // flag, so an explicit opt-out wins over turbo's force. const optimisticStart = params.lazyStepInput !== undefined && + // Stale-sensitive guarded batches await the claim so the 412 fence + // covers the body, not just durable writes — see + // StepExecutorParams.suppressOptimisticStart. + params.suppressOptimisticStart !== true && (isOptimisticInlineStartEnabled() || (params.forceOptimisticStart === true && !isOptimisticInlineStartExplicitlyDisabled())); @@ -434,20 +505,30 @@ export async function executeStep( // RSFS measures the run_started-to-POST stretch, and the barrier // wait IS part of that stretch under turbo. stepStartPostSentAtMs = Date.now(); - return world.events.create(workflowRunId, { - eventType: 'step_started', - specVersion: SPEC_VERSION_CURRENT, - correlationId: stepId, - eventData: { - stepName, - workflowName, - input: params.lazyStepInput, - // Inline-ownership stamp — see StepExecutorParams.ownerMessageId. - ...(params.ownerMessageId !== undefined - ? { ownerMessageId: params.ownerMessageId } - : {}), + return world.events.create( + workflowRunId, + { + eventType: 'step_started', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { + stepName, + workflowName, + input: params.lazyStepInput, + // Inline-ownership stamp — see StepExecutorParams.ownerMessageId. + ...(params.ownerMessageId !== undefined + ? { ownerMessageId: params.ownerMessageId } + : {}), + }, }, - }); + // Guard the claim — see StepExecutorParams.stateUpdatedAt. A stale + // (412) rejection surfaces via reconcileOptimisticStart as a + // non-translatable error: the body result is discarded and the + // rejection propagates to the caller. + params.stateUpdatedAt !== undefined + ? { stateUpdatedAt: params.stateUpdatedAt } + : undefined + ); } ); optimisticStartSettled = startedPromise.then( @@ -486,20 +567,30 @@ export async function executeStep( ? { ownerMessageId: params.ownerMessageId } : {}; stepStartPostSentAtMs = Date.now(); - const startResult = await world.events.create(workflowRunId, { - eventType: 'step_started', - specVersion: SPEC_VERSION_CURRENT, - correlationId: stepId, - eventData: - params.lazyStepInput !== undefined - ? { - stepName, - workflowName, - input: params.lazyStepInput, - ...ownershipStamp, - } - : { stepName, ...ownershipStamp }, - }); + const startResult = await world.events.create( + workflowRunId, + { + eventType: 'step_started', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: + params.lazyStepInput !== undefined + ? { + stepName, + workflowName, + input: params.lazyStepInput, + ...ownershipStamp, + } + : { stepName, ...ownershipStamp }, + }, + // Guard the claim — see StepExecutorParams.stateUpdatedAt. A stale + // (412) rejection is intentionally NOT translated by + // startErrorToResult below, so it propagates to the caller for a + // fresh replay. + params.stateUpdatedAt !== undefined + ? { stateUpdatedAt: params.stateUpdatedAt } + : undefined + ); if (!startResult.step) { throw new WorkflowRuntimeError( @@ -1120,6 +1211,111 @@ export async function executeStep( return { type: 'retry', timeoutSeconds }; } + const completedEventData = { + stepName, + workflowName, + result: result as Uint8Array, + ...latencyEventData, + }; + + // Deferred terminal write (WORKFLOW_ASYNC_STEP_COMPLETED): initiate the + // step_completed create but return before it settles, so the caller can + // consume this step's result (via the synthesized events below) and start + // the next step's body while the write is in flight. Only on the clean + // path: the step completed, its ops flushed, and it was a lazy inline + // start (the synthesis needs the input). Failure handling moves to the + // caller: an EntityConflict / RunExpired rejection means another writer + // owns the outcome, and since the caller has already speculated on OUR + // result it must force a fresh replay rather than treat it as the benign + // skip the awaited path below maps it to. + if ( + params.deferCompletedWrite && + opsSettled && + params.lazyStepInput !== undefined + ) { + const deltaRequested = params.inlineDeltaSinceCursor !== undefined; + const write = world.events + .create( + workflowRunId, + { + eventType: 'step_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: completedEventData, + }, + deltaRequested + ? { sinceCursor: params.inlineDeltaSinceCursor } + : undefined + ) + .then((res) => extractInlineDelta(res, deltaRequested)); + // Keep the platform alive until the write settles even if the handler + // exits abnormally. The caller consumes the real rejection (every ack + // path settles the deferred write first), so this only logs. + safeWaitUntil(write, (err) => { + runtimeLogger.warn('Deferred step_completed write failed', { + workflowRunId, + stepId, + error: err instanceof Error ? err.message : String(err), + }); + }); + // Local stand-ins for the canonical events this step produces (the + // World's lazy-start synthesizes step_created + step_started from the + // started create; step_completed comes from the write above). + // Content-equal to the durable events; event IDs and timestamps are + // locally minted and only valid for one replay pass — see + // DeferredCompletedWrite. + const now = new Date(); + const synthesizedEvents: Event[] = [ + { + eventId: `evnt_local_${stepId}_created`, + runId: workflowRunId, + createdAt: now, + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { + stepName, + workflowName, + input: params.lazyStepInput, + }, + }, + { + eventId: `evnt_local_${stepId}_started`, + runId: workflowRunId, + createdAt: now, + eventType: 'step_started', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { + stepName, + attempt: 1, + workflowName, + ...(params.ownerMessageId !== undefined + ? { ownerMessageId: params.ownerMessageId } + : {}), + }, + }, + { + eventId: `evnt_local_${stepId}_completed`, + runId: workflowRunId, + createdAt: now, + eventType: 'step_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: completedEventData, + }, + ]; + span?.setAttributes({ + ...Attribute.StepStatus('completed'), + ...Attribute.StepResultType(typeof result), + }); + return { + type: 'completed', + hasPendingOps: false, + deferredCompleted: { write, synthesizedEvents }, + }; + } + // Create step_completed event outside the step execution failure path: // persistence failures are infrastructure errors and should redeliver the // queue message, not become user step_retrying/step_failed events. @@ -1131,12 +1327,7 @@ export async function executeStep( eventType: 'step_completed', specVersion: SPEC_VERSION_CURRENT, correlationId: stepId, - eventData: { - stepName, - workflowName, - result: result as Uint8Array, - ...latencyEventData, - }, + eventData: completedEventData, }, params.inlineDeltaSinceCursor !== undefined ? { sinceCursor: params.inlineDeltaSinceCursor } diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index cf5b821652..1304cce25c 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -1305,13 +1305,16 @@ describe('executeStep optimistic inline start', () => { expect(result.type).toBe('completed'); expect(mockStepFn).toHaveBeenCalledTimes(1); // The lazy step_started carries the input so the world creates the step. + // The third argument is the CreateEventParams slot — undefined here + // because no precondition-guard snapshot (`stateUpdatedAt`) was passed. expect(mockEventsCreate).toHaveBeenCalledWith( 'wrun_test123', expect.objectContaining({ eventType: 'step_started', correlationId: 'step_abc', eventData: expect.objectContaining({ stepName: 'myStep', input: [] }), - }) + }), + undefined ); }); @@ -1521,9 +1524,8 @@ describe('executeStep optimistic inline start', () => { (event as { eventType: string }).eventType === 'step_completed' ); expect(completedWrite).toBeDefined(); - const eventData = ( - completedWrite![1] as { eventData: { rsfs?: number } } - ).eventData; + const eventData = (completedWrite![1] as { eventData: { rsfs?: number } }) + .eventData; expect(eventData.rsfs).toBeDefined(); expect(eventData.rsfs).toBeGreaterThanOrEqual(0); }); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 291a45f57d..28437856cc 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -56,6 +56,22 @@ export interface SuspensionHandlerParams { * where `run_started` was already awaited up front. */ runReadyBarrier?: Promise; + /** + * Fail-closed ordering barrier for a deferred `step_completed` write + * (`WORKFLOW_ASYNC_STEP_COMPLETED`): when present, every event creation this + * suspension performs awaits it first and — unlike `runReadyBarrier`, whose + * rejection is swallowed — REJECTS if it rejects. A rejected barrier means + * the previous step's terminal write never became durable, so committing any + * downstream event (a hook/wait created after consuming that step's result) + * would record effects of a result the log doesn't contain. The rejection + * propagates out of handleSuspension; the runtime maps it to a fresh replay + * or a queue redelivery. The barrier also carries the reconcile step that + * merges the deferred write's canonical events into `eventLog`, so awaiting + * it guarantees the `stateUpdatedAt` snapshot sent with each create reflects + * the run's own latest terminal event (no self-inflicted 412 under the + * precondition guard). + */ + preWriteBarrier?: Promise; } /** @@ -205,6 +221,7 @@ export async function handleSuspension({ requestId, eventLog, runReadyBarrier, + preWriteBarrier, }: SuspensionHandlerParams): Promise { const runId = run.runId; @@ -231,10 +248,16 @@ export async function handleSuspension({ // a replay snapshot, e.g. tests). The guard reloads + retries on a stale // (412) rejection, keeping `eventLog` current in place. All suspension events // are non-run_created events on this run's `runId`. - const createGuarded = ( + const createGuarded = async ( data: CreateEventRequest, params?: CreateEventParams ): Promise => { + // Fail-closed: a rejected preWriteBarrier (the previous step's deferred + // step_completed write failed) must abort this write, not be swallowed + // like the run-ready barrier — see SuspensionHandlerParams.preWriteBarrier. + if (preWriteBarrier) { + await preWriteBarrier; + } if (!eventLog) { return world.events.create(runId, data, params); } @@ -614,6 +637,12 @@ export async function handleSuspension({ (async () => { try { await ensureRunReady(); + // Same fail-closed ordering as createGuarded: never commit an + // attr_set downstream of a deferred step_completed that has not + // become durable (see SuspensionHandlerParams.preWriteBarrier). + if (preWriteBarrier) { + await preWriteBarrier; + } await world.events.create( runId, { diff --git a/packages/world-vercel/src/index.ts b/packages/world-vercel/src/index.ts index 6aef96ff56..989c53c278 100644 --- a/packages/world-vercel/src/index.ts +++ b/packages/world-vercel/src/index.ts @@ -37,6 +37,17 @@ export function createWorld(config?: APIConfig): World { // at least it — workflow-server declared spec-5 support in // vercel/workflow-server#520. specVersion: SPEC_VERSION_SUPPORTS_COMPRESSION, + capabilities: { + // workflow-server enforces the `stateUpdatedAt` optimistic-concurrency + // guard: creations carrying a stale snapshot are rejected with 412 + // (PreconditionFailedError) when the run's outside-event marker is + // newer. See vercel/workflow-server#484. + preconditionGuard: true, + // Vercel Queues supports maxConcurrency-limited consumers, which + // WORKFLOW_SEQUENTIAL_REPLAYS=1 uses for per-run `maxConcurrency: 1` + // flow topics (see queue.ts and @workflow/builders). + maxConcurrency: true, + }, // On Vercel the platform fails the function invocation when the // process exits non-zero, and VQS redelivers the queue message via a // fresh invocation. The core runtime uses this to decide whether diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index ef0fecd4a7..40410a40cd 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -290,6 +290,43 @@ export interface Storage { }; } +/** + * Optional feature capabilities a World implementation declares so the core + * runtime can enable optimizations that depend on backend behavior, instead + * of inferring support from environment variables alone. Every capability + * defaults to "unsupported" when absent — runtime fast paths that rely on + * one must fail closed (keep their conservative behavior) unless the World + * explicitly declares it. + */ +export interface WorldCapabilities { + /** + * The World enforces the optimistic-concurrency precondition guard: an + * event creation carrying a `stateUpdatedAt` snapshot is rejected with a + * `PreconditionFailedError` (412) when a newer out-of-band event (e.g. a + * received hook) was recorded after that snapshot. Worlds that accept but + * ignore `stateUpdatedAt` must leave this unset so runtime optimizations + * that rely on the 412 fence (see `WORKFLOW_PRECONDITION_GUARD`) are not + * enabled without an actual fence behind them. + */ + preconditionGuard?: boolean; + + /** + * The World's queue supports `maxConcurrency`-limited consumption — in + * particular the per-run flow topics consumed with `maxConcurrency: 1` + * that `WORKFLOW_SEQUENTIAL_REPLAYS=1` uses to serialize a run's + * orchestrator invocations. Worlds whose queue has no concurrency-limit + * concept must leave this unset. + * + * Note this declares queue *support*, not deployed configuration: the + * serialization also requires the build-time half (a flow trigger emitted + * with `maxConcurrency: 1`), which a runtime process cannot verify today. + * The core runtime therefore does not yet take any fast path from this + * capability alone — it exists so a future build-verified signal can be + * combined with it (and so Worlds document the contract explicitly). + */ + maxConcurrency?: boolean; +} + /** * The "World" interface represents how Workflows are able to communicate with the outside world. */ @@ -312,6 +349,13 @@ export interface World extends Queue, Streamer, Storage { */ specVersion: number; + /** + * Feature capabilities this World implementation supports — see + * {@link WorldCapabilities}. Absent (or absent members) means + * "unsupported": runtime optimizations gated on a capability fail closed. + */ + capabilities?: WorldCapabilities; + /** * Whether calling `process.exit(1)` from a queue handler is observed by * the World as a delivery failure that will be retried.