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.ts b/packages/core/src/telemetry.ts index 1ae00dc7e9..d1a1548633 100644 --- a/packages/core/src/telemetry.ts +++ b/packages/core/src/telemetry.ts @@ -273,6 +273,52 @@ export async function trace( }); } +/** Starts a child span without installing it as the active context. */ +export async function startTraceSpan(spanName: string) { + const [tracer, otel] = await Promise.all([Tracer.value, OtelApi.value]); + if (!tracer || !otel) return { end() {}, fail() {} }; + + const span = tracer.startSpan(spanName); + let ended = false; + const finish = (status: api.SpanStatus) => { + if (ended) return; + ended = true; + span.setStatus(status); + span.end(); + }; + + return { + end: () => finish({ code: otel.SpanStatusCode.OK }), + fail: (error: unknown) => + finish({ + code: otel.SpanStatusCode.ERROR, + message: (error as Error).message, + }), + }; +} + +/** Keeps a parked workflow's ambient trace context aligned with each resume. */ +export async function createRefreshableTraceContext() { + const otel = await OtelApi.value; + if (!otel) { + return { refresh() {}, run: (fn: () => T): T => fn() }; + } + + let current = otel.context.active(); + const context: api.Context = { + getValue: (key) => current.getValue(key), + setValue: (key, value) => current.setValue(key, value), + deleteValue: (key) => current.deleteValue(key), + }; + + return { + refresh: () => { + current = otel.context.active(); + }, + run: (fn: () => T): T => otel.context.with(context, fn), + }; +} + /** * Emit a span whose start is back-dated to `startEpochMs` and whose end is now, * so its duration reflects an interval only measurable at its end (e.g. 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..3d2a1553a7 --- /dev/null +++ b/packages/core/src/workflow-tracing.test.ts @@ -0,0 +1,212 @@ +import { + context, + trace as otelTrace, + SpanStatusCode, +} from '@opentelemetry/api'; +import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import type { Event, WorkflowRun } from '@workflow/world'; +import { + afterAll, + afterEach, + assert, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { ReplayPayloadCache } from './replay-payload-cache.js'; +import { + dehydrateStepReturnValue, + dehydrateWorkflowArguments, +} from './serialization.js'; +import { createContext } from './vm/index.js'; +import { clearWorkflowScriptCache } from './vm/script-cache.js'; +import { replayWorkflow, resumeWorkflow, runWorkflow } from './workflow.js'; + +vi.mock('./vm/index.js', async (importActual) => { + const actual = await importActual(); + return { ...actual, createContext: vi.fn(actual.createContext) }; +}); + +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(); + vi.restoreAllMocks(); +}); + +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', + }; +} + +function spans(name: string) { + return exporter.getFinishedSpans().filter((span) => span.name === name); +} + +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 allSpans = exporter.getFinishedSpans(); + const [workflowRun] = spans('workflow.run workflow'); + expect(workflowRun).toBeDefined(); + + const childNames = allSpans + .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('records VM bootstrap failures on the create-context span', async () => { + vi.mocked(createContext).mockImplementationOnce(() => { + throw new Error('test bootstrap failure'); + }); + + await expect( + runWorkflow(workflowCode, await makeRun(), [], undefined) + ).rejects.toThrow('test bootstrap failure'); + + expect(spans('workflow.vm.create_context')[0]?.status).toEqual({ + code: SpanStatusCode.ERROR, + message: 'test bootstrap failure', + }); + }); + + it('parents retained workflow continuations to the retained run', async () => { + const run = await makeRun(); + const code = `const step = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step"); + async function workflow() { await step(); console.log("resumed"); } + globalThis.__private_workflows = new Map([["workflow", workflow]]); + `; + const activeSpanIds: (string | undefined)[] = []; + vi.spyOn(console, 'log').mockImplementation((message) => { + if (message === 'resumed') { + activeSpanIds.push(otelTrace.getActiveSpan()?.spanContext().spanId); + } + }); + + const first = await replayWorkflow({ + workflowCode: code, + workflowRun: run, + events: [], + encryptionKey: undefined, + replayPayloadCache: new ReplayPayloadCache(undefined), + }); + assert(first.type === 'suspended'); + expect(spans('workflow.replay.execute')).toHaveLength(1); + const step = first.suspension.steps[0]; + assert(step?.type === 'step'); + const result = await dehydrateStepReturnValue( + undefined, + run.runId, + undefined + ); + + const completed = await resumeWorkflow(first.session, [ + { + eventId: 'event-step-completed', + runId: run.runId, + eventType: 'step_completed', + correlationId: step.correlationId, + eventData: { stepName: 'step', result }, + createdAt: run.updatedAt, + }, + ] as Event[]); + assert(completed.type === 'completed'); + + expect(spans('workflow.replay.execute')).toHaveLength(1); + const retainedRun = spans('workflow.run workflow').find( + (span) => span.attributes['workflow.execution.mode'] === 'retained' + ); + expect(activeSpanIds).toEqual([retainedRun?.spanContext().spanId]); + }); + + 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 = spans('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 = spans('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..9ce4a39883 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -42,10 +42,15 @@ import { WORKFLOW_USE_STEP, } from './symbols.js'; import * as Attribute from './telemetry/semantic-conventions.js'; -import { applyWorkflowSuspensionToSpan, trace } from './telemetry.js'; +import { + applyWorkflowSuspensionToSpan, + createRefreshableTraceContext, + startTraceSpan, + 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, @@ -305,15 +310,26 @@ export async function runWorkflow( return result.output; } -async function createWorkflowSession({ - workflowCode, - workflowRun, - events, - encryptionKey, - replayPayloadCache, - runReadyBarrier, - worldCapabilities, -}: WorkflowSessionOptions): Promise<{ +async function createWorkflowSession(options: WorkflowSessionOptions) { + const vmTrace = await startTraceSpan('workflow.vm.create_context'); + return createWorkflowSessionInner(options, vmTrace.end).catch((error) => { + vmTrace.fail(error); + throw error; + }); +} + +async function createWorkflowSessionInner( + { + workflowCode, + workflowRun, + events, + encryptionKey, + replayPayloadCache, + runReadyBarrier, + worldCapabilities, + }: WorkflowSessionOptions, + endVmTrace: () => void +): Promise<{ session: WorkflowSession; execution: Promise; }> { @@ -344,7 +360,10 @@ async function createWorkflowSession({ ? `https://${process.env.VERCEL_URL}` : `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 { context, globalThis: vmGlobalThis, @@ -1058,22 +1077,40 @@ async function createWorkflowSession({ vmGlobalThis[SYMBOL_FOR_REQ_CONTEXT] = (globalThis as any)[ SYMBOL_FOR_REQ_CONTEXT ]; + 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`. - 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,25 +1122,22 @@ 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); - })(); - const failWorkflow = async (error: unknown): Promise => { // Control-flow signals are handled by the runtime and do not mean the // workflow has terminally failed. `onWorkflowError` usually already moved @@ -1130,6 +1164,7 @@ async function createWorkflowSession({ }; const waitForExecution = async ( + workflowBody: Promise, interruption: PromiseWithResolvers ): Promise => { let result: unknown; @@ -1192,6 +1227,12 @@ async function createWorkflowSession({ } }; + const workflowTraceContext = await createRefreshableTraceContext(); + const replayTrace = await startTraceSpan('workflow.replay.execute'); + const workflowBody = workflowTraceContext.run(async () => + workflowFn(...args) + ); + const session: WorkflowSession = { workflowRun, argumentCount: args.length, @@ -1216,8 +1257,9 @@ async function createWorkflowSession({ const interruption = withResolvers(); state = { type: 'running', interruption }; workflowContext.suspensionGeneration++; + workflowTraceContext.refresh(); eventsConsumer.append(nextEvents.slice(knownEvents.length)); - return waitForExecution(interruption); + return waitForExecution(workflowBody, interruption); } case 'replay': return { type: 'replay' }; @@ -1231,8 +1273,14 @@ async function createWorkflowSession({ }, }; + // The replay span measures the user function without becoming its ambient + // context. The workflow promise stays pending across retained resumes, so an + // active replay span here would remain captured after that span has ended. + const execution = waitForExecution(workflowBody, initialInterruption); + void execution.then(replayTrace.end, replayTrace.fail); + return { session, - execution: waitForExecution(initialInterruption), + execution, }; }