diff --git a/.changeset/bright-baboons-trade.md b/.changeset/bright-baboons-trade.md new file mode 100644 index 0000000000..cd1f2f0586 --- /dev/null +++ b/.changeset/bright-baboons-trade.md @@ -0,0 +1,7 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +'@workflow/world-vercel': patch +--- + +Clarify runtime invariants and compression behavior. diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index a48e99fde2..517d3ca7ae 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -282,16 +282,10 @@ function rootRunIdFrom( * `wait_completed`, which the wait timer can resolve with * `wait_completed`). * - * This gates the inline-delta fast path (per kind — see the gate) and the - * turbo forced-optimistic-start latch. The delta returned by the - * step-terminal write is the event log as of that write; it is consumed - * on the NEXT loop iteration, so any event a concurrent writer appends in - * that window would be present in a real `events.list` fetch but absent - * from the stale delta — the replay observes it one iteration later than - * the fetch path would. When no hook or wait is open, the only - * out-of-band writer is cancellation, which is benign to observe one - * iteration late (the next entity write is rejected against the terminal - * run and the run is already terminal), so the fast path is safe. + * This gates the inline-delta fast path and turbo's forced optimistic start. + * A terminal-step delta can omit an event appended concurrently after that + * write. With no open hook or wait, only cancellation can do so, and observing + * it one replay late is safe because the next entity write is rejected. * * Step-body `attr_set` writes are NOT a concern: they land before the * step's terminal write and are therefore already inside the returned @@ -481,30 +475,9 @@ export function workflowEntrypoint( traceContext ); - // --- Replay budget bookkeeping --- - // The replay budget bounds the *non-step* portion of a single - // handler invocation: deterministic event-log replay, workflow-VM - // execution between step boundaries, suspension handling, queue - // round-trips, etc. Inline step bodies (`"use step"` functions - // invoked via `executeStep`) are intentionally excluded — they are - // bounded by the platform's function `maxDuration` and the - // `NO_INLINE_REPLAY_AFTER_MS` early-return guard below. - // - // The budget is checked at loop boundaries (top of each `while` - // iteration). Note this is *less responsive* than the old - // `setTimeout`-based approach: a single pathological `runWorkflow` - // call processing a huge event log can overshoot the budget by up - // to one iteration before bailing. In practice the headroom built - // into `MAX_REPLAY_TIMEOUT_MS` (and the platform `maxDuration` - // SIGTERM as ultimate backstop) gives us slack — the previous - // `setTimeout` approach also relied on the platform kill as the - // hard backstop. Do *not* "fix" this by adding a `setInterval`; - // it would risk the same bug we just removed (bounding step - // bodies). - // - // Earlier versions (pre-#2009 fix) used a single `setTimeout` - // that also bounded step bodies, which broke any workflow with a - // single step longer than the budget. + // The replay budget covers orchestration work between steps, not inline + // step bodies. It is checked between loop iterations; step bodies use + // the platform timeout and NO_INLINE_REPLAY_AFTER_MS guard instead. const replayBudget = new ReplayBudget(); // In linked mode the run-origin context is NOT restored as the @@ -2588,11 +2561,8 @@ export function workflowEntrypoint( ); } - // If any inline step had pending background ops (e.g., - // stream writes to S3), break the loop and queue a plain - // continuation so waitUntil can flush them before the - // next replay reads them. This matches V1 behavior where - // each step ran in a separate function invocation. + // Let pending background operations flush before the next + // replay reads their results. if (anyPendingOps) { runtimeLogger.debug( 'Breaking loop: inline step has pending ops', @@ -2858,12 +2828,12 @@ export function workflowEntrypoint( return; } } - } // End while loop + } } - ); // End trace + ); } - ); // End withWorkflowBaggage - }); // End withTraceContext + ); + }); } ); diff --git a/packages/core/src/runtime/resume-hook.ts b/packages/core/src/runtime/resume-hook.ts index 60df8d3f7f..465aea690a 100644 --- a/packages/core/src/runtime/resume-hook.ts +++ b/packages/core/src/runtime/resume-hook.ts @@ -150,9 +150,8 @@ export async function resumeHook( encryptionKey = undefined; } - // Compress the payload only when the target run is marked as - // possibly containing compressed payloads (specVersion >= 5) AND - // its deployment can decode the 'gzip' format. + // Compress only when the target run and its deployment support the + // compression formats introduced with spec version 5. const compression = (workflowRun.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION && capabilities.supportedFormats.has(SerializationFormat.GZIP); diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index 70b7a8a9fb..70da5598c4 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -88,8 +88,8 @@ export interface StepExecutorParams { encryptionKey?: CryptoKey; /** * The workflow run's specVersion, used to gate payload compression. - * Step outputs/errors are only gzip-compressed when the run is marked - * as possibly containing compressed payloads (specVersion >= 5). + * Step outputs/errors are only compressed when the run is marked as + * compression-capable (specVersion >= 5). */ runSpecVersion?: number; /** @@ -250,8 +250,7 @@ export async function executeStep( stepName, } = params; const isVercel = process.env.VERCEL_URL !== undefined; - // Gate payload compression on the run's specVersion: only runs marked - // as possibly containing compressed payloads (spec >= 5) get gzip data. + // Gate payload compression on the run's specVersion. const compression = (params.runSpecVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION; diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index a9c102caeb..921986d7bd 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -285,8 +285,7 @@ export async function handleSuspension({ const rawKey = await world.getEncryptionKeyForRun?.(run); const encryptionKey = rawKey ? await importKey(rawKey) : undefined; - // Gate payload compression on the run's specVersion: only runs marked - // as possibly containing compressed payloads (spec >= 5) get gzip data. + // Gate payload compression on the run's specVersion. const compression = (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION; diff --git a/packages/core/src/serialization/codec.ts b/packages/core/src/serialization/codec.ts index ec66ca2a8a..7411d6b80e 100644 --- a/packages/core/src/serialization/codec.ts +++ b/packages/core/src/serialization/codec.ts @@ -57,9 +57,11 @@ export interface CodecOptions { extraRevivers?: Record any>; /** - * Whether to gzip-compress the serialized payload (write side only; - * reads always handle both compressed and uncompressed data). Must - * only be enabled when the target run supports compressed payloads: + * Whether to compress the serialized payload (write side only; zstd is + * preferred and gzip is the portable fallback). Reads dispatch compressed + * payloads by format prefix; zstd decoding still requires runtime or + * registered decoder support. Must only be enabled when the target run + * supports compressed payloads: * run specVersion >= SPEC_VERSION_SUPPORTS_COMPRESSION, and for * cross-deployment writes the target deployment's capabilities (see * `getRunCapabilities` in capabilities.ts). Defaults to `false`. diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index f8cf522bb7..d66faa9fb1 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -408,7 +408,7 @@ export const QueueSerializeTimeMs = SemanticConvention( 'workflow.queue.serialize_time_ms' ); -// Payload compression attributes (gzip codec, specVersion >= 5) +// Payload compression attributes (zstd preferred, gzip fallback; specVersion >= 5) // // Sizes are measured at the compression boundary: before encryption on the // write path and after decryption on the read path. They therefore reflect diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 6c4a1ff5e7..4dfd069e4d 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -53,31 +53,15 @@ import { createSleep } from './workflow/sleep.js'; /** * Drain pending queue items at workflow completion (success or failure). * - * Treats end-of-run like a final suspension: any operation the workflow code - * spawned but didn't `await` — abort hook resumes, hook creations/disposals, - * sleep waits, step queueings — gets committed to the event log via the - * suspension handler before the run is marked terminal. + * Treats completion as a final suspension so unawaited hook, wait, and + * attribute operations are recorded before the run becomes terminal. This + * also lets a final `controller.abort()` reach in-flight steps. * - * This matches normal JS semantics where `setTimeout(fn, ...)` etc. continue - * running after the surrounding function returns. Most importantly, it ensures - * `controller.abort()` called as the last statement of a workflow actually - * propagates to in-flight steps on other compute instances — without this, - * the abort hook is created but never resumed and the cancellation never - * reaches the running step. + * It records `*_created` and native attribute events, but never executes step + * bodies: `step_started` is rejected after the run becomes terminal. A + * fire-and-forget step therefore needs a later suspension to be scheduled. * - * NOTE: drain commits native attribute events and the `*_created` events; it - * does NOT enqueue step - * bodies for execution. The combined runtime handler rejects `step_started` - * for runs that have already transitioned to terminal (`RunExpiredError`), - * so a step queued here would be skipped anyway. Fire-and-forget step calls - * with side effects therefore work only when followed by some later `await` - * on a runtime primitive that triggers a real suspension (the normal - * runtime loop in `runtime.ts` queues the step there). Native attribute writes - * do not have this limitation because their event is the durable write. - * - * Drain failures are swallowed: the workflow's own outcome (the user's return - * value or thrown error) is the source of truth; secondary cleanup that fails - * shouldn't change the run's terminal state. + * Drain failures do not change the workflow's terminal outcome. */ async function drainPendingQueueItems( runId: string, @@ -86,13 +70,8 @@ async function drainPendingQueueItems( workflowRun: WorkflowRun, outcome: 'completed' | 'failed', /** - * Turbo mode only: resolves once the backgrounded `run_started` has landed. - * The drain runs at workflow completion *inside* `runWorkflow`, before the - * caller's terminal `awaitRunReady()` — so a workflow that creates a - * fire-and-forget hook (or wait/attribute) and then returns synchronously - * would otherwise have its `*_created` write race ahead of `run_started`. - * Threading the barrier into the suspension handler gates those writes the - * same way the normal suspension path is gated. Undefined outside turbo. + * In turbo mode, gates final `*_created` writes on backgrounded + * `run_started`. Undefined when `run_started` is awaited. */ runReadyBarrier?: Promise ): Promise { @@ -174,26 +153,12 @@ export async function runWorkflow( ); } - // The deterministic RNG seed is derived from identifiers that are all - // known the instant the queue message arrives — `runId`, `workflowName`, - // and `deploymentId` — with no timestamp component. `runId` alone already - // makes the seed unique-per-run and replay-stable; `workflowName` and - // `deploymentId` are included for extra entropy. Dropping the timestamp - // means the seed no longer depends on `startedAt`/`createdAt`, so it (and - // the VM context) can be computed before any server round-trip. - // - // The VM's initial fixed clock is derived from the run's creation time, - // recovered from the ULID embedded in `runId` (also available immediately), - // falling back to the run snapshot's `createdAt` for non-ULID ids. This - // initial `fixedTimestamp` only governs `Date.now()` / `new Date()` in the - // window before the first event is consumed; thereafter `updateTimestamp` - // advances the VM clock to each consumed event's `createdAt` (see the - // EventsConsumer below), starting with `run_created`. + // Seed and initial clock must be available before I/O and remain stable on + // replay. After the first event, EventsConsumer advances the VM clock from + // each event's `createdAt`. const fixedTimestamp = runIdCreatedAt(workflowRun.runId) ?? +workflowRun.createdAt; - // Get the port before creating VM context to avoid async operations - // affecting the deterministic timestamp const isVercel = process.env.VERCEL_URL !== undefined; // Load getPort lazily to prevent Turbopack from tracing get-port's // fs ops (readdir, readFile) into the flow route bundle. The resolved @@ -248,15 +213,8 @@ export async function runWorkflow( globalThis: vmGlobalThis, onWorkflowError: workflowDiscontinuation.reject, eventsConsumer, - // Correlation IDs (step_/wait_/hook_) are derived from `generateUlid`, so - // the time prefix fed to `ulid()` MUST be replay-stable across every - // delivery — otherwise a redelivery regenerates different correlation IDs - // and replay throws ReplayDivergenceError. `startedAt` is NOT safe here: - // under turbo the first delivery synthesizes `startedAt` from the local - // clock, but later (non-turbo) deliveries load the server-canonical - // `startedAt`, which differs by >=1ms. Use the same replay-stable value - // that already seeds the RNG and the VM clock (`fixedTimestamp`, recovered - // from the run ID's ULID and known the instant the message arrives). + // Correlation IDs must be replay-stable. `startedAt` differs between a + // turbo delivery and a later server-backed replay, so use fixedTimestamp. generateUlid: () => ulid(fixedTimestamp), generateNanoid, invocationsQueue: new Map(), @@ -335,8 +293,7 @@ export async function runWorkflow( // @ts-expect-error - `@types/node` says symbol is not valid, but it does work vmGlobalThis[STABLE_ULID] = ulid; - // NOTE: Will have a config override to use the custom fetch step. - // For now `fetch` must be explicitly imported from `workflow`. + // Workflow code must import the deterministic `fetch` step from `workflow`. vmGlobalThis.fetch = () => { throw new vmGlobalThis.Error( `Global "fetch" is unavailable in workflow functions. Use the "fetch" step function from "workflow" to make HTTP requests.\n\nLearn more: https://workflow-sdk.dev/err/${ERROR_SLUGS.FETCH_IN_WORKFLOW_FUNCTION}` @@ -812,11 +769,9 @@ export async function runWorkflow( } vmGlobalThis.TransformStream = TransformStream; - // Eventually we'll probably want to provide our own `console` object, - // but for now we'll just expose the global one. vmGlobalThis.console = globalThis.console; - // HACK: propagate symbol needed for AI gateway usage + // Expose the request-context symbol required by AI Gateway. const SYMBOL_FOR_REQ_CONTEXT = Symbol.for('@vercel/request-context'); // @ts-expect-error - `@types/node` says symbol is not valid, but it does work vmGlobalThis[SYMBOL_FOR_REQ_CONTEXT] = (globalThis as any)[ @@ -829,24 +784,9 @@ export async function runWorkflow( const parsedName = parseWorkflowName(workflowRun.workflowName); const filename = parsedName?.moduleSpecifier || workflowRun.workflowName; - // Evaluate the workflow bundle against the fresh context using a - // process-wide cache of the compiled `vm.Script`. The bundle is the same - // string for every replay and every invocation in this process, and - // compilation is a pure function of `(code, filename)`, so reusing the - // compiled Script across replays is determinism-safe: it produces the same - // workflow function and the same `filename` source attribution as - // re-parsing the bundle every time, but skips the (expensive) re-parse. - // Evaluating the bundle registers every workflow on - // `globalThis.__private_workflows`; the trailing lookup expression then - // retrieves the requested workflow function. The lookup is evaluated as a - // separate cached Script under the same `filename`, so error stack frames - // still attribute to the workflow's source file (`remapErrorStack` keys on - // `filename`). The one behavioural difference from the previous - // single-combined-string approach is the *line number* of an error thrown - // by the lookup expression itself: it now reports line 1 of the lookup - // Script rather than the line just past the end of the bundle. That path - // is rare (it requires the lookup `?.get(...)` expression to throw) and - // does not affect the workflow function or replay determinism. + // Reuse compiled scripts by `(code, filename)`: compilation is deterministic + // and the filename preserves workflow source attribution in stack traces. + // The bundle registers workflows on `globalThis.__private_workflows`. runCachedWorkflowScript(workflowCode, filename, context); const workflowFn = runCachedWorkflowScript( `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`, @@ -896,9 +836,8 @@ export async function runWorkflow( encryptionKey, vmGlobalThis, false, - // Gate payload compression on the run's specVersion: only runs - // marked as possibly containing compressed payloads (spec >= 5) - // get gzip data. + // Only compression-capable runs (spec >= 5) may write compressed + // payloads. The serializer uses zstd when supported and may use gzip. (workflowRun.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION ); diff --git a/packages/world-vercel/src/index.ts b/packages/world-vercel/src/index.ts index 1df709509a..902885222d 100644 --- a/packages/world-vercel/src/index.ts +++ b/packages/world-vercel/src/index.ts @@ -29,13 +29,8 @@ export function createWorld(config?: APIConfig): World { config?.projectConfig?.projectId || process.env.VERCEL_PROJECT_ID; return { - // Spec v5: new runs may carry gzip-compressed payloads (compression is - // entirely client-side — the workflow-server stores payloads opaquely - // via RemoteRef and never deserializes them). Spec 5 is a superset of - // spec 4, so native `attr_set` events and initial run attributes still - // work. New runs are stamped with this version; the server must support - // at least it — workflow-server declared spec-5 support in - // vercel/workflow-server#520. + // Spec v5 adds client-side zstd/gzip payload compression. The server stores + // those payloads opaquely, and v5 remains a superset of v4 attributes. specVersion: SPEC_VERSION_SUPPORTS_COMPRESSION, capabilities: { // workflow-server enforces the `stateUpdatedAt` optimistic-concurrency diff --git a/packages/world/src/spec-version.test.ts b/packages/world/src/spec-version.test.ts index 75e607611b..880c43180d 100644 --- a/packages/world/src/spec-version.test.ts +++ b/packages/world/src/spec-version.test.ts @@ -29,7 +29,7 @@ describe('requiresNewerWorld', () => { // payloads they cannot decode: a spec-5 run read by an SDK whose // SPEC_VERSION_CURRENT is 4 fails this check up front (with // RunNotSupportedError at the storage layer) instead of failing on - // individual gzip payloads. + // individual compressed payloads. expect(requiresNewerWorld(SPEC_VERSION_CURRENT + 1)).toBe(true); }); diff --git a/packages/world/src/spec-version.ts b/packages/world/src/spec-version.ts index 78b9d531f5..cf516b772b 100644 --- a/packages/world/src/spec-version.ts +++ b/packages/world/src/spec-version.ts @@ -25,10 +25,9 @@ export const SPEC_VERSION_SUPPORTS_EVENT_SOURCING = 2 as SpecVersion; export const SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT = 3 as SpecVersion; export const SPEC_VERSION_SUPPORTS_ATTRIBUTES = 4 as SpecVersion; /** - * Runs at this spec version or later may contain gzip-compressed payloads - * ('gzip' format prefix). Readers older than this version cannot decode - * such payloads and reject the run via `requiresNewerWorld()` instead of - * failing on individual payloads. + * Runs at this spec version or later may contain zstd- or gzip-compressed + * payloads. Readers older than this version reject the run via + * `requiresNewerWorld()` instead of failing on individual payloads. */ export const SPEC_VERSION_SUPPORTS_COMPRESSION = 5 as SpecVersion;