diff --git a/.changeset/overlap-workflow-compile.md b/.changeset/overlap-workflow-compile.md new file mode 100644 index 0000000000..ec1dd241f7 --- /dev/null +++ b/.changeset/overlap-workflow-compile.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Compile Node.js workflow scripts while loading the replay event log. diff --git a/packages/core/src/runtime-trace-mode.test.ts b/packages/core/src/runtime-trace-mode.test.ts index 22721b4961..1d2d83389e 100644 --- a/packages/core/src/runtime-trace-mode.test.ts +++ b/packages/core/src/runtime-trace-mode.test.ts @@ -81,7 +81,10 @@ const simpleWorkflow = `async function workflow() { return 'done'; }${getWorkflowTransformCode('workflow')}`; -async function makeRunningRun(runId: string): Promise { +async function makeRunningRun( + runId: string, + executionContext?: WorkflowRun['executionContext'] +): Promise { return { runId, workflowName: 'workflow', @@ -91,6 +94,7 @@ async function makeRunningRun(runId: string): Promise { updatedAt: new Date('2024-01-01T00:00:00.000Z'), startedAt: new Date('2024-01-01T00:00:00.000Z'), deploymentId: 'test-deployment', + executionContext, }; } @@ -105,12 +109,16 @@ async function driveHandler(opts: { workflowCode: string; traceCarrier?: Record; routeModuleBodyStartedAt?: number; + includeRunInput?: boolean; + executionContext?: WorkflowRun['executionContext']; + whileRunStartedPending?: () => Promise; }) { - const workflowRun = await makeRunningRun(opts.runId); + const workflowRun = await makeRunningRun(opts.runId, opts.executionContext); const queuedMessages: any[] = []; const eventsCreate = vi.fn(async (_runId: string, data: any) => { if (data.eventType === 'run_started') { + await opts.whileRunStartedPending?.(); return { run: workflowRun, events: [] as Event[] }; } return { @@ -136,10 +144,23 @@ async function driveHandler(opts: { runId: workflowRun.runId, requestedAt: new Date('2024-01-01T00:00:00.000Z'), traceCarrier: opts.traceCarrier, + ...(opts.includeRunInput + ? { + runInput: { + input: workflowRun.input, + deploymentId: workflowRun.deploymentId, + workflowName: workflowRun.workflowName, + specVersion: SPEC_VERSION_CURRENT, + executionContext: workflowRun.executionContext, + }, + } + : {}), }, { requestId: 'req_test', - attempt: 1, + // Keep this trace harness on the awaited run_started path even + // when a test supplies runInput for pre-response VM selection. + attempt: opts.includeRunInput ? 2 : 1, queueName: '__wkf_workflow_workflow', messageId: 'msg_test', } @@ -246,6 +267,57 @@ describe('getWorkflowTraceMode', () => { }); describe('workflowEntrypoint trace modes', () => { + it('compiles while run_started is loading, without evaluating early', async () => { + let observedOverlap = false; + await driveHandler({ + runId: 'wrun_trace_compile_overlap', + workflowCode: simpleWorkflow, + includeRunInput: true, + whileRunStartedPending: async () => { + expect( + exporter + .getFinishedSpans() + .find((span) => span.name === 'workflow.bundle.evaluate') + ).toBeUndefined(); + await vi.waitFor(() => { + expect( + exporter + .getFinishedSpans() + .find((span) => span.name === 'workflow.bundle.compile') + ).toBeDefined(); + }); + expect( + exporter + .getFinishedSpans() + .find((span) => span.name === 'workflow.bundle.evaluate') + ).toBeUndefined(); + observedOverlap = true; + }, + }); + + expect(observedOverlap).toBe(true); + expect( + exporter + .getFinishedSpans() + .find((span) => span.name === 'workflow.bundle.evaluate') + ).toBeDefined(); + }); + + it('does not compile a Node bundle for a known QuickJS run', async () => { + await driveHandler({ + runId: 'wrun_trace_quickjs_compile', + workflowCode: simpleWorkflow, + includeRunInput: true, + executionContext: { workflowVm: 'quickjs' }, + }); + + expect( + exporter + .getFinishedSpans() + .find((span) => span.name === 'workflow.bundle.compile') + ).toBeUndefined(); + }); + it('linked (default): nests under the flow route context with a link to the run-origin context', async () => { const { workflowSpan, diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 6cc4ce4415..15e756f679 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -134,6 +134,7 @@ import { import { getErrorName, getErrorStack, normalizeUnknownError } from './types.js'; import { buildWorkflowSuspensionMessage } from './util.js'; import { + compileWorkflowBundle, replayWorkflow, resumeWorkflow, type WorkflowResumeResult, @@ -948,6 +949,26 @@ export function workflowEntrypoint( const replayRecoveryReporter = replayDivergence ? new ReplayRecoveryReporter(replayDivergence.count) : ReplayRecoveryReporter.inert(); + // Compilation is useful only for the Node VM. Wait until the + // run's engine selection is known so QuickJS deliveries never + // parse and cache an unused node:vm Script. The promise is + // invocation-scoped and reused by every cold replay; + // evaluation still waits for a fresh VM context. + let compiledWorkflowScripts: + | ReturnType + | undefined; + const startWorkflowCompile = ( + workflow?: Pick + ) => { + if (!workflow || useQuickJSVm(workflow)) return; + compiledWorkflowScripts ??= compileWorkflowBundle( + workflowCode, + workflowName + ); + // Terminal runs can return without awaiting compilation. + void compiledWorkflowScripts.catch(() => {}); + return compiledWorkflowScripts; + }; // Every write this loop makes carries the cursor of the log // it was computed against, and folds a complete returned // delta into that log. @@ -1892,6 +1913,11 @@ export function workflowEntrypoint( // All steps done: fall through to the main replay loop. // Set up shared state so the loop can continue. + // The step body itself needs no workflow VM. Start Node + // compilation only now, once this delivery is known to + // continue into a workflow replay rather than return for + // a still-pending sibling. + startWorkflowCompile(bgRun); runtimeLogger.debug( 'All parallel steps done, replaying inline after background step', { workflowRunId: runId } @@ -2052,7 +2078,7 @@ export function workflowEntrypoint( span?.addEvent('workflow.hook_received.create.start', { 'workflow.hook_received.preload_events': true, }); - const result = await traceReplayLoad('hook_preload', () => + const replayLoad = traceReplayLoad('hook_preload', () => createEvent( { eventType: 'hook_received', @@ -2072,6 +2098,7 @@ export function workflowEntrypoint( } ) ); + const result = await replayLoad; hookEnsured = true; // Note: unlike the re-ensure below, this hoisted write // does NOT set HookResilientResumeMaterialized: it @@ -2161,6 +2188,7 @@ export function workflowEntrypoint( return; } workflowRun = result.run; + startWorkflowCompile(workflowRun); maxEventsLimit = clampMaxEvents(result.maxEvents); // Anchors RSFS, see the declaration above. This // response plays run_started's role on this path. @@ -2276,6 +2304,7 @@ export function workflowEntrypoint( // wasted list+resolve it would otherwise compute. { requestId, skipPreload: true } ); + startWorkflowCompile(runInput); runReadyBarrier = startedPromise; // Turbo backgrounds run_started, so the non-turbo // assignment below never runs. Thread the per-run event @@ -2344,10 +2373,14 @@ export function workflowEntrypoint( span?.addEvent('workflow.run_started.create.start', { 'workflow.run_started.skip_preload': false, }); - const result = await traceReplayLoad( - 'run_started', - () => createEvent(runStartedEvent, { requestId }) + const replayLoad = traceReplayLoad('run_started', () => + createEvent(runStartedEvent, { requestId }) ); + // Initial deliveries carry runInput, so Node compilation + // can overlap this load without guessing the VM engine. + // Continuations learn the engine from result.run below. + startWorkflowCompile(runInput); + const result = await replayLoad; workflowRun = result.run; maxEventsLimit = clampMaxEvents(result.maxEvents); // Anchors RSFS, see the declaration above. @@ -2392,6 +2425,7 @@ export function workflowEntrypoint( return; } + startWorkflowCompile(workflowRun); } catch (err) { // Run was concurrently completed/failed/canceled if ( @@ -3084,12 +3118,18 @@ export function workflowEntrypoint( if (workflowResult.type === 'replay') { retainedSession = null; + const compiled = startWorkflowCompile(workflowRun); + assert( + compiled, + 'Node workflow replay requires compiled scripts' + ); workflowResult = await replayWorkflow({ workflowCode, workflowRun, events: eventLog.events, encryptionKey, replayPayloadCache, + compiledWorkflowScripts: await compiled, // Turbo: the end-of-run drain inside workflow // execution commits fire-and-forget `*_created` // events before the terminal `awaitRunReady()` below. diff --git a/packages/core/src/runtime/vm-mode.ts b/packages/core/src/runtime/vm-mode.ts index 1ea4349af7..876bd56d4e 100644 --- a/packages/core/src/runtime/vm-mode.ts +++ b/packages/core/src/runtime/vm-mode.ts @@ -58,7 +58,9 @@ export function getWorkflowVmFromEnv( * Throws if `WORKFLOW_VM` or `executionContext.workflowVm` is set to an * unknown value. */ -export function useQuickJSVm(workflowRun: WorkflowRun): boolean { +export function useQuickJSVm( + workflowRun: Pick +): boolean { const vmFromRun = ( workflowRun.executionContext as { workflowVm?: string } | undefined )?.workflowVm; diff --git a/packages/core/src/vm/script-cache.ts b/packages/core/src/vm/script-cache.ts index 09db402497..f1f6da93a9 100644 --- a/packages/core/src/vm/script-cache.ts +++ b/packages/core/src/vm/script-cache.ts @@ -39,27 +39,21 @@ import { globalSingleton } from '@workflow/utils'; * function bodies the duplicated work is the (cheap) top-level parse, not full * per-workflow codegen. * - * We use a nested Map (code -> filename -> Script) so that evicting a bundle - * (e.g. a new deployment/hot-reload producing a different `code`) drops the old - * code string and all of its per-filename scripts together. + * We use a nested Map (code -> filename -> Script) so that evicting a source + * string drops all of its per-filename scripts together. Most entries are full + * workflow bundles; `compileWorkflowBundle` also caches its tiny workflow-name + * lookup snippets here. * * Bounding * -------- * The top-level (`code`-keyed) map is an insertion-ordered LRU capped at - * `MAX_BUNDLES` entries. In production this bound is never reached: a - * deployment is its own process serving exactly one build-time bundle literal - * (skew protection runs old versions as separate processes), so there is a - * single `code` key for the process lifetime. The bound exists for dev/watch - * mode, where the dev route re-reads `workflowCode` from disk and re-invokes - * the entrypoint on every edit: each edit produces a NEW bundle string, which - * without a bound would pin every historical version forever (~0.8MB per edit, - * growing monotonically with edit count). The dev path only ever needs the - * latest bundle, so an LRU that keeps the few most-recent bundles and evicts - * the rest preserves the pre-cache GC behaviour while still serving the - * steady-state single-bundle case for free. The per-`filename` inner map is not - * separately bounded: it is naturally bounded by the (small) number of workflow - * source files in a bundle and is dropped wholesale when its parent `code` - * entry is evicted. + * `MAX_SCRIPT_SOURCES` entries. A production deployment has one large bundle + * source plus small lookup sources; the bundle is touched immediately before + * its lookup on every compilation, so lookup churn cannot evict the expensive + * entry in normal use. The bound primarily protects dev/watch mode, where every + * edit produces a new bundle string that would otherwise pin all historical + * versions. The per-`filename` inner map is naturally bounded by the workflows + * compiled from that source and is dropped wholesale with its parent entry. */ // On `globalThis` (see `globalSingleton`): compiling a bundle is the expensive // part this cache exists to skip, and per-copy caches would pay it once per @@ -69,13 +63,10 @@ const scripts = globalSingleton('@workflow/core//vmScriptCache', 1, () => ({ })); /** - * Max number of distinct bundle (`code`) versions to retain. One is enough for - * production; a handful covers pathological dev hot-reload / repeated-rebuild - * churn within a single long-lived process (e.g. a watch session or a test - * file) without unbounded growth. Kept deliberately small: there is no value - * in retaining stale bundles, only a memory cost. + * Maximum number of distinct script source strings to retain. Kept deliberately + * small because stale bundles and one-off lookup snippets have no lasting value. */ -const MAX_BUNDLES = 8; +const MAX_SCRIPT_SOURCES = 8; /** * Looks up the per-filename map for `code`, marking it most-recently-used. @@ -83,7 +74,7 @@ const MAX_BUNDLES = 8; * existing key moves it to the end (newest), so the first key is always the * least-recently-used eviction candidate. */ -function touchBundle(code: string): Map | undefined { +function touchScriptSource(code: string): Map | undefined { const byFilename = scripts.byCode.get(code); if (byFilename === undefined) { return undefined; @@ -95,9 +86,9 @@ function touchBundle(code: string): Map | undefined { } /** - * Returns a compiled `vm.Script` for the given workflow bundle code and - * filename, compiling and caching it on first use. Subsequent calls with the - * same `(code, filename)` return the cached `Script`. + * Returns a compiled `vm.Script` for the given source code and filename, + * compiling and caching it on first use. Subsequent calls with the same + * `(code, filename)` return the cached `Script`. * * The returned `Script` is not yet bound to any context; the caller runs it * against a specific VM context via `script.runInContext(context)`. This is @@ -108,13 +99,13 @@ export function getCachedWorkflowScript( code: string, filename: string ): { script: Script; cacheHit: boolean } { - let byFilename = touchBundle(code); + let byFilename = touchScriptSource(code); if (byFilename === undefined) { byFilename = new Map(); scripts.byCode.set(code, byFilename); - // Evict the least-recently-used bundle(s) when over the cap. New bundles + // Evict the least-recently-used source(s) when over the cap. New sources // are appended at the end, so the oldest live at the front. - while (scripts.byCode.size > MAX_BUNDLES) { + while (scripts.byCode.size > MAX_SCRIPT_SOURCES) { const oldest = scripts.byCode.keys().next().value; if (oldest === undefined) { break; @@ -140,7 +131,7 @@ export function clearWorkflowScriptCache(): void { } /** - * Number of distinct bundle (`code`) versions currently retained. Exposed for + * Number of distinct script source strings currently retained. Exposed for * tests asserting the LRU bound; not used on the hot path. */ export function workflowScriptCacheSize(): number { diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index f138806fd9..ac97eb94b6 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -1,3 +1,4 @@ +import type { Script } from 'node:vm'; import type { Span } from '@opentelemetry/api'; import { ERROR_SLUGS, @@ -130,10 +131,49 @@ interface WorkflowSessionOptions { readonly events: Event[]; readonly encryptionKey: PayloadKey | undefined; readonly replayPayloadCache: ReplayPayloadCache; + readonly compiledWorkflowScripts?: CompiledWorkflowScripts; readonly runReadyBarrier?: Promise; readonly worldCapabilities?: WorldCapabilities; } +/** Context-independent V8 scripts that can be evaluated in any fresh VM. */ +export interface CompiledWorkflowScripts { + readonly bundleScript: Script; + readonly workflowLookupScript: Script; +} + +/** + * Compile the workflow bundle before its run snapshot is available. + * + * Compilation depends only on the route's bundle string and workflow name, + * not the event log or VM context. The runtime starts this promise while + * `run_started` loads the replay snapshot, then evaluates the scripts only + * after it has created the fresh context. + */ +export function compileWorkflowBundle( + workflowCode: string, + workflowName: string +): Promise { + const parsedName = parseWorkflowName(workflowName); + const filename = parsedName?.moduleSpecifier || workflowName; + const workflowLookupCode = `globalThis.__private_workflows?.get(${JSON.stringify(workflowName)})`; + + return trace('workflow.bundle.compile', async (span) => { + const bundle = getCachedWorkflowScript(workflowCode, filename); + const lookup = getCachedWorkflowScript(workflowLookupCode, filename); + span?.setAttributes({ + // This attribute intentionally describes the workflow bundle. The tiny + // lookup script may miss when another workflow from the same source file + // runs, but that does not mean V8 recompiled the application bundle. + ...Attribute.WorkflowBundleCompileCacheHit(bundle.cacheHit), + }); + return { + bundleScript: bundle.script, + workflowLookupScript: lookup.script, + }; + }); +} + /** * A live workflow VM, parked at a suspension boundary. `resume` advances it * by appending events instead of replaying from scratch. @@ -325,6 +365,7 @@ async function createWorkflowSessionInner( events, encryptionKey, replayPayloadCache, + compiledWorkflowScripts, runReadyBarrier, worldCapabilities, }: WorkflowSessionOptions, @@ -1092,34 +1133,12 @@ async function createWorkflowSessionInner( ]; endVmTrace(); - // Get a reference to the user-defined workflow function. - // The filename parameter ensures stack traces show a meaningful name - // (e.g., "example/workflows/99_e2e.ts") instead of "evalmachine.". - const parsedName = parseWorkflowName(workflowRun.workflowName); - const filename = parsedName?.moduleSpecifier || workflowRun.workflowName; - const workflowLookupCode = `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`; - // Reuse compiled scripts by `(code, filename)`: compilation is deterministic // and the filename preserves workflow source attribution in stack traces. // The bundle registers workflows on `globalThis.__private_workflows`. - const { bundleScript, workflowLookupScript } = await trace( - 'workflow.bundle.compile', - async (span) => { - const bundle = getCachedWorkflowScript(workflowCode, filename); - const lookup = getCachedWorkflowScript(workflowLookupCode, filename); - span?.setAttributes({ - // This attribute intentionally describes the workflow bundle. The - // tiny workflow-name lookup script has its own cache entry and may - // miss when another workflow from the same source file runs, but that - // does not mean V8 recompiled the application bundle. - ...Attribute.WorkflowBundleCompileCacheHit(bundle.cacheHit), - }); - return { - bundleScript: bundle.script, - workflowLookupScript: lookup.script, - }; - } - ); + const { bundleScript, workflowLookupScript } = + compiledWorkflowScripts ?? + (await compileWorkflowBundle(workflowCode, workflowRun.workflowName)); const workflowFn = await trace('workflow.bundle.evaluate', async () => { bundleScript.runInContext(context); return workflowLookupScript.runInContext(context);