diff --git a/.changeset/trace-replay-phases.md b/.changeset/trace-replay-phases.md new file mode 100644 index 0000000000..96334a0ae5 --- /dev/null +++ b/.changeset/trace-replay-phases.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Trace workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution. diff --git a/packages/core/src/runtime-trace-mode.test.ts b/packages/core/src/runtime-trace-mode.test.ts index bf119ac4ac..8f71a1d2b6 100644 --- a/packages/core/src/runtime-trace-mode.test.ts +++ b/packages/core/src/runtime-trace-mode.test.ts @@ -201,7 +201,6 @@ async function driveHandler(opts: { const getWorldSpan = exporter .getFinishedSpans() .find((s) => s.name === 'workflow.route.get_world'); - return { workflowSpan, routeSpan, @@ -287,7 +286,6 @@ describe('workflowEntrypoint trace modes', () => { ); expect(getWorldSpan).toBeDefined(); expect(getWorldSpan?.parentSpanId).toBe(routeSpan?.spanContext().spanId); - expect(workflowSpan).toBeDefined(); // Child of the local /flow route span — same trace, so one // invocation is a single bounded trace rather than a new root. diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index 824935feb3..32a16480c9 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -82,6 +82,11 @@ export const WorkflowExecutionMode = SemanticConvention<'replay' | 'retained'>( 'workflow.execution.mode' ); +/** Whether every script needed for workflow bundle evaluation was cached. */ +export const WorkflowBundleCompileCacheHit = SemanticConvention( + 'workflow.bundle.compile.cache_hit' +); + /** * Events the replay walked past that no consumer claimed, still held when the * replay stopped. diff --git a/packages/core/src/vm/script-cache.test.ts b/packages/core/src/vm/script-cache.test.ts index 399f6b2c69..1dc2b34db5 100644 --- a/packages/core/src/vm/script-cache.test.ts +++ b/packages/core/src/vm/script-cache.test.ts @@ -1,10 +1,9 @@ -import { runInContext } from 'node:vm'; +import { type Context, runInContext } from 'node:vm'; import { afterEach, describe, expect, it } from 'vitest'; import { createContext } from './index.js'; import { clearWorkflowScriptCache, getCachedWorkflowScript, - runCachedWorkflowScript, workflowScriptCacheSize, } from './script-cache.js'; @@ -37,26 +36,43 @@ function buildBundle(marker: string, workflowCount = 12): string { return `globalThis.__private_workflows = new Map();\n${defs.join('\n')}\n`; } +function getScript(code: string, filename: string) { + return getCachedWorkflowScript(code, filename).script; +} + +function runScript(code: string, filename: string, context: Context) { + return getScript(code, filename).runInContext(context); +} + describe('script-cache', () => { afterEach(() => { clearWorkflowScriptCache(); }); it('returns the same compiled Script for identical (code, filename)', () => { - const a = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); - const b = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + const a = getScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + const b = getScript(SAMPLE_BUNDLE, 'workflows/a.ts'); expect(a).toBe(b); }); + it('reports whether compilation was served from cache', () => { + const first = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + const second = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + + expect(first.cacheHit).toBe(false); + expect(second.cacheHit).toBe(true); + expect(second.script).toBe(first.script); + }); + it('returns distinct Scripts for the same code under different filenames', () => { - const a = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); - const b = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/b.ts'); + const a = getScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + const b = getScript(SAMPLE_BUNDLE, 'workflows/b.ts'); expect(a).not.toBe(b); }); it('returns distinct Scripts for different code under the same filename', () => { - const a = getCachedWorkflowScript('1 + 1', 'workflows/a.ts'); - const b = getCachedWorkflowScript('2 + 2', 'workflows/a.ts'); + const a = getScript('1 + 1', 'workflows/a.ts'); + const b = getScript('2 + 2', 'workflows/a.ts'); expect(a).not.toBe(b); }); @@ -64,8 +80,8 @@ describe('script-cache', () => { // Cached path: run the bundle then look up the workflow, mirroring // runWorkflow's two-step evaluation. const { context: cachedCtx } = createContext({ seed, fixedTimestamp }); - runCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts', cachedCtx); - const cachedFn = runCachedWorkflowScript( + runScript(SAMPLE_BUNDLE, 'workflows/a.ts', cachedCtx); + const cachedFn = runScript( `globalThis.__private_workflows?.get('my/workflow')`, 'workflows/a.ts', cachedCtx @@ -90,16 +106,14 @@ describe('script-cache', () => { }); it('reuses the compiled Script across multiple runs against fresh contexts', async () => { - const script = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts'); + const script = getScript(SAMPLE_BUNDLE, 'workflows/a.ts'); const results: string[] = []; for (let i = 0; i < 3; i++) { const { context } = createContext({ seed, fixedTimestamp }); - runCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts', context); + runScript(SAMPLE_BUNDLE, 'workflows/a.ts', context); // The same cached Script object is used every iteration. - expect(getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts')).toBe( - script - ); + expect(getScript(SAMPLE_BUNDLE, 'workflows/a.ts')).toBe(script); const fn = runInContext( `globalThis.__private_workflows?.get('my/workflow')`, context @@ -119,7 +133,7 @@ describe('script-cache', () => { const editCount = 100; const filename = 'workflows/a.ts'; for (let i = 0; i < editCount; i++) { - getCachedWorkflowScript(buildBundle(`edit-${i}`), filename); + getScript(buildBundle(`edit-${i}`), filename); } const size = workflowScriptCacheSize(); @@ -130,9 +144,7 @@ describe('script-cache', () => { // The cache still serves correctly after heavy churn: the most-recently // inserted bundle is retained and repeated lookups return the same Script. const latest = buildBundle(`edit-${editCount - 1}`); - expect(getCachedWorkflowScript(latest, filename)).toBe( - getCachedWorkflowScript(latest, filename) - ); + expect(getScript(latest, filename)).toBe(getScript(latest, filename)); }); it('keeps the most-recently-used bundle and evicts the stale one', () => { @@ -141,18 +153,18 @@ describe('script-cache', () => { // unrelated bundles churn through. LRU must NOT evict the bundle we keep // using, even though it was inserted first. const hot = buildBundle('hot'); - const hotScript = getCachedWorkflowScript(hot, filename); + const hotScript = getScript(hot, filename); for (let i = 0; i < 50; i++) { - getCachedWorkflowScript(buildBundle(`cold-${i}`), filename); + getScript(buildBundle(`cold-${i}`), filename); // Re-access the hot bundle so it stays most-recently-used. - expect(getCachedWorkflowScript(hot, filename)).toBe(hotScript); + expect(getScript(hot, filename)).toBe(hotScript); } // After all that churn the hot bundle is still the *same* cached Script — // proving LRU recency (touch-on-access), not mere insertion order, governs // eviction. - expect(getCachedWorkflowScript(hot, filename)).toBe(hotScript); + expect(getScript(hot, filename)).toBe(hotScript); }); it('never returns the wrong Script across realistic multi-workflow bundles', async () => { @@ -165,10 +177,10 @@ describe('script-cache', () => { const fileA = 'workflows/a.ts'; const fileB = 'workflows/b.ts'; - const xa = getCachedWorkflowScript(bundleX, fileA); - const xb = getCachedWorkflowScript(bundleX, fileB); - const ya = getCachedWorkflowScript(bundleY, fileA); - const yb = getCachedWorkflowScript(bundleY, fileB); + const xa = getScript(bundleX, fileA); + const xb = getScript(bundleX, fileB); + const ya = getScript(bundleY, fileA); + const yb = getScript(bundleY, fileB); // All four (code, filename) combinations are distinct Script objects. const scripts = [xa, xb, ya, yb]; @@ -179,12 +191,12 @@ describe('script-cache', () => { } // Same (code, filename) is stable across lookups. - expect(getCachedWorkflowScript(bundleX, fileA)).toBe(xa); - expect(getCachedWorkflowScript(bundleY, fileB)).toBe(yb); + expect(getScript(bundleX, fileA)).toBe(xa); + expect(getScript(bundleY, fileB)).toBe(yb); // Running each bundle yields its OWN marker, confirming no cross-wiring. const { context: ctxX } = createContext({ seed, fixedTimestamp }); - runCachedWorkflowScript(bundleX, fileA, ctxX); + runScript(bundleX, fileA, ctxX); const fnX = runInContext( `globalThis.__private_workflows?.get('app/workflow-3')`, ctxX @@ -192,7 +204,7 @@ describe('script-cache', () => { expect(await fnX('z')).toContain('bundle-X:3:z'); const { context: ctxY } = createContext({ seed, fixedTimestamp }); - runCachedWorkflowScript(bundleY, fileA, ctxY); + runScript(bundleY, fileA, ctxY); const fnY = runInContext( `globalThis.__private_workflows?.get('app/workflow-3')`, ctxY diff --git a/packages/core/src/vm/script-cache.ts b/packages/core/src/vm/script-cache.ts index d23bbc1624..85f4ad3014 100644 --- a/packages/core/src/vm/script-cache.ts +++ b/packages/core/src/vm/script-cache.ts @@ -1,4 +1,4 @@ -import { type Context, Script } from 'node:vm'; +import { Script } from 'node:vm'; /** * Module-level cache of compiled workflow-bundle `vm.Script` objects. @@ -101,7 +101,7 @@ function touchBundle(code: string): Map | undefined { export function getCachedWorkflowScript( code: string, filename: string -): Script { +): { script: Script; cacheHit: boolean } { let byFilename = touchBundle(code); if (byFilename === undefined) { byFilename = new Map(); @@ -117,23 +117,12 @@ export function getCachedWorkflowScript( } } let script = byFilename.get(filename); + const cacheHit = script !== undefined; if (script === undefined) { script = new Script(code, { filename }); byFilename.set(filename, script); } - return script; -} - -/** - * Runs the cached workflow-bundle `Script` against `context`. Compiles and - * caches the `Script` on first use for the given `(code, filename)`. - */ -export function runCachedWorkflowScript( - code: string, - filename: string, - context: Context -): unknown { - return getCachedWorkflowScript(code, filename).runInContext(context); + return { script, cacheHit }; } /** diff --git a/packages/core/src/workflow-tracing.test.ts b/packages/core/src/workflow-tracing.test.ts new file mode 100644 index 0000000000..d187f64dd7 --- /dev/null +++ b/packages/core/src/workflow-tracing.test.ts @@ -0,0 +1,133 @@ +import { context, trace as otelTrace } from '@opentelemetry/api'; +import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import type { WorkflowRun } from '@workflow/world'; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, +} from 'vitest'; +import { dehydrateWorkflowArguments } from './serialization.js'; +import { clearWorkflowScriptCache } from './vm/script-cache.js'; +import { runWorkflow } from './workflow.js'; + +const exporter = new InMemorySpanExporter(); +const provider = new BasicTracerProvider(); +const contextManager = new AsyncLocalStorageContextManager(); + +beforeAll(() => { + provider.addSpanProcessor(new SimpleSpanProcessor(exporter)); + contextManager.enable(); + context.setGlobalContextManager(contextManager); + otelTrace.setGlobalTracerProvider(provider); +}); + +afterAll(async () => { + await provider.shutdown(); + context.disable(); + otelTrace.disable(); +}); + +beforeEach(() => { + clearWorkflowScriptCache(); +}); + +afterEach(() => { + exporter.reset(); +}); + +async function makeRun(): Promise { + const runId = 'wrun_trace_replay'; + return { + runId, + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments(['hello'], runId, undefined, []), + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; +} + +const workflowCode = ` +async function workflow(value) { return value; } +globalThis.__private_workflows = new Map(); +globalThis.__private_workflows.set('workflow', workflow); +`; + +describe('fresh replay tracing', () => { + it('breaks workflow.run into blocking replay phases', async () => { + const run = await makeRun(); + await runWorkflow(workflowCode, run, [], undefined); + + const spans = exporter.getFinishedSpans(); + const workflowRun = spans.find( + (span) => span.name === 'workflow.run workflow' + ); + expect(workflowRun).toBeDefined(); + + const childNames = spans + .filter((span) => span.parentSpanId === workflowRun?.spanContext().spanId) + .map((span) => span.name); + expect(childNames).toEqual( + expect.arrayContaining([ + 'workflow.vm.create_context', + 'workflow.bundle.compile', + 'workflow.bundle.evaluate', + 'workflow.input.hydrate', + 'workflow.replay.execute', + ]) + ); + }); + + it('marks bundle compilation cache hits on later fresh replays', async () => { + const run = await makeRun(); + await runWorkflow(workflowCode, run, [], undefined); + await runWorkflow(workflowCode, run, [], undefined); + + const compileSpans = exporter + .getFinishedSpans() + .filter((span) => span.name === 'workflow.bundle.compile'); + expect(compileSpans).toHaveLength(2); + expect( + compileSpans.map( + (span) => span.attributes['workflow.bundle.compile.cache_hit'] + ) + ).toEqual([false, true]); + }); + + it('reports a bundle hit when only a different workflow lookup compiles', async () => { + const firstName = 'workflow//./workflows/shared//first'; + const secondName = 'workflow//./workflows/shared//second'; + const sharedBundle = ` +async function first(value) { return value; } +async function second(value) { return value; } +globalThis.__private_workflows = new Map(); +globalThis.__private_workflows.set(${JSON.stringify(firstName)}, first); +globalThis.__private_workflows.set(${JSON.stringify(secondName)}, second); +`; + const firstRun = { ...(await makeRun()), workflowName: firstName }; + const secondRun = { ...(await makeRun()), workflowName: secondName }; + + await runWorkflow(sharedBundle, firstRun, [], undefined); + await runWorkflow(sharedBundle, secondRun, [], undefined); + + const compileSpans = exporter + .getFinishedSpans() + .filter((span) => span.name === 'workflow.bundle.compile'); + expect( + compileSpans.map( + (span) => span.attributes['workflow.bundle.compile.cache_hit'] + ) + ).toEqual([false, true]); + }); +}); diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index de9a248cf1..10e4a375ee 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -42,10 +42,14 @@ import { WORKFLOW_USE_STEP, } from './symbols.js'; import * as Attribute from './telemetry/semantic-conventions.js'; -import { applyWorkflowSuspensionToSpan, trace } from './telemetry.js'; +import { + applyWorkflowSuspensionToSpan, + recordElapsedSpan, + trace, +} from './telemetry.js'; import { getWorkflowRunStreamId } from './util.js'; import { createContext } from './vm/index.js'; -import { runCachedWorkflowScript } from './vm/script-cache.js'; +import { getCachedWorkflowScript } from './vm/script-cache.js'; import { createAbortSignalStatics, createCreateAbortController, @@ -345,6 +349,11 @@ async function createWorkflowSession({ : `http://localhost:${(await getPortLazy()) ?? 3000}` ); + // Include both node:vm's context creation and the host-side sandbox wiring + // below. Most of the bootstrap lives in this function (EventsConsumer, + // workflow globals, Web API shims), so tracing createContext() alone would + // materially under-report VM startup. + const vmBootstrapStartedAt = Date.now(); const { context, globalThis: vmGlobalThis, @@ -1058,22 +1067,40 @@ async function createWorkflowSession({ vmGlobalThis[SYMBOL_FOR_REQ_CONTEXT] = (globalThis as any)[ SYMBOL_FOR_REQ_CONTEXT ]; + await recordElapsedSpan('workflow.vm.create_context', vmBootstrapStartedAt); // 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`. - runCachedWorkflowScript(workflowCode, filename, context); - const workflowFn = runCachedWorkflowScript( - `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`, - filename, - context + 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 workflowFn = await trace('workflow.bundle.evaluate', async () => { + bundleScript.runInContext(context); + return workflowLookupScript.runInContext(context); + }); if (typeof workflowFn !== 'function') { throw new WorkflowNotRegisteredError(workflowRun.workflowName); @@ -1085,24 +1112,26 @@ async function createWorkflowSession({ // workflow function subscribing its first step callbacks. let args: unknown[] = []; workflowContext.promiseQueue = workflowContext.promiseQueue.then(async () => { - const prepared = await replayPayloadCache.getWorkflowInput(workflowRun); - args = await hydrateWorkflowArguments( - workflowRun.input, - workflowRun.runId, - encryptionKey, - vmGlobalThis, - {}, - prepared - ); + // Include any residual preparation that did not finish while the event log + // was streaming, plus VM-local deserialization, in the blocking boundary. + args = await trace('workflow.input.hydrate', async () => { + const prepared = await replayPayloadCache.getWorkflowInput(workflowRun); + return hydrateWorkflowArguments( + workflowRun.input, + workflowRun.runId, + encryptionKey, + vmGlobalThis, + {}, + prepared + ); + }); }); await workflowContext.promiseQueue; // The user function's promise. It may stay pending across many resumes // (each parked step promise holds it up) and is raced against the current // attempt's interruption in waitForExecution. - const workflowBody = (async (): Promise => { - return await workflowFn(...args); - })(); + let workflowBody: Promise; const failWorkflow = async (error: unknown): Promise => { // Control-flow signals are handled by the runtime and do not mean the @@ -1231,8 +1260,20 @@ async function createWorkflowSession({ }, }; + // Start the user function inside the span, rather than wrapping the already + // running promise: an async workflow executes synchronously until its first + // await, and that work is part of replay. The span ends at the first + // suspension/completion; later retained resumes get their own workflow.run + // span and do not leave this replay span open while the VM is parked. + const execution = trace('workflow.replay.execute', async () => { + workflowBody = (async (): Promise => { + return await workflowFn(...args); + })(); + return waitForExecution(initialInterruption); + }); + return { session, - execution: waitForExecution(initialInterruption), + execution, }; }