diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index c5bf715128..605021f660 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1945,6 +1945,16 @@ export function workflowEntrypoint( maxEventsLimit, namespace, nextTraceCarrier, + // Inline-step ownership plumbing: redeliveries of + // this message drive crash recovery for steps an + // earlier invocation claimed inline (see the + // backstop pass in the entrypoint's loop), and + // the message id is stamped as `ownerMessageId` + // on inline lazy step claims so wake replays + // defer to the in-flight body instead of + // requeueing the step. + deliveryAttempt: metadata.attempt, + ownerMessageId: metadata.messageId, }); if (quickjsResult?.timeoutSeconds !== undefined) { // Use `reinvoke` rather than returning diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index a28dda7b78..864344f914 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -25,6 +25,7 @@ import { import { parseWorkflowName } from '@workflow/utils/parse-name'; import { type Event, + ROOT_RUN_ID_ATTRIBUTE, type RunInput, SPEC_VERSION_CURRENT, type WorkflowRun, @@ -45,6 +46,10 @@ import { import { remapErrorStack, stripInlineSourceMap } from '../source-map.js'; import * as Attribute from '../telemetry/semantic-conventions.js'; import { serializeTraceCarrier } from '../telemetry.js'; +import { + getInlineOwnershipLeaseSeconds, + getMaxInlineSteps, +} from './constants.js'; import { getPortLazy } from './get-port-lazy.js'; import { getWorkflowQueueName, queueMessage } from './helpers.js'; import { @@ -54,8 +59,11 @@ import { type PendingOperation, type PendingStep, type PendingWait, - runQuickJSWorkflow, + startQuickJSWorkflow, } from './quickjs-runtime.js'; +import { ReplayBudget } from './replay-budget.js'; +import { executeStep, type StepExecutionResult } from './step-executor.js'; +import { runStepSingleFlight } from './step-single-flight.js'; import { getWaitContinuationDispatch } from './wait-continuation.js'; import { getWorld } from './world.js'; @@ -92,18 +100,87 @@ export function isFirstInvocation( ); } +/** + * Queue a step for background execution via the unified workflow queue + * (V2 architecture). The combined handler in runtime.ts dispatches + * messages with `stepId` to executeStep, which works for both VM engines. + * `delaySeconds` supports retry/throttle backoff. + */ +async function queueStepMessage(params: { + world: Awaited>; + runId: string; + workflowRun: WorkflowRun; + step: PendingStep; + delaySeconds?: number; + /** Queue namespace for the publish (see runtime.ts). */ + namespace: string | undefined; + /** Run-origin trace carrier accessor (see runWorkflowWithQuickJS). */ + nextTraceCarrier: () => Promise>; + /** + * Publish purpose, used to bucket the idempotency key. Worlds retire + * used keys (VQS retention TTL, world-postgres completed-keys cache), + * so a key shared across purposes silently swallows the second + * publish — see wait-continuation.ts for the same hazard on wait + * keys. `dispatch` is the plain background handoff (overflow / crash + * recovery) and keeps the bare correlationId so it stays mutually + * exclusive with the node engine's dispatch of the same step; + * `backstop:` covers delayed crash backstops, scoped to the + * ownership epoch so a refreshed lease re-arms a NEW backstop instead + * of being absorbed by the in-flight one; `retry:` covers delayed + * retry/throttle re-enqueues, scoped to the attempt so each backoff + * hop is enqueueable. + */ + purpose: 'dispatch' | `backstop:${string}` | `retry:${number}`; + wfdiag: (checkpoint: string, fields: Record) => void; +}): Promise { + const { + world, + runId, + workflowRun, + step, + delaySeconds, + namespace, + nextTraceCarrier, + purpose, + wfdiag, + } = params; + const traceCarrier = await nextTraceCarrier(); + await queueMessage( + world, + getWorkflowQueueName(workflowRun.workflowName, namespace), + { + runId, + stepId: step.correlationId, + stepName: step.stepId, + traceCarrier, + requestedAt: new Date(), + }, + { + idempotencyKey: + purpose === 'dispatch' + ? step.correlationId + : `${step.correlationId}:${purpose}`, + ...(delaySeconds && delaySeconds > 0 ? { delaySeconds } : {}), + } + ); + wfdiag('step_queued', { + stepId: step.stepId, + correlationId: step.correlationId, + purpose, + delaySeconds: delaySeconds ?? 0, + }); +} + /** * Dispatch durable side effects for a set of pending VM operations: * step_created (+ optional queueing), hook_created / hook_received (aborts), * attr_set, hook_disposed, and wait_created events. * - * Used in two modes: - * - suspension (queueSteps: true): normal suspension processing; new steps - * are queued for execution. - * - terminal drain (queueSteps: false): flush leftover side effects when - * the workflow completed or failed — mirrors the node:vm engine's - * drainPendingQueueItems. Steps are created but NOT queued, and the run - * is never requeued. + * Steps are created but never queued here — queueing (or inline + * execution) is the caller's decision. Used both for suspension + * processing (the inline loop) and for the terminal drain (flushing + * leftover side effects when the workflow completed or failed, mirroring + * the node:vm engine's drainPendingQueueItems). */ /** * Rejects retained Hooks before registration when the configured World @@ -145,7 +222,13 @@ async function dispatchPendingOps(params: { workflowRun: WorkflowRun; encryptionKey: RunPayloadKeys | undefined; pendingOperations: PendingOperation[]; - queueSteps: boolean; + /** + * Step cids whose `step_created` must NOT be written here: the inline + * loop claims these atomically via a lazy `step_started` (carrying the + * input), so a concurrent claimant loses with EntityConflictError + * instead of both invocations bare-starting the same step. + */ + skipStepCreation?: Set; /** Queue namespace for all message publishes (see runtime.ts). */ namespace: string | undefined; /** @@ -171,6 +254,7 @@ async function dispatchPendingOps(params: { namespace, nextTraceCarrier, } = params; + const skipStepCreation = params.skipStepCreation; const wfdiag = params.wfdiag; // Set when a hook with a parked getConflict() awaiter had its // hook_created written this invocation. The workflow must be re-invoked @@ -380,7 +464,11 @@ async function dispatchPendingOps(params: { } for (const op of pendingOperations) { - if (op.type === 'step' && !op.hasCreatedEvent) { + if ( + op.type === 'step' && + !op.hasCreatedEvent && + !skipStepCreation?.has(op.correlationId) + ) { const step = op as PendingStep; opsPromises.push( (async () => { @@ -406,35 +494,9 @@ async function dispatchPendingOps(params: { throw err; } - // Queue the step execution via the unified workflow queue - // (V2 architecture). The combined handler in runtime.ts - // dispatches messages with `stepId` to executeStep, which - // works for both VM engines — so the QuickJS engine reuses - // the same step execution path as the node:vm engine - // instead of needing a separate step route. Skipped in - // terminal-drain mode (the workflow already finished; the - // event is the durable record, matching the node:vm drain). - if (params.queueSteps) { - const traceCarrier = await nextTraceCarrier(); - await queueMessage( - world, - getWorkflowQueueName(workflowRun.workflowName, namespace), - { - runId, - stepId: step.correlationId, - stepName: step.stepId, - traceCarrier, - requestedAt: new Date(), - }, - { - idempotencyKey: step.correlationId, - } - ); - wfdiag('step_queued', { - stepId: step.stepId, - correlationId: step.correlationId, - }); - } + // NOTE: step queueing is the caller's decision — the inline + // loop executes fresh steps in the live VM and only queues the + // overflow / retry / backstop cases (see queueStepMessage). })() ); } else if (op.type === 'attribute' && !op.hasCreatedEvent) { @@ -542,6 +604,19 @@ export async function runWorkflowWithQuickJS(params: { * records run_failed with MAX_EVENTS_EXCEEDED. */ maxEventsLimit?: number; + /** + * Queue delivery attempt of the message driving this invocation (from + * the queue handler's metadata; 1 = first delivery). Surfaced in + * diagnostics; crash recovery itself is driven by the ownership-lease + * decision table in the loop, not by the attempt count. + */ + deliveryAttempt?: number; + /** + * Queue message ID of the delivery driving this invocation, stamped as + * `ownerMessageId` on inline lazy step claims so wake replays defer to + * the in-flight body instead of requeueing the step. + */ + ownerMessageId?: string; /** * Queue namespace resolved at route registration (runtime.ts). Must be * threaded into every message publish: the builders bake the namespace @@ -567,6 +642,8 @@ export async function runWorkflowWithQuickJS(params: { runInput, parentSpan, maxEventsLimit, + deliveryAttempt, + ownerMessageId, namespace, } = params; // Standalone-caller fallback (tests): without a runtime.ts carrier @@ -607,6 +684,7 @@ export async function runWorkflowWithQuickJS(params: { wfdiag('enter', { workflowName, + deliveryAttempt, hasPreloadedEvents: Array.isArray(preloadedEvents) && preloadedEvents.length > 0, preloadedEventCount: preloadedEvents?.length ?? 0, @@ -736,7 +814,7 @@ export async function runWorkflowWithQuickJS(params: { eventCount: events.length, }); - const result = await runQuickJSWorkflow({ + const session = await startQuickJSWorkflow({ // Pass the STRIPPED bundle to the VM so the inline source map // doesn't end up in the QuickJS heap. The original (unstripped) // `workflowCode` is still kept in this host-side scope and is used @@ -749,6 +827,7 @@ export async function runWorkflowWithQuickJS(params: { port, runInput, }); + let result = session.result; runtimeLogger.debug('QuickJS runtime: VM returned', { workflowRunId: runId, @@ -777,6 +856,552 @@ export async function runWorkflowWithQuickJS(params: { failureName: result.failed?.name, }); + // ---- Inline continuation loop ---- + // + // While the workflow is suspended, this loop keeps the VM alive and + // makes as much forward progress as possible within one invocation: + // + // 1. Dispatch durable side effects for the suspension's pending ops + // (step_created / hook_created / attr_set / wait_created / + // hook_received for aborts) and complete elapsed waits. + // 2. Feed all newly recorded events (attr_set, hook_created, elapsed + // wait_completed, terminals written by concurrent invocations, ...) + // into the LIVE VM via session.continueWithEvents — resuming + // execution exactly where it left off, no fresh-VM re-replay. + // Cheap progress is fed BEFORE running step bodies so promise + // chains that are not gated on steps (hook.getConflict(), + // setAttributes(), racing sleeps) advance first and can surface + // additional pending steps for the same inline batch. + // 3. Once no cheap progress remains, execute up to + // getMaxInlineSteps() steps created by THIS invocation inline (no + // queue round-trip), in parallel, with the replay budget paused + // during step bodies — mirroring the node:vm engine's inline + // replay loop. Overflow and retry/throttled steps are queued for + // background execution. A delayed wait-continuation message is + // enqueued for the soonest pending wait first, so racing timers + // fire on time (in a separate invocation) while step bodies block + // this one. + // + // The loop exits when the workflow settles, no forward progress is + // possible in-process, the replay budget is exhausted, or the run is + // gone. + const seenEventIds = new Set(); + for (const e of events) { + if (e.eventId) seenEventIds.add(e.eventId); + } + // Step cids already executed inline by this invocation. + const executedStepIds = new Set(); + // Steps for which THIS invocation already sent a queue message. + const queuedStepIds = new Set(); + // Aborts THIS invocation already recorded (hook_received written) — + // guards against re-recording when the VM-side flag has not been + // cleared yet within the same iteration. + const recordedAbortIds = new Set(); + // Waits for which THIS invocation already completed/scheduled work. + const completedWaitIds2 = new Set(); + // Inline-ownership state per step correlationId, derived from every + // event this invocation observes (initial log + every feed) — the + // quickjs analog of the replay-derived ownership on the node engine's + // StepInvocationQueueItem (see step-ownership.ts). Latest-wins: + // events arrive in log order, so a later step_started overwrites the + // stamp; a step_retrying lapses ownership permanently for the id. + const stepOwnership = new Map< + string, + { owner?: string; startedAtMs?: number; sawRetrying: boolean } + >(); + const observeEventsForOwnership = (observed: Event[]): void => { + for (const e of observed) { + if (e.correlationId === undefined) continue; + if (e.eventType === 'step_started') { + const owner = + 'eventData' in e && + e.eventData && + 'ownerMessageId' in e.eventData && + typeof e.eventData.ownerMessageId === 'string' + ? e.eventData.ownerMessageId + : undefined; + const prior = stepOwnership.get(e.correlationId); + stepOwnership.set(e.correlationId, { + owner, + startedAtMs: e.createdAt ? +new Date(e.createdAt) : undefined, + sawRetrying: prior?.sawRetrying ?? false, + }); + } else if (e.eventType === 'step_retrying') { + const prior = stepOwnership.get(e.correlationId); + stepOwnership.set(e.correlationId, { + ...(prior ?? {}), + sawRetrying: true, + }); + } + } + }; + observeEventsForOwnership(events); + const scheduledWaitContinuations = new Set(); + const maxInlineSteps = getMaxInlineSteps(); + const budget = new ReplayBudget(); + const workflowStartedAt = workflowRun.startedAt + ? +workflowRun.startedAt + : Date.now(); + const rootRunId = + (workflowRun.attributes as Record | undefined)?.[ + ROOT_RUN_ID_ATTRIBUTE + ] ?? runId; + let inlineStepsExecuted = 0; + let runGone = false; + // Set when this invocation wrote an event the workflow must consume to + // make progress (attr_set, getConflict-awaited hook_created) and the + // loop has not yet read it back — eventually-consistent listings can + // return 0 new events right after a write. If it is still set when the + // loop exits suspended, the entrypoint requeues immediately instead of + // exiting awaiting_external with the unblocking event already written + // and nothing scheduled to read it. + let pendingRequeueSignal = false; + + /** Fetch all events not yet processed by the live VM (log order). */ + const fetchUnseenEvents = async (): Promise => { + const unseen: Event[] = []; + let cursor: string | null = null; + let hasMore = true; + while (hasMore) { + const response = await world.events.list({ + runId, + pagination: { + sortOrder: 'asc', + cursor: cursor ?? undefined, + limit: 1000, + }, + }); + for (const e of response.data) { + if (e.eventId && seenEventIds.has(e.eventId)) continue; + if (e.eventId) seenEventIds.add(e.eventId); + unseen.push(e); + } + if (response.cursor) cursor = response.cursor; + hasMore = response.data.length > 0 && response.cursor != null; + } + observeEventsForOwnership(unseen); + return unseen; + }; + + try { + let iteration = 0; + while (result.suspended && !runGone && !budget.isExhausted()) { + iteration++; + // Re-check the event ceiling every turn: the loop appends events on + // each continueWithEvents, so a single invocation can otherwise grow + // the log arbitrarily far past the operator's limit (the node engine + // re-checks per replay for the same reason). `seenEventIds` counts + // every event this invocation has observed — initial log + all + // feeds. + if (maxEventsLimit !== undefined && seenEventIds.size >= maxEventsLimit) { + throw new MaxEventsExceededError(seenEventIds.size, maxEventsLimit); + } + const pendingOperations = result.suspended.pendingOperations; + + // Select this turn's inline candidates BEFORE dispatch: fresh steps + // (no step_created yet) that this invocation hasn't already handled. + // Their step_created is deliberately NOT written by dispatch — the + // inline claim below is a lazy step_started carrying the input, + // which the world applies as an atomic create-claim. A concurrent + // invocation racing on the same fresh step loses that claim with + // EntityConflictError and skips, so step bodies cannot double-run + // (previously both invocations bare-started the step after one lost + // the swallowed step_created race). + const freshSteps = pendingOperations.filter( + (op): op is PendingStep => + op.type === 'step' && + !op.hasCreatedEvent && + !executedStepIds.has(op.correlationId) && + !queuedStepIds.has(op.correlationId) + ); + const inlineCandidates = + maxInlineSteps <= 0 ? [] : freshSteps.slice(0, maxInlineSteps); + const inlineClaimCids = new Set( + inlineCandidates.map((step) => step.correlationId) + ); + + // 1. Durable side effects for this suspension's pending ops. + const opsToDispatch = pendingOperations.map((op) => + op.type === 'hook' && + (op as PendingHook).abortRequested && + recordedAbortIds.has(op.correlationId) + ? ({ ...op, abortRequested: false } as PendingOperation) + : op + ); + for (const op of pendingOperations) { + if (op.type === 'hook' && (op as PendingHook).abortRequested) { + recordedAbortIds.add(op.correlationId); + } + } + const dispatched = await dispatchPendingOps({ + world, + runId, + workflowRun, + encryptionKey, + namespace, + nextTraceCarrier, + pendingOperations: opsToDispatch, + skipStepCreation: inlineClaimCids, + wfdiag, + }); + if ( + dispatched.createdAttributeEvent || + dispatched.createdGetConflictHook + ) { + pendingRequeueSignal = true; + } + + // Hand steps beyond the inline cap to the queue NOW — in the same + // turn their step_created was written by the dispatch above. This + // must happen BEFORE the event feed below: the feed always observes + // those very step_created writes as unseen events and `continue`s, + // so a handoff placed after it is unreachable on the only iteration + // that still classifies these steps as fresh — next turn they carry + // hasCreatedEvent and would never be queued at all (the wedge behind + // promiseRaceStressTestWorkflow hanging in the quickjs CI legs). The + // bare-correlationId idempotency key makes repeats harmless. + const overflowSteps = freshSteps.slice(inlineCandidates.length); + for (const step of overflowSteps) { + queuedStepIds.add(step.correlationId); + await queueStepMessage({ + world, + runId, + workflowRun, + step, + namespace, + nextTraceCarrier, + purpose: 'dispatch', + wfdiag, + }); + } + + // Complete elapsed waits so their wait_completed events are picked + // up by the feed below (instead of a queue re-invocation). + const waitCompletePromises: Promise[] = []; + for (const op of pendingOperations) { + if (op.type !== 'wait') continue; + const wait = op as PendingWait; + if (completedWaitIds2.has(wait.correlationId)) continue; + if (new Date(wait.resumeAt).getTime() - Date.now() > 0) continue; + completedWaitIds2.add(wait.correlationId); + waitCompletePromises.push( + (async () => { + try { + await world.events.create(runId, { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: wait.correlationId, + }); + } catch (err) { + if (EntityConflictError.is(err)) return; + throw err; + } + })() + ); + } + if (waitCompletePromises.length > 0) { + await Promise.all(waitCompletePromises); + } + + // 2. Cheap progress first: feed newly recorded events into the live + // VM before blocking on step bodies. + { + const newEvents = await fetchUnseenEvents(); + if (newEvents.length > 0) { + // The listing caught up with this invocation's writes — any + // attr_set / getConflict hook_created has been (or is being) + // consumed by the live VM, so no external requeue is needed. + pendingRequeueSignal = false; + result = await session.continueWithEvents(newEvents); + wfdiag('inline_iteration', { + iteration, + phase: 'feed', + fedEvents: newEvents.length, + outcome: result.completed + ? 'completed' + : result.failed + ? 'failed' + : 'suspended', + }); + continue; + } + } + + // 3. No cheap progress left — execute steps inline. + const stepOps = pendingOperations.filter( + (op): op is PendingStep => op.type === 'step' + ); + // Steps created by an EARLIER invocation (or an earlier turn) that + // are still pending, with no work owned by THIS invocation. Mirror + // the node engine's ownership decision table (step-ownership.ts) — + // NOT a deliveryAttempt gate: worlds advance the attempt counter on + // routine redeliveries (world-local counts every handled response), + // so attempt > 1 is the common case and would fire backstops at + // steps actively executing inline in a live invocation. + // + // - Ownership lease ACTIVE, held by ANOTHER message → the step is + // (presumably) executing inline in a live invocation. Arm a + // DELAYED backstop for the lease remainder, keyed to the + // ownership epoch (a refreshed lease re-arms a fresh backstop + // instead of deduping against the in-flight one). If the owner + // completes normally, the backstop delivery resolves the step + // as 'skipped'. + // - Ownership lease ACTIVE, held by THIS message → this delivery + // is the owner's redelivery; the claimant crashed + // mid-execution. Dispatch immediately for background recovery. + // - No stamp / lease EXPIRED / step_retrying observed → the step + // is queue-owned or orphaned. Dispatch immediately; the + // bare-correlationId idempotency key dedupes against the + // original handoff. + const nowMs = Date.now(); + for (const step of stepOps) { + if (!step.hasCreatedEvent) continue; + if (executedStepIds.has(step.correlationId)) continue; + if (queuedStepIds.has(step.correlationId)) continue; + const ownership = stepOwnership.get(step.correlationId); + const ownershipActive = + ownership !== undefined && + ownership.owner !== undefined && + !ownership.sawRetrying; + let leaseRemainingSeconds = 0; + if (ownershipActive && ownership.startedAtMs !== undefined) { + const leaseSeconds = getInlineOwnershipLeaseSeconds(); + leaseRemainingSeconds = Math.min( + leaseSeconds, + Math.max( + 0, + Math.ceil( + (ownership.startedAtMs + leaseSeconds * 1000 - nowMs) / 1000 + ) + ) + ); + } + if ( + ownershipActive && + ownership.owner !== ownerMessageId && + leaseRemainingSeconds > 0 + ) { + queuedStepIds.add(step.correlationId); + await queueStepMessage({ + world, + runId, + workflowRun, + step, + delaySeconds: leaseRemainingSeconds, + namespace, + nextTraceCarrier, + purpose: `backstop:${ownership.startedAtMs}`, + wfdiag, + }); + } else { + queuedStepIds.add(step.correlationId); + await queueStepMessage({ + world, + runId, + workflowRun, + step, + namespace, + nextTraceCarrier, + purpose: 'dispatch', + wfdiag, + }); + } + } + + if (inlineCandidates.length === 0) { + // No in-process progress possible — the run awaits an external + // stimulus (hook payload, queued step, wait timer). + break; + } + + // Racing timers must fire on time while step bodies block this + // invocation: enqueue a delayed continuation for the soonest + // pending wait (a separate invocation writes its wait_completed at + // the right log position — same mechanism as the node:vm engine's + // wait-continuation dispatch). + let soonestWait: { correlationId: string; seconds: number } | undefined; + for (const op of pendingOperations) { + if (op.type !== 'wait') continue; + const wait = op as PendingWait; + if (scheduledWaitContinuations.has(wait.correlationId)) continue; + // Waits whose wait_completed THIS invocation already wrote (the + // elapsed-wait pass above) are done — the event just hasn't fed + // back into the VM yet. No continuation needed. + if (completedWaitIds2.has(wait.correlationId)) continue; + const resumeMs = new Date(wait.resumeAt).getTime() - Date.now(); + // An already-elapsed wait MUST still get a continuation (clamped + // to the 1s minimum, exactly like the node engine's + // `Math.max(1000, resumeAtMs - now)`), not be skipped: a wait + // whose deadline falls between this iteration's elapsed-wait + // pass (which saw it as still pending and wrote nothing) and + // this sweep would otherwise get NEITHER a wait_completed NOR a + // continuation — and the inline batch below then blocks this + // invocation for the full step duration with no wake armed + // anywhere. For `Promise.race(step, sleep)` that silently hands + // the race to the step: the sleep's wait_completed is never + // written and the run completes with the wrong winner. The + // window between the two checks spans this iteration's dispatch + // + feed round-trips, so on network-backed worlds (world-vercel) + // a short sleep lands in it routinely — observed as a ~50% + // sleepWinsRaceWorkflow failure rate in the Vercel e2e legs, + // while world-local's sub-ms round-trips masked it locally. The + // continuation invocation's pre-VM elapsed check writes the + // wait_completed ~1s later. + const seconds = Math.max(1, Math.ceil(resumeMs / 1000)); + if (!soonestWait || seconds < soonestWait.seconds) { + soonestWait = { correlationId: wait.correlationId, seconds }; + } + } + if (soonestWait) { + scheduledWaitContinuations.add(soonestWait.correlationId); + await queueMessage( + world, + getWorkflowQueueName(workflowRun.workflowName, namespace), + { + runId, + traceCarrier: await nextTraceCarrier(), + requestedAt: new Date(), + }, + getWaitContinuationDispatch( + soonestWait.seconds, + soonestWait.correlationId + ) + ); + wfdiag('wait_continuation_scheduled', { + correlationId: soonestWait.correlationId, + delaySeconds: soonestWait.seconds, + }); + } + + // Execute the inline batch in parallel. The replay budget is + // paused while step bodies run — step duration is bounded by the + // platform function duration, not the replay timeout. NOTE (by + // design): with the budget parked per batch, the only bound on how + // many inline steps one invocation can chain is the platform's + // function timeout — the SDK deliberately imposes no cap of its + // own, matching the node:vm engine, where a long sequential + // workflow likewise runs step-by-step until the platform reclaims + // the invocation and a redelivery resumes from the log. + budget.pause(); + let outcomes: StepExecutionResult[]; + try { + outcomes = await Promise.all( + inlineCandidates.map((step) => + runStepSingleFlight(runId, step.correlationId, () => + (async () => + executeStep({ + world, + workflowRunId: runId, + workflowDeploymentId: workflowRun.deploymentId, + workflowName: workflowRun.workflowName, + workflowStartedAt, + rootRunId, + stepId: step.correlationId, + stepName: step.stepId, + encryptionKey, + runSpecVersion: workflowRun.specVersion, + // Lazy inline claim: step_created is deferred (dispatch + // skipped it) and this step_started carries the input, + // so the world creates the step atomically — + // exactly-one-owner. A concurrent claimant gets + // EntityConflictError → { type: 'skipped' } and never + // runs the body. Mirrors the node engine's inline path. + lazyStepInput: await encryptSerializedData( + step.input, + encryptionKey + ), + // Ownership stamp: wake replays see the body as in + // flight in this invocation and arm a delayed backstop + // instead of immediately requeueing the step. + ownerMessageId, + // A lazy step is brand-new by construction — first + // attempt. + authoritativeAttempt: 1, + }))() + ) + ) + ); + } finally { + budget.resume(); + } + inlineStepsExecuted += inlineCandidates.length; + + for (let i = 0; i < inlineCandidates.length; i++) { + const step = inlineCandidates[i]; + const outcome = outcomes[i]; + executedStepIds.add(step.correlationId); + if (outcome.type === 'retry' || outcome.type === 'throttled') { + // Hand the step to the queue with the requested backoff — + // background delivery drives the retry from here. + queuedStepIds.add(step.correlationId); + await queueStepMessage({ + world, + runId, + workflowRun, + step, + delaySeconds: outcome.timeoutSeconds, + namespace, + nextTraceCarrier, + // Suffixed key: this step was inline-claimed, so no dispatch + // publish exists under the bare correlationId — but suffixing + // keeps the retry enqueueable even if a world retired a + // historical key for this step (see the purpose docs above). + purpose: 'retry:1', + wfdiag, + }); + } else if (outcome.type === 'gone') { + runGone = true; + } + // 'skipped': a concurrent invocation won the lazy create-claim and + // owns the body. Marked executed above so this invocation never + // re-claims it; the winner's terminal events arrive via the feed + // (or drive a separate invocation). + } + wfdiag('inline_steps_executed', { + iteration, + count: inlineCandidates.length, + outcomes: outcomes.map((o) => o.type), + }); + + // Feed the inline batch's terminal events into the live VM. When + // the eventually-consistent listing has not surfaced them yet, + // exiting must NOT ack silently: the terminals this invocation just + // caused are durably written with no queue message left to consume + // them (inline steps have none), so an awaiting_external exit would + // park the run 'running' with all its steps complete. Raise the + // requeue signal so the suspended exit schedules a fresh immediate + // invocation whose fresh read picks the terminals up. Outcomes that + // wrote no terminal ('skipped' — a concurrent claimant owns the + // body; 'gone', retry/throttled — a queue message exists) don't + // need it, but signaling on them too only costs a no-op invocation + // in an already-rare lag window. + const newEvents = await fetchUnseenEvents(); + if (newEvents.length === 0) { + pendingRequeueSignal = true; + break; + } + result = await session.continueWithEvents(newEvents); + + wfdiag('inline_iteration', { + iteration, + phase: 'steps', + fedEvents: newEvents.length, + outcome: result.completed + ? 'completed' + : result.failed + ? 'failed' + : 'suspended', + budgetExhausted: budget.isExhausted(), + }); + } + } finally { + session.dispose(); + } + + parentSpan?.setAttributes({ + ...Attribute.QuickJSInlineSteps(inlineStepsExecuted), + }); + if (result.completed) { // Workflow completed runtimeLogger.info('QuickJS runtime: workflow completed', { @@ -812,7 +1437,6 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, pendingOperations: result.completed.drainOperations, - queueSteps: false, wfdiag, }); } catch (err) { @@ -857,24 +1481,21 @@ export async function runWorkflowWithQuickJS(params: { throw err; } } else if (result.suspended) { - // Workflow suspended + // Workflow still suspended after the inline loop. All durable side + // effects for the final suspension state were already dispatched by + // the loop; what remains is deciding how the run gets re-invoked. const { pendingOperations } = result.suspended; runtimeLogger.info('QuickJS runtime: workflow suspended', { workflowRunId: runId, + inlineStepsExecuted, pendingSteps: pendingOperations.filter((p) => p.type === 'step').length, pendingWaits: pendingOperations.filter((p) => p.type === 'wait').length, pendingOps: pendingOperations.map((p) => ({ type: p.type, correlationId: p.correlationId, hasCreatedEvent: p.hasCreatedEvent, - ...(p.type === 'step' - ? { - stepId: (p as PendingStep).stepId, - inputType: typeof (p as PendingStep).input, - inputIsUint8Array: (p as PendingStep).input instanceof Uint8Array, - } - : {}), + ...(p.type === 'step' ? { stepId: (p as PendingStep).stepId } : {}), })), }); @@ -883,56 +1504,56 @@ export async function runWorkflowWithQuickJS(params: { ...Attribute.QuickJSPendingOpsCount(pendingOperations.length), }); - // Build per-pending-op promises so events.create + queueMessage - // calls fan out in parallel rather than serially. This mirrors - // the node:vm engine's `Promise.all(ops)` pattern in - // suspension-handler.ts and significantly reduces wall-clock time - // on cloud worlds (e.g. Vercel) where each storage call is a - // network round-trip. - let soonestWait: { seconds: number; correlationId: string } | undefined; - const { createdAttributeEvent, createdGetConflictHook } = - await dispatchPendingOps({ + if (runGone) { + // The run no longer exists (expired / deleted) — nothing to drive. + wfdiag('exit_suspended', { action: 'run_gone' }); + return; + } + + // Exit requeues are FRESH messages, never a `{ timeoutSeconds }` + // visibility-redelivery of the current message. Redelivering the + // CURRENT message is a trap: a hook-resume delivery carries + // `hookInput`, and its redelivery re-runs the lazy-resume re-ensure + // in the handler prologue — if the workflow disposed that hook + // during this invocation (dispose → sleep), a world that rejects + // the re-ensure would ack the message as "nothing left to resume" + // and the continuation it carried is silently lost. A fresh message + // carries only `runId`, so its delivery always reaches replay (and + // under turbo a reschedule would re-engage turbo against a stale + // preloaded log — see the reinvoke() docs in runtime.ts). + const requeueImmediately = async (): Promise => { + await queueMessage( world, - runId, - workflowRun, - encryptionKey, - namespace, - nextTraceCarrier, - pendingOperations, - queueSteps: true, - wfdiag, - }); + getWorkflowQueueName(workflowRun.workflowName, namespace), + { + runId, + traceCarrier: await nextTraceCarrier(), + requestedAt: new Date(), + } + ); + }; - // Handle pending waits — both newly created and still-pending from - // earlier invocations. For each wait, either create a wait_completed - // event (if elapsed) or track the soonest pending wait so a delayed - // continuation can be enqueued below. - let needsRequeue = false; - const waitCompletePromises: Promise[] = []; + if (budget.isExhausted()) { + // The loop stopped on the replay budget with progress still + // possible — continue in a fresh invocation. + wfdiag('exit_suspended', { action: 'budget_exhausted_requeue' }); + await requeueImmediately(); + return; + } + + // Exit wait sweep. A wait that elapsed in the window since the + // loop's last check requeues immediately (its wait_completed is + // written by the next invocation's elapsed-wait pass); pending waits + // whose continuation the loop already enqueued are skipped. + let soonestWait: { seconds: number; correlationId: string } | undefined; + let hasElapsedWait = false; for (const op of pendingOperations) { if (op.type !== 'wait') continue; const wait = op as PendingWait; const resumeMs = new Date(wait.resumeAt).getTime() - Date.now(); - if (resumeMs <= 0) { - // Wait has elapsed — create wait_completed and re-queue. - waitCompletePromises.push( - (async () => { - try { - await world.events.create(runId, { - eventType: 'wait_completed', - specVersion: SPEC_VERSION_CURRENT, - correlationId: wait.correlationId, - }); - needsRequeue = true; - } catch (err) { - if (EntityConflictError.is(err)) return; - throw err; - } - })() - ); - } else { - // Wait hasn't elapsed yet — track the soonest one. + hasElapsedWait = true; + } else if (!scheduledWaitContinuations.has(wait.correlationId)) { const timeoutSeconds = Math.max(1, Math.ceil(resumeMs / 1000)); if (!soonestWait || timeoutSeconds < soonestWait.seconds) { soonestWait = { @@ -942,57 +1563,40 @@ export async function runWorkflowWithQuickJS(params: { } } } - if (waitCompletePromises.length > 0) { - await Promise.all(waitCompletePromises); + + if (hasElapsedWait) { + wfdiag('exit_suspended', { action: 'wait_elapsed_requeue' }); + await requeueImmediately(); + return; } - // Progress and wait continuations are enqueued as FRESH messages - // rather than returned as `{ timeoutSeconds }` visibility-redelivery - // of the current message (which is what the node engine's suspension - // handler does too — see the wait-continuation dispatch in - // runtime.ts). Redelivering the CURRENT message is a trap: a - // hook-resume delivery carries `hookInput`, and its redelivery - // re-runs the lazy-resume re-ensure in the handler prologue. If the - // workflow disposed that hook during this invocation (dispose → - // sleep), the re-ensure gets HookNotFound, the prologue acks the - // message as "nothing left to resume", and the wait timer it was - // carrying is silently lost — the run wedges. A fresh continuation - // message carries only `runId`, so its delivery always reaches - // replay. - if (needsRequeue || createdAttributeEvent || createdGetConflictHook) { - // An elapsed wait was completed, a new attr_set event was written, - // or a getConflict()-awaited hook was created — re-queue immediately - // so the next invocation can process the new event. - wfdiag('exit_suspended', { - action: needsRequeue - ? 'wait_elapsed_requeue' - : createdAttributeEvent - ? 'attr_set_requeue' - : 'get_conflict_requeue', - timeoutSeconds: 0, - }); - await queueMessage( - world, - getWorkflowQueueName(workflowRun.workflowName, namespace), - { - runId, - traceCarrier: await nextTraceCarrier(), - requestedAt: new Date(), - } - ); + if (pendingRequeueSignal) { + // This invocation wrote events the workflow needs to consume + // (attr_set / getConflict-awaited hook_created / inline step + // terminals) but the eventually-consistent listing never returned + // them before the loop exited. Without a requeue the run would + // park awaiting_external with its unblocking events already + // durably written and no future invocation coming — requeue + // immediately so a fresh read picks them up. In the common case + // the loop's own feed observes the writes and clears this flag, so + // this only fires when the read actually lagged. + wfdiag('exit_suspended', { action: 'unread_self_write_requeue' }); + await requeueImmediately(); return; } if (soonestWait) { - // Delayed continuation for the soonest pending wait. The dispatch - // helper handles delay clamping (long waits chain across hops) and - // idempotency-key dedup of re-observations of the same pending - // wait — see runtime/wait-continuation.ts. + // Delayed continuation for the soonest pending wait the loop has + // not already scheduled. The dispatch helper handles delay + // clamping (long waits chain across hops) and idempotency-key + // dedup of re-observations of the same pending wait — see + // runtime/wait-continuation.ts. wfdiag('exit_suspended', { action: 'schedule_wait_timeout', timeoutSeconds: soonestWait.seconds, waitCorrelationId: soonestWait.correlationId, }); + scheduledWaitContinuations.add(soonestWait.correlationId); await queueMessage( world, getWorkflowQueueName(workflowRun.workflowName, namespace), @@ -1071,7 +1675,6 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, pendingOperations: result.failed.drainOperations, - queueSteps: false, wfdiag, }); } catch (err) { diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index 8043564a05..b885142be3 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -29,7 +29,12 @@ import type { Event, RunInput, WorkflowRun } from '@workflow/world'; import * as nanoid from 'nanoid'; -import { JSException, QuickJS, type WasiOptions } from 'quickjs-wasi'; +import { + type ExtensionDescriptor, + JSException, + QuickJS, + type WasiOptions, +} from 'quickjs-wasi'; import seedrandom from 'seedrandom'; import { runtimeLogger } from '../logger.js'; import { decompress } from '../serialization/compression.js'; @@ -239,6 +244,46 @@ globalThis.__workflowError = undefined; // This mirrors the event-replay runtime's payloadsQueue in hook.ts. globalThis.__hookPayloadBuffer = {}; +// Buffer for step/wait/attr terminal outcomes that arrive before this VM +// has constructed the corresponding awaiting promise. In fresh-VM replay +// the multi-pass event scan makes this unreachable (awaits are +// reconstructed before their terminals are re-scanned), but the live +// continuation path (continueWithEvents) scans each delta exactly once — +// a concurrent invocation's terminal arriving before this VM reaches the +// await would otherwise be dropped on the floor and the await would +// never settle (the feed's seen-set means it is never re-delivered). +// Mirrors __hookPayloadBuffer, which exists for exactly this reason on +// the hook path. Keyed by correlationId → single terminal (steps, waits +// and attrs settle exactly once). +globalThis.__terminalBuffer = {}; + +// Registers a resolver for an awaited primitive, first draining any +// buffered terminal recorded for the correlationId. Entries are prepared +// host-side (bytes already decrypted; see processEvents). +globalThis.__registerResolver = function(correlationId, resolve, reject) { + var buffered = globalThis.__terminalBuffer[correlationId]; + if (buffered) { + delete globalThis.__terminalBuffer[correlationId]; + if (buffered.kind === "resolve_bytes") { + resolve(globalThis[Symbol.for("workflow-deserialize")](buffered.bytes)); + } else if (buffered.kind === "resolve_value") { + resolve(buffered.value); + } else if (buffered.kind === "reject_bytes") { + reject(globalThis[Symbol.for("workflow-deserialize")](buffered.bytes)); + } else if (buffered.kind === "reject_error") { + var e = new Error(buffered.message); + e.name = "FatalError"; + e.fatal = true; + if (buffered.stack) e.stack = buffered.stack; + reject(e); + } else { + resolve(undefined); + } + return; + } + globalThis.__resolvers[correlationId] = { resolve: resolve, reject: reject }; +}; + // Stubs for Web APIs that the workflow bundle may reference but are not // available in QuickJS. Native C extensions (encoding, headers, url, // structured-clone) provide the real implementations; these are minimal @@ -395,7 +440,7 @@ globalThis[Symbol.for("WORKFLOW_USE_STEP")] = function(stepId, closureVarsFn) { hasCreatedEvent: false, }); return new Promise(function(resolve, reject) { - globalThis.__resolvers[correlationId] = { resolve: resolve, reject: reject }; + globalThis.__registerResolver(correlationId, resolve, reject); }); }; // Set stepId on the proxy so the StepFunction reducer can detect and @@ -478,7 +523,7 @@ globalThis[Symbol.for("WORKFLOW_SLEEP")] = function(param) { hasCreatedEvent: false, }); return new Promise(function(resolve, reject) { - globalThis.__resolvers[correlationId] = { resolve: resolve, reject: reject }; + globalThis.__registerResolver(correlationId, resolve, reject); }); }; @@ -739,7 +784,7 @@ globalThis[Symbol.for("WORKFLOW_SET_ATTRIBUTES")] = function(changes, options) { hasCreatedEvent: false, }); return new Promise(function(resolve, reject) { - globalThis.__resolvers[correlationId] = { resolve: resolve, reject: reject }; + globalThis.__registerResolver(correlationId, resolve, reject); }); }; @@ -936,7 +981,65 @@ globalThis[Symbol.for("WORKFLOW_GET_STREAM_ID")] = function(namespace) { * new runs can restore from it instead of paying VM creation + eval cost * (`QuickJS.restore` accepts the same wasi override). */ -async function initWorkflowVM(getNowMs: () => number): Promise { +/** + * Loosely-typed accessor for the `WebAssembly` global. The package + * tsconfig's `lib: ["es2022"]` does not include the DOM lib where the + * `WebAssembly` namespace types live; the runtime global is available on + * every WASM-capable platform this engine targets. + */ +const WebAssemblyGlobal = (globalThis as any).WebAssembly as { + compile(bytes: Uint8Array): Promise; +}; + +type CompiledExtension = Omit & { + wasm: ExtensionDescriptor['wasm']; +}; + +/** + * Process-wide cache of the compiled `WebAssembly.Module`s for the main + * QuickJS runtime and its native extensions. `WebAssembly.compile` of the + * ~600 KB runtime binary is the most expensive part of VM creation and is + * pure (no per-VM state — instantiation binds the per-VM memory), so it + * only needs to happen once per process. The promise is cached (not the + * result) so concurrent first invocations share a single compilation. + */ +let compiledAssetsPromise: + | Promise<{ + wasm: object; + extensions: CompiledExtension[]; + }> + | undefined; + +function getCompiledAssets() { + if (!compiledAssetsPromise) { + compiledAssetsPromise = (async () => { + const [wasm, ...extensionModules] = await Promise.all([ + WebAssemblyGlobal.compile(quickjsWasm), + ...quickjsExtensions.map((ext) => + WebAssemblyGlobal.compile(ext.wasm as Uint8Array) + ), + ]); + return { + wasm, + extensions: quickjsExtensions.map((ext, i) => ({ + ...ext, + wasm: extensionModules[i] as ExtensionDescriptor['wasm'], + })), + }; + })(); + // On failure, clear the cache so a later invocation can retry rather + // than being stuck with a rejected promise forever. + compiledAssetsPromise.catch(() => { + compiledAssetsPromise = undefined; + }); + } + return compiledAssetsPromise; +} + +async function initWorkflowVM( + getNowMs: () => number, + interruptBudget: InterruptBudget +): Promise { // Deterministic replay clock: Date.now() / new Date() inside the VM // read the host-controlled clock instead of wall time. Replay // re-executes the workflow from the top on every invocation, so the @@ -950,11 +1053,12 @@ async function initWorkflowVM(getNowMs: () => number): Promise { }, }); + const assets = await getCompiledAssets(); const vm = await QuickJS.create({ - wasm: quickjsWasm, + wasm: assets.wasm as never, memoryLimit: 256 * 1024 * 1024, - interruptHandler: createInterruptHandler(), - extensions: quickjsExtensions, + interruptHandler: createInterruptHandler(interruptBudget), + extensions: assets.extensions, wasi, }); @@ -967,9 +1071,43 @@ async function initWorkflowVM(getNowMs: () => number): Promise { return vm; } +/** + * A live QuickJS workflow invocation. When the initial `result` is + * `suspended`, the VM is kept alive so the caller can feed newly recorded + * events (e.g. terminal events of inline-executed steps) into the SAME VM + * via `continueWithEvents` — resuming execution exactly where it left off + * without a fresh-VM re-replay. Terminal results dispose the VM + * automatically; `dispose()` must be called when abandoning a suspended + * session (idempotent). + */ +export interface QuickJSWorkflowSession { + result: QuickJSRuntimeResult; + /** + * Process newly recorded events in the live VM and re-evaluate the + * workflow state. Only valid while the last result was `suspended`. + * Resets the VM's interrupt budget for the new execution burst. + */ + continueWithEvents(newEvents: Event[]): Promise; + /** Dispose the VM if it is still alive. Safe to call multiple times. */ + dispose(): void; +} + +/** + * Run a workflow invocation to its first settled state and dispose the + * VM. Convenience wrapper over {@link startQuickJSWorkflow} for callers + * (and tests) that don't use live-VM continuation. + */ export async function runQuickJSWorkflow( options: QuickJSRuntimeOptions ): Promise { + const session = await startQuickJSWorkflow(options); + session.dispose(); + return session.result; +} + +export async function startQuickJSWorkflow( + options: QuickJSRuntimeOptions +): Promise { const { workflowCode, workflowId, workflowRun, events } = options; const startedAt = workflowRun.startedAt ? +workflowRun.startedAt : Date.now(); @@ -1017,7 +1155,8 @@ export async function runQuickJSWorkflow( }; // ---- Phase 1: static initialization ---- - const vm = await initWorkflowVM(() => vmNowMs); + const interruptBudget: InterruptBudget = { start: Date.now() }; + const vm = await initWorkflowVM(() => vmNowMs, interruptBudget); // Any throw between here and the terminal paths (which dispose the VM // inside checkWorkflowState / extractError before RETURNING) would leak @@ -1036,7 +1175,7 @@ export async function runQuickJSWorkflow( } // ---- Phase 2: per-run initialization ---- - async function runWorkflowInVM(): Promise { + async function runWorkflowInVM(): Promise { // Seeded Math.random { using randomFn = vm.newFunction('random', () => vm.newNumber(rng())); @@ -1087,7 +1226,9 @@ export async function runQuickJSWorkflow( try { vm.evalCode(workflowCode, workflowId || 'workflow.js').dispose(); } catch (err) { - return extractError(vm, err, 'Workflow evaluation failed'); + return makeSettledSession( + extractError(vm, err, 'Workflow evaluation failed') + ); } // Extract workflow arguments. Prefer the run_created event; fall back @@ -1196,7 +1337,9 @@ export async function runQuickJSWorkflow( ); `).dispose(); } catch (err) { - return extractError(vm, err, 'Failed to start workflow'); + return makeSettledSession( + extractError(vm, err, 'Failed to start workflow') + ); } // Process events and drain jobs in a loop. Events may resolve promises @@ -1235,10 +1378,93 @@ export async function runQuickJSWorkflow( } // ---- Check result ---- - return checkWorkflowState(vm); + return makeLiveSession( + vm, + interruptBudget, + advanceClock, + options.encryptionKey + ); } } +/** Session wrapper for a result whose VM is already settled/disposed. */ +function makeSettledSession( + result: QuickJSRuntimeResult +): QuickJSWorkflowSession { + return { + result, + continueWithEvents: () => { + throw new Error( + 'QuickJS workflow session is settled — continueWithEvents is only valid while suspended' + ); + }, + dispose: () => {}, + }; +} + +/** + * Evaluate the VM's state and wrap it in a live session. While suspended, + * the VM stays alive so `continueWithEvents` can resume it in place; + * terminal states dispose the VM immediately (inside checkWorkflowState). + */ +function makeLiveSession( + vm: QuickJS, + interruptBudget: InterruptBudget, + advanceClock: (ms: number) => void, + encryptionKey?: DecryptionKey +): QuickJSWorkflowSession { + const result = checkWorkflowState(vm, { keepAliveOnSuspend: true }); + let alive = !!result.suspended; + + const session: QuickJSWorkflowSession = { + result, + async continueWithEvents( + newEvents: Event[] + ): Promise { + if (!alive) { + throw new Error( + 'QuickJS workflow session is not alive — continueWithEvents is only valid while suspended' + ); + } + // Fresh execution burst — the interrupt budget bounds VM compute, + // not wall time spent waiting on inline steps between bursts. + interruptBudget.start = Date.now(); + + let maxIterations = 100; + let madeProgress: boolean; + do { + madeProgress = await processEvents( + vm, + newEvents, + advanceClock, + encryptionKey + ); + let batch: number; + do { + batch = vm.executePendingJobs(); + if (batch > 0) madeProgress = true; + } while (batch > 0); + } while (madeProgress && --maxIterations > 0); + + const next = checkWorkflowState(vm, { keepAliveOnSuspend: true }); + if (!next.suspended) alive = false; + session.result = next; + return next; + }, + dispose(): void { + if (alive) { + alive = false; + try { + vm.dispose(); + } catch { + // Already disposed — ignore. + } + } + }, + }; + return session; +} + // ---- Event Processing ---- async function processEvents( @@ -1324,6 +1550,31 @@ async function processEvents( b = vm.executePendingJobs(); } while (b > 0); } + } else { + // No resolver yet — buffer the prepared outcome so the promise + // settles the moment the VM constructs it (see __terminalBuffer + // in the bootstrap). Without this, the live-continuation path + // (which scans each delta exactly once) drops the terminal and + // the await never settles. + if (rawOutput instanceof Uint8Array) { + const decryptedOutput = await prepareBytesForVM( + rawOutput, + encryptionKey + ); + const bytesHandle = vm.newUint8Array(decryptedOutput); + vm.setProp(vm.global, '__tmp_buf', bytesHandle); + bytesHandle.dispose(); + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "resolve_bytes", bytes: globalThis.__tmp_buf };` + + `delete globalThis.__tmp_buf;` + ).dispose(); + } else { + const serialized = + rawOutput !== undefined ? JSON.stringify(rawOutput) : 'undefined'; + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "resolve_value", value: ${serialized} };` + ).dispose(); + } } markCreated(vm, cidJs); break; @@ -1386,6 +1637,36 @@ async function processEvents( b = vm.executePendingJobs(); } while (b > 0); } + } else { + // No resolver yet — buffer the prepared rejection (see the + // step_completed branch above for the rationale). + const errorData = eventData?.error; + if (errorData instanceof Uint8Array) { + const decrypted = await prepareBytesForVM(errorData, encryptionKey); + const bytesHandle = vm.newUint8Array(decrypted); + vm.setProp(vm.global, '__tmp_buf', bytesHandle); + bytesHandle.dispose(); + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "reject_bytes", bytes: globalThis.__tmp_buf };` + + `delete globalThis.__tmp_buf;` + ).dispose(); + } else { + const isErrorObject = + typeof errorData === 'object' && errorData !== null; + const msg = isErrorObject + ? (((errorData as Record).message as string) ?? + 'Step failed') + : typeof errorData === 'string' + ? errorData + : 'Step failed'; + const errorStack = + (isErrorObject + ? (errorData as Record).stack + : undefined) ?? (eventData?.stack as string | undefined); + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "reject_error", message: ${JSON.stringify(msg)}, stack: ${errorStack ? JSON.stringify(errorStack) : 'undefined'} };` + ).dispose(); + } } markCreated(vm, cidJs); break; @@ -1406,6 +1687,11 @@ async function processEvents( b = vm.executePendingJobs(); } while (b > 0); } + } else { + // No resolver yet — buffer (see step_completed above). + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "resolve_undefined" };` + ).dispose(); } markCreated(vm, cidJs); break; @@ -1420,6 +1706,12 @@ async function processEvents( const hasResolver = vm.dump( vm.evalCode(`!!globalThis.__resolvers[${cidJs}]`) ); + if (!hasResolver) { + // No resolver yet — buffer (see step_completed above). + vm.evalCode( + `globalThis.__terminalBuffer[${cidJs}] = { kind: "resolve_undefined" };` + ).dispose(); + } if (hasResolver) { vm.evalCode( `globalThis.__resolvers[${cidJs}].resolve();` + @@ -1532,6 +1824,17 @@ async function processEvents( `globalThis.__abortSignals[${cidJs}]._setAborted(undefined);` ).dispose(); } + // The abort is durably recorded — clear the pending op's + // abortRequested marker so the host doesn't re-record it (the + // workflow's own abort() call can set the flag before this + // event is processed when it happens later in replay order, + // and hook_received events are not unique per correlationId). + vm.evalCode( + `(function(){` + + `var p=globalThis.__pending.find(function(q){return q.correlationId===${JSON.stringify(cid)}&&q.type==="hook";});` + + `if(p)p.abortRequested=false;` + + `})()` + ).dispose(); if (event.eventId) { vm.evalCode( `(globalThis.__hookPayloadBuffer.__processedEventIds = globalThis.__hookPayloadBuffer.__processedEventIds || {})[${JSON.stringify(event.eventId)}] = true;` @@ -1816,7 +2119,10 @@ function collectDrainOperations(vm: QuickJS): PendingOperation[] { return vm.dump(h) as PendingOperation[]; } -function checkWorkflowState(vm: QuickJS): QuickJSRuntimeResult { +function checkWorkflowState( + vm: QuickJS, + opts: { keepAliveOnSuspend?: boolean } = {} +): QuickJSRuntimeResult { // Check completed — __workflowResult is a format-prefixed Uint8Array { using h = vm.evalCode('globalThis.__workflowResult'); @@ -1885,7 +2191,7 @@ function checkWorkflowState(vm: QuickJS): QuickJSRuntimeResult { `globalThis.__pending.filter(function(p){return!!globalThis.__resolvers[p.correlationId] || !p.hasCreatedEvent || p.abortRequested;})` ); const pendingOps = vm.dump(pendingH) as PendingOperation[]; - vm.dispose(); + if (!opts.keepAliveOnSuspend) vm.dispose(); return { suspended: { @@ -1928,14 +2234,26 @@ function extractError( }; } -function createInterruptHandler(): () => boolean { - const start = Date.now(); - // Same configurable budget as the node engine's ReplayBudget - // (REPLAY_TIMEOUT_MS, default 240s): a workflow whose replay the node - // engine handles fine must not be interrupted here by a lower - // hardcoded ceiling. The interrupt error escapes runQuickJSWorkflow - // and reaches the replay loop's catch in runtime.ts, which records - // run_failed. +/** + * Mutable interrupt budget for a VM. QuickJS polls the interrupt handler + * during JS execution; when it returns true, execution aborts. The budget + * bounds a single host->VM execution burst (bundle eval + event + * processing), not total VM lifetime — the inline-step loop keeps a VM + * alive across step executions that can legitimately take minutes, so the + * host resets the budget before each re-entry (see resetBudget calls). + * + * The per-burst ceiling is the same configurable budget as the node + * engine's ReplayBudget (REPLAY_TIMEOUT_MS, default 240s): a workflow + * whose replay the node engine handles fine must not be interrupted here + * by a lower hardcoded ceiling. The interrupt error escapes + * runQuickJSWorkflow and reaches the replay loop's catch in runtime.ts, + * which records run_failed. + */ +interface InterruptBudget { + start: number; +} + +function createInterruptHandler(budget: InterruptBudget): () => boolean { const timeout = getReplayTimeoutMs(); - return () => Date.now() - start > timeout; + return () => Date.now() - budget.start > timeout; } diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index 9ca2860001..f830197d8a 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -127,6 +127,11 @@ export const QuickJSPendingOpsCount = SemanticConvention( 'workflow.vm.pending_ops_count' ); +/** Number of steps executed inline (live-VM continuation) this invocation */ +export const QuickJSInlineSteps = SemanticConvention( + 'quickjs.inline_steps' +); + /** Active trace-correlation mode for this invocation (linked or continuous) */ export const WorkflowTraceMode = SemanticConvention<'linked' | 'continuous'>( 'workflow.trace.mode'