From c07d011d838ef635f5b0bfff913a39201c676077 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 12:04:06 -0700 Subject: [PATCH 1/8] perf(core): parallel inline steps + optimistic lazy step start Build two latency wins on top of lazy inline step start (#2478), both client-side only (no world/backend changes): 1. Inline up to WORKFLOW_MAX_INLINE_STEPS (default 3) steps per suspension, in parallel. The suspension handler defers step_created for the first N uncreated steps (lazyInlineSteps) and the runtime runs them via Promise.all of lazy step_started events; steps beyond the cap keep their eager step_created and are queued as before. The inline-delta fast path stays single-step only. WORKFLOW_MAX_INLINE_STEPS=1 reproduces prior behavior. 2. Optimistic inline start (WORKFLOW_OPTIMISTIC_INLINE_START, default on): executeStep fires step_started without awaiting and runs the body against locally-synthesized attempt-1 state, reconciling the in-flight start before the terminal write. A lost create-claim (409) / gone / throttle discards the body result; a transient failure propagates for redelivery. Exactly-one terminal write is preserved; bodies may run more than once under contention so inline steps must be idempotent (disable with =0). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../parallel-inline-optimistic-start.md | 6 + .../docs/v5/changelog/lazy-event-creation.md | 114 ++++++++ docs/content/docs/v5/changelog/meta.json | 7 +- packages/core/src/runtime.test.ts | 30 +- packages/core/src/runtime.ts | 270 +++++++++++------- packages/core/src/runtime/constants.test.ts | 95 ++++++ packages/core/src/runtime/constants.ts | 79 +++++ packages/core/src/runtime/step-executor.ts | 146 ++++++++-- .../core/src/runtime/step-handler.test.ts | 120 +++++++- .../src/runtime/suspension-handler.test.ts | 113 ++++++++ .../core/src/runtime/suspension-handler.ts | 65 +++-- 11 files changed, 874 insertions(+), 171 deletions(-) create mode 100644 .changeset/parallel-inline-optimistic-start.md create mode 100644 docs/content/docs/v5/changelog/lazy-event-creation.md diff --git a/.changeset/parallel-inline-optimistic-start.md b/.changeset/parallel-inline-optimistic-start.md new file mode 100644 index 0000000000..a6a8ba75e0 --- /dev/null +++ b/.changeset/parallel-inline-optimistic-start.md @@ -0,0 +1,6 @@ +--- +'workflow': minor +'@workflow/core': minor +--- + +Inline execution now runs up to `WORKFLOW_MAX_INLINE_STEPS` (default 3) steps in parallel per suspension (each lazily created), and starts step bodies optimistically before `step_started` is confirmed (`WORKFLOW_OPTIMISTIC_INLINE_START`, default on) — reconciling the in-flight start before the terminal write so a lost create-claim is discarded. Optimistic bodies may run more than once under contention, so inline steps must be idempotent; disable with `WORKFLOW_OPTIMISTIC_INLINE_START=0`. diff --git a/docs/content/docs/v5/changelog/lazy-event-creation.md b/docs/content/docs/v5/changelog/lazy-event-creation.md new file mode 100644 index 0000000000..96fff126d9 --- /dev/null +++ b/docs/content/docs/v5/changelog/lazy-event-creation.md @@ -0,0 +1,114 @@ +--- +title: Lazy inline step start +description: Defer step_created for the inline step and fold it into a single step_started, saving one world round-trip per inline step. +--- + +# Lazy inline step start + +## Motivation + +The owned-inline runtime path used to write two separate world events for a step it already owns and is about to run inline: + +1. `step_created` — written by the suspension handler (`suspension-handler.ts`) +2. `step_started` — written by `executeStep` (`step-executor.ts`) +3. `step_completed` / `step_failed` — written by `executeStep` + +On the Vercel world each `world.events.create` is a network round-trip, so for a simple sequential `"use step"` workflow this is pure latency between steps. Steps (1) and (2) are two round-trips for a step we already own and are about to execute in the same invocation. + +This change defers `step_created` for that one inline step: `executeStep` sends a single `step_started` carrying the step input, and the world creates the step on the fly — materializing the step entity **and** a synthetic `step_created` event so replay still observes it. **Two writes per inline step instead of three.** It mirrors the existing [resilient `run_started` → `run_created`](./resilient-start) pattern. + +Steps that are *queued* (not run inline) keep their eager `step_created` and are unchanged. Only the single inline step per suspension is made lazy. + +## Design + +### Suspension handler + +- `handleSuspension` selects exactly one step to defer — the first uncreated step (`stepItems.find(item => stepsNeedingCreation.has(...))`), which matches the inline candidate the caller would have picked. +- For that step it **skips** the `step_created` write and instead returns it as `lazyInlineStep = { correlationId, stepName, dehydratedInput }`. It is **not** added to `createdStepCorrelationIds` — ownership is no longer decided here. +- A `lazyInlineStep` is designated only when there is no `hook.getConflict()` awaiter (`hasAwaitedHookCreation === false`). With an awaiter present nothing runs inline, so nothing is deferred. + +### `executeStep` + +- A new `lazyStepInput` parameter carries the already-dehydrated input. When present, the `step_started` event includes `input` so the world creates the step on the fly. When absent, `step_started` carries only `stepName` (the legacy contract). +- For an **unregistered** step on the lazy path, `executeStep` sends the lazy `step_started` first to materialize the step, *then* writes `step_failed` (see [Materialize before failing](#materialize-before-failing-unregistered-steps) below). + +### World contract + +- `step_started` accepts an optional `input`. When provided for a non-existent step, the world creates the step entity plus a **synthetic `step_created`** event (so the event log reads `created → started → completed`), then records `step_started` — atomically. +- A `stepCreated` signal is added to the event result so callers can tell whether the lazy `step_started` created the step or attached to an existing one. +- Worlds updated: `world-local`, `world-postgres`, `world-vercel`. + +## Interactions with inline execution + +### Exactly-one-owner is preserved (the race moved) + +The guarantee that exactly one handler runs a step's body inline is intact — it just resolves at a different event: + +- **Before:** ownership was won at the atomic `step_created` claim in the suspension handler; the loser caught `EntityConflictError` and queued instead. +- **After:** two concurrent handlers may both *select* the same `lazyInlineStep` (selection is optimistic, before any race). The race is now the world's atomic create-claim inside the lazy `step_started` (lock file in world-local, `onConflictDoNothing` + unique index in world-postgres, `attribute_not_exists` on the server). The loser gets `EntityConflictError`, which `executeStep` maps to `{ type: 'skipped' }`, so it never runs the body. + +This is safe **because a lazy `step_started` is only ever sent for a brand-new step** — the suspension handler defers only steps with no prior `step_created` (`!hasCreatedEvent`). + +### Crash recovery is unchanged + +On crash recovery the step already has a `step_created` event in the log (`hasCreatedEvent === true`), so it is **not** a lazy candidate. It is re-queued and re-run via a normal **non-lazy** `step_started`, which re-starts a `running` step. At-least-once execution is preserved exactly as before. + +### Materialize before failing unregistered steps + +The inline path previously assumed an invariant: *by the time `executeStep` runs, the step entity already exists* (the suspension handler created it). The unregistered-step branch relied on this — it writes `step_failed` directly, with no preceding `step_started`. + +With the `step_created` deferred, the entity no longer exists when `executeStep` bails for an unregistered step, so the `step_failed` write hits the world's "step must exist" ordering guard and is rejected — wedging the run until it times out. The fix: on the lazy path, send the lazy `step_started` first (creating the entity + synthetic `step_created`), then write `step_failed`. The lazy `step_started`'s atomic create-claim still preserves exactly-one-owner — a concurrent winner makes our create reject with `EntityConflictError` → `skipped`, so the failure is never written twice. + +This is the general rule the deferral introduces: **any inline path that writes a terminal step event must first ensure the deferred step has been materialized.** + +### Replay correctness and the inline-delta fast path + +The client step consumer (`step.ts`) sets `hasCreatedEvent` only when it observes a `step_created` event, and checks step-name divergence against `stepName`. The lazy path stays replay-correct only because the world writes a **synthetic `step_created`**: replay still sees `created → started → completed`. The input lives on the synthetic `step_created`; the `step_started` row drops the input but keeps `stepName` for the divergence check. + +This intersects with the inline-delta optimization. The delta returned on the `step_completed` write is consumed by the *next* replay in place of an `events.list`, diffed against `preInlineWriteCursor` (snapshotted before replay). Because the synthetic `step_created`, `step_started`, and `step_completed` are all written *after* that cursor, the world's "events since cursor" delta carries the full triple — so the next replay does not diverge. The delta gate (one step, no hooks/waits, the lone pending step is the inline one) is unchanged. + +### Pre-emption by attributes / hook conflicts + +When `attr_set` events or a hook conflict force an immediate re-invocation, the handler returns *before* the dispatch loop. The deferred step is therefore **neither created nor queued** on that pass; it is recreated and run on the re-invocation (where it is still a lazy candidate). + +This is a small behavioral improvement: previously the eager `step_created` left an orphan "created but never started" event when a step lost an attribute/hook race (e.g. `Promise.race([setAttributes(), step()])` where the attribute write wins and completes the run). With deferral, a step that loses the race is never created at all — less event-log garbage. + +### `hook.getConflict()` awaiter + +When a `hook.getConflict()` awaiter is present, no `lazyInlineStep` is designated, nothing runs inline, every step gets its eager `step_created` and is queued, and the handler re-invokes immediately so replay resolves the awaiter. This is identical to the pre-change behavior — the deferral never serializes the awaiter's parallel continuation behind an inline step. + +## Rollout and compatibility + +Server-first. The matching world-vercel backend change must deploy before this ships; the Vercel world targets a single backend whose spec version is always at least the SDK's, so the new SDK only ever talks to an already-upgraded backend. An old SDK against a new backend is safe because the lazy path is strictly additive — it triggers only when `step_started` carries both `stepName` and `input`, which old SDKs never send. For `world-local` / `world-postgres` the world ships in the same package as the runtime, so there is no version skew. Detection is by `input` presence on the event, mirroring resilient `run_started` — no capability negotiation is needed. + +# Parallel inline steps + optimistic start + +A follow-up builds two more latency wins on top of lazy inline start. Both are client-side only — they reuse the world's lazy create-on-`step_started` support and need no further world/backend changes. + +## Inline up to N steps in parallel + +Previously the owned-inline path ran **exactly one** step inline per suspension and queued the rest. For a `Promise.all([stepA(), stepB(), stepC()])` fan-out that meant one branch ran inline while the others paid a queue round-trip each before making any progress. + +The suspension handler now defers `step_created` for up to **`WORKFLOW_MAX_INLINE_STEPS` (default 3)** steps and returns them as `lazyInlineSteps`. The runtime runs that batch inline **in parallel** (`Promise.all`), each via its own lazy `step_started`, and queues only the steps beyond the cap. + +- **Selection:** the first N uncreated steps, matching the previous single-step inline candidate. Steps beyond N keep their eager `step_created` and are queued exactly as before. +- **Result aggregation:** steps that need to run again (`retry`/`throttled`) are re-queued per-step with their own delay; the runtime only loops back to replay once every inline step has reached a terminal state. A lone throttled inline step still delays redelivery of the orchestrator message (preserving the single-step backpressure contract). +- **Inline-delta fast path:** still used only for the single-step sequential case (`lazyInlineSteps.length === 1`). With more than one inline step each writes its own events, so a per-write delta would be partial; multi-step batches fall back to a normal incremental `events.list`. +- **Config:** `WORKFLOW_MAX_INLINE_STEPS` is clamped to 1..16. Setting it to `1` reproduces the previous single-inline-step behavior exactly (a useful kill-switch). Inline bodies run in parallel within one function invocation, so the cap also bounds per-handler memory/CPU fan-out. + +## Optimistic inline start + +Normally `executeStep` awaits `step_started` (the lazy create-claim round-trip) before running the body. Because the inline path already holds the step input locally, it doesn't actually need that round-trip to begin. + +When `WORKFLOW_OPTIMISTIC_INLINE_START` is enabled (**default on**), an inline step fires `step_started` **without awaiting it** and starts running the body immediately against locally-synthesized state. A lazy step is always brand-new, so attempt is 1, there is no prior error, and `startedAt` is now — all known without the server. The in-flight `step_started` is reconciled just before the terminal write: + +- **Lost the create-claim (409 / `EntityConflictError`)** → discard the body result and return `skipped`; the winning handler owns the terminal write. +- **Run gone / throttled / too-early** → discard the body result and surface `gone` / `throttled` / `retry`. +- **Transient (non-translatable) failure** → propagate it, so the queue redelivers — exactly as the await path does today. +- **Success** → write `step_completed` / `step_failed` / `step_retrying` as usual. Awaiting `step_started` before the terminal write keeps the event log ordered (`created → started → completed`). + +### Safety and the idempotency tradeoff + +- **Exactly-one terminal write is preserved.** Optimistic start changes only *when the body runs*, never who writes the terminal event — that is still gated by the lazy `step_started` create-claim, which is awaited before the terminal write. Losers return `skipped`. +- **Bounded to attempt 1.** Only brand-new (`!hasCreatedEvent`) steps are lazy; a retried step already has a `step_created`, so it takes the normal await-then-run path with the real attempt counter. Synthesizing `attempt = 1` locally is therefore always correct. +- **Wider double-execution.** Running the body before confirming ownership means two handlers racing into the same batch boundary can *both* run the side effects before either wins (previously the loser 409'd on `step_created` and skipped before running anything). This is an explicit, accepted tradeoff: **inline step bodies must be idempotent.** Set `WORKFLOW_OPTIMISTIC_INLINE_START=0` (or `false`) to restore the await-`step_started`-then-run behavior. diff --git a/docs/content/docs/v5/changelog/meta.json b/docs/content/docs/v5/changelog/meta.json index 0c01dff133..b65b777d83 100644 --- a/docs/content/docs/v5/changelog/meta.json +++ b/docs/content/docs/v5/changelog/meta.json @@ -1,5 +1,10 @@ { "title": "Changelog", - "pages": ["index", "eager-processing", "resilient-start"], + "pages": [ + "index", + "eager-processing", + "resilient-start", + "lazy-event-creation" + ], "defaultOpen": false } diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index b7aca42635..3a4337ae35 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -4,7 +4,7 @@ import { SPEC_VERSION_CURRENT, type WorkflowRun, } from '@workflow/world'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { registerStepFunction } from './private.js'; import { REPLAY_DIVERGENCE_MAX_RETRIES } from './runtime/constants.js'; import { setWorld } from './runtime/world.js'; @@ -889,7 +889,15 @@ describe('workflowEntrypoint replay guards', () => { }); describe('workflowEntrypoint step-dispatch ack ordering', () => { + // Pin to a single inline step so exactly one of the two parallel steps is + // queued — these tests assert the dispatch→ack ordering for that QUEUED step, + // which is independent of how many steps run inline. (With the default of + // `getMaxInlineSteps()` both would run inline and nothing would be queued.) + beforeEach(() => { + process.env.WORKFLOW_MAX_INLINE_STEPS = '1'; + }); afterEach(() => { + delete process.env.WORKFLOW_MAX_INLINE_STEPS; setWorld(undefined); vi.clearAllMocks(); waitUntilPromises.length = 0; @@ -1181,4 +1189,24 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { // promise (which would become an unhandled rejection / process exit 128). expect(await anyWaitUntilPromiseRejected()).toBe(false); }); + + it('runs BOTH parallel steps inline (none queued) when the inline cap allows it', async () => { + // Override the per-suite cap of 1: with a cap of 3 both `add` and `addB` + // are deferred and run inline via lazy step_started, so neither is eagerly + // created or dispatched to a background handler. Only the sleep's wait + // continuation is queued (it carries no stepId). + process.env.WORKFLOW_MAX_INLINE_STEPS = '3'; + + const { handlerPromise, order } = await driveHandler({ + runId: 'wrun_multi_inline', + queueImpl: async () => ({ messageId: null }), + }); + + const res = (await handlerPromise) as Response; + expect(res.status).toBe(204); + + // No eager step_created and no step-dispatch send: both steps went inline. + expect(order).not.toContain('step_created'); + expect(order).not.toContain('queue_dispatch_start'); + }); }); diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index c378664b43..4026b23e5b 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1247,19 +1247,20 @@ export function workflowEntrypoint( // Inline execution is gated on ownership. The // suspension handler deferred the step_created write for - // exactly one step (`lazyInlineStep`) so we can run it - // inline via a lazy `step_started` that creates the step - // on the fly — saving one world round-trip. Ownership is - // still atomic and exactly-one: the world's - // create-claim inside that step_started returns + // up to `getMaxInlineSteps()` steps (`lazyInlineSteps`) + // so we can run them inline — in parallel — via lazy + // `step_started` events that create each step on the fly, + // saving one world round-trip per inline step. Ownership + // is still atomic and exactly-one per step: the world's + // create-claim inside each step_started returns // `EntityConflictError` (→ executeStep `skipped`) to any - // concurrent loser, so only one handler ever runs the + // concurrent loser, so only one handler ever runs a given // body. Every other pending step keeps its eager // step_created (in `createdStepCorrelationIds`) and is // queued below. // - // The suspension handler only designates a - // `lazyInlineStep` when no `hook.getConflict()` awaiter + // The suspension handler only designates + // `lazyInlineSteps` when no `hook.getConflict()` awaiter // is present. That awaiter case must execute nothing // inline: an inline `await executeStep(...)` blocks this // handler for the full step duration, so the awaiter's @@ -1267,20 +1268,15 @@ export function workflowEntrypoint( // would be serialized behind the step — defeating work // the workflow expressed as parallel (e.g. // `hook.getConflict().then(() => stepB())` racing `await - // stepA()`). In that case `lazyInlineStep` is undefined - // and every step is queued for re-invocation, which - // replays over the just-committed hook_created and - // resolves the awaiter while queued steps run in - // parallel invocations. - const lazyInline = suspensionResult.lazyInlineStep; - const inlineStep: - | (typeof pendingSteps)[number] - | undefined = lazyInline - ? pendingSteps.find( - (s) => - s.correlationId === lazyInline.correlationId - ) - : undefined; + // stepA()`). In that case `lazyInlineSteps` is empty and + // every step is queued for re-invocation, which replays + // over the just-committed hook_created and resolves the + // awaiter while queued steps run in parallel invocations. + const lazyInlineSteps = + suspensionResult.lazyInlineSteps; + const inlineCorrelationIds = new Set( + lazyInlineSteps.map((s) => s.correlationId) + ); // Unified queue dispatch for everything we are NOT // inline-executing. Steps are queued with stepId so @@ -1317,10 +1313,7 @@ export function workflowEntrypoint( const traceCarrier = await nextTraceCarrier(); const dispatches: Promise[] = []; for (const step of pendingSteps) { - if ( - inlineStep && - step.correlationId === inlineStep.correlationId - ) { + if (inlineCorrelationIds.has(step.correlationId)) { continue; } dispatches.push( @@ -1362,7 +1355,7 @@ export function workflowEntrypoint( // Nothing to execute inline — everything has been // queued (or no work needs scheduling). Exit and let // the queue drive subsequent replays. - if (!inlineStep) { + if (lazyInlineSteps.length === 0) { // A `hook.getConflict()` awaiter needs an immediate // re-invocation: the replay consumes the // just-committed hook_created and resolves the @@ -1393,10 +1386,10 @@ export function workflowEntrypoint( // 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 it is the one we - // are running inline (`inlineStep` — no parallel - // siblings queued to background handlers, which would - // write their own events out of band). + // (`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 @@ -1410,91 +1403,134 @@ export function workflowEntrypoint( // the only out-of-band writer is cancellation, which // is safe to observe one iteration late. See // hasOpenHookOrWait. + // + // 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' && err.stepCount === 1 && err.hookCount === 0 && err.waitCount === 0 && pendingSteps.length === 1 && - inlineStep !== undefined && + lazyInlineSteps.length === 1 && !suspensionResult.waitTimeout && !hasOpenHookOrWait(cachedEvents ?? []); + // Execute the inline steps in parallel. The replay + // budget is paused for the whole batch — step duration is + // bounded by the platform's function maxDuration, not the + // replay timeout — so the budget check at the top of the + // next loop iteration doesn't charge the step bodies. replayBudget.pause(); - let stepResult: Awaited>; + let stepResults: Awaited< + ReturnType + >[]; try { - stepResult = await executeStep({ - world, - workflowRunId: runId, - workflowDeploymentId: workflowRun.deploymentId, - workflowName, - workflowStartedAt, - stepId: inlineStep.correlationId, - stepName: inlineStep.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. lazyInline is defined whenever inlineStep - // is (both derive from suspensionResult.lazyInlineStep). - lazyStepInput: lazyInline?.dehydratedInput, - ...(requestInlineDelta && preInlineWriteCursor - ? { inlineDeltaSinceCursor: preInlineWriteCursor } - : {}), - }); + stepResults = await Promise.all( + lazyInlineSteps.map((s) => + executeStep({ + world, + workflowRunId: runId, + workflowDeploymentId: workflowRun.deploymentId, + workflowName, + workflowStartedAt, + 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. + lazyStepInput: s.dehydratedInput, + ...(requestInlineDelta && preInlineWriteCursor + ? { + inlineDeltaSinceCursor: + preInlineWriteCursor, + } + : {}), + }) + ) + ); } finally { replayBudget.resume(); } - if (stepResult.type === 'retry') { - // Step needs retry — queue self with stepId for retry. - // Any pending wait timer was already enqueued as part - // of the unified dispatch above, so we can return - // unconditionally here. - const retryTraceCarrier = await nextTraceCarrier(); - await queueMessage( - world, - getWorkflowQueueName(workflowName, namespace), - { - runId, - stepId: inlineStep.correlationId, - stepName: inlineStep.stepName, - traceCarrier: retryTraceCarrier, - requestedAt: new Date(), - }, - { - delaySeconds: stepResult.timeoutSeconds, + // Aggregate the batch results. Steps that need to run + // again (`retry`/`throttled`) are re-queued per-step with + // their own delay; completed/failed steps already wrote + // their terminal events. We only loop back to replay when + // every inline step reached a terminal state — otherwise + // the still-pending steps will be re-run by their queued + // retry messages and the background-step handler replays + // once all steps are done. + const toRetry: { + step: (typeof lazyInlineSteps)[number]; + delaySeconds: number; + }[] = []; + let anyPendingOps = false; + // Preserve the single-step backpressure contract: a lone + // throttled inline step delays redelivery of THIS + // orchestrator message (rather than re-queuing per-step). + let soleThrottleTimeout: number | undefined; + for (let i = 0; i < lazyInlineSteps.length; i++) { + const r = stepResults[i]; + const s = lazyInlineSteps[i]; + if (r.type === 'retry') { + toRetry.push({ + step: s, + delaySeconds: r.timeoutSeconds, + }); + } else if (r.type === 'throttled') { + if (lazyInlineSteps.length === 1) { + soleThrottleTimeout = r.timeoutSeconds; + } else { + toRetry.push({ + step: s, + delaySeconds: r.timeoutSeconds, + }); } - ); - return; + } else if ( + r.type === 'completed' && + r.hasPendingOps + ) { + anyPendingOps = true; + } } - if (stepResult.type === 'throttled') { - return { - timeoutSeconds: stepResult.timeoutSeconds, - }; + if (soleThrottleTimeout !== undefined) { + return { timeoutSeconds: soleThrottleTimeout }; } - // Step completed or failed — loop back to replay - // (gone/skipped also loop back since the workflow - // will see the completed/failed event on replay) + if (toRetry.length > 0) { + const retryTraceCarrier = await nextTraceCarrier(); + await Promise.all( + toRetry.map(({ step, delaySeconds }) => + queueMessage( + world, + getWorkflowQueueName(workflowName, namespace), + { + runId, + stepId: step.correlationId, + stepName: step.stepName, + traceCarrier: retryTraceCarrier, + requestedAt: new Date(), + }, + { delaySeconds } + ) + ) + ); + } - // If the step had pending background ops (e.g., stream - // writes to S3), break the loop and return so waitUntil - // can flush them. This matches V1 behavior where each - // step ran in a separate function invocation. Without - // this, the inline loop continues and the stream data - // may not reach S3 before the test tries to read it. - if ( - stepResult.type === 'completed' && - stepResult.hasPendingOps - ) { + // If any inline step had pending background ops (e.g., + // stream writes to S3), break the loop and queue a plain + // continuation so waitUntil can flush them before the + // next replay reads them. This matches V1 behavior where + // each step ran in a separate function invocation. + if (anyPendingOps) { runtimeLogger.debug( - 'Breaking loop: step has pending ops', - { - workflowRunId: runId, - loopIteration, - stepName: inlineStep.stepName, - } + 'Breaking loop: inline step has pending ops', + { workflowRunId: runId, loopIteration } ); await queueMessage( world, @@ -1508,22 +1544,36 @@ export function workflowEntrypoint( return; } - // Looping back to replay. If this step's terminal write - // returned an inline delta (supporting World + the - // single-step gate above), stash it so the next - // iteration's load consumes it instead of issuing an - // incremental events.list. Only the completed path - // carries a delta; on any other terminal type the next - // iteration falls back to the normal fetch. - if ( - stepResult.type === 'completed' && - stepResult.inlineDelta && - !stepResult.inlineDelta.hasMore - ) { - pendingInlineDelta = { - events: stepResult.inlineDelta.events, - cursor: stepResult.inlineDelta.cursor, - }; + if (toRetry.length > 0) { + // Some inline steps will be re-run via their queued + // retry messages; the background-step handler replays + // once all steps are terminal. Don't loop here — the + // retrying steps have no terminal event to observe yet. + return; + } + + // All inline steps reached a terminal state + // (completed/failed/skipped/gone) — loop back to replay + // (the workflow observes the terminal events on replay). + // + // If the single inline step's terminal write returned an + // inline delta (supporting World + the single-step gate + // above), stash it so the next iteration's load consumes + // it instead of issuing an incremental events.list. Only + // the completed path carries a delta; multi-step batches + // never request one. + if (lazyInlineSteps.length === 1) { + const only = stepResults[0]; + if ( + only.type === 'completed' && + only.inlineDelta && + !only.inlineDelta.hasMore + ) { + pendingInlineDelta = { + events: only.inlineDelta.events, + cursor: only.inlineDelta.cursor, + }; + } } } else { let terminalError = err; diff --git a/packages/core/src/runtime/constants.test.ts b/packages/core/src/runtime/constants.test.ts index fe71939262..a4d80c369d 100644 --- a/packages/core/src/runtime/constants.test.ts +++ b/packages/core/src/runtime/constants.test.ts @@ -2,8 +2,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { runtimeLogger } from '../logger.js'; import { _resetReplayTimeoutWarnCacheForTests, + getMaxInlineSteps, getReplayTimeoutMs, + isOptimisticInlineStartEnabled, + MAX_INLINE_STEPS, + MAX_MAX_INLINE_STEPS, MAX_REPLAY_TIMEOUT_MS, + MIN_MAX_INLINE_STEPS, MIN_REPLAY_TIMEOUT_MS, REPLAY_TIMEOUT_MS, } from './constants.js'; @@ -109,3 +114,93 @@ describe('getReplayTimeoutMs', () => { expect(warnSpy).toHaveBeenCalledTimes(1); }); }); + +describe('getMaxInlineSteps', () => { + const originalEnv = process.env.WORKFLOW_MAX_INLINE_STEPS; + let warnSpy: ReturnType; + + beforeEach(() => { + delete process.env.WORKFLOW_MAX_INLINE_STEPS; + warnSpy = vi.spyOn(runtimeLogger, 'warn').mockImplementation(() => {}); + }); + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.WORKFLOW_MAX_INLINE_STEPS; + } else { + process.env.WORKFLOW_MAX_INLINE_STEPS = originalEnv; + } + warnSpy.mockRestore(); + }); + + it('returns the default when the env var is unset', () => { + expect(getMaxInlineSteps()).toBe(MAX_INLINE_STEPS); + expect(warnSpy).not.toHaveBeenCalled(); + }); + + it('returns a valid in-range override', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = '5'; + expect(getMaxInlineSteps()).toBe(5); + expect(warnSpy).not.toHaveBeenCalled(); + }); + + it('clamps to the minimum (1 = single inline step)', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = '1'; + expect(getMaxInlineSteps()).toBe(MIN_MAX_INLINE_STEPS); + }); + + it('clamps values above the maximum and warns', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = String(MAX_MAX_INLINE_STEPS + 100); + expect(getMaxInlineSteps()).toBe(MAX_MAX_INLINE_STEPS); + expect(warnSpy).toHaveBeenCalledTimes(1); + }); + + it('falls back to the default on a non-integer and warns', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = '2.5'; + expect(getMaxInlineSteps()).toBe(MAX_INLINE_STEPS); + expect(warnSpy).toHaveBeenCalledTimes(1); + }); + + it('falls back to the default on a non-numeric value and warns', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = 'lots'; + expect(getMaxInlineSteps()).toBe(MAX_INLINE_STEPS); + expect(warnSpy).toHaveBeenCalledTimes(1); + }); + + it('falls back to the default on a non-positive value', () => { + process.env.WORKFLOW_MAX_INLINE_STEPS = '0'; + expect(getMaxInlineSteps()).toBe(MAX_INLINE_STEPS); + }); +}); + +describe('isOptimisticInlineStartEnabled', () => { + const originalEnv = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + } else { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = originalEnv; + } + }); + + it('defaults to enabled when unset', () => { + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + expect(isOptimisticInlineStartEnabled()).toBe(true); + }); + + it('is disabled by "0"', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '0'; + expect(isOptimisticInlineStartEnabled()).toBe(false); + }); + + it('is disabled by "false" (case-insensitive)', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = 'FALSE'; + expect(isOptimisticInlineStartEnabled()).toBe(false); + }); + + it('stays enabled for any other value', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; + expect(isOptimisticInlineStartEnabled()).toBe(true); + }); +}); diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index b29df7bd15..3bcaecc35e 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -118,6 +118,85 @@ export function _resetReplayTimeoutWarnCacheForTests(): void { // On the next attempt the run is marked as failed. export const REPLAY_TIMEOUT_MAX_RETRIES = 3; +/** + * Default maximum number of steps the owned-inline path runs inline (in + * parallel) per suspension. The rest are queued to background handlers. Each + * inline step is created lazily — its `step_created` is folded into the + * `step_started` that `executeStep` sends — so inlining N steps saves N queue + * round-trips for a `Promise.all`-style fan-out. `1` reproduces the + * single-inline-step behavior exactly (useful kill-switch). + * + * Override via `WORKFLOW_MAX_INLINE_STEPS` (clamped to + * `MIN_MAX_INLINE_STEPS`..`MAX_MAX_INLINE_STEPS`). + */ +export const MAX_INLINE_STEPS = 3; + +/** Lower bound for the inline-steps env override (1 = single inline step). */ +export const MIN_MAX_INLINE_STEPS = 1; + +/** + * Upper bound for the inline-steps env override. Inline bodies run in parallel + * within one function invocation, so this caps memory/CPU fan-out per handler. + */ +export const MAX_MAX_INLINE_STEPS = 16; + +// Warn-once cache for WORKFLOW_MAX_INLINE_STEPS, keyed by raw env value. +const warnedMaxInlineStepsValues = new Set(); + +/** + * Resolve the effective max number of inline steps for the current process. + * + * Reads `process.env.WORKFLOW_MAX_INLINE_STEPS` lazily so tests and + * deployments can override per invocation. Invalid / out-of-range values fall + * back to a safe value (no throw — the env var is an escape hatch) and emit a + * one-time warning so misconfiguration is observable. + */ +export function getMaxInlineSteps(): number { + const raw = process.env.WORKFLOW_MAX_INLINE_STEPS; + if (!raw) return MAX_INLINE_STEPS; + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed <= 0) { + if (!warnedMaxInlineStepsValues.has(raw)) { + warnedMaxInlineStepsValues.add(raw); + runtimeLogger.warn( + 'Ignoring WORKFLOW_MAX_INLINE_STEPS: not a positive integer; using default', + { raw, defaultValue: MAX_INLINE_STEPS } + ); + } + return MAX_INLINE_STEPS; + } + if (parsed < MIN_MAX_INLINE_STEPS) return MIN_MAX_INLINE_STEPS; + if (parsed > MAX_MAX_INLINE_STEPS) { + if (!warnedMaxInlineStepsValues.has(raw)) { + warnedMaxInlineStepsValues.add(raw); + runtimeLogger.warn('WORKFLOW_MAX_INLINE_STEPS above maximum; clamped', { + raw, + clampedValue: MAX_MAX_INLINE_STEPS, + maxValue: MAX_MAX_INLINE_STEPS, + }); + } + return MAX_MAX_INLINE_STEPS; + } + return parsed; +} + +/** + * Whether optimistic inline step start is enabled. When on, the owned-inline + * path begins running a brand-new step's body *before* its lazy `step_started` + * network call resolves (the input is already known locally), awaiting the + * `step_started` only before the terminal write. This can run a step body more + * than once when handlers race — inline step bodies must be idempotent. + * + * Reads `process.env.WORKFLOW_OPTIMISTIC_INLINE_START` lazily. Default ON; + * disabled only by an explicit `'0'` / `'false'`, which restores the + * await-`step_started`-then-run behavior. + */ +export function isOptimisticInlineStartEnabled(): boolean { + const raw = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + if (raw === undefined || raw === '') return true; + return raw !== '0' && raw.toLowerCase() !== 'false'; +} + // A replay-consumer mismatch can be caused by a transient divergent replay // rather than an invalid persisted history. Queue bounded recovery replays // before recording terminal corruption for a run that cannot replay. diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index 6155c4b2fc..e510c5bb06 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -33,6 +33,7 @@ import { promoteAbortErrorToFatal, } from '../types.js'; +import { isOptimisticInlineStartEnabled } from './constants.js'; import { getPortLazy } from './get-port-lazy.js'; import { memoizeEncryptionKey } from './helpers.js'; import { safeWaitUntil } from './wait-until.js'; @@ -252,32 +253,12 @@ export async function executeStep( ...Attribute.StepMaxRetries(maxRetries), }); - // step_started validates state and returns the step entity. On the lazy - // inline path we also carry the step `input` so the world creates the step - // on the fly (no separate step_created round-trip). The world's atomic - // create-claim makes this exactly-one-owner: a concurrent loser gets - // EntityConflictError, mapped to `{ type: 'skipped' }` below, so it never - // runs the body. When `lazyStepInput` is absent this is the legacy - // step_started (step already created, no payload). - let step: Step; - try { - const startResult = await world.events.create(workflowRunId, { - eventType: 'step_started', - specVersion: SPEC_VERSION_CURRENT, - correlationId: stepId, - eventData: - params.lazyStepInput !== undefined - ? { stepName, input: params.lazyStepInput } - : { stepName }, - }); - - if (!startResult.step) { - throw new WorkflowRuntimeError( - `step_started event for "${stepId}" did not return step entity` - ); - } - step = startResult.step; - } catch (err) { + // Maps a `step_started` rejection to a terminal StepExecutionResult, + // shared by the await path (below) and the optimistic-start reconciliation. + // Returns undefined when the error is not one we translate (caller rethrows). + const startErrorToResult = ( + err: unknown + ): StepExecutionResult | undefined => { if (ThrottleError.is(err)) { const retryAfter = Math.max( 1, @@ -299,7 +280,7 @@ export async function executeStep( stepName, stepId, workflowRunId, - error: err.message, + error: err instanceof Error ? err.message : String(err), }); span?.setAttributes({ ...Attribute.StepSkipped(true), @@ -316,7 +297,99 @@ export async function executeStep( }); return { type: 'retry', timeoutSeconds }; } - throw err; + return undefined; + }; + + // Optimistic inline start: when we hold the step input locally (lazy inline + // path) and the optimization is enabled, fire `step_started` WITHOUT + // awaiting and run the body against locally-synthesized state. A lazy step + // is always brand-new ⇒ attempt 1, no prior error, started now — so we + // don't need the server round-trip to begin. We reconcile the in-flight + // `step_started` before any terminal write (`reconcileOptimisticStart`): if + // it lost the atomic create-claim (409) or the run is gone/throttled, we + // discard the body result. Running the body before confirming ownership can + // execute a step more than once when handlers race — inline step bodies + // must be idempotent; disable via WORKFLOW_OPTIMISTIC_INLINE_START=0. + const optimisticStart = + params.lazyStepInput !== undefined && isOptimisticInlineStartEnabled(); + + let step: Step; + // Settled outcome of the in-flight optimistic `step_started`. Handlers are + // attached synchronously (`.then(ok, err)`) so a fast rejection never + // surfaces as an unhandledRejection while the body runs. + let optimisticStartSettled: + | Promise<{ ok: true } | { ok: false; err: unknown }> + | undefined; + // Await the optimistic `step_started` outcome and translate a lost race / + // terminal run / throttle into a result that short-circuits the body + // output. Returns undefined when we own the step and may write its terminal + // event. A non-translatable rejection is rethrown (so a transient + // step_started failure propagates to the queue handler for redelivery, + // exactly as on the await path). Idempotent — safe to call more than once. + const reconcileOptimisticStart = async (): Promise< + StepExecutionResult | undefined + > => { + if (!optimisticStartSettled) return undefined; + const settled = await optimisticStartSettled; + if (settled.ok) return undefined; + const mapped = startErrorToResult(settled.err); + if (!mapped) throw settled.err; + return mapped; + }; + + if (optimisticStart) { + const startedPromise = world.events.create(workflowRunId, { + eventType: 'step_started', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { stepName, input: params.lazyStepInput }, + }); + optimisticStartSettled = startedPromise.then( + () => ({ ok: true as const }), + (err) => ({ ok: false as const, err }) + ); + const now = new Date(); + step = { + runId: workflowRunId, + stepId, + stepName, + status: 'running', + input: params.lazyStepInput, + attempt: 1, + startedAt: now, + createdAt: now, + updatedAt: now, + }; + } else { + // step_started validates state and returns the step entity. On the lazy + // inline path we also carry the step `input` so the world creates the + // step on the fly (no separate step_created round-trip). The world's + // atomic create-claim makes this exactly-one-owner: a concurrent loser + // gets EntityConflictError, mapped to `{ type: 'skipped' }`, so it never + // runs the body. When `lazyStepInput` is absent this is the legacy + // step_started (step already created, no payload). + try { + const startResult = await world.events.create(workflowRunId, { + eventType: 'step_started', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: + params.lazyStepInput !== undefined + ? { stepName, input: params.lazyStepInput } + : { stepName }, + }); + + if (!startResult.step) { + throw new WorkflowRuntimeError( + `step_started event for "${stepId}" did not return step entity` + ); + } + step = startResult.step; + } catch (err) { + const mapped = startErrorToResult(err); + if (mapped) return mapped; + throw err; + } } runtimeLogger.debug('Step execution details', { @@ -532,6 +605,14 @@ export async function executeStep( ]); } + // Optimistic start: the body ran before `step_started` was confirmed. + // Reconcile it now — if we lost the create-claim (or the run is + // gone/throttled) discard this result and don't write step_completed. + if (optimisticStart) { + const reconcile = await reconcileOptimisticStart(); + if (reconcile) return reconcile; + } + // Create step_completed event. When the caller supplied a // sinceCursor (inline sequential execution), thread it through so a // supporting World returns the event-log delta on the result, @@ -597,6 +678,15 @@ export async function executeStep( // and queue a continuation so waitUntil can flush them. return { type: 'completed', hasPendingOps: !opsSettled, inlineDelta }; } catch (err: unknown) { + // Optimistic start: the body threw before `step_started` was confirmed. + // Reconcile first — if we lost the create-claim (or the run is + // gone/throttled) the body error is moot; discard it and don't write a + // terminal event (the winning handler owns the outcome). + if (optimisticStart) { + const reconcile = await reconcileOptimisticStart(); + if (reconcile) return reconcile; + } + const effectiveErr = promoteAbortErrorToFatal(err); const normalizedError = await normalizeUnknownError(effectiveErr); diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index b3a15b01e0..6cea41af21 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -1,6 +1,7 @@ import { EntityConflictError, FatalError, + ThrottleError, WorkflowWorldError, } from '@workflow/errors'; import { @@ -189,11 +190,11 @@ import { normalizeUnknownError, } from '../types.js'; import { MAX_QUEUE_DELIVERIES } from './constants.js'; +import { executeStep } from './step-executor.js'; // Import the module AFTER all mocks are set up // Since getWorldHandlers is now async, we need to call stepEntrypoint // to trigger createQueueHandler and populate capturedHandlerRef import { stepEntrypoint } from './step-handler.js'; -import { executeStep } from './step-executor.js'; import { getWorld } from './world.js'; function capturedHandler( @@ -1175,3 +1176,120 @@ describe('executeStep inline-delta threading', () => { expect(result.inlineDelta).toBeUndefined(); }); }); + +describe('executeStep optimistic inline start', () => { + const baseParams = { + workflowRunId: 'wrun_test123', + workflowName: 'test-workflow', + workflowStartedAt: Date.now(), + stepId: 'step_abc', + stepName: 'myStep', + // Empty dehydrated input — the world round-trips this back as step.input. + lazyStepInput: [] as never, + }; + + beforeEach(() => { + vi.clearAllMocks(); + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + vi.mocked(getStepFunction).mockReturnValue(mockStepFn); + vi.mocked(normalizeUnknownError).mockImplementation( + async (err: unknown) => ({ + message: err instanceof Error ? err.message : String(err), + name: err instanceof Error ? err.name : 'Error', + stack: err instanceof Error ? err.stack : undefined, + }) + ); + mockStepFn.mockReset().mockResolvedValue('step-result'); + mockStepFn.maxRetries = 3; + }); + + afterEach(() => { + delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + vi.restoreAllMocks(); + }); + + it('sends step_started carrying the input and completes (default on)', async () => { + mockEventsCreate + .mockReset() + .mockImplementation((_runId: string, event: { eventType: string }) => + Promise.resolve({ event: {} }) + ); + + const world = await getWorld(); + const result = await executeStep({ world: world as never, ...baseParams }); + + expect(result.type).toBe('completed'); + expect(mockStepFn).toHaveBeenCalledTimes(1); + // The lazy step_started carries the input so the world creates the step. + expect(mockEventsCreate).toHaveBeenCalledWith( + 'wrun_test123', + expect.objectContaining({ + eventType: 'step_started', + correlationId: 'step_abc', + eventData: expect.objectContaining({ stepName: 'myStep', input: [] }), + }) + ); + }); + + it('runs the body optimistically but discards the result when step_started loses the create race (409)', async () => { + mockEventsCreate + .mockReset() + .mockImplementation((_runId: string, event: { eventType: string }) => { + if (event.eventType === 'step_started') { + return Promise.reject(new EntityConflictError('lost create race')); + } + return Promise.resolve({ event: {} }); + }); + + const world = await getWorld(); + const result = await executeStep({ world: world as never, ...baseParams }); + + // The body ran (optimistic execution before the start was confirmed)... + expect(mockStepFn).toHaveBeenCalledTimes(1); + // ...but we lost the race, so the result is discarded and no terminal + // event is written. + expect(result.type).toBe('skipped'); + const completedWrites = mockEventsCreate.mock.calls.filter( + ([, event]) => + (event as { eventType: string }).eventType === 'step_completed' + ); + expect(completedWrites).toHaveLength(0); + }); + + it('returns throttled (discarding the body result) when step_started is throttled', async () => { + mockEventsCreate + .mockReset() + .mockImplementation((_runId: string, event: { eventType: string }) => { + if (event.eventType === 'step_started') { + return Promise.reject( + new ThrottleError('slow down', { retryAfter: 7 }) + ); + } + return Promise.resolve({ event: {} }); + }); + + const world = await getWorld(); + const result = await executeStep({ world: world as never, ...baseParams }); + + expect(result).toEqual({ type: 'throttled', timeoutSeconds: 7 }); + }); + + it('does NOT run the body before confirming start when the flag is disabled', async () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '0'; + mockEventsCreate + .mockReset() + .mockImplementation((_runId: string, event: { eventType: string }) => { + if (event.eventType === 'step_started') { + return Promise.reject(new EntityConflictError('already running')); + } + return Promise.resolve({ event: {} }); + }); + + const world = await getWorld(); + const result = await executeStep({ world: world as never, ...baseParams }); + + // Await-first path: the conflict short-circuits before the body runs. + expect(result.type).toBe('skipped'); + expect(mockStepFn).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 8bc6ba67c9..9acf9b609d 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -107,6 +107,119 @@ describe('handleSuspension', () => { expect(result.createdStepCorrelationIds).toContain('step_parallel'); }); + it('defers up to getMaxInlineSteps() uncreated steps and eagerly creates the rest', async () => { + // Default getMaxInlineSteps() is 3. With 4 uncreated parallel steps, the + // first 3 are deferred for lazy inline start (no step_created written) and + // the 4th keeps its eager step_created and is owned for queuing. + const eventsCreate = vi.fn().mockResolvedValue({ + event: { eventType: 'step_created' }, + }); + const world = createWorld(eventsCreate); + const pending = new Map( + ['s1', 's2', 's3', 's4'].map((id) => [ + id, + { type: 'step' as const, correlationId: id, stepName: id, args: [] }, + ]) + ); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([ + 's1', + 's2', + 's3', + ]); + // Only the non-deferred step writes a step_created and is owned. + expect(eventsCreate).toHaveBeenCalledTimes(1); + expect(eventsCreate).toHaveBeenCalledWith( + run.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's4', + }), + expect.anything() + ); + expect([...result.createdStepCorrelationIds]).toEqual(['s4']); + }); + + it('honors WORKFLOW_MAX_INLINE_STEPS as the inline cap', async () => { + const prev = process.env.WORKFLOW_MAX_INLINE_STEPS; + process.env.WORKFLOW_MAX_INLINE_STEPS = '1'; + try { + const eventsCreate = vi.fn().mockResolvedValue({ + event: { eventType: 'step_created' }, + }); + const world = createWorld(eventsCreate); + const pending = new Map( + ['s1', 's2', 's3'].map((id) => [ + id, + { type: 'step' as const, correlationId: id, stepName: id, args: [] }, + ]) + ); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + // Cap of 1: only the first step is deferred; s2 and s3 are eager-created. + expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([ + 's1', + ]); + expect(eventsCreate).toHaveBeenCalledTimes(2); + expect([...result.createdStepCorrelationIds].sort()).toEqual([ + 's2', + 's3', + ]); + } finally { + if (prev === undefined) delete process.env.WORKFLOW_MAX_INLINE_STEPS; + else process.env.WORKFLOW_MAX_INLINE_STEPS = prev; + } + }); + + it('defers no inline steps when a hook.getConflict() awaiter is present', async () => { + const eventsCreate = vi.fn().mockResolvedValue({ + event: { eventType: 'hook_created' }, + }); + const world = createWorld(eventsCreate); + const pending = new Map([ + [ + 's1', + { + type: 'step' as const, + correlationId: 's1', + stepName: 's1', + args: [], + }, + ], + [ + 'hook_awaited', + { + type: 'hook' as const, + correlationId: 'hook_awaited', + token: 'claim-token', + hasConflictAwaiter: true, + }, + ], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + // Nothing runs inline: the step keeps its eager step_created (owned) and is + // queued; the caller re-invokes immediately to resolve the awaiter. + expect(result.lazyInlineSteps).toEqual([]); + expect(result.createdStepCorrelationIds).toContain('s1'); + }); + it('does not immediately continue after creating a hook without a getConflict awaiter', async () => { const eventsCreate = vi.fn().mockResolvedValue({ event: { diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 5d0efa3935..ebd766198c 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -27,6 +27,7 @@ import { runtimeLogger } from '../logger.js'; import { dehydrateStepArguments } from '../serialization.js'; import * as Attribute from '../telemetry/semantic-conventions.js'; import { getAbortStreamIdFromToken } from '../util.js'; +import { getMaxInlineSteps } from './constants.js'; export interface SuspensionHandlerParams { suspension: WorkflowSuspension; @@ -53,22 +54,23 @@ export interface SuspensionHandlerResult { */ createdStepCorrelationIds: Set; /** - * The single step whose `step_created` write was intentionally deferred so - * the caller can run it inline via a lazy `step_started` (which creates the - * step on the fly), saving one world round-trip per inline step. Undefined - * when no step was deferred (nothing pending, or a `hook.getConflict()` - * awaiter is present so nothing is executed inline). The caller passes - * `dehydratedInput` straight to `executeStep`, which sends it as the - * `step_started` payload. The atomic create-claim inside that `step_started` - * is the exactly-one-owner gate that the standalone `step_created` provided - * before: the loser of the race gets `EntityConflictError` → `skipped` and - * does not run the body. + * The steps whose `step_created` writes were intentionally deferred so the + * caller can run them inline via lazy `step_started` events (which create + * the step on the fly), saving one world round-trip per inline step. Up to + * `getMaxInlineSteps()` steps are deferred; the caller runs them inline in + * parallel and queues the rest. Empty when no step was deferred (nothing + * pending, or a `hook.getConflict()` awaiter is present so nothing is + * executed inline). The caller passes each `dehydratedInput` straight to + * `executeStep`, which sends it as the `step_started` payload. The atomic + * create-claim inside each `step_started` is the exactly-one-owner gate that + * the standalone `step_created` provided before: the loser of the race gets + * `EntityConflictError` → `skipped` and does not run the body. */ - lazyInlineStep?: { + lazyInlineSteps: Array<{ correlationId: string; stepName: string; dehydratedInput: SerializedData; - }; + }>; /** * The soonest pending wait, if any: seconds until it elapses and the * correlationId of the wait that produced that timeout. The @@ -387,21 +389,24 @@ export async function handleSuspension({ // racing with concurrent handlers on step execution. const createdStepCorrelationIds = new Set(); - // Lazy inline start: defer the step_created write for ONE step the caller - // will run inline. Its step is created on the fly by the lazy `step_started` - // executeStep sends (saving a round-trip). We never defer when a - // `hook.getConflict()` awaiter is present, because in that case the caller - // executes nothing inline (it re-invokes immediately to resolve the - // awaiter), so deferring would leave the step uncreated and unqueued. We - // pick the first uncreated step — matching the caller's `ownedPendingSteps[0]` - // inline-candidate selection — and dehydrate its input here so executeStep - // can ship it as the step_started payload. - const lazyInlineCorrelationId = + // 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 + // (saving a round-trip per step). We never defer when a `hook.getConflict()` + // awaiter is present, because in that case the caller executes nothing inline + // (it re-invokes immediately to resolve the awaiter), so deferring would + // leave the steps uncreated and unqueued. We pick the first N uncreated steps + // — matching the caller's inline-candidate selection — and dehydrate their + // input here so executeStep can ship it as the step_started payload. + const lazyInlineCorrelationIds = new Set( hasAwaitedHookCreation === false - ? stepItems.find((item) => stepsNeedingCreation.has(item.correlationId)) - ?.correlationId - : undefined; - let lazyInlineStep: SuspensionHandlerResult['lazyInlineStep']; + ? stepItems + .filter((item) => stepsNeedingCreation.has(item.correlationId)) + .slice(0, getMaxInlineSteps()) + .map((item) => item.correlationId) + : [] + ); + const lazyInlineSteps: SuspensionHandlerResult['lazyInlineSteps'] = []; const ops: Promise[] = []; @@ -428,12 +433,12 @@ export async function handleSuspension({ // step_created event) atomically. We do NOT add it to // createdStepCorrelationIds; ownership is decided by that lazy // step_started's atomic create-claim instead. - if (queueItem.correlationId === lazyInlineCorrelationId) { - lazyInlineStep = { + if (lazyInlineCorrelationIds.has(queueItem.correlationId)) { + lazyInlineSteps.push({ correlationId: queueItem.correlationId, stepName: queueItem.stepName, dehydratedInput: dehydratedInput as SerializedData, - }; + }); return; } const stepEvent: CreateEventRequest = { @@ -583,7 +588,7 @@ export async function handleSuspension({ return { pendingSteps: stepItems, createdStepCorrelationIds, - lazyInlineStep, + lazyInlineSteps, // On hook conflict the caller re-invokes immediately and never reads // the wait timeout, so don't report one. waitTimeout: hasHookConflict ? undefined : soonestWait, From 356f03130757496c3d4a0c4c8068ee9e0ea015a8 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Thu, 18 Jun 2026 19:28:38 +0000 Subject: [PATCH 2/8] Fix: A throttled step in a multi-step inline batch is re-queued as an input-less background step, but since a throttle rejects the lazy `step_started` before the step is created, the re-delivery sends a bare `step_started` that the world rejects with "Step not found", causing repeated redeliveries until MAX_QUEUE_DELIVERIES marks the step (and run) failed. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit fixes the issue reported at packages/core/src/runtime.ts:1484 ## Bug In `packages/core/src/runtime.ts`, the inline parallel-step batch aggregation handled a `throttled` result like this: ```ts } else if (r.type === 'throttled') { if (lazyInlineSteps.length === 1) { soleThrottleTimeout = r.timeoutSeconds; // re-delivers orchestrator (safe) } else { toRetry.push({ step: s, delaySeconds: r.timeoutSeconds }); // re-queues as background step (broken) } } ``` `toRetry` entries are re-queued as workflow-queue messages carrying only `{ runId, stepId, stepName }` (no input). They are consumed by the background-step branch (~line 510), which calls `executeStep` **without** `lazyStepInput`. ### Why a throttled step has no input to recover A `throttled` result is produced by `startErrorToResult` in `step-executor.ts` when the lazy `step_started` (the atomic create-claim) is rejected with a `ThrottleError`. That means the step was **never created** — there is no `step_created` event and no step entity. When the background-step branch later runs `executeStep` without `lazyStepInput`, `executeStep` sends a bare `step_started` with `eventData: { stepName }` (no `input`). In `packages/world-local/src/storage/events-storage.ts`, the lazy-start detector requires both `stepName` **and** `input`: ```ts const lazyStepStart = data.eventType === 'step_started' && ... && (data.eventData as { input?: unknown }).input !== undefined; ... if (!validatedStep && !lazyStepStart) { throw new WorkflowWorldError(`Step "${data.correlationId}" not found`); } ``` So a bare `step_started` for a never-created step throws `WorkflowWorldError("Step not found")`. That error is not one of the translatable errors (Throttle/RunExpired/EntityConflict/TooEarly), so `executeStep` rethrows it, the queue message is redelivered, and it fails repeatedly until `MAX_QUEUE_DELIVERIES`, at which point the step is marked failed and the run can fail. ### Trigger Two or more steps run inline in parallel (`lazyInlineSteps.length > 1`) and at least one of them is rate-limited on its lazy `step_started` create-claim (returns `throttled`). The single-step path was safe only because it returned `soleThrottleTimeout`, re-delivering the orchestrator so the step is re-attempted inline **with** its input on replay. ## Fix Generalized the single-step `soleThrottleTimeout` contract to all batch sizes. A throttled step is no longer pushed into `toRetry`; instead the batch tracks the longest throttle backoff (`throttleTimeout = Math.max(...)`) and returns `{ timeoutSeconds: throttleTimeout }`, deferring redelivery of the orchestrator message. On replay: - still-throttled steps are re-attempted **inline with their input** (their `step_created` is deferred anew), - `retry` steps from the same batch are re-dispatched as background steps by the suspension handler (they already exist, so a bare `step_started` is valid; their `retryAfter` is honored), - terminal steps (completed/failed/skipped/gone) are observed from their events and not re-run, - pending background ops are flushed via `waitUntil` before the replay reads them. Because the orchestrator replay drives all remaining work, the early `return` deliberately skips re-queuing `toRetry`, avoiding double-dispatch of those steps. `retry` results remain safe to re-queue as background steps because a retry implies `step_started` already succeeded and the step exists — only `throttled` (which fails on the create-claim itself) leaves the step uncreated. Co-authored-by: Vercel Co-authored-by: VaguelySerious --- packages/core/src/runtime.ts | 72 +++++++++++++++++++++++++----------- 1 file changed, 50 insertions(+), 22 deletions(-) diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 4026b23e5b..cdd20d9b7a 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1456,23 +1456,42 @@ export function workflowEntrypoint( replayBudget.resume(); } - // Aggregate the batch results. Steps that need to run - // again (`retry`/`throttled`) are re-queued per-step with - // their own delay; completed/failed steps already wrote - // their terminal events. We only loop back to replay when - // every inline step reached a terminal state — otherwise - // the still-pending steps will be re-run by their queued - // retry messages and the background-step handler replays - // once all steps are done. + // Aggregate the batch results. `retry` steps (which + // already exist — their `step_started` succeeded) are + // re-queued per-step as background steps with their own + // delay; `throttled` steps (rejected on the create-claim, + // so never created) instead defer redelivery of this + // orchestrator message so they re-run inline with input + // on replay; completed/failed steps already wrote their + // terminal events. We only loop back to replay when every + // inline step reached a terminal state — otherwise the + // still-pending steps will be re-run by their queued retry + // messages and the background-step handler replays once + // all steps are done. const toRetry: { step: (typeof lazyInlineSteps)[number]; delaySeconds: number; }[] = []; let anyPendingOps = false; - // Preserve the single-step backpressure contract: a lone - // throttled inline step delays redelivery of THIS - // orchestrator message (rather than re-queuing per-step). - let soleThrottleTimeout: number | undefined; + // A throttled inline step delays redelivery of THIS + // orchestrator message rather than being re-queued as a + // background step. Crucially, a `throttled` result means + // the lazy `step_started` was rejected on the atomic + // create-claim — so the step was NEVER created (no + // `step_created`, no step entity). Re-queuing it as a + // background step would send a bare `step_started` (no + // input), which the world rejects with `Step "" not + // found` because it cannot lazily create the step without + // its input; that error isn't translatable, so the + // message redelivers until MAX_QUEUE_DELIVERIES and the + // step (and run) fail. Deferring redelivery of the + // orchestrator instead re-attempts the throttled step + // inline WITH its input on replay. We track the longest + // backoff so a batch with multiple throttles waits the + // max. Note: `retry` results are safe to re-queue as + // background steps because a retry implies `step_started` + // already succeeded and the step exists. + let throttleTimeout: number | undefined; for (let i = 0; i < lazyInlineSteps.length; i++) { const r = stepResults[i]; const s = lazyInlineSteps[i]; @@ -1482,14 +1501,10 @@ export function workflowEntrypoint( delaySeconds: r.timeoutSeconds, }); } else if (r.type === 'throttled') { - if (lazyInlineSteps.length === 1) { - soleThrottleTimeout = r.timeoutSeconds; - } else { - toRetry.push({ - step: s, - delaySeconds: r.timeoutSeconds, - }); - } + throttleTimeout = Math.max( + throttleTimeout ?? 0, + r.timeoutSeconds + ); } else if ( r.type === 'completed' && r.hasPendingOps @@ -1498,8 +1513,21 @@ export function workflowEntrypoint( } } - if (soleThrottleTimeout !== undefined) { - return { timeoutSeconds: soleThrottleTimeout }; + if (throttleTimeout !== undefined) { + // Defer redelivery of the orchestrator after the + // throttle backoff. On replay every non-terminal step + // is re-dispatched by the suspension handler: the + // still-throttled steps run inline again WITH their + // input (their `step_created` is deferred anew), and + // any `retry` steps in this batch are queued as + // background steps with their own retryAfter honored. + // Terminal steps (completed/failed/skipped/gone) are + // observed from their events and not re-run; pending + // background ops are flushed via waitUntil before the + // replay reads them. Because the replay drives all + // remaining work, we must NOT also re-queue `toRetry` + // here — that would double-dispatch those steps. + return { timeoutSeconds: throttleTimeout }; } if (toRetry.length > 0) { From 01290dfaaadca8fd99ee4fe11cf782ddc7dba6c9 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 12:43:03 -0700 Subject: [PATCH 3/8] test(core): regression test for throttled inline step not re-queued as background Locks in the fix in 356f031307: a throttled step in a multi-step inline batch must defer the orchestrator (re-run inline with input on replay), never be re-queued as an input-less background step (which the world rejects with "Step not found"). Verified the test fails on the pre-fix aggregation. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/src/runtime.test.ts | 122 +++++++++++++++++++++++++++++- 1 file changed, 121 insertions(+), 1 deletion(-) diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 3a4337ae35..cd6b87ab12 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -1,4 +1,8 @@ -import { RUN_ERROR_CODES, WorkflowWorldError } from '@workflow/errors'; +import { + RUN_ERROR_CODES, + ThrottleError, + WorkflowWorldError, +} from '@workflow/errors'; import { type Event, SPEC_VERSION_CURRENT, @@ -1209,4 +1213,120 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { expect(order).not.toContain('step_created'); expect(order).not.toContain('queue_dispatch_start'); }); + + it('does not re-queue a throttled inline step as an input-less background step', async () => { + // Regression: a `throttled` result means the lazy step_started lost on the + // atomic create-claim, so the step was never created and has no input to + // recover. Re-queuing it as a background step would send a bare + // step_started that the world rejects with "Step not found", redelivering + // until MAX_QUEUE_DELIVERIES fails the run. The runtime must instead defer + // the orchestrator (return a timeout) so the step re-runs inline WITH its + // input on replay — never enqueue a stepId message for the throttled step. + process.env.WORKFLOW_MAX_INLINE_STEPS = '3'; + registerStepFunction('tA', async () => undefined); + registerStepFunction('tB', async () => undefined); + const wf = `const tA = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("tA"); + const tB = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("tB"); + async function workflow() { + const r = await Promise.all([tA(), tB()]); + return r; + }${getWorkflowTransformCode('workflow')}`; + + const workflowRun = await makeRunningRun('wrun_throttle_inline'); + const durableEvents: Event[] = []; + let seq = 0; + const rec = (data: any): Event => { + seq += 1; + const e = { + eventId: `e-${seq}`, + runId: workflowRun.runId, + createdAt: new Date(), + ...data, + } as Event; + durableEvents.push(e); + return e; + }; + // The SECOND lazy step_started to arrive is throttled (rejected on the + // create-claim); the first completes normally. Keyed by arrival order so we + // don't depend on which correlationId the runtime starts first. + let startedSeen = 0; + const eventsCreate = vi.fn(async (_runId: string, data: any) => { + if (data.eventType === 'run_started') + return { run: workflowRun, events: [] as Event[] }; + if (data.eventType === 'step_started') { + const d = data.eventData as { stepName?: string; input?: unknown }; + startedSeen += 1; + if (startedSeen === 2) { + throw new ThrottleError('rate limited', { retryAfter: 5 }); + } + 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: workflowRun.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 } : {}), + }; + } + return { event: rec(data) }; + }); + const stepIdMessages: unknown[] = []; + const queue = vi.fn(async (_queueName: string, message: any) => { + if (message && typeof message === 'object' && 'stepId' in message) { + stepIdMessages.push(message.stepId); + } + return { messageId: null }; + }); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn( + (_p: string, handler: (m: unknown, md: unknown) => Promise) => + async () => { + await handler( + { runId: workflowRun.runId, requestedAt: new Date() }, + { + requestId: 'req', + attempt: 1, + queueName: '__wkf_workflow_workflow', + messageId: 'msg', + } + ); + return new Response(null, { status: 204 }); + } + ), + events: { + create: eventsCreate, + list: vi.fn(async () => ({ + data: [...durableEvents], + hasMore: false, + cursor: 'c', + })), + }, + runs: { get: vi.fn(async () => workflowRun) }, + queue, + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as any); + + const res = (await workflowEntrypoint(wf)( + new Request('https://example.test') + )) as Response; + expect(res.status).toBe(204); + // The throttled step is NOT re-queued as a background (stepId) message — + // the orchestrator is deferred instead so it re-runs inline with input. + expect(stepIdMessages).toHaveLength(0); + }); }); From d267b5665808191d493561be98f9066299d424e1 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 12:49:00 -0700 Subject: [PATCH 4/8] docs(changelog): document inline-step queue behavior and correct throttle note Inline steps that complete cost zero queue round-trips (in-process replay loop); only steps that run as their own background invocations create flow-route messages. Also correct the throttle bullet to match the fix: a throttled inline step defers the orchestrator rather than being re-queued. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../docs/v5/changelog/lazy-event-creation.md | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/docs/content/docs/v5/changelog/lazy-event-creation.md b/docs/content/docs/v5/changelog/lazy-event-creation.md index 96fff126d9..7cf765126b 100644 --- a/docs/content/docs/v5/changelog/lazy-event-creation.md +++ b/docs/content/docs/v5/changelog/lazy-event-creation.md @@ -92,7 +92,7 @@ Previously the owned-inline path ran **exactly one** step inline per suspension The suspension handler now defers `step_created` for up to **`WORKFLOW_MAX_INLINE_STEPS` (default 3)** steps and returns them as `lazyInlineSteps`. The runtime runs that batch inline **in parallel** (`Promise.all`), each via its own lazy `step_started`, and queues only the steps beyond the cap. - **Selection:** the first N uncreated steps, matching the previous single-step inline candidate. Steps beyond N keep their eager `step_created` and are queued exactly as before. -- **Result aggregation:** steps that need to run again (`retry`/`throttled`) are re-queued per-step with their own delay; the runtime only loops back to replay once every inline step has reached a terminal state. A lone throttled inline step still delays redelivery of the orchestrator message (preserving the single-step backpressure contract). +- **Result aggregation:** `retry` steps (whose `step_started` succeeded, so the step exists) are re-queued per-step as background steps with their own delay. `throttled` steps are different: a throttle rejects the lazy `step_started` on the create-claim, so the step was *never created* and has no recoverable input — re-queuing it as an input-less background step would make the world reject the bare `step_started` with "Step not found" and redeliver until it fails. So any throttle instead **defers redelivery of the orchestrator** (by the longest throttle backoff in the batch), which re-runs the throttled step inline *with its input* on replay. The runtime only loops back to replay in-process once every inline step has reached a terminal state. - **Inline-delta fast path:** still used only for the single-step sequential case (`lazyInlineSteps.length === 1`). With more than one inline step each writes its own events, so a per-write delta would be partial; multi-step batches fall back to a normal incremental `events.list`. - **Config:** `WORKFLOW_MAX_INLINE_STEPS` is clamped to 1..16. Setting it to `1` reproduces the previous single-inline-step behavior exactly (a useful kill-switch). Inline bodies run in parallel within one function invocation, so the cap also bounds per-handler memory/CPU fan-out. @@ -112,3 +112,16 @@ When `WORKFLOW_OPTIMISTIC_INLINE_START` is enabled (**default on**), an inline s - **Exactly-one terminal write is preserved.** Optimistic start changes only *when the body runs*, never who writes the terminal event — that is still gated by the lazy `step_started` create-claim, which is awaited before the terminal write. Losers return `skipped`. - **Bounded to attempt 1.** Only brand-new (`!hasCreatedEvent`) steps are lazy; a retried step already has a `step_created`, so it takes the normal await-then-run path with the real attempt counter. Synthesizing `attempt = 1` locally is therefore always correct. - **Wider double-execution.** Running the body before confirming ownership means two handlers racing into the same batch boundary can *both* run the side effects before either wins (previously the loser 409'd on `step_created` and skipped before running anything). This is an explicit, accepted tradeoff: **inline step bodies must be idempotent.** Set `WORKFLOW_OPTIMISTIC_INLINE_START=0` (or `false`) to restore the await-`step_started`-then-run behavior. + +## Queue messages: inline steps don't pay a round-trip + +Inline steps that **complete** never enqueue a per-step flow-route message. When every step in an inline batch reaches a terminal state with no pending background ops, the runtime simply continues its in-process loop and replays — so a sequential chain (or a clean parallel fan-out) of inline steps runs entirely within one invocation with **zero** queue messages. Verified: a workflow whose only work is three parallel inline steps issues no `queue()` calls. + +The only flow-route messages produced around an inline batch are: + +- **Pre-batch dispatch** — the steps *beyond* the inline cap (and any pending wait/sleep continuation). Inline steps are explicitly excluded from this dispatch. +- **`retry` results** — one delayed message per retried step. A retry *is* the step becoming its own background invocation, so this is expected. +- **`throttled` results** — a single deferral of the orchestrator message (see above). +- **Pending background ops** — if any inline step left unflushed stream writes (e.g. output streams to blob storage), the loop breaks and enqueues **one** continuation (aggregated across the batch, not per-step) so `waitUntil` can flush before the next replay reads them. + +In other words: completed inline steps cost no queue round-trips; only steps that genuinely run as their own background invocations create new flow-route messages. From 145a84831c84352d61551d6b073b12145972cfc2 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 13:41:01 -0700 Subject: [PATCH 5/8] fix(core): default WORKFLOW_OPTIMISTIC_INLINE_START to off MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optimistic inline start can run a step body more than once when handlers race for the create-claim. For non-idempotent steps this is unsafe — notably, two concurrent runs of a step writing to the workflow stream (e.g. an AI agent streaming tokens) can interleave and corrupt the stream. Make the optimization opt-in (enable with WORKFLOW_OPTIMISTIC_INLINE_START=1); parallel inline execution (WORKFLOW_MAX_INLINE_STEPS) stays on by default. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../parallel-inline-optimistic-start.md | 2 +- .../docs/v5/changelog/lazy-event-creation.md | 6 ++--- packages/core/src/runtime/constants.test.ts | 22 +++++++++---------- packages/core/src/runtime/constants.ts | 19 ++++++++++------ .../core/src/runtime/step-handler.test.ts | 6 +++-- 5 files changed, 31 insertions(+), 24 deletions(-) diff --git a/.changeset/parallel-inline-optimistic-start.md b/.changeset/parallel-inline-optimistic-start.md index a6a8ba75e0..a5cf4c8ce3 100644 --- a/.changeset/parallel-inline-optimistic-start.md +++ b/.changeset/parallel-inline-optimistic-start.md @@ -3,4 +3,4 @@ '@workflow/core': minor --- -Inline execution now runs up to `WORKFLOW_MAX_INLINE_STEPS` (default 3) steps in parallel per suspension (each lazily created), and starts step bodies optimistically before `step_started` is confirmed (`WORKFLOW_OPTIMISTIC_INLINE_START`, default on) — reconciling the in-flight start before the terminal write so a lost create-claim is discarded. Optimistic bodies may run more than once under contention, so inline steps must be idempotent; disable with `WORKFLOW_OPTIMISTIC_INLINE_START=0`. +Inline execution now runs up to `WORKFLOW_MAX_INLINE_STEPS` (default 3) steps in parallel per suspension, each lazily created. An opt-in `WORKFLOW_OPTIMISTIC_INLINE_START` (default off) additionally starts step bodies before `step_started` is confirmed, reconciling the in-flight start before the terminal write so a lost create-claim is discarded; it is off by default because under contention a step body can run more than once (e.g. two runs writing to the workflow stream can corrupt it), so only enable it for idempotent steps. diff --git a/docs/content/docs/v5/changelog/lazy-event-creation.md b/docs/content/docs/v5/changelog/lazy-event-creation.md index 7cf765126b..bfaedccb41 100644 --- a/docs/content/docs/v5/changelog/lazy-event-creation.md +++ b/docs/content/docs/v5/changelog/lazy-event-creation.md @@ -96,11 +96,11 @@ The suspension handler now defers `step_created` for up to **`WORKFLOW_MAX_INLIN - **Inline-delta fast path:** still used only for the single-step sequential case (`lazyInlineSteps.length === 1`). With more than one inline step each writes its own events, so a per-write delta would be partial; multi-step batches fall back to a normal incremental `events.list`. - **Config:** `WORKFLOW_MAX_INLINE_STEPS` is clamped to 1..16. Setting it to `1` reproduces the previous single-inline-step behavior exactly (a useful kill-switch). Inline bodies run in parallel within one function invocation, so the cap also bounds per-handler memory/CPU fan-out. -## Optimistic inline start +## Optimistic inline start (opt-in, off by default) Normally `executeStep` awaits `step_started` (the lazy create-claim round-trip) before running the body. Because the inline path already holds the step input locally, it doesn't actually need that round-trip to begin. -When `WORKFLOW_OPTIMISTIC_INLINE_START` is enabled (**default on**), an inline step fires `step_started` **without awaiting it** and starts running the body immediately against locally-synthesized state. A lazy step is always brand-new, so attempt is 1, there is no prior error, and `startedAt` is now — all known without the server. The in-flight `step_started` is reconciled just before the terminal write: +When `WORKFLOW_OPTIMISTIC_INLINE_START` is enabled (set it to `1`/`true` — it is **off by default**), an inline step fires `step_started` **without awaiting it** and starts running the body immediately against locally-synthesized state. A lazy step is always brand-new, so attempt is 1, there is no prior error, and `startedAt` is now — all known without the server. The in-flight `step_started` is reconciled just before the terminal write: - **Lost the create-claim (409 / `EntityConflictError`)** → discard the body result and return `skipped`; the winning handler owns the terminal write. - **Run gone / throttled / too-early** → discard the body result and surface `gone` / `throttled` / `retry`. @@ -111,7 +111,7 @@ When `WORKFLOW_OPTIMISTIC_INLINE_START` is enabled (**default on**), an inline s - **Exactly-one terminal write is preserved.** Optimistic start changes only *when the body runs*, never who writes the terminal event — that is still gated by the lazy `step_started` create-claim, which is awaited before the terminal write. Losers return `skipped`. - **Bounded to attempt 1.** Only brand-new (`!hasCreatedEvent`) steps are lazy; a retried step already has a `step_created`, so it takes the normal await-then-run path with the real attempt counter. Synthesizing `attempt = 1` locally is therefore always correct. -- **Wider double-execution.** Running the body before confirming ownership means two handlers racing into the same batch boundary can *both* run the side effects before either wins (previously the loser 409'd on `step_created` and skipped before running anything). This is an explicit, accepted tradeoff: **inline step bodies must be idempotent.** Set `WORKFLOW_OPTIMISTIC_INLINE_START=0` (or `false`) to restore the await-`step_started`-then-run behavior. +- **Wider double-execution — why it's off by default.** Running the body before confirming ownership means two handlers racing for the same step's create-claim can *both* run the side effects before either wins (previously the loser 409'd on `step_created` and skipped before running anything). This is unsafe for non-idempotent steps: in particular, two concurrent runs of a step that writes to the **workflow stream** (e.g. an AI agent streaming tokens) can interleave and **corrupt the stream data**. So the optimization ships **disabled**; enable it (`WORKFLOW_OPTIMISTIC_INLINE_START=1`) only for deployments whose inline step bodies are idempotent and stream-safe. ## Queue messages: inline steps don't pay a round-trip diff --git a/packages/core/src/runtime/constants.test.ts b/packages/core/src/runtime/constants.test.ts index a4d80c369d..441e19d547 100644 --- a/packages/core/src/runtime/constants.test.ts +++ b/packages/core/src/runtime/constants.test.ts @@ -184,23 +184,23 @@ describe('isOptimisticInlineStartEnabled', () => { } }); - it('defaults to enabled when unset', () => { + it('defaults to disabled when unset', () => { delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; - expect(isOptimisticInlineStartEnabled()).toBe(true); - }); - - it('is disabled by "0"', () => { - process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '0'; expect(isOptimisticInlineStartEnabled()).toBe(false); }); - it('is disabled by "false" (case-insensitive)', () => { - process.env.WORKFLOW_OPTIMISTIC_INLINE_START = 'FALSE'; - expect(isOptimisticInlineStartEnabled()).toBe(false); + it('is enabled by an explicit "1"', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; + expect(isOptimisticInlineStartEnabled()).toBe(true); }); - it('stays enabled for any other value', () => { - process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; + it('is enabled by "true" (case-insensitive)', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = 'TRUE'; expect(isOptimisticInlineStartEnabled()).toBe(true); }); + + it('stays disabled for any other value', () => { + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = 'yes'; + expect(isOptimisticInlineStartEnabled()).toBe(false); + }); }); diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 3bcaecc35e..92bdf96a7e 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -184,17 +184,22 @@ export function getMaxInlineSteps(): number { * Whether optimistic inline step start is enabled. When on, the owned-inline * path begins running a brand-new step's body *before* its lazy `step_started` * network call resolves (the input is already known locally), awaiting the - * `step_started` only before the terminal write. This can run a step body more - * than once when handlers race — inline step bodies must be idempotent. + * `step_started` only before the terminal write. * - * Reads `process.env.WORKFLOW_OPTIMISTIC_INLINE_START` lazily. Default ON; - * disabled only by an explicit `'0'` / `'false'`, which restores the - * await-`step_started`-then-run behavior. + * This can run a step body more than once when handlers race for the same + * step's create-claim — both run the body before one wins. That is unsafe for + * steps with non-idempotent side effects; in particular, two concurrent runs + * of a step that writes to the workflow stream (e.g. an AI agent streaming + * tokens) can interleave and corrupt the stream data. So the optimization is + * **off by default** and must be explicitly opted into per deployment. + * + * Reads `process.env.WORKFLOW_OPTIMISTIC_INLINE_START` lazily. Default OFF; + * enabled only by an explicit `'1'` / `'true'`. */ export function isOptimisticInlineStartEnabled(): boolean { const raw = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; - if (raw === undefined || raw === '') return true; - return raw !== '0' && raw.toLowerCase() !== 'false'; + if (raw === undefined || raw === '') return false; + return raw === '1' || raw.toLowerCase() === 'true'; } // A replay-consumer mismatch can be caused by a transient divergent replay diff --git a/packages/core/src/runtime/step-handler.test.ts b/packages/core/src/runtime/step-handler.test.ts index 6cea41af21..b4cb64df43 100644 --- a/packages/core/src/runtime/step-handler.test.ts +++ b/packages/core/src/runtime/step-handler.test.ts @@ -1190,7 +1190,9 @@ describe('executeStep optimistic inline start', () => { beforeEach(() => { vi.clearAllMocks(); - delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; + // Optimistic start is OFF by default — explicitly enable it so these tests + // exercise the optimistic path. (The disabled-path test overrides to '0'.) + process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; vi.mocked(getStepFunction).mockReturnValue(mockStepFn); vi.mocked(normalizeUnknownError).mockImplementation( async (err: unknown) => ({ @@ -1208,7 +1210,7 @@ describe('executeStep optimistic inline start', () => { vi.restoreAllMocks(); }); - it('sends step_started carrying the input and completes (default on)', async () => { + it('sends step_started carrying the input and completes (when enabled)', async () => { mockEventsCreate .mockReset() .mockImplementation((_runId: string, event: { eventType: string }) => From 583d3ca5342f194b4b86508d42ce6e85037d8a88 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 14:00:20 -0700 Subject: [PATCH 6/8] chore: re-trigger CI (synchronize event dropped for prior pushes) From 91ad308a9e555488b35bbf94a65ffbb50dcce21b Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Thu, 18 Jun 2026 14:33:46 -0700 Subject: [PATCH 7/8] refactor(core): deterministic inline-batch order + throttle/flush note Two review follow-ups: - suspension-handler: collect deferred lazy inline steps into a Map keyed by correlationId and rebuild lazyInlineSteps in deterministic lazyInlineCorrelationIds (ordered first-N slice) order after the concurrent dehydration ops settle, instead of push()-ing in ops-completion order. No behavior change (results are zipped by index in the runtime), but the batch order is now stable regardless of dehydration timing. - runtime: document why the throttle-defer early return is safe to skip the anyPendingOps flush continuation (throttle backoff >= 1s always exceeds the <= 500ms + waitUntil in-invocation flush window). Co-Authored-By: Claude Opus 4.8 --- packages/core/src/runtime.ts | 18 ++++++++++++----- .../core/src/runtime/suspension-handler.ts | 20 +++++++++++++++++-- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index cdd20d9b7a..e10fde5089 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1522,11 +1522,19 @@ export function workflowEntrypoint( // any `retry` steps in this batch are queued as // background steps with their own retryAfter honored. // Terminal steps (completed/failed/skipped/gone) are - // observed from their events and not re-run; pending - // background ops are flushed via waitUntil before the - // replay reads them. Because the replay drives all - // remaining work, we must NOT also re-queue `toRetry` - // here — that would double-dispatch those steps. + // observed from their events and not re-run. Because + // the replay drives all remaining work, we must NOT + // also re-queue `toRetry` here — that would + // double-dispatch those steps. + // + // This returns BEFORE the `anyPendingOps` branch + // below, so a batch that mixes a throttle with a + // completed step that left unflushed ops does not + // queue the explicit flush continuation. That is safe + // because the throttle backoff (>= 1s) always exceeds + // the in-invocation flush window (<= 500ms + waitUntil), + // so ops settle before the post-backoff redelivery + // replays and reads them. return { timeoutSeconds: throttleTimeout }; } diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 55f5ec6c25..e31b8ba42a 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -406,7 +406,14 @@ export async function handleSuspension({ .map((item) => item.correlationId) : [] ); - const lazyInlineSteps: SuspensionHandlerResult['lazyInlineSteps'] = []; + // Collected by correlationId because the per-step ops below run concurrently + // and settle out of order. We rebuild the array in deterministic + // `lazyInlineCorrelationIds` order (the ordered slice above) after the ops + // settle, so the inline batch order is stable regardless of dehydration timing. + const lazyInlineByCorrelationId = new Map< + string, + SuspensionHandlerResult['lazyInlineSteps'][number] + >(); const ops: Promise[] = []; @@ -434,7 +441,7 @@ export async function handleSuspension({ // createdStepCorrelationIds; ownership is decided by that lazy // step_started's atomic create-claim instead. if (lazyInlineCorrelationIds.has(queueItem.correlationId)) { - lazyInlineSteps.push({ + lazyInlineByCorrelationId.set(queueItem.correlationId, { correlationId: queueItem.correlationId, stepName: queueItem.stepName, dehydratedInput: dehydratedInput as SerializedData, @@ -564,6 +571,15 @@ export async function handleSuspension({ // step_created and re-dispatches, and recovers the run instead of orphaning it. await Promise.all(ops); + // 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. + const lazyInlineSteps: SuspensionHandlerResult['lazyInlineSteps'] = []; + for (const correlationId of lazyInlineCorrelationIds) { + const lazyStep = lazyInlineByCorrelationId.get(correlationId); + if (lazyStep) lazyInlineSteps.push(lazyStep); + } + // Find the soonest pending wait (minimum timeout) const now = Date.now(); let soonestWait: { seconds: number; correlationId: string } | undefined; From ba626b63512813b67687a3f517ffde063fa60064 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Thu, 18 Jun 2026 14:49:38 -0700 Subject: [PATCH 8/8] Update packages/core/src/runtime.ts Co-authored-by: vercel[bot] <35613825+vercel[bot]@users.noreply.github.com> Signed-off-by: Peter Wielander --- packages/core/src/runtime.ts | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index e10fde5089..71a3ca4726 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1552,7 +1552,33 @@ export function workflowEntrypoint( traceCarrier: retryTraceCarrier, requestedAt: new Date(), }, - { delaySeconds } + { + delaySeconds, + // Key the delayed retry on the step's + // correlationId so it dedupes against the + // keyed re-dispatch the suspension handler + // performs on replay (it also uses + // `idempotencyKey: step.correlationId`). + // + // Without this, a mixed batch where one step + // `completed` with unflushed background ops + // (`anyPendingOps`) and another step is + // retrying would double-dispatch the retry: + // the `anyPendingOps` branch below queues an + // immediate plain continuation, whose replay + // sees the still-`retrying` step as pending + // and re-dispatches it *immediately* and + // *with* a key. Since this delayed retry had + // no key, the two messages wouldn't dedupe — + // the step would run twice, the configured + // retry backoff would be ignored (plain + // `Error` retries persist no `retryAfter`, so + // the world has no `TooEarly` guard), and the + // retry body could run early/concurrently. + // Sharing the key lets the earlier delayed + // message win, honoring the backoff. + idempotencyKey: step.correlationId, + } ) ) );