From ad89c0275d583d08cec75778847e8cfe5954d16b Mon Sep 17 00:00:00 2001 From: Nathan Rajlich Date: Wed, 22 Jul 2026 08:32:41 -0700 Subject: [PATCH 1/2] Add threshold-based VM-memory snapshotting to the QuickJS engine (WORKFLOW_SNAPSHOT_THRESHOLD) --- .changeset/quickjs-threshold-snapshots.md | 6 + .github/workflows/tests.yml | 3 + .../docs/v5/configuration/runtime-tuning.mdx | 12 + .../core/src/runtime/quickjs-entrypoint.ts | 217 ++++++++++++++++-- .../core/src/runtime/quickjs-runtime.test.ts | 164 +++++++++++++ packages/core/src/runtime/quickjs-runtime.ts | 163 +++++++++++-- packages/core/src/runtime/start.ts | 17 +- packages/core/src/runtime/vm-mode.test.ts | 74 +++++- packages/core/src/runtime/vm-mode.ts | 57 +++++ scripts/create-test-matrix.mjs | 18 ++ 10 files changed, 688 insertions(+), 43 deletions(-) create mode 100644 .changeset/quickjs-threshold-snapshots.md diff --git a/.changeset/quickjs-threshold-snapshots.md b/.changeset/quickjs-threshold-snapshots.md new file mode 100644 index 0000000000..29c6edfa3f --- /dev/null +++ b/.changeset/quickjs-threshold-snapshots.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': minor +'workflow': minor +--- + +Add threshold-based VM-memory snapshotting to the QuickJS engine via `WORKFLOW_SNAPSHOT_THRESHOLD` (or per-run `executionContext.snapshotThreshold`). Once the configured number of events has been processed since the last snapshot, suspensions persist a compressed (and encrypted, when configured) VM snapshot through `world.snapshots`; resumptions restore the VM and replay only the delta events, with automatic fallback to full replay on any load/restore failure. `0` (default) disables snapshotting. diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index c191c30a74..84bb27400e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -701,6 +701,7 @@ jobs: WORKFLOW_DEV_HMR_LOGS: "1" NEXT_CANARY: ${{ matrix.app.canary && '1' || '' }} WORKFLOW_VM: ${{ matrix.app.vm || '' }} + WORKFLOW_SNAPSHOT_THRESHOLD: ${{ matrix.app.snapshotThreshold || '' }} - name: Generate E2E summary if: always() @@ -790,6 +791,7 @@ jobs: DEPLOYMENT_URL: "http://localhost:${{ matrix.app.name == 'sveltekit' && '4173' || (matrix.app.name == 'astro' && '4321' || '3000') }}" NEXT_CANARY: ${{ matrix.app.canary && '1' || '' }} WORKFLOW_VM: ${{ matrix.app.vm || '' }} + WORKFLOW_SNAPSHOT_THRESHOLD: ${{ matrix.app.snapshotThreshold || '' }} - name: Generate E2E summary if: always() @@ -899,6 +901,7 @@ jobs: DEPLOYMENT_URL: "http://localhost:${{ matrix.app.name == 'sveltekit' && '4173' || (matrix.app.name == 'astro' && '4321' || '3000') }}" NEXT_CANARY: ${{ matrix.app.canary && '1' || '' }} WORKFLOW_VM: ${{ matrix.app.vm || '' }} + WORKFLOW_SNAPSHOT_THRESHOLD: ${{ matrix.app.snapshotThreshold || '' }} - name: Generate E2E summary if: always() diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index f4c94a3327..85672dfc86 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -126,6 +126,18 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - The engine choice is stamped into the run's `executionContext` when the run starts, so a run keeps executing on the engine it started on even if the deployment's `WORKFLOW_VM` changes. Runs without a stamped engine use the handler's `WORKFLOW_VM` value. - Unknown values throw at startup. +### `WORKFLOW_SNAPSHOT_THRESHOLD` + +- Default: `0` (disabled) +- Values: non-negative integer +- Only used by the QuickJS engine (`WORKFLOW_VM=quickjs`). +- When set above `0`, the runtime persists a **VM-memory snapshot** at a suspension once at least this many events have been processed since the last snapshot. Subsequent invocations restore the VM from the snapshot and replay only the events recorded since — instead of re-executing the workflow from the top against the full event log. +- Short-lived runs below the threshold never pay the snapshot cost; long-running or unbounded runs stop scaling their resume cost with total event-log length. `1` snapshots at every qualifying suspension. +- Snapshots are an optimization, not a source of truth: the event log remains authoritative, and a missing, corrupt, or incompatible snapshot automatically falls back to a full replay. +- Snapshot bytes are compressed (zstd, gzip fallback) and encrypted with the run's encryption key (when configured) before being handed to the World's `snapshots` storage. They are deleted when the run reaches a terminal state. +- Like `WORKFLOW_VM`, the policy is stamped into the run's `executionContext` at start, so a run keeps the snapshot policy it started with. +- Invalid values throw at startup. + ## Compression and tracing ### `WORKFLOW_DISABLE_COMPRESSION` diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index 480b927f76..94108c6a59 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -33,7 +33,9 @@ import { import { decodeTime } from 'ulid'; import { classifyRunError } from '../classify-error.js'; import { runtimeLogger } from '../logger.js'; +import { compress, decompress } from '../serialization/compression.js'; import { + decrypt as decryptSerializedData, deriveRunPayloadKeys, encrypt as encryptSerializedData, type RunPayloadKeys, @@ -61,6 +63,7 @@ import { import { ReplayBudget } from './replay-budget.js'; import { executeStep, type StepExecutionResult } from './step-executor.js'; import { runStepSingleFlight } from './step-single-flight.js'; +import { getSnapshotThreshold } from './vm-mode.js'; import { getWaitContinuationDispatch } from './wait-continuation.js'; import { getWorld } from './world.js'; @@ -598,17 +601,66 @@ export async function runWorkflowWithQuickJS(params: { const rawKey = await world.getEncryptionKeyForRun?.(workflowRun); const encryptionKey = rawKey ? await deriveRunPayloadKeys(rawKey) : undefined; - // Load the FULL event log for the run. On first invocation the - // preloaded events from the run_started response are the complete log - // and save the events.list round-trips. + // VM-memory snapshotting policy for this run. 0 = disabled (pure + // replay). When enabled, suspensions persist a snapshot once at least + // `snapshotThreshold` events have been processed since the last one, + // and resumptions restore the VM and replay only the delta events. + const snapshotThreshold = getSnapshotThreshold(workflowRun); + + // Try to load a persisted snapshot. Skipped on the first invocation + // (nothing can have been saved yet) and on any load/decode failure — + // the fresh-boot full replay below is always a correct fallback (the + // event log remains the source of truth; snapshots are an optimization). + let existingSnapshot: { + data: Uint8Array; + metadata: import('@workflow/world').SnapshotMetadata; + } | null = null; + if (snapshotThreshold > 0 && !isFirstInvocation(preloadedEvents)) { + try { + const loaded = await world.snapshots.load(runId); + if (loaded) { + // Inverse of the save pipeline: decrypt → decompress. + const decrypted = await decryptSerializedData( + loaded.data, + encryptionKey + ); + const decompressed = await decompress(decrypted); + if (decompressed instanceof Uint8Array) { + existingSnapshot = { data: decompressed, metadata: loaded.metadata }; + } + } + } catch (err) { + runtimeLogger.warn( + 'QuickJS runtime: snapshot load failed, falling back to full replay', + { workflowRunId: runId, message: (err as Error)?.message } + ); + } + } + wfdiag('snapshot_load', { + threshold: snapshotThreshold, + restored: !!existingSnapshot, + eventsCursor: existingSnapshot?.metadata.eventsCursor ?? null, + }); + + // Load the event log. With a restored snapshot only the delta after + // its cursor is needed. Otherwise load the FULL log — on first + // invocation the preloaded events from the run_started response are the + // complete log and save the events.list round-trips (only usable when + // snapshotting is off: snapshot metadata needs a cursor, which + // preloaded events don't carry). let events: Event[]; let eventsFetchedPages = 0; - const usePreloaded = isFirstInvocation(preloadedEvents); + // Cursor after the last event the VM has processed — persisted as the + // snapshot's eventsCursor so restores fetch only the delta. + let lastEventsCursor: string | null = + existingSnapshot?.metadata.eventsCursor ?? null; + const usePreloaded = + snapshotThreshold === 0 && isFirstInvocation(preloadedEvents); if (usePreloaded && preloadedEvents) { events = preloadedEvents; } else { const allEvents: Event[] = []; - let cursor: string | null = null; + let cursor: string | null = lastEventsCursor; let hasMore = true; while (hasMore) { @@ -632,6 +684,7 @@ export async function runWorkflowWithQuickJS(params: { } events = allEvents; + if (cursor) lastEventsCursor = cursor; } // --- Resilient resume: materialize missing hook_received --- @@ -802,19 +855,65 @@ export async function runWorkflowWithQuickJS(params: { eventCount: events.length, }); - 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 - // by `remapErrorStack` on workflow failures below. - workflowCode: workflowCodeForVM, - workflowId, - workflowRun, - events, - encryptionKey, - port, - runInput, - }); + let session: Awaited>; + try { + 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 + // by `remapErrorStack` on workflow failures below. + workflowCode: workflowCodeForVM, + workflowId, + workflowRun, + events, + existingSnapshot, + encryptionKey, + port, + runInput, + }); + } catch (err) { + if (!existingSnapshot) throw err; + // Snapshot restore failed (corrupt bytes, incompatible quickjs-wasi + // build across a redeploy without version-skew protection, ...). + // Fall back to a fresh boot + full event replay — always correct, + // since the event log is the source of truth. + runtimeLogger.warn( + 'QuickJS runtime: snapshot restore failed, falling back to full replay', + { workflowRunId: runId, message: (err as Error)?.message } + ); + wfdiag('snapshot_restore_failed', { message: (err as Error)?.message }); + existingSnapshot = null; + lastEventsCursor = null; + // Refetch the FULL log (the earlier fetch started at the snapshot's + // cursor). + const allEvents: 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, + }, + }); + allEvents.push(...response.data); + if (response.cursor) cursor = response.cursor; + hasMore = response.data.length > 0 && response.cursor != null; + } + events = allEvents; + if (cursor) lastEventsCursor = cursor; + session = await startQuickJSWorkflow({ + workflowCode: workflowCodeForVM, + workflowId, + workflowRun, + events, + encryptionKey, + port, + runInput, + }); + } let result = session.result; runtimeLogger.debug('QuickJS runtime: VM returned', { @@ -877,6 +976,10 @@ export async function runWorkflowWithQuickJS(params: { for (const e of events) { if (e.eventId) seenEventIds.add(e.eventId); } + // Events processed since the restored snapshot (or since run start when + // booting fresh) — compared against snapshotThreshold at suspension + // exit to decide whether to persist a new snapshot. + let eventsProcessedSinceSnapshot = events.length; // Step cids already executed inline by this invocation. const executedStepIds = new Set(); // Steps for which THIS invocation already sent a queue message. @@ -907,6 +1010,9 @@ export async function runWorkflowWithQuickJS(params: { // exiting awaiting_external with the unblocking event already written // and nothing scheduled to read it. let pendingRequeueSignal = false; + // Snapshot bytes captured at suspension exit (threshold met), persisted + // after the VM is disposed. + let capturedSnapshot: Uint8Array | undefined; /** Fetch all events not yet processed by the live VM (log order). */ const fetchUnseenEvents = async (): Promise => { @@ -927,7 +1033,12 @@ export async function runWorkflowWithQuickJS(params: { if (e.eventId) seenEventIds.add(e.eventId); unseen.push(e); } - if (response.cursor) cursor = response.cursor; + if (response.cursor) { + cursor = response.cursor; + // Every listed event is either already processed or about to be + // fed, so the page cursor always tracks the VM's frontier. + lastEventsCursor = response.cursor; + } hasMore = response.data.length > 0 && response.cursor != null; } return unseen; @@ -1036,6 +1147,7 @@ export async function runWorkflowWithQuickJS(params: { // attr_set / getConflict hook_created has been (or is being) // consumed by the live VM, so no external requeue is needed. pendingRequeueSignal = false; + eventsProcessedSinceSnapshot += newEvents.length; result = await session.continueWithEvents(newEvents); wfdiag('inline_iteration', { iteration, @@ -1213,6 +1325,7 @@ export async function runWorkflowWithQuickJS(params: { // Feed the inline batch's terminal events into the live VM. const newEvents = await fetchUnseenEvents(); if (newEvents.length === 0) break; + eventsProcessedSinceSnapshot += newEvents.length; result = await session.continueWithEvents(newEvents); wfdiag('inline_iteration', { @@ -1227,14 +1340,78 @@ export async function runWorkflowWithQuickJS(params: { budgetExhausted: budget.isExhausted(), }); } + // Capture the VM memory for persistence while the session is still + // alive. The (compress → encrypt → save) pipeline runs after the VM + // is disposed — only the byte capture needs the live session. + if ( + snapshotThreshold > 0 && + result.suspended && + !runGone && + eventsProcessedSinceSnapshot >= snapshotThreshold + ) { + try { + capturedSnapshot = session.snapshot(); + } catch (err) { + runtimeLogger.warn('QuickJS runtime: snapshot capture failed', { + workflowRunId: runId, + message: (err as Error)?.message, + }); + } + } } finally { session.dispose(); } + if (capturedSnapshot) { + // Persist: compress (QuickJS heaps compress ~4x) → encrypt → save. + // Compression goes BEFORE encryption because ciphertext is ~random + // and doesn't compress. Failures are non-fatal — the run still makes + // progress via full replay; the next qualifying suspension retries. + try { + const t0 = tick(); + const compressed = await compress(capturedSnapshot, true); + const toStore = (await encryptSerializedData( + compressed as Uint8Array, + encryptionKey + )) as Uint8Array; + await world.snapshots.save(runId, toStore, { + eventsCursor: lastEventsCursor, + createdAt: new Date(), + }); + wfdiag('snapshot_saved', { + plaintextBytes: capturedSnapshot.byteLength, + storedBytes: toStore.byteLength, + eventsCursor: lastEventsCursor, + eventsProcessedSinceSnapshot, + durationMs: Math.round(tick() - t0), + }); + } catch (err) { + runtimeLogger.warn('QuickJS runtime: snapshot save failed', { + workflowRunId: runId, + message: (err as Error)?.message, + }); + } + } + parentSpan?.setAttributes({ ...Attribute.QuickJSInlineSteps(inlineStepsExecuted), }); + // The run reached a terminal state — its snapshot (if any) is dead + // weight; delete best-effort. Guarded on the policy so replay-only + // deployments never issue delete round-trips. + const deleteSnapshotIfAny = async (): Promise => { + if (snapshotThreshold <= 0) return; + try { + await world.snapshots.delete(runId); + } catch (err) { + runtimeLogger.debug('QuickJS runtime: snapshot delete failed', { + workflowRunId: runId, + message: (err as Error)?.message, + }); + } + }; + if (result.completed) { // Workflow completed runtimeLogger.info('QuickJS runtime: workflow completed', { @@ -1243,6 +1420,7 @@ export async function runWorkflowWithQuickJS(params: { parentSpan?.setAttributes({ ...Attribute.QuickJSOutcome('completed'), }); + await deleteSnapshotIfAny(); // Flush leftover pending side effects (abort recordings, system-hook // disposals, fire-and-forget attribute/hook events) BEFORE writing @@ -1437,6 +1615,7 @@ export async function runWorkflowWithQuickJS(params: { parentSpan?.setAttributes({ ...Attribute.QuickJSOutcome('failed'), }); + await deleteSnapshotIfAny(); // Flush leftover pending side effects before writing run_failed — // same drain semantics as the completed branch. diff --git a/packages/core/src/runtime/quickjs-runtime.test.ts b/packages/core/src/runtime/quickjs-runtime.test.ts index e88179f858..45173926ab 100644 --- a/packages/core/src/runtime/quickjs-runtime.test.ts +++ b/packages/core/src/runtime/quickjs-runtime.test.ts @@ -980,3 +980,167 @@ describe('global surface parity', () => { expect(value.plainCompare).toBeLessThan(0); }); }); + +describe('VM snapshot/restore', () => { + const twoStepCode = ` + var add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step//test//add"); + async function workflow() { + var a = await add(10, 7); + var b = await add(a, 8); + return b; + } + workflow.workflowId = "workflow//test//workflow"; + globalThis.__private_workflows.set("workflow//test//workflow", workflow); + `; + + async function suspendFresh(run: ReturnType) { + const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); + const session = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [], + }); + expect(session.result.suspended).toBeDefined(); + return session; + } + + function stepEvents( + run: { runId: string }, + cid: string, + result: number, + idPrefix: string + ) { + return [ + { + eventId: `${idPrefix}_created`, + runId: run.runId, + eventType: 'step_created' as const, + correlationId: cid, + eventData: { stepName: 'step//test//add' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + }, + { + eventId: `${idPrefix}_completed`, + runId: run.runId, + eventType: 'step_completed' as const, + correlationId: cid, + eventData: { result }, + createdAt: new Date('2025-01-01T00:00:02Z'), + }, + ]; + } + + it('restores a snapshot and resumes at the suspension point', async () => { + const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); + const run = makeRun(); + const s1 = await suspendFresh(run); + const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; + const snapshotBytes = s1.snapshot(); + s1.dispose(); + expect(snapshotBytes).toBeInstanceOf(Uint8Array); + expect(snapshotBytes.byteLength).toBeGreaterThan(1000); + + // Restore in a "fresh process": only the delta events are supplied. + const s2 = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: stepEvents(run, step1Cid, 17, 'evnt_s1'), + existingSnapshot: { + data: snapshotBytes, + metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, + }, + }); + expect(s2.result.suspended).toBeDefined(); + const step2Cid = s2.result.suspended!.pendingOperations[0].correlationId; + expect(step2Cid).toMatch(/^step_[0-9A-Z]{26}$/); + expect(step2Cid).not.toBe(step1Cid); + + // Feed step 2's completion into the restored live VM. + const final = await s2.continueWithEvents( + stepEvents(run, step2Cid, 25, 'evnt_s2') + ); + expect(unwrapResult(final.completed!.result)).toBe(25); + s2.dispose(); + }); + + it('produces identical post-restore correlationIds for concurrent resumes from the same snapshot', async () => { + const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); + const run = makeRun(); + const s1 = await suspendFresh(run); + const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; + const snapshotBytes = s1.snapshot(); + s1.dispose(); + + const delta = stepEvents(run, step1Cid, 17, 'evnt_s1'); + const meta = { eventsCursor: 'cursor_1', createdAt: new Date() }; + const [ra, rb] = await Promise.all([ + startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: delta, + existingSnapshot: { data: snapshotBytes, metadata: meta }, + }), + startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: delta, + existingSnapshot: { data: snapshotBytes, metadata: meta }, + }), + ]); + expect(ra.result.suspended!.pendingOperations[0].correlationId).toBe( + rb.result.suspended!.pendingOperations[0].correlationId + ); + ra.dispose(); + rb.dispose(); + }); + + it('supports restore from an OLDER snapshot with multi-suspension partial replay (threshold model)', async () => { + const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); + const run = makeRun(); + + // Take a snapshot at suspension 1 only. + const s1 = await suspendFresh(run); + const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; + const snapshotBytes = s1.snapshot(); + s1.dispose(); + + // Simulate a later invocation that resumed from that snapshot WITHOUT + // saving a newer one: it consumed step 1's results and recorded step 2. + const mid = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: stepEvents(run, step1Cid, 17, 'evnt_s1'), + existingSnapshot: { + data: snapshotBytes, + metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, + }, + }); + const step2Cid = mid.result.suspended!.pendingOperations[0].correlationId; + mid.dispose(); + + // A fresh resume still restores the OLD snapshot but the delta now + // spans TWO suspensions (step 1 and step 2 events). The restored heap + // must regenerate step 2's id identically to consume its events. + const late = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + ...stepEvents(run, step1Cid, 17, 'evnt_s1'), + ...stepEvents(run, step2Cid, 25, 'evnt_s2'), + ], + existingSnapshot: { + data: snapshotBytes, + metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, + }, + }); + expect(late.result.completed).toBeDefined(); + expect(unwrapResult(late.result.completed!.result)).toBe(25); + late.dispose(); + }); +}); diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index 9371aff338..6f1a2a910c 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -27,7 +27,12 @@ * `node:vm` engine's replay determinism. */ -import type { Event, RunInput, WorkflowRun } from '@workflow/world'; +import type { + Event, + RunInput, + SnapshotMetadata, + WorkflowRun, +} from '@workflow/world'; import * as nanoid from 'nanoid'; import { type ExtensionDescriptor, @@ -190,10 +195,26 @@ export interface QuickJSRuntimeOptions { /** The workflow run entity */ workflowRun: WorkflowRun; /** - * The full event log for the run. Every invocation replays the complete - * log from the start (same replay semantics as the `node:vm` engine). + * The event log to process. Without a snapshot this is the FULL log and + * every invocation replays it from the start (same replay semantics as + * the `node:vm` engine). With `existingSnapshot`, this is the delta of + * events recorded at/after the snapshot's `eventsCursor` — feeding + * already-consumed events is harmless (consumed resolvers are gone and + * hook deliveries are deduped in the VM heap), so an imprecise cursor + * only costs redundant scanning. */ events: Event[]; + /** + * A previously persisted VM-memory snapshot to restore from, or + * null/undefined for a fresh boot + full replay. Restoring skips VM + * bootstrap, bundle evaluation, and pre-snapshot re-execution entirely: + * the WASM heap resumes at the exact suspension point it was captured + * at, and only `events` are processed on top. + */ + existingSnapshot?: { + data: Uint8Array; + metadata: SnapshotMetadata; + } | null; /** Encryption key for decrypting event payloads (undefined if unencrypted) */ encryptionKey?: DecryptionKey; /** @@ -966,30 +987,34 @@ function getCompiledAssets() { 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 - // clock must be derived from the event log (not real time) for the - // workflow to observe stable timestamps across invocations. - const wasi: WasiOptions = (memory) => ({ +/** + * Deterministic replay clock: Date.now() / new Date() inside the VM read + * the host-controlled clock instead of wall time. Replay re-executes the + * workflow (or resumes a restored heap) against the event log, so the + * clock must be derived from the log — not real time — for the workflow + * to observe stable timestamps across invocations. + */ +function buildWasiClock(getNowMs: () => number): WasiOptions { + return (memory) => ({ clock_time_get(_clockId: number, _precision: bigint, resultPtr: number) { const timeNs = BigInt(Math.round(getNowMs())) * 1_000_000n; new DataView(memory.buffer).setBigUint64(resultPtr, timeNs, true); return 0; }, }); +} +async function initWorkflowVM( + getNowMs: () => number, + interruptBudget: InterruptBudget +): Promise { const assets = await getCompiledAssets(); const vm = await QuickJS.create({ wasm: assets.wasm as never, memoryLimit: 256 * 1024 * 1024, interruptHandler: createInterruptHandler(interruptBudget), extensions: assets.extensions, - wasi, + wasi: buildWasiClock(getNowMs), }); // Evaluate the VM serde bundle @@ -1001,6 +1026,30 @@ async function initWorkflowVM( return vm; } +/** + * Restore a VM from persisted snapshot bytes. The restored WASM heap + * resumes at the exact suspension point it was captured at — the serde + * bundle, workflow bundle, and all workflow state are already inside it, + * so no bootstrap or bundle evaluation happens here. Host callbacks are + * name-registered by the caller (they live host-side and do not survive + * serialization). + */ +async function restoreWorkflowVM( + data: Uint8Array, + getNowMs: () => number, + interruptBudget: InterruptBudget +): Promise { + const assets = await getCompiledAssets(); + const snapshot = QuickJS.deserializeSnapshot(data); + return QuickJS.restore(snapshot, { + wasm: assets.wasm as never, + memoryLimit: 256 * 1024 * 1024, + interruptHandler: createInterruptHandler(interruptBudget), + extensions: assets.extensions, + wasi: buildWasiClock(getNowMs), + }); +} + /** * A live QuickJS workflow invocation. When the initial `result` is * `suspended`, the VM is kept alive so the caller can feed newly recorded @@ -1018,6 +1067,13 @@ export interface QuickJSWorkflowSession { * Resets the VM's interrupt budget for the new execution burst. */ continueWithEvents(newEvents: Event[]): Promise; + /** + * Capture and serialize the live VM's memory. Only valid while the + * last result was `suspended`. The returned bytes restore via + * `existingSnapshot` on a later invocation (pair them with the events + * cursor at capture time). + */ + snapshot(): Uint8Array; /** Dispose the VM if it is still alive. Safe to call multiple times. */ dispose(): void; } @@ -1059,11 +1115,23 @@ export async function startQuickJSWorkflow( // synthesized run object whose timestamps differ from the durably // stored ones that later invocations load. Matches the node:vm // engine's seed (workflow.ts). - const seed = [ + // + // When restoring from a snapshot, the snapshot's events cursor is + // mixed into the seed: the restored heap already consumed some number + // of PRNG draws, so re-seeding from the base would replay the first-N + // draws and collide with correlationIds recorded before the snapshot. + // The cursor is stable for every resumption from the SAME snapshot + // (concurrent resumes still produce identical ids, preserving the + // world's dedup) but advances when a newer snapshot is taken. + const seedParts = [ workflowRun.runId, workflowRun.workflowName, workflowRun.deploymentId, - ].join(':'); + ]; + if (options.existingSnapshot?.metadata.eventsCursor) { + seedParts.push(options.existingSnapshot.metadata.eventsCursor); + } + const seed = seedParts.join(':'); const rng = seedrandom(seed); // Seeded nanoid generator — uses the same nanoid package and seeded PRNG @@ -1084,8 +1152,55 @@ export async function startQuickJSWorkflow( if (Number.isFinite(ms)) vmNowMs = Math.max(vmNowMs, ms); }; - // ---- Phase 1: static initialization ---- const interruptBudget: InterruptBudget = { start: Date.now() }; + + if (options.existingSnapshot) { + // ---- RESTORE from a persisted VM snapshot ---- + const vm = await restoreWorkflowVM( + options.existingSnapshot.data, + () => vmNowMs, + interruptBudget + ); + + // Re-register host callbacks after restore. Host functions are + // referenced from the WASM heap by name; the host-side registry is + // empty in a fresh process, so each callback must be re-registered + // under the same name used during newFunction() on the fresh-boot + // path. (The in-VM serde functions survive in the heap — no + // re-registration needed for them.) + vm.registerHostCallback('random', () => vm.newNumber(rng())); + vm.registerHostCallback('__generateNanoid', () => + vm.newString(generateNanoid()) + ); + + // Process the delta events and drain jobs. + { + let maxIterations = 100; + let madeProgress: boolean; + do { + madeProgress = await processEvents( + vm, + events, + advanceClock, + options.encryptionKey + ); + let batch: number; + do { + batch = vm.executePendingJobs(); + if (batch > 0) madeProgress = true; + } while (batch > 0); + } while (madeProgress && --maxIterations > 0); + } + + return makeLiveSession( + vm, + interruptBudget, + advanceClock, + options.encryptionKey + ); + } + + // ---- Phase 1: static initialization ---- const vm = await initWorkflowVM(() => vmNowMs, interruptBudget); // Any throw between here and the terminal paths (which dispose the VM @@ -1326,6 +1441,11 @@ function makeSettledSession( 'QuickJS workflow session is settled — continueWithEvents is only valid while suspended' ); }, + snapshot: () => { + throw new Error( + 'QuickJS workflow session is settled — snapshot is only valid while suspended' + ); + }, dispose: () => {}, }; } @@ -1379,6 +1499,15 @@ function makeLiveSession( session.result = next; return next; }, + snapshot(): Uint8Array { + if (!alive) { + throw new Error( + 'QuickJS workflow session is not alive — snapshot is only valid while suspended' + ); + } + const snap = vm.snapshot(); + return QuickJS.serializeSnapshot(snap); + }, dispose(): void { if (alive) { alive = false; diff --git a/packages/core/src/runtime/start.ts b/packages/core/src/runtime/start.ts index e5b43f35a2..27d40e89fe 100644 --- a/packages/core/src/runtime/start.ts +++ b/packages/core/src/runtime/start.ts @@ -35,7 +35,10 @@ import { version as workflowCoreVersion } from '../version.js'; import { getWorldLazy } from './get-world-lazy.js'; import { getWorkflowQueueName, healthCheck } from './helpers.js'; import { Run } from './run.js'; -import { getWorkflowVmFromEnv } from './vm-mode.js'; +import { + getSnapshotThresholdFromEnv, + getWorkflowVmFromEnv, +} from './vm-mode.js'; import { safeWaitUntil, waitedUntil } from './wait-until.js'; import { assertWorldSupportsRuntimeProtocol } from './world-compatibility.js'; @@ -526,18 +529,20 @@ export async function start( // is simply absent. const creatorEnvironment = world.getEnvironment?.(); - // If WORKFLOW_VM is set on the client starting the run, stamp the - // engine choice into the run's executionContext so the run keeps - // executing on the engine it started on (the same deployment can - // serve both VM engines). Unknown values throw — see - // getWorkflowVmFromEnv(). + // If WORKFLOW_VM / WORKFLOW_SNAPSHOT_THRESHOLD are set on the + // client starting the run, stamp them into the run's + // executionContext so the run keeps the engine and snapshot policy + // it started with (the same deployment can serve both VM engines). + // Unknown values throw — see vm-mode.ts. const workflowVm = getWorkflowVmFromEnv(); + const snapshotThreshold = getSnapshotThresholdFromEnv(); const executionContext = { traceCarrier, workflowCoreVersion, features: { encryption: !!encryptionKey }, ...(workflowVm ? { workflowVm } : {}), + ...(snapshotThreshold !== undefined ? { snapshotThreshold } : {}), ...(opts.replayedFromRunId ? { replayedFromRunId: opts.replayedFromRunId } : {}), diff --git a/packages/core/src/runtime/vm-mode.test.ts b/packages/core/src/runtime/vm-mode.test.ts index 1f451e016b..003f3329af 100644 --- a/packages/core/src/runtime/vm-mode.test.ts +++ b/packages/core/src/runtime/vm-mode.test.ts @@ -1,7 +1,13 @@ import { WorkflowRuntimeError } from '@workflow/errors'; import type { WorkflowRun } from '@workflow/world'; import { afterEach, describe, expect, it } from 'vitest'; -import { getWorkflowVmFromEnv, useQuickJSVm, WORKFLOW_VMS } from './vm-mode.js'; +import { + getSnapshotThreshold, + getSnapshotThresholdFromEnv, + getWorkflowVmFromEnv, + useQuickJSVm, + WORKFLOW_VMS, +} from './vm-mode.js'; describe('getWorkflowVmFromEnv', () => { it('returns undefined when WORKFLOW_VM is not set', () => { @@ -107,3 +113,69 @@ describe('useQuickJSVm', () => { expect(() => useQuickJSVm(makeRun())).toThrow(WorkflowRuntimeError); }); }); + +describe('getSnapshotThresholdFromEnv', () => { + it('returns undefined when unset or empty', () => { + expect(getSnapshotThresholdFromEnv({})).toBeUndefined(); + expect( + getSnapshotThresholdFromEnv({ WORKFLOW_SNAPSHOT_THRESHOLD: '' }) + ).toBeUndefined(); + }); + + it('parses non-negative integers', () => { + expect( + getSnapshotThresholdFromEnv({ WORKFLOW_SNAPSHOT_THRESHOLD: '0' }) + ).toBe(0); + expect( + getSnapshotThresholdFromEnv({ WORKFLOW_SNAPSHOT_THRESHOLD: '1' }) + ).toBe(1); + expect( + getSnapshotThresholdFromEnv({ WORKFLOW_SNAPSHOT_THRESHOLD: '250' }) + ).toBe(250); + }); + + it('throws on invalid values', () => { + for (const bad of ['-1', '1.5', 'abc', 'Infinity']) { + expect(() => + getSnapshotThresholdFromEnv({ WORKFLOW_SNAPSHOT_THRESHOLD: bad }) + ).toThrow(WorkflowRuntimeError); + } + }); +}); + +describe('getSnapshotThreshold', () => { + const makeRun = (executionContext?: Record) => + ({ + runId: 'wrun_test', + workflowName: 'test', + executionContext, + }) as unknown as WorkflowRun; + + afterEach(() => { + delete process.env.WORKFLOW_SNAPSHOT_THRESHOLD; + }); + + it('defaults to 0 (disabled)', () => { + expect(getSnapshotThreshold(makeRun())).toBe(0); + }); + + it('reads the env var when the run has no stamped policy', () => { + process.env.WORKFLOW_SNAPSHOT_THRESHOLD = '100'; + expect(getSnapshotThreshold(makeRun())).toBe(100); + }); + + it('executionContext.snapshotThreshold wins over env (run affinity)', () => { + process.env.WORKFLOW_SNAPSHOT_THRESHOLD = '100'; + expect(getSnapshotThreshold(makeRun({ snapshotThreshold: 5 }))).toBe(5); + expect(getSnapshotThreshold(makeRun({ snapshotThreshold: 0 }))).toBe(0); + }); + + it('throws on invalid stamped values', () => { + expect(() => + getSnapshotThreshold(makeRun({ snapshotThreshold: -1 })) + ).toThrow(WorkflowRuntimeError); + expect(() => + getSnapshotThreshold(makeRun({ snapshotThreshold: 'x' })) + ).toThrow(WorkflowRuntimeError); + }); +}); diff --git a/packages/core/src/runtime/vm-mode.ts b/packages/core/src/runtime/vm-mode.ts index e8a7864003..8dcecefa13 100644 --- a/packages/core/src/runtime/vm-mode.ts +++ b/packages/core/src/runtime/vm-mode.ts @@ -73,3 +73,60 @@ export function useQuickJSVm(workflowRun: WorkflowRun): boolean { } return getWorkflowVmFromEnv() === 'quickjs'; } + +/** + * Read and validate the `WORKFLOW_SNAPSHOT_THRESHOLD` env var. + * + * The threshold is the number of processed events after which the QuickJS + * engine persists a VM-memory snapshot at suspension, so subsequent + * invocations restore the VM and replay only the events recorded since — + * instead of re-executing the workflow from the top against the full log. + * + * `0` (or unset) disables snapshotting entirely: short-lived runs never + * pay the snapshot cost, while long/forever runs can opt in. `1` + * effectively snapshots at every suspension. + * + * Returns the configured threshold, or `undefined` if unset/empty. + * Throws {@link WorkflowRuntimeError} for non-integer or negative values. + */ +export function getSnapshotThresholdFromEnv( + env: NodeJS.ProcessEnv = process.env +): number | undefined { + const raw = env.WORKFLOW_SNAPSHOT_THRESHOLD; + if (raw === undefined || raw === '') return undefined; + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed < 0) { + throw new WorkflowRuntimeError( + `Invalid WORKFLOW_SNAPSHOT_THRESHOLD value: "${raw}". ` + + 'Expected a non-negative integer (0 disables snapshotting).' + ); + } + return parsed; +} + +/** + * Resolve the snapshot threshold for a run: the run's stamped + * `executionContext.snapshotThreshold` (set by the SDK at `start()` when + * `WORKFLOW_SNAPSHOT_THRESHOLD` is set on the client) wins so a run keeps + * the policy it started with; otherwise the handler's env var; otherwise + * `0` (disabled). Only consulted by the QuickJS engine. + */ +export function getSnapshotThreshold(workflowRun: WorkflowRun): number { + const fromRun = ( + workflowRun.executionContext as { snapshotThreshold?: unknown } | undefined + )?.snapshotThreshold; + if (fromRun !== undefined) { + if ( + typeof fromRun !== 'number' || + !Number.isInteger(fromRun) || + fromRun < 0 + ) { + throw new WorkflowRuntimeError( + `Invalid executionContext.snapshotThreshold value: "${fromRun}". ` + + 'Expected a non-negative integer.' + ); + } + return fromRun; + } + return getSnapshotThresholdFromEnv() ?? 0; +} diff --git a/scripts/create-test-matrix.mjs b/scripts/create-test-matrix.mjs index 3d056d912b..27f9456aee 100644 --- a/scripts/create-test-matrix.mjs +++ b/scripts/create-test-matrix.mjs @@ -180,4 +180,22 @@ matrix.app = matrix.app.flatMap((app) => })) ); +// QuickJS engine with VM-memory snapshotting at maximum churn +// (WORKFLOW_SNAPSHOT_THRESHOLD=1 snapshots at every qualifying +// suspension) — exercises the save/restore/delete lifecycle and the +// restore + partial-replay determinism on every run. +matrix.app.push( + createMatrixEntry( + 'nextjs-turbopack', + 'example-nextjs-workflow-turbopack', + DEV_TEST_CONFIGS['nextjs-turbopack'], + { + vm: 'quickjs', + snapshotThreshold: '1', + runLabel: 'quickjs-snapshot', + artifactSuffix: 'quickjs-snapshot', + } + ) +); + console.log(JSON.stringify(matrix)); From c139c2a098e9c42b411cf6d1d17c55ac4b67e366 Mon Sep 17 00:00:00 2001 From: Nathan Rajlich Date: Fri, 31 Jul 2026 14:25:20 -0700 Subject: [PATCH 2/2] =?UTF-8?q?Address=20review:=20snapshot=20correctness?= =?UTF-8?q?=20=E2=80=94=20total=20event=20ceiling,=20position-based=20PRNG?= =?UTF-8?q?=20fast-forward,=20unified=20host-callback=20list,=20lifecycle?= =?UTF-8?q?=20hardening?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SnapshotMetadata gains eventCount, rngDraws and formatVersion. The max-events guard now compares restored total + delta (both at entry and per loop turn) — previously a run that kept snapshotting could never accumulate enough delta to trip the ceiling it exists for. - Correlation-id generation is position-based across snapshots: the runtime seeds from the BASE seed and fast-forwards the persisted draw count instead of mixing the snapshot cursor into the seed. Ids are now identical across snapshot generations AND identical to a no-snapshot run, so overlapping invocations straddling a snapshot save still collide on the world's dedup (new test pins restored ids == full-replay ids). Snapshots without a draw count fall back to full replay. - Host callbacks are declared in ONE list that drives both the fresh-boot install and the restore re-registration, so adding a callback can't silently skip the restore path. - Preloaded events are used again with snapshotting enabled (the first qualifying suspension skips its save — no cursor yet); short runs keep the zero-round-trip fast path. - Snapshot persist runs off the response path (waitUntil), with a 32MB plaintext size ceiling (skip + warn). Loads that fail format/shape checks warn instead of silently miming a miss. Terminal deletes are gated on a snapshot actually existing and now also fire on the runGone path; a server-side TTL remains the backstop for unobserved cancellations. --- .../core/src/runtime/quickjs-entrypoint.ts | 191 +++++++++++++----- .../core/src/runtime/quickjs-runtime.test.ts | 91 +++++++-- packages/core/src/runtime/quickjs-runtime.ts | 118 +++++++---- packages/world/src/index.ts | 5 +- packages/world/src/snapshots.ts | 32 +++ 5 files changed, 328 insertions(+), 109 deletions(-) diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index 94108c6a59..a4b8ee7f33 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -27,6 +27,7 @@ import { type HookInput, ROOT_RUN_ID_ATTRIBUTE, type RunInput, + SNAPSHOT_FORMAT_VERSION, SPEC_VERSION_CURRENT, type WorkflowRun, } from '@workflow/world'; @@ -65,6 +66,7 @@ import { executeStep, type StepExecutionResult } from './step-executor.js'; import { runStepSingleFlight } from './step-single-flight.js'; import { getSnapshotThreshold } from './vm-mode.js'; import { getWaitContinuationDispatch } from './wait-continuation.js'; +import { safeWaitUntil } from './wait-until.js'; import { getWorld } from './world.js'; /** Tiny ms timer using performance.now() — already monotonic on Node. */ @@ -149,6 +151,14 @@ async function queueStepMessage(params: { * leftover side effects when the workflow completed or failed, mirroring * the node:vm engine's drainPendingQueueItems). */ +/** + * Plaintext size ceiling for persisted VM snapshots. A heap beyond this + * costs more to store, encrypt and decompress than the replay it saves — + * oversized captures are skipped (with a warn) and the run keeps making + * progress via full replay. + */ +const MAX_SNAPSHOT_PLAINTEXT_BYTES = 32 * 1024 * 1024; + async function dispatchPendingOps(params: { world: Awaited>; runId: string; @@ -619,14 +629,43 @@ export async function runWorkflowWithQuickJS(params: { try { const loaded = await world.snapshots.load(runId); if (loaded) { - // Inverse of the save pipeline: decrypt → decompress. - const decrypted = await decryptSerializedData( - loaded.data, - encryptionKey - ); - const decompressed = await decompress(decrypted); - if (decompressed instanceof Uint8Array) { - existingSnapshot = { data: decompressed, metadata: loaded.metadata }; + const version = loaded.metadata.formatVersion; + if ( + (version !== undefined && version !== SNAPSHOT_FORMAT_VERSION) || + loaded.metadata.rngDraws === undefined + ) { + // Unknown format OR a snapshot without the PRNG draw count — + // restoring the latter would reset id generation to the base + // seed and collide with pre-snapshot correlation ids. + runtimeLogger.warn( + 'QuickJS runtime: snapshot format version mismatch, falling back to full replay', + { + workflowRunId: runId, + snapshotVersion: version, + expectedVersion: SNAPSHOT_FORMAT_VERSION, + } + ); + } else { + // Inverse of the save pipeline: decrypt → decompress. + const decrypted = await decryptSerializedData( + loaded.data, + encryptionKey + ); + const decompressed = await decompress(decrypted); + if (decompressed instanceof Uint8Array) { + existingSnapshot = { + data: decompressed, + metadata: loaded.metadata, + }; + } else { + // A stored snapshot that doesn't decode is a failure worth + // seeing — 100% miss rate must not look like a working + // system. + runtimeLogger.warn( + 'QuickJS runtime: snapshot decoded to an unexpected shape, falling back to full replay', + { workflowRunId: runId, decodedType: typeof decompressed } + ); + } } } } catch (err) { @@ -645,17 +684,19 @@ export async function runWorkflowWithQuickJS(params: { // Load the event log. With a restored snapshot only the delta after // its cursor is needed. Otherwise load the FULL log — on first // invocation the preloaded events from the run_started response are the - // complete log and save the events.list round-trips (only usable when - // snapshotting is off: snapshot metadata needs a cursor, which - // preloaded events don't carry). + // complete log and save the events.list round-trips. Preload is used + // even with snapshotting enabled: it carries no cursor, so the FIRST + // qualifying suspension simply skips its snapshot save (the persist + // path requires a cursor) and the next one — whose feed loop has + // observed a cursor — snapshots normally. Short-lived runs keep the + // zero-overhead fast path either way. let events: Event[]; let eventsFetchedPages = 0; // Cursor after the last event the VM has processed — persisted as the // snapshot's eventsCursor so restores fetch only the delta. let lastEventsCursor: string | null = existingSnapshot?.metadata.eventsCursor ?? null; - const usePreloaded = - snapshotThreshold === 0 && isFirstInvocation(preloadedEvents); + const usePreloaded = isFirstInvocation(preloadedEvents); if (usePreloaded && preloadedEvents) { events = preloadedEvents; } else { @@ -786,10 +827,20 @@ export async function runWorkflowWithQuickJS(params: { } // Event-limit guard: fail a runaway run once its log reaches the - // server-supplied ceiling — same enforcement point as the node:vm - // engine's replay loop. - if (maxEventsLimit !== undefined && events.length >= maxEventsLimit) { - throw new MaxEventsExceededError(events.length, maxEventsLimit); + // server-supplied ceiling. With a restored snapshot `events` is only + // the delta after the snapshot cursor, so the guard compares the TOTAL + // (pre-snapshot count persisted in the metadata + delta) — otherwise a + // run that keeps snapshotting would never accumulate enough delta to + // trip the ceiling it exists to enforce. + const restoredEventCount = existingSnapshot?.metadata.eventCount ?? 0; + if ( + maxEventsLimit !== undefined && + restoredEventCount + events.length >= maxEventsLimit + ) { + throw new MaxEventsExceededError( + restoredEventCount + events.length, + maxEventsLimit + ); } parentSpan?.setAttributes({ @@ -1010,9 +1061,9 @@ export async function runWorkflowWithQuickJS(params: { // exiting awaiting_external with the unblocking event already written // and nothing scheduled to read it. let pendingRequeueSignal = false; - // Snapshot bytes captured at suspension exit (threshold met), persisted + // Snapshot captured at suspension exit (threshold met), persisted // after the VM is disposed. - let capturedSnapshot: Uint8Array | undefined; + let capturedSnapshot: { data: Uint8Array; rngDraws: number } | undefined; /** Fetch all events not yet processed by the live VM (log order). */ const fetchUnseenEvents = async (): Promise => { @@ -1054,8 +1105,14 @@ export async function runWorkflowWithQuickJS(params: { // 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); + if ( + maxEventsLimit !== undefined && + restoredEventCount + seenEventIds.size >= maxEventsLimit + ) { + throw new MaxEventsExceededError( + restoredEventCount + seenEventIds.size, + maxEventsLimit + ); } const pendingOperations = result.suspended.pendingOperations; @@ -1351,6 +1408,20 @@ export async function runWorkflowWithQuickJS(params: { ) { try { capturedSnapshot = session.snapshot(); + if (capturedSnapshot.data.byteLength > MAX_SNAPSHOT_PLAINTEXT_BYTES) { + // A heap this large costs more to store/decompress than the + // replay it saves — skip the save (full replay remains correct) + // and make the skip visible. + runtimeLogger.warn( + 'QuickJS runtime: snapshot exceeds the size ceiling, skipping persist', + { + workflowRunId: runId, + plaintextBytes: capturedSnapshot.data.byteLength, + maxBytes: MAX_SNAPSHOT_PLAINTEXT_BYTES, + } + ); + capturedSnapshot = undefined; + } } catch (err) { runtimeLogger.warn('QuickJS runtime: snapshot capture failed', { workflowRunId: runId, @@ -1362,35 +1433,49 @@ export async function runWorkflowWithQuickJS(params: { session.dispose(); } - if (capturedSnapshot) { + if (capturedSnapshot && lastEventsCursor !== null) { // Persist: compress (QuickJS heaps compress ~4x) → encrypt → save. // Compression goes BEFORE encryption because ciphertext is ~random // and doesn't compress. Failures are non-fatal — the run still makes // progress via full replay; the next qualifying suspension retries. - try { - const t0 = tick(); - const compressed = await compress(capturedSnapshot, true); - const toStore = (await encryptSerializedData( - compressed as Uint8Array, - encryptionKey - )) as Uint8Array; - await world.snapshots.save(runId, toStore, { - eventsCursor: lastEventsCursor, - createdAt: new Date(), - }); - wfdiag('snapshot_saved', { - plaintextBytes: capturedSnapshot.byteLength, - storedBytes: toStore.byteLength, - eventsCursor: lastEventsCursor, - eventsProcessedSinceSnapshot, - durationMs: Math.round(tick() - t0), - }); - } catch (err) { - runtimeLogger.warn('QuickJS runtime: snapshot save failed', { - workflowRunId: runId, - message: (err as Error)?.message, - }); - } + // Moved off the response path via waitUntil: only the byte capture + // needed the live session; the pipeline runs post-response so a + // multi-MB heap doesn't delay the next step's pickup. (Skipped when + // no cursor exists yet — a preloaded first invocation snapshots at + // its next qualifying suspension instead.) + const snapshot = capturedSnapshot; + const totalEventCount = restoredEventCount + seenEventIds.size; + safeWaitUntil( + (async () => { + const t0 = tick(); + const compressed = await compress(snapshot.data, true); + const toStore = (await encryptSerializedData( + compressed as Uint8Array, + encryptionKey + )) as Uint8Array; + await world.snapshots.save(runId, toStore, { + eventsCursor: lastEventsCursor, + createdAt: new Date(), + eventCount: totalEventCount, + rngDraws: snapshot.rngDraws, + formatVersion: SNAPSHOT_FORMAT_VERSION, + }); + wfdiag('snapshot_saved', { + plaintextBytes: snapshot.data.byteLength, + storedBytes: toStore.byteLength, + eventsCursor: lastEventsCursor, + eventsProcessedSinceSnapshot, + rngDraws: snapshot.rngDraws, + durationMs: Math.round(tick() - t0), + }); + })(), + (err) => { + runtimeLogger.warn('QuickJS runtime: snapshot save failed', { + workflowRunId: runId, + message: (err as Error)?.message, + }); + } + ); } parentSpan?.setAttributes({ @@ -1398,10 +1483,16 @@ export async function runWorkflowWithQuickJS(params: { }); // The run reached a terminal state — its snapshot (if any) is dead - // weight; delete best-effort. Guarded on the policy so replay-only - // deployments never issue delete round-trips. + // weight; delete best-effort. Guarded on the policy AND on whether a + // snapshot can actually exist (one was restored, or this invocation + // just persisted one), so the common short-run case never pays the + // delete round-trip. A transient load failure earlier can leave an + // orphan behind; a server-side TTL on snapshot records is the + // self-correcting backstop for those (and for runs cancelled without + // any invocation observing it). const deleteSnapshotIfAny = async (): Promise => { if (snapshotThreshold <= 0) return; + if (!existingSnapshot && !capturedSnapshot) return; try { await world.snapshots.delete(runId); } catch (err) { @@ -1503,7 +1594,9 @@ export async function runWorkflowWithQuickJS(params: { }); if (runGone) { - // The run no longer exists (expired / deleted) — nothing to drive. + // The run no longer exists (expired / cancelled / deleted) — + // nothing to drive, and its snapshot is dead weight. + await deleteSnapshotIfAny(); wfdiag('exit_suspended', { action: 'run_gone' }); return; } diff --git a/packages/core/src/runtime/quickjs-runtime.test.ts b/packages/core/src/runtime/quickjs-runtime.test.ts index 45173926ab..c8a847c0bc 100644 --- a/packages/core/src/runtime/quickjs-runtime.test.ts +++ b/packages/core/src/runtime/quickjs-runtime.test.ts @@ -1036,10 +1036,11 @@ describe('VM snapshot/restore', () => { const run = makeRun(); const s1 = await suspendFresh(run); const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; - const snapshotBytes = s1.snapshot(); + const captured = s1.snapshot(); s1.dispose(); - expect(snapshotBytes).toBeInstanceOf(Uint8Array); - expect(snapshotBytes.byteLength).toBeGreaterThan(1000); + expect(captured.data).toBeInstanceOf(Uint8Array); + expect(captured.data.byteLength).toBeGreaterThan(1000); + expect(captured.rngDraws).toBeGreaterThan(0); // Restore in a "fresh process": only the delta events are supplied. const s2 = await startQuickJSWorkflow({ @@ -1048,8 +1049,12 @@ describe('VM snapshot/restore', () => { workflowRun: run, events: stepEvents(run, step1Cid, 17, 'evnt_s1'), existingSnapshot: { - data: snapshotBytes, - metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, + data: captured.data, + metadata: { + eventsCursor: 'cursor_1', + createdAt: new Date(), + rngDraws: captured.rngDraws, + }, }, }); expect(s2.result.suspended).toBeDefined(); @@ -1070,25 +1075,29 @@ describe('VM snapshot/restore', () => { const run = makeRun(); const s1 = await suspendFresh(run); const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; - const snapshotBytes = s1.snapshot(); + const captured = s1.snapshot(); s1.dispose(); const delta = stepEvents(run, step1Cid, 17, 'evnt_s1'); - const meta = { eventsCursor: 'cursor_1', createdAt: new Date() }; + const meta = { + eventsCursor: 'cursor_1', + createdAt: new Date(), + rngDraws: captured.rngDraws, + }; const [ra, rb] = await Promise.all([ startQuickJSWorkflow({ workflowCode: twoStepCode, workflowId: 'workflow//test//workflow', workflowRun: run, events: delta, - existingSnapshot: { data: snapshotBytes, metadata: meta }, + existingSnapshot: { data: captured.data, metadata: meta }, }), startQuickJSWorkflow({ workflowCode: twoStepCode, workflowId: 'workflow//test//workflow', workflowRun: run, events: delta, - existingSnapshot: { data: snapshotBytes, metadata: meta }, + existingSnapshot: { data: captured.data, metadata: meta }, }), ]); expect(ra.result.suspended!.pendingOperations[0].correlationId).toBe( @@ -1098,6 +1107,53 @@ describe('VM snapshot/restore', () => { rb.dispose(); }); + it('post-restore correlationIds equal the no-snapshot run (position-based fast-forward)', async () => { + // The definitive dedup property: an invocation restored from a + // snapshot and an invocation that full-replayed from scratch must + // generate the SAME id for the same logical step — otherwise a queue + // redelivery straddling a snapshot save double-executes the step. + const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); + const run = makeRun(); + + // No-snapshot reference: full replay past step 1. + const s1 = await suspendFresh(run); + const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; + const captured = s1.snapshot(); + s1.dispose(); + const reference = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + runCreatedEvent(run), + ...stepEvents(run, step1Cid, 17, 'evnt_s1'), + ], + }); + const referenceStep2Cid = + reference.result.suspended!.pendingOperations[0].correlationId; + reference.dispose(); + + // Snapshot-restored invocation reaching the same logical step. + const restored = await startQuickJSWorkflow({ + workflowCode: twoStepCode, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: stepEvents(run, step1Cid, 17, 'evnt_s1'), + existingSnapshot: { + data: captured.data, + metadata: { + eventsCursor: 'cursor_1', + createdAt: new Date(), + rngDraws: captured.rngDraws, + }, + }, + }); + expect(restored.result.suspended!.pendingOperations[0].correlationId).toBe( + referenceStep2Cid + ); + restored.dispose(); + }); + it('supports restore from an OLDER snapshot with multi-suspension partial replay (threshold model)', async () => { const { startQuickJSWorkflow } = await import('./quickjs-runtime.js'); const run = makeRun(); @@ -1105,8 +1161,13 @@ describe('VM snapshot/restore', () => { // Take a snapshot at suspension 1 only. const s1 = await suspendFresh(run); const step1Cid = s1.result.suspended!.pendingOperations[0].correlationId; - const snapshotBytes = s1.snapshot(); + const captured = s1.snapshot(); s1.dispose(); + const meta = { + eventsCursor: 'cursor_1', + createdAt: new Date(), + rngDraws: captured.rngDraws, + }; // Simulate a later invocation that resumed from that snapshot WITHOUT // saving a newer one: it consumed step 1's results and recorded step 2. @@ -1115,10 +1176,7 @@ describe('VM snapshot/restore', () => { workflowId: 'workflow//test//workflow', workflowRun: run, events: stepEvents(run, step1Cid, 17, 'evnt_s1'), - existingSnapshot: { - data: snapshotBytes, - metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, - }, + existingSnapshot: { data: captured.data, metadata: meta }, }); const step2Cid = mid.result.suspended!.pendingOperations[0].correlationId; mid.dispose(); @@ -1134,10 +1192,7 @@ describe('VM snapshot/restore', () => { ...stepEvents(run, step1Cid, 17, 'evnt_s1'), ...stepEvents(run, step2Cid, 25, 'evnt_s2'), ], - existingSnapshot: { - data: snapshotBytes, - metadata: { eventsCursor: 'cursor_1', createdAt: new Date() }, - }, + existingSnapshot: { data: captured.data, metadata: meta }, }); expect(late.result.completed).toBeDefined(); expect(unwrapResult(late.result.completed!.result)).toBe(25); diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index 6f1a2a910c..32f9b125fb 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -1071,9 +1071,11 @@ export interface QuickJSWorkflowSession { * Capture and serialize the live VM's memory. Only valid while the * last result was `suspended`. The returned bytes restore via * `existingSnapshot` on a later invocation (pair them with the events - * cursor at capture time). + * cursor at capture time). `rngDraws` is the seeded PRNG's draw count + * at capture — persisted in the snapshot metadata so a restore + * fast-forwards the base seed to the same position. */ - snapshot(): Uint8Array; + snapshot(): { data: Uint8Array; rngDraws: number }; /** Dispose the VM if it is still alive. Safe to call multiple times. */ dispose(): void; } @@ -1116,23 +1118,32 @@ export async function startQuickJSWorkflow( // stored ones that later invocations load. Matches the node:vm // engine's seed (workflow.ts). // - // When restoring from a snapshot, the snapshot's events cursor is - // mixed into the seed: the restored heap already consumed some number - // of PRNG draws, so re-seeding from the base would replay the first-N - // draws and collide with correlationIds recorded before the snapshot. - // The cursor is stable for every resumption from the SAME snapshot - // (concurrent resumes still produce identical ids, preserving the - // world's dedup) but advances when a newer snapshot is taken. - const seedParts = [ + // When restoring from a snapshot, the restored heap already consumed + // some number of PRNG draws. The seed stays the BASE seed and the + // runtime fast-forwards the recorded draw count (metadata.rngDraws) + // instead of mixing the snapshot cursor into the seed: a cursor-mixed + // seed made ids depend on WHICH snapshot generation an invocation + // restored from, so two overlapping invocations straddling a snapshot + // save (queue redelivery of an in-flight invocation) generated + // DIFFERENT ids for the same logical step and both step_created writes + // landed — the exact double-execution the seeding exists to prevent. + // Position-based fast-forward keeps ids identical across snapshot + // generations and identical to a no-snapshot run. + const seed = [ workflowRun.runId, workflowRun.workflowName, workflowRun.deploymentId, - ]; - if (options.existingSnapshot?.metadata.eventsCursor) { - seedParts.push(options.existingSnapshot.metadata.eventsCursor); - } - const seed = seedParts.join(':'); - const rng = seedrandom(seed); + ].join(':'); + const baseRng = seedrandom(seed); + const restoredDraws = options.existingSnapshot?.metadata.rngDraws ?? 0; + for (let i = 0; i < restoredDraws; i++) baseRng(); + // Every consumer (Math.random, nanoid, the VM ULID factory via + // Math.random) draws through this counter so the total is exact. + let rngDraws = restoredDraws; + const rng = () => { + rngDraws++; + return baseRng(); + }; // Seeded nanoid generator — uses the same nanoid package and seeded PRNG // as the node:vm engine for consistent token generation. @@ -1154,6 +1165,39 @@ export async function startQuickJSWorkflow( const interruptBudget: InterruptBudget = { start: Date.now() }; + // ---- Host callbacks ---- + // ONE list drives both the fresh-boot path (newFunction + install) and + // the snapshot-restore path (registerHostCallback): host functions are + // referenced from the WASM heap by name and the host-side registry is + // empty in a fresh process, so a callback added to the boot path but + // not re-registered on restore resolves to nothing after a restore. + // Add new host callbacks HERE, never inline at either site. + const hostCallbacks: { + name: string; + fn: (vm: QuickJS) => () => ReturnType; + /** How the fresh-boot path exposes the function to guest code. */ + install: ( + vm: QuickJS, + fnHandle: ReturnType + ) => void; + }[] = [ + { + name: 'random', + fn: (vm) => () => vm.newNumber(rng()), + install: (vm, fnHandle) => { + using math = vm.global.getProp('Math'); + math.setProp('random', fnHandle); + }, + }, + { + name: '__generateNanoid', + fn: (vm) => () => vm.newString(generateNanoid()) as never, + install: (vm, fnHandle) => { + vm.setProp(vm.global, '__generateNanoid', fnHandle); + }, + }, + ]; + if (options.existingSnapshot) { // ---- RESTORE from a persisted VM snapshot ---- const vm = await restoreWorkflowVM( @@ -1162,16 +1206,12 @@ export async function startQuickJSWorkflow( interruptBudget ); - // Re-register host callbacks after restore. Host functions are - // referenced from the WASM heap by name; the host-side registry is - // empty in a fresh process, so each callback must be re-registered - // under the same name used during newFunction() on the fresh-boot - // path. (The in-VM serde functions survive in the heap — no - // re-registration needed for them.) - vm.registerHostCallback('random', () => vm.newNumber(rng())); - vm.registerHostCallback('__generateNanoid', () => - vm.newString(generateNanoid()) - ); + // Re-register every host callback from the shared list. (The in-VM + // serde functions survive in the heap — no re-registration needed + // for them.) + for (const callback of hostCallbacks) { + vm.registerHostCallback(callback.name, callback.fn(vm)); + } // Process the delta events and drain jobs. { @@ -1196,6 +1236,7 @@ export async function startQuickJSWorkflow( vm, interruptBudget, advanceClock, + () => rngDraws, options.encryptionKey ); } @@ -1221,19 +1262,12 @@ export async function startQuickJSWorkflow( // ---- Phase 2: per-run initialization ---- async function runWorkflowInVM(): Promise { - // Seeded Math.random - { - using randomFn = vm.newFunction('random', () => vm.newNumber(rng())); - using math = vm.global.getProp('Math'); - math.setProp('random', randomFn); - } - - // Seeded nanoid generator - { - using nanoidFn = vm.newFunction('__generateNanoid', () => - vm.newString(generateNanoid()) - ); - vm.setProp(vm.global, '__generateNanoid', nanoidFn); + // Install every host callback from the shared list (see + // hostCallbacks above — the restore path re-registers from the same + // list, so the two can't drift). + for (const callback of hostCallbacks) { + using fnHandle = vm.newFunction(callback.name, callback.fn(vm)); + callback.install(vm, fnHandle); } // Inject a deterministic timestamp for the VM's ULID factory. ULIDs @@ -1425,6 +1459,7 @@ export async function startQuickJSWorkflow( vm, interruptBudget, advanceClock, + () => rngDraws, options.encryptionKey ); } @@ -1459,6 +1494,7 @@ function makeLiveSession( vm: QuickJS, interruptBudget: InterruptBudget, advanceClock: (ms: number) => void, + getRngDraws: () => number, encryptionKey?: DecryptionKey ): QuickJSWorkflowSession { const result = checkWorkflowState(vm, { keepAliveOnSuspend: true }); @@ -1499,14 +1535,14 @@ function makeLiveSession( session.result = next; return next; }, - snapshot(): Uint8Array { + snapshot(): { data: Uint8Array; rngDraws: number } { if (!alive) { throw new Error( 'QuickJS workflow session is not alive — snapshot is only valid while suspended' ); } const snap = vm.snapshot(); - return QuickJS.serializeSnapshot(snap); + return { data: QuickJS.serializeSnapshot(snap), rngDraws: getRngDraws() }; }, dispose(): void { if (alive) { diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index 647c4d6ab0..fd6d1193c9 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -102,7 +102,10 @@ export { StructuredErrorSchema, } from './shared.js'; export type * from './snapshots.js'; -export { SnapshotMetadataSchema } from './snapshots.js'; +export { + SNAPSHOT_FORMAT_VERSION, + SnapshotMetadataSchema, +} from './snapshots.js'; export type { SpecVersion } from './spec-version.js'; export { isLegacySpecVersion, diff --git a/packages/world/src/snapshots.ts b/packages/world/src/snapshots.ts index 5b3998d5b3..975ae11ff9 100644 --- a/packages/world/src/snapshots.ts +++ b/packages/world/src/snapshots.ts @@ -9,6 +9,38 @@ export const SnapshotMetadataSchema = z.object({ eventsCursor: z.string().nullable(), /** Timestamp when the snapshot was created */ createdAt: z.coerce.date(), + /** + * Number of events the run had processed when the snapshot was taken. + * Restores see only the delta after `eventsCursor`, so guards that need + * the TOTAL log size (the server-supplied max-events ceiling) add this + * to the delta. Optional for snapshots written before the field existed + * (treated as 0 — the ceiling degrades to delta-only for those, exactly + * the pre-field behavior). + */ + eventCount: z.number().int().nonnegative().optional(), + /** + * Number of draws the run's seeded PRNG had consumed when the snapshot + * was taken. On restore the runtime re-seeds from the run's BASE seed + * and fast-forwards this many draws, so correlation-id generation + * continues at the exact position full replay would have reached — + * keeping ids identical across snapshot generations AND identical to a + * no-snapshot run (which is what makes concurrent invocations restored + * from different snapshots of the same run still collide on the world's + * per-(runId, correlationId) dedup). + */ + rngDraws: z.number().int().nonnegative().optional(), + /** + * Snapshot format tag. A reader that doesn't recognize the version + * treats the snapshot as a clean miss (full replay) instead of handing + * an incompatible heap to the WASM engine. + */ + formatVersion: z.number().int().optional(), }); +/** + * Current snapshot format version, bumped when the heap layout or the + * metadata contract changes incompatibly. + */ +export const SNAPSHOT_FORMAT_VERSION = 1; + export type SnapshotMetadata = z.infer;