diff --git a/.changeset/bound-decoded-collections.md b/.changeset/bound-decoded-collections.md new file mode 100644 index 0000000000..f52cd89bb2 --- /dev/null +++ b/.changeset/bound-decoded-collections.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Reject compact sparse-array payloads with logical lengths above 100,000 during hydration. diff --git a/.changeset/pre.json b/.changeset/pre.json index 0cc7719137..b25e608af4 100644 --- a/.changeset/pre.json +++ b/.changeset/pre.json @@ -48,7 +48,9 @@ "@workflow/vitest": "4.0.0", "@workflow/vitest-workbench": "4.0.0", "tarballs": "0.0.0", - "@workflow/example-tanstack-start": "0.0.0" + "@workflow/example-tanstack-start": "0.0.0", + "@workflow/world-sim": "0.0.0", + "@workflow/sim-world-workbench": "0.0.1" }, "changesets": [ "abort-e2e-flakes", @@ -97,6 +99,7 @@ "bundle-aliased-project-local-helpers", "byte-stream-wire-framing", "cache-step-port-discovery", + "calm-dodos-retry", "calm-events-guard", "calm-geese-observe", "cancel-v4-frame-stream", @@ -117,6 +120,7 @@ "cold-lines-change", "compression-telemetry-attributes", "compute-instance-event-field", + "continue-run-started-preload", "cookbook-child-workflows-hook-resume", "cool-cups-greet", "copyable-data-block-tests", @@ -129,6 +133,7 @@ "curvy-dingos-cry", "cyan-impalas-drop", "data-inspector-expanded-brackets", + "date-subclass-vm", "decouple-vm-seed-from-startedat", "dedupe-otel-api", "dense-stream-viewer", @@ -173,6 +178,7 @@ "event-log-race-repro-infra", "event-log-race-repro-local", "event-precondition-guard", + "events-create-cursor-delta", "events-exact-id-search", "experimental-attributes-docs", "extend-cn-tailwind-merge", @@ -181,6 +187,7 @@ "fail-cross-deployment-runs", "fast-next-workflow-hmr", "fast-workflow-discovery", + "fatal-hook-world-validation", "fatal-retryable-error-serialization", "features-encryption-metadata", "few-cups-share", @@ -294,6 +301,7 @@ "lazy-hook-resumption", "lazy-inline-step-start", "lazy-next-step-imports", + "lazy-resume-deployment-affinity", "lazy-stream-key-request-diagnostics", "leading-edge-dispatch", "little-sites-cover", @@ -303,6 +311,7 @@ "local-port-cache", "local-world-vercel-warning", "log-queue-handler-retries", + "loud-pugs-recycle", "lucky-windows-smash", "many-peas-jog", "mean-cameras-hope", @@ -341,6 +350,7 @@ "olive-hands-repeat", "olive-lions-measure", "olive-pandas-brush", + "olive-pugs-repeat", "optional-input-on-snapshots", "orphan-ui-chunks-on-resume", "otel-api-optional-peer-external", @@ -348,7 +358,6 @@ "outcome-event-workflowname", "parallel-inline-optimistic-start", "pending-trace-viewer-gray-indicator", - "per-kind-correlation-ids", "perf-cached-workflow-script", "perf-memoize-step-hydration", "perf-prepare-replay-payloads", @@ -370,12 +379,15 @@ "queue-namespace-primitive", "queued-for-uses-first-step-started", "quick-local-replay", + "quickjs-baseline-snapshot", "quickjs-hook-capability", "quickjs-hook-retention", + "quickjs-host-serde", "quickjs-inline-steps", "quickjs-vm-engine", "quiet-analytics-readers", "quiet-build-logs", + "quiet-donkeys-repeat", "quiet-events-merge", "quiet-lamps-parse", "quiet-marker-shadow", @@ -450,6 +462,7 @@ "skip-changeset-release-deploys", "skip-community-worlds-main", "skip-world-local-nov-ghosts", + "slot-event-ids", "slow-bottles-pull", "small-cycles-rest", "smart-baths-make", @@ -475,10 +488,12 @@ "step-delivery-ordering", "step-maxretries-timeout", "step-run-required", + "stream-all-replay-events", "stream-close-retry", "stream-failure-diagnostics", "stream-otel-spans", "stream-read-v3-reconnect", + "stream-run-started-page", "stream-write-flush-span", "stream-writemulti-batching", "strong-cups-sip", @@ -501,6 +516,8 @@ "test-limit-env-config", "test-promise-race-prefix-determinism", "tidy-badgers-build", + "tidy-buttons-swim", + "tidy-dodos-observe", "tidy-hoops-open", "tidy-lamps-sort", "timeline-queued-connector", @@ -525,6 +542,7 @@ "turbo-next-workbench-vercel-output", "turbo-setattr-run-ready", "turbo-skip-run-started-preload", + "type-event-create-results", "typography-system-tokens", "update-queue-client-version", "update-vercel-queue", @@ -538,6 +556,7 @@ "v4-trace-propagation", "validate-step-id-duplicates", "vast-oranges-fail", + "vercel-hook-min-retention", "vercel-world-custom-dispatcher", "vercel-world-missing-deployment-id", "versioning-docs", @@ -588,6 +607,7 @@ "world-vercel-trusted-sources", "world-zod-dependency", "writable-group-commit", + "ws-events-transport", "yellow-pianos-relax", "zstd-step-error-display", "zstd-web-decoder" diff --git a/.changeset/resilient-step-dispatch.md b/.changeset/resilient-step-dispatch.md new file mode 100644 index 0000000000..dadc178baa --- /dev/null +++ b/.changeset/resilient-step-dispatch.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': minor +'@workflow/world-vercel': minor +'@workflow/core': minor +--- + +Resilient step dispatch: newly created steps are published to the queue in parallel with their `step_created` event write, with the serialized input carried on the message (`stepInput`) so the consumer re-ensures the event if the direct write failed transiently — mirroring resilient start (`runInput`) and resilient hook resume (`hookInput`). Worlds that enforce the precondition guard keep the sequential create-then-publish dispatch (only sequencing gives the message a happens-after edge over the create's guard verdict); step-dispatch idempotency keys are now step-identity-scoped. Disable via `WORKFLOW_RESILIENT_STEP_DISPATCH=0`. diff --git a/.changeset/stream-expired-errors.md b/.changeset/stream-expired-errors.md new file mode 100644 index 0000000000..f276b6f7f6 --- /dev/null +++ b/.changeset/stream-expired-errors.md @@ -0,0 +1,7 @@ +--- +'@workflow/core': patch +'@workflow/errors': patch +'@workflow/world-vercel': patch +--- + +Surface typed terminal errors when stream retention expires. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 53a59e7759..b3e1c85824 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -71,6 +71,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Only applies to Worlds with atomic, immutable deployments (the Vercel World). A run whose pinned deployment cannot be reached at all fails immediately regardless of this value. - Transient or unknown queue publishing failures use normal queue redelivery and do not consume this budget. +### `WORKFLOW_RESILIENT_STEP_DISPATCH` + +- Default: enabled +- When a suspension hands newly created steps to the queue, the runtime publishes each step's execution message in parallel with its `step_created` event write instead of sequencing them, cutting a round trip per dispatched step. The message also carries the serialized step input (`stepInput`), so a transient `step_created` write failure (429 / 5xx / transport) still executes the step — the queue consumer idempotently re-ensures the event before running it, converging with the producer's write on the step's correlation ID. This mirrors resilient start (`runInput`) and the lazy hook resume (`hookInput`). +- The runtime falls back to the sequential create-then-publish dispatch automatically when the step input is too large to inline on the queue message, when the run's queue transport cannot carry binary payloads (pre-CBOR spec versions), or — on the `node` VM engine, whose suspension writes carry the [precondition guard](#workflow_precondition_guard) snapshot — when the World enforces that guard (`capabilities.preconditionGuard`; the Vercel World does): a guard-rejected `step_created` must not be materializable through the queue side-channel, and only sequencing the publish after the create gives the message a happens-after edge over the create's guard verdict. The `quickjs` engine's suspension writes are unguarded, so it uses resilient dispatch against every World. +- Producer-side recoveries are reported on the suspension span as `workflow.step.resilient_dispatch_recovered`; a consumer that materialized the event reports `workflow.step.resilient_dispatch_materialized`. +- Set `0` to force the sequential dispatch as a kill switch. + ### `WORKFLOW_PRECONDITION_GUARD` - Default: enabled diff --git a/packages/astro/CHANGELOG.md b/packages/astro/CHANGELOG.md index 1c4812c48f..338f71bd58 100644 --- a/packages/astro/CHANGELOG.md +++ b/packages/astro/CHANGELOG.md @@ -1,5 +1,14 @@ # @workflow/astro +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/builders@5.0.0-beta.41 + - @workflow/rollup@5.0.0-beta.41 + - @workflow/vite@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/astro/package.json b/packages/astro/package.json index b38e76dbfa..04862678ce 100644 --- a/packages/astro/package.json +++ b/packages/astro/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/astro", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Astro integration for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/builders/CHANGELOG.md b/packages/builders/CHANGELOG.md index 3d65454663..cb860091bc 100644 --- a/packages/builders/CHANGELOG.md +++ b/packages/builders/CHANGELOG.md @@ -1,5 +1,19 @@ # @workflow/builders +## 5.0.0-beta.41 + +### Minor Changes + +- [#3163](https://github.com/vercel/workflow/pull/3163) [`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba) Thanks [@lucamaraschi](https://github.com/lucamaraschi)! - Add an optional observer for accepted workflow SWC transform results. + +### Patch Changes + +- [#3084](https://github.com/vercel/workflow/pull/3084) [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727) Thanks [@shalabhc](https://github.com/shalabhc)! - Add an opt-in WebSocket transport for event writes during step execution, enabled with `WORKFLOW_EVENTS_TRANSPORT=ws`. Defaults to HTTP; no behavior change unless explicitly enabled. + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/builders/package.json b/packages/builders/package.json index 9b6f8766c9..72a65ac800 100644 --- a/packages/builders/package.json +++ b/packages/builders/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/builders", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Shared builder infrastructure for Workflow SDK", "type": "module", "main": "./dist/index.js", diff --git a/packages/cli/CHANGELOG.md b/packages/cli/CHANGELOG.md index bfb69ce75f..7439662d18 100644 --- a/packages/cli/CHANGELOG.md +++ b/packages/cli/CHANGELOG.md @@ -1,5 +1,18 @@ # @workflow/cli +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`bf4dda6`](https://github.com/vercel/workflow/commit/bf4dda64780f5f8e13840871960cf2a647cdd329), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`665110b`](https://github.com/vercel/workflow/commit/665110b3a255c920a3612dec521234efcf34d4f8), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/world@5.0.0-beta.26 + - @workflow/world-vercel@5.0.0-beta.37 + - @workflow/world-local@5.0.0-beta.35 + - @workflow/builders@5.0.0-beta.41 + - @workflow/web@5.0.0-beta.41 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.40 ### Minor Changes diff --git a/packages/cli/package.json b/packages/cli/package.json index 397d896891..b3c9dba263 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/cli", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Command-line interface for Workflow SDK", "type": "module", "bin": { diff --git a/packages/core/CHANGELOG.md b/packages/core/CHANGELOG.md index 6b40923945..077283bf00 100644 --- a/packages/core/CHANGELOG.md +++ b/packages/core/CHANGELOG.md @@ -1,5 +1,41 @@ # @workflow/core +## 5.0.0-beta.41 + +### Patch Changes + +- [#3385](https://github.com/vercel/workflow/pull/3385) [`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Retry replay timeouts through normal queue redelivery instead of exiting the process, and keep Postgres jobs retryable through Core's terminal delivery limit. + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Continue partial `run_started` event preloads from their cursor instead of reloading the event log from the beginning. + +- [#3372](https://github.com/vercel/workflow/pull/3372) [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536) Thanks [@ar-tama](https://github.com/ar-tama)! - Fix `Date` subclassing inside workflow functions. The deterministic `Date` override in the workflow VM now forwards `new.target` via `Reflect.construct`, so subclasses like `TZDate` from `@date-fns/tz` keep their identity, methods, and fields. Calling `Date()` without `new` now returns the (fixed) time string per spec, instead of a `Date` object. + +- [#3382](https://github.com/vercel/workflow/pull/3382) [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip + +- [#3286](https://github.com/vercel/workflow/pull/3286) [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Fail a workflow when its World deterministically rejects Hook creation. + +- [#3374](https://github.com/vercel/workflow/pull/3374) [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040) Thanks [@karthikscale3](https://github.com/karthikscale3)! - Carry the run's pinned deployment on lazy resume messages so misrouted deliveries re-route before the `hook_received` write. + +- [#3342](https://github.com/vercel/workflow/pull/3342) [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac) Thanks [@TooTallNate](https://github.com/TooTallNate)! - QuickJS engine: start invocations by restoring a bundle-hydrated VM-memory snapshot instead of re-evaluating the workflow bundle (~25× faster VM startup; bundles with module-scope randomness or clock reads automatically fall back to fresh evaluation). Disable with `WORKFLOW_QUICKJS_BASELINE_SNAPSHOT=0`. + +- [#3263](https://github.com/vercel/workflow/pull/3263) [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595) Thanks [@TooTallNate](https://github.com/TooTallNate)! - QuickJS engine: move serialization out of the VM onto the host, replacing the in-VM serde bundle with side-effect-free handle introspection (same wire format, 2–100× faster). + +- [#3406](https://github.com/vercel/workflow/pull/3406) [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fix replay divergence when a step result overtook an earlier sleep or hook delivery that was parked behind an unread hook's payload + +- [#3389](https://github.com/vercel/workflow/pull/3389) [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - **Breaking**: SpecVersion 6: Event IDs are now a dense per-run slot number, allocated by the world at publish time so a rejected write leaves no gap in the event log. A replay tells the world how many events it had read and gets back the ones it did not see, so an event that arrives from outside the replay and lands ahead of an event the replay wrote no longer fails the run with `CORRUPTED_EVENT_LOG`: it is held for whichever part of the workflow awaits it. A gap in the numbering fails the run instead of being replayed over. + +- [#3205](https://github.com/vercel/workflow/pull/3205) [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load replay-event suffixes through one World request. + +- [#3439](https://github.com/vercel/workflow/pull/3439) [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Stop reporting replay divergence for an event the workflow is still on its way to consuming, by waiting for in-flight step and hook deliveries instead of a fixed delay + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Infer required run and step entities, including their start times, from + `events.create` request types. +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`bf4dda6`](https://github.com/vercel/workflow/commit/bf4dda64780f5f8e13840871960cf2a647cdd329), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`665110b`](https://github.com/vercel/workflow/commit/665110b3a255c920a3612dec521234efcf34d4f8), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/world@5.0.0-beta.26 + - @workflow/world-vercel@5.0.0-beta.37 + - @workflow/world-local@5.0.0-beta.35 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.40 ### Minor Changes diff --git a/packages/core/package.json b/packages/core/package.json index c69222315a..c1f9a07993 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/core", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Core runtime and engine for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/core/src/reconnecting-framed-stream.test.ts b/packages/core/src/reconnecting-framed-stream.test.ts index 6dfa098caa..9431ed8bf0 100644 --- a/packages/core/src/reconnecting-framed-stream.test.ts +++ b/packages/core/src/reconnecting-framed-stream.test.ts @@ -1,3 +1,4 @@ +import { StreamExpiredError } from '@workflow/errors'; import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; import { afterEach, describe, expect, it, vi } from 'vitest'; @@ -224,6 +225,37 @@ describe('createReconnectingFramedStream', () => { expect(calls).toEqual([0, 2, 2]); }); + it('surfaces retention expiry from a reconnect without retrying it', async () => { + const expired = new StreamExpiredError( + 'stream expired', + RUN_ID, + 's', + new Date('2026-08-10T14:40:00.000Z') + ); + let calls = 0; + const world = { + specVersion: SPEC_VERSION_CURRENT, + streams: { + get: vi.fn(async () => { + calls++; + if (calls === 1) { + return scriptedStream([ + { kind: 'value', value: payloadFrame(1) }, + { kind: 'error', err: new Error('connection dropped') }, + ]); + } + throw expired; + }), + }, + } as unknown as World; + setWorld(world); + + const reader = createReconnectingFramedStream(RUN_ID, 's', 0).getReader(); + expect((await reader.read()).value).toEqual(payloadFrame(1)); + await expect(reader.read()).rejects.toBe(expired); + expect(calls).toBe(2); + }); + it('respects an initial non-zero startIndex on reconnect', async () => { const { world, calls } = makeWorldWithScriptedStreams({ 10: () => diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 76724c199a..8a2e153e95 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -1,4 +1,5 @@ import { + EntityConflictError, PreconditionFailedError, RUN_ERROR_CODES, ThrottleError, @@ -20,6 +21,7 @@ import { import { setWorld } from './runtime/world.js'; import { workflowEntrypoint } from './runtime.js'; import { + dehydrateStepArguments, dehydrateStepReturnValue, dehydrateWorkflowArguments, hydrateRunError, @@ -1790,6 +1792,337 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { }); }); +describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)', () => { + afterEach(() => { + setWorld(undefined); + vi.clearAllMocks(); + }); + + const getWorkflowTransformCode = (workflowName: string) => + `;globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set(${JSON.stringify(workflowName)}, ${workflowName});`; + + // The workflow body is never replayed by these tests: the seeded log keeps + // an unrelated step pending, so the background-step path returns right + // after executing the message's step. + const resilientWorkflow = `const resilientAdd = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("resilientAdd"); + async function workflow() { + return await resilientAdd(2, 3); + }${getWorkflowTransformCode('workflow')}`; + + const stepBodySpy = vi.fn(async (a: number, b: number) => a + b); + registerStepFunction('resilientAdd', stepBodySpy); + + /** + * Drives the handler with a background-step message carrying `stepInput`. + * The event log is seeded with a pending unrelated step so the handler + * returns after the step executes (no full workflow replay to converge). + */ + async function driveStepMessage(opts: { + runId: string; + attempt: number; + /** Reject the step_created re-ensure with this error. */ + ensureError?: Error; + omitStepInput?: boolean; + /** + * Simulate the delivery beating the producer's parallel step_created: + * bare step_started rejects with this error until a step_created for the + * step has been written (the in-band re-ensure path). + */ + stepMissingError?: Error; + }) { + const stepId = 'step_resilient_1'; + const dehydratedInput = (await dehydrateStepArguments( + { args: [2, 3], closureVars: [], thisVal: null }, + opts.runId, + undefined + )) as Uint8Array; + + const workflowRun: WorkflowRun = { + runId: opts.runId, + workflowName: 'workflow', + status: 'running', + specVersion: SPEC_VERSION_CURRENT, + input: await dehydrateWorkflowArguments([], opts.runId, undefined, []), + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; + + let eventSeq = 0; + const durableEvents: Event[] = [ + // An unrelated pending step: keeps the run un-replayable so the handler + // returns right after the background step completes. + { + eventId: 'event-other', + runId: opts.runId, + createdAt: new Date(), + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: 'step_other', + eventData: { stepName: 'otherStep', input: dehydratedInput }, + } as unknown as Event, + ]; + const recordEvent = (data: any): Event => { + eventSeq += 1; + const created = { + eventId: `event-${eventSeq}`, + runId: opts.runId, + createdAt: new Date(), + ...data, + } as Event; + durableEvents.push(created); + return created; + }; + + const createdEvents: any[] = []; + const createdEventParams: any[] = []; + let stepEntityExists = false; + const eventsCreate = vi.fn( + async (_runId: string, data: any, params?: any) => { + createdEvents.push(data); + createdEventParams.push(params); + if (data.eventType === 'step_created') { + if (opts.ensureError) throw opts.ensureError; + stepEntityExists = true; + return { event: recordEvent(data) }; + } + if (data.eventType === 'step_started') { + if (opts.stepMissingError && !stepEntityExists) { + throw opts.stepMissingError; + } + return { + event: recordEvent(data), + step: { + runId: opts.runId, + stepId, + stepName: 'resilientAdd', + status: 'running' as const, + attempt: 1, + input: dehydratedInput, + startedAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }, + }; + } + return { event: recordEvent(data) }; + } + ); + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn( + ( + _prefix: string, + handler: (message: unknown, metadata: unknown) => Promise + ) => { + return async () => { + await handler( + { + runId: opts.runId, + stepId, + stepName: 'resilientAdd', + requestedAt: new Date('2024-01-01T00:00:00.000Z'), + ...(opts.omitStepInput + ? {} + : { stepInput: { input: dehydratedInput } }), + }, + { + requestId: 'req_test', + attempt: opts.attempt, + queueName: '__wkf_workflow_workflow', + messageId: 'msg_test', + } + ); + return new Response(null, { status: 204 }); + }; + } + ), + events: { + create: eventsCreate, + list: vi.fn(async () => ({ + data: [...durableEvents], + hasMore: false, + cursor: 'cursor_test', + })), + }, + runs: { + get: vi.fn(async () => workflowRun), + }, + queue: vi.fn(async () => ({ messageId: null })), + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as any); + + const handler = workflowEntrypoint(resilientWorkflow); + const response = (await handler( + new Request('https://example.test') + )) as Response; + return { response, createdEvents, createdEventParams, dehydratedInput }; + } + + it('materializes step_created from stepInput on a redelivery before executing', async () => { + const { response, createdEvents, createdEventParams, dehydratedInput } = + await driveStepMessage({ + runId: 'wrun_resilient_step_materialize', + attempt: 2, + }); + + expect(response.status).toBe(204); + // The re-ensure wrote the step_created with the message's payload… + expect(createdEvents).toContainEqual( + expect.objectContaining({ + eventType: 'step_created', + correlationId: 'step_resilient_1', + eventData: expect.objectContaining({ + stepName: 'resilientAdd', + input: dehydratedInput, + }), + }) + ); + // …marked as a dispatch re-ensure so a guard-enforcing backend can refuse + // it when the producer's write was 412-rejected (dispatch revoked). + const ensureParamIdx = createdEvents.findIndex( + (e) => e.eventType === 'step_created' + ); + expect(createdEventParams[ensureParamIdx]).toMatchObject({ + viaStepDispatch: true, + }); + // …and it preceded the step's start. + const createdIdx = createdEvents.findIndex( + (e) => e.eventType === 'step_created' + ); + const startedIdx = createdEvents.findIndex( + (e) => e.eventType === 'step_started' + ); + expect(createdIdx).toBeGreaterThanOrEqual(0); + expect(createdIdx).toBeLessThan(startedIdx); + // The step body ran and its terminal event was written. + expect(stepBodySpy).toHaveBeenCalledWith(2, 3); + expect(createdEvents).toContainEqual( + expect.objectContaining({ + eventType: 'step_completed', + correlationId: 'step_resilient_1', + }) + ); + }); + + it('skips the re-ensure on a first delivery (no per-step write overhead)', async () => { + const { response, createdEvents } = await driveStepMessage({ + runId: 'wrun_resilient_step_first_delivery', + attempt: 1, + }); + + expect(response.status).toBe(204); + expect( + createdEvents.filter((e) => e.eventType === 'step_created') + ).toHaveLength(0); + expect(createdEvents).toContainEqual( + expect.objectContaining({ + eventType: 'step_completed', + correlationId: 'step_resilient_1', + }) + ); + }); + + it('treats an EntityConflict re-ensure as the common already-created case', async () => { + const { response, createdEvents } = await driveStepMessage({ + runId: 'wrun_resilient_step_conflict', + attempt: 2, + ensureError: new EntityConflictError('already exists'), + }); + + expect(response.status).toBe(204); + // The conflict is swallowed and the step still executes to completion. + expect(createdEvents).toContainEqual( + expect.objectContaining({ + eventType: 'step_completed', + correlationId: 'step_resilient_1', + }) + ); + }); + + it('does not re-ensure when the message carries no stepInput (legacy dispatch)', async () => { + const { response, createdEvents } = await driveStepMessage({ + runId: 'wrun_resilient_step_legacy', + attempt: 2, + omitStepInput: true, + }); + + expect(response.status).toBe(204); + expect( + createdEvents.filter((e) => e.eventType === 'step_created') + ).toHaveLength(0); + }); + + // The load-bearing recovery: a FIRST delivery that beats (or outlives a + // transient failure of) the producer's parallel step_created must + // materialize the step and execute it within the same delivery. It cannot + // wait for a redelivery — world-vercel's failure retries re-enqueue fresh + // messages whose attempt resets to 1, so an attempt-gated recovery would + // stall the step until the original message's ~300s visibility-timeout + // redelivery (measured exactly so in the durabench parallel sweeps). + it('recovers in-band on attempt 1 when the bare start rejects with step-not-found (world-vercel shape)', async () => { + const { response, createdEvents, createdEventParams } = + await driveStepMessage({ + runId: 'wrun_resilient_step_inband_vercel', + attempt: 1, + stepMissingError: new WorkflowWorldError( + 'workflow step step_resilient_1 not found', + { status: 404 } + ), + }); + + expect(response.status).toBe(204); + // Order: failed bare start → re-ensured step_created (viaStepDispatch) → + // successful start → completion, all in this delivery. + const types = createdEvents.map((e) => e.eventType); + expect(types).toEqual([ + 'step_started', + 'step_created', + 'step_started', + 'step_completed', + ]); + const ensureIdx = types.indexOf('step_created'); + expect(createdEventParams[ensureIdx]).toMatchObject({ + viaStepDispatch: true, + }); + }); + + it('recovers in-band on attempt 1 with the local-world error shape (no status)', async () => { + const { response, createdEvents } = await driveStepMessage({ + runId: 'wrun_resilient_step_inband_local', + attempt: 1, + stepMissingError: new WorkflowWorldError( + 'Step "step_resilient_1" not found' + ), + }); + + expect(response.status).toBe(204); + expect(createdEvents.map((e) => e.eventType)).toEqual([ + 'step_started', + 'step_created', + 'step_started', + 'step_completed', + ]); + }); + + it('propagates step-not-found without stepInput (nothing to recover from)', async () => { + await expect( + driveStepMessage({ + runId: 'wrun_resilient_step_inband_legacy', + attempt: 1, + omitStepInput: true, + stepMissingError: new WorkflowWorldError( + 'workflow step step_resilient_1 not found', + { status: 404 } + ), + }) + ).rejects.toThrow('not found'); + }); +}); + describe('workflowEntrypoint turbo mode', () => { const ORIG_TURBO = process.env.WORKFLOW_TURBO; const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 5b34574433..4d19bb2a8b 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -12,6 +12,7 @@ import { type RunErrorCode, RunExpiredError, WorkflowRuntimeError, + WorkflowWorldError, } from '@workflow/errors'; import { once, setWorkflowBasePath } from '@workflow/utils'; import { @@ -83,6 +84,7 @@ import { preconditionSnapshotParams, queueMessage, settleEventSlotGap, + stepDispatchIdempotencyKey, withHealthCheck, } from './runtime/helpers.js'; import { @@ -341,6 +343,27 @@ function getWorkflowSetupErrorCode(err: unknown): RunErrorCode | null { return null; } +/** + * Whether a step execution rejected because the step entity does not exist — + * the signature of a resilient step dispatch message whose delivery beat (or + * outlived a transient failure of) the producer's parallel `step_created` + * write. Every World surfaces it as a `WorkflowWorldError` naming the missing + * step: world-vercel maps the backend's 404 to + * `workflow step step_… not found`, world-local and world-postgres throw + * `Step "step_…" not found` directly. The message match is deliberately loose + * across those shapes; the status check narrows the remote case without + * excluding the local ones (which carry no status). + * + * Used only when the message carries `stepInput` — a bare dispatch without a + * payload has nothing to recover from, and the error keeps propagating for + * queue-driven recovery exactly as before. + */ +function isStepMissingError(err: unknown): boolean { + if (!WorkflowWorldError.is(err)) return false; + if (err.status !== undefined && err.status !== 404) return false; + return /step/i.test(err.message) && /not found/i.test(err.message); +} + async function recordFatalRunError({ world, workflowRun, @@ -644,6 +667,7 @@ export function workflowEntrypoint( deploymentMismatchRetryCount, runInput, hookInput, + stepInput, } = WorkflowInvokePayloadSchema.parse(message_); // `start()` always attaches a trace carrier, but // serializeTraceCarrier() returns `{}` when no OTEL SDK is registered @@ -1376,9 +1400,116 @@ export function workflowEntrypoint( // will pick up the replay. if (incomingStepId && incomingStepName) { try { - const bgRun = await world.runs.get(runId, { - resolveData: 'none', - }); + // Resilient step dispatch: the producer parallelized the + // `step_created` write with this queue publish, so the + // step entity may not exist yet when this delivery + // executes — the delivery beat the write, or the write + // failed transiently and this message carries the only + // copy of the input. Idempotently re-ensure the event + // from the message's `stepInput` — keyed by the step's + // correlation id, so the producer's write and this + // re-ensure converge on exactly one event. + // + // Invoked from two places: + // + // - IN-BAND (the load-bearing path): when the bare + // `step_started` below rejects with "step not found", + // the executor catch materializes the step and + // retries once, all within this delivery. This must + // not rely on delivery attempts: world-vercel's + // failure-retry path re-enqueues a FRESH message + // (attempt resets to 1), so an attempt-gated recovery + // is unreachable on the retry chain and the step + // would stall until the ORIGINAL message's + // ~300s visibility-timeout redelivery — measured + // exactly so in the durabench parallel sweeps before + // this path existed. + // - EAGERLY on a genuine redelivery (attempt > 1), + // in parallel with the run fetch below — a + // redelivered dispatch already had its create race + // resolved either way, so this saves the failed + // start round-trip at no wall-time cost. First + // deliveries skip it: the producer's write almost + // always lands, and an eager ensure would burn a + // conditional write per step. + const ensureStepFromMessage = async (): Promise< + 'ok' | 'gone' + > => { + if (!stepInput) return 'ok'; + try { + await world.events.create( + runId, + { + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: incomingStepId, + eventData: { + stepName: incomingStepName, + workflowName, + // Typed Uint8Array by StepDispatchInputSchema: + // a non-binary (mangled) payload fails the + // message parse above and never reaches this + // write. + input: stepInput.input, + }, + }, + { + requestId, + // Marks this create as a dispatch re-ensure so + // a guard-enforcing backend can refuse it when + // the producer's write was 412-rejected (the + // dispatch was revoked). Surfaces as + // RunExpiredError → 'gone' below, acking the + // message. Worlds without the guard ignore it. + viaStepDispatch: true, + } + ); + // This delivery materialized the step — the + // completion of the producer's recovery path. + span?.setAttributes( + Attribute.StepResilientDispatchMaterialized(true) + ); + runtimeLogger.warn( + 'Materialized step_created from the queue message — the producer\u2019s direct write did not land', + { + workflowRunId: runId, + stepId: incomingStepId, + stepName: incomingStepName, + } + ); + } catch (err) { + // The common case: the producer's write (or a + // concurrent re-ensure) already landed. + if (EntityConflictError.is(err)) return 'ok'; + // Nothing left to execute: the run went terminal + // (matches the run-status check below), or a + // guard-enforcing backend revoked this dispatch + // (410 `step-dispatch-revoked` — the producer's + // write was 412-rejected and the replay restarted + // with a corrected schedule). + if (RunExpiredError.is(err)) return 'gone'; + // Transient — rethrow so the queue redelivers and a + // later attempt converges instead of executing (and + // acking) a step that may not exist. + throw err; + } + return 'ok'; + }; + const [bgRun, ensureOutcome] = await Promise.all([ + world.runs.get(runId, { + resolveData: 'none', + }), + stepInput && metadata.attempt > 1 + ? ensureStepFromMessage() + : ('ok' as const), + ]); + if (ensureOutcome === 'gone') { + runtimeLogger.debug( + 'Run already finished, skipping background step', + { workflowRunId: runId } + ); + return; + } if (bgRun.status !== 'running') { runtimeLogger.debug( 'Run already finished, skipping background step', @@ -1455,25 +1586,47 @@ export function workflowEntrypoint( // step_started of a queue-driven execution // intentionally clears inline ownership (the step is // queue-owned from this point). + const executeQueuedStep = () => + executeStep({ + world, + workflowRunId: runId, + workflowDeploymentId: bgRun.deploymentId, + workflowName, + workflowStartedAt: bgStartedAt, + rootRunId: rootRunIdFrom(bgRun.attributes, runId), + stepId: incomingStepId, + stepName: incomingStepName, + runSpecVersion: bgRun.specVersion, + // Retry ceiling: the queue delivery count as a fast + // gate, verified against the recorded step_started + // count once it crosses the ceiling (see above). + authoritativeAttempt: bgAuthoritativeAttempt, + }); stepResult = await runStepSingleFlight( runId, incomingStepId, - () => - executeStep({ - world, - workflowRunId: runId, - workflowDeploymentId: bgRun.deploymentId, - workflowName, - workflowStartedAt: bgStartedAt, - rootRunId: rootRunIdFrom(bgRun.attributes, runId), - stepId: incomingStepId, - stepName: incomingStepName, - runSpecVersion: bgRun.specVersion, - // Retry ceiling: the queue delivery count as a fast - // gate, verified against the recorded step_started - // count once it crosses the ceiling (see above). - authoritativeAttempt: bgAuthoritativeAttempt, - }) + async () => { + try { + return await executeQueuedStep(); + } catch (err) { + // In-band resilient recovery: a missing step on + // a stepInput-carrying message means this + // delivery outran (or outlived a transient + // failure of) the producer's parallel + // step_created write. Materialize the event + // from the payload and retry ONCE, within this + // delivery — see ensureStepFromMessage for why + // this cannot wait for a redelivery. A second + // failure propagates as before. + if (!stepInput || !isStepMissingError(err)) { + throw err; + } + if ((await ensureStepFromMessage()) === 'gone') { + return { type: 'gone' as const }; + } + return await executeQueuedStep(); + } + } ); } finally { replayBudget.resume(); @@ -2844,6 +2997,19 @@ export function workflowEntrypoint( eventLog, runReadyBarrier, replayRecoveryReporter, + // Resilient step dispatch: lets eligible newly + // created steps publish their step-execution + // message (carrying `stepInput`) in parallel with + // the step_created write. Steps queued there are + // reported back in `queuedStepCorrelationIds` and + // skipped by the dispatch pass below. + stepDispatch: { + queueName: getWorkflowQueueName( + workflowName, + namespace + ), + getTraceCarrier: nextTraceCarrier, + }, }); } catch (suspensionError) { // A suspension create was rejected as stale: re-derive @@ -3111,8 +3277,8 @@ export function workflowEntrypoint( // kill-switched) → immediate enqueue, exactly as // before. This covers crash recovery: if a prior // handler wrote step_created but crashed before - // queueing, a later handler queues it; - // idempotencyKey on correlationId dedupes + // queueing, a later handler queues it; the + // step-identity-scoped idempotencyKey dedupes // redundant queues across concurrent handlers. // // The wait continuation is what makes @@ -3143,6 +3309,20 @@ export function workflowEntrypoint( if (inlineCorrelationIds.has(step.correlationId)) { continue; } + // Already published by the suspension handler's + // resilient dispatch (create + queue in parallel, + // message carrying `stepInput`). A re-publish here + // would dedupe on the idempotency key anyway, but + // skip the wasted round-trip. Ownership never + // applies to these: they were created this pass, so + // no step_started stamp can exist yet. + if ( + suspensionResult.queuedStepCorrelationIds.has( + step.correlationId + ) + ) { + continue; + } const ownershipActive = inlineOwnership && isStepOwnershipActive(step); if ( @@ -3204,7 +3384,17 @@ export function workflowEntrypoint( requestedAt: new Date(), }, { - idempotencyKey: step.correlationId, + // Step-identity-scoped: dedupes against every + // other dispatch of THIS step (concurrent + // handlers, crash-recovery re-dispatch, the + // suspension handler's resilient publish) + // without absorbing a dispatch of a different + // step under a reassigned correlation id — + // see stepDispatchIdempotencyKey. + idempotencyKey: stepDispatchIdempotencyKey( + step.correlationId, + step.stepName + ), } ) ); @@ -3779,10 +3969,10 @@ export function workflowEntrypoint( { delaySeconds, // Key the delayed retry on the step's - // correlationId so it dedupes against the + // dispatch key so it dedupes against the // keyed re-dispatch the suspension handler - // performs on replay (it also uses - // `idempotencyKey: step.correlationId`). + // performs on replay (it uses the same + // stepDispatchIdempotencyKey). // // Without this, a mixed batch where one step // `completed` with unflushed background ops @@ -3801,7 +3991,10 @@ export function workflowEntrypoint( // retry body could run early/concurrently. // Sharing the key lets the earlier delayed // message win, honoring the backoff. - idempotencyKey: step.correlationId, + idempotencyKey: stepDispatchIdempotencyKey( + step.correlationId, + step.stepName + ), } ) ) diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 37a82d1d82..a6b71eb839 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -215,6 +215,38 @@ export function getMaxInlineSteps(): number { return parsed; } +/** + * Upper bound on the serialized step input that resilient step dispatch will + * inline into the queue message's `stepInput`. + * + * Vercel Queues has no hard message-size cap (bodies above its ~256 KB + * inline threshold transparently spill to S3-backed storage), so this bound + * is a cost/latency choice, not a rejection guard: the message also carries + * the runId, stepId, stepName, and trace carrier alongside CBOR framing + * overhead, and staying under the queue's inline threshold keeps step + * messages on its fast inline path instead of paying an S3 store+fetch + * double-hop for bytes that already live in the event log. Above this size + * the dispatch falls back to the sequential path (`step_created` write, then + * a payload-less queue message). Matches `MAX_INLINE_RESUME_PAYLOAD_BYTES` + * on the resilient hook resume path. + */ +export const MAX_RESILIENT_STEP_INPUT_BYTES = 128 * 1024; + +/** + * Whether resilient step dispatch is enabled: the suspension handler + * parallelizes each newly created step's `step_created` event write with its + * step-execution queue publish, carrying the serialized step input in the + * queue message (`stepInput`) so the consumer can idempotently re-ensure the + * event if the direct write failed transiently. Mirrors the resilient start + * (`runInput`) and resilient hook resume (`hookInput`) patterns. + * + * **On by default.** Disable via `WORKFLOW_RESILIENT_STEP_DISPATCH=0` to + * restore the sequential create-then-queue dispatch. + */ +export function isResilientStepDispatchEnabled(): boolean { + return process.env.WORKFLOW_RESILIENT_STEP_DISPATCH !== '0'; +} + const warnedMaxEventsValues = new Set(); /** diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index c2483d6285..7c8d1cfcee 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -1133,6 +1133,45 @@ export function withHealthCheck( }; } +/** FNV-1a 32-bit hash of a string, as 8 hex chars. Tiny, deterministic, and + * dependency-free — used only to scope idempotency keys, not for security. */ +function fnv1a32Hex(value: string): string { + let hash = 0x811c9dc5; + for (let i = 0; i < value.length; i++) { + hash ^= value.charCodeAt(i); + hash = Math.imul(hash, 0x01000193); + } + return (hash >>> 0).toString(16).padStart(8, '0'); +} + +/** + * Idempotency key for a step's background-dispatch queue message, scoped to + * the step's IDENTITY — correlation id plus (hashed) step name — rather than + * the bare correlation id. + * + * The scoping matters for resilient step dispatch under the precondition + * guard: a guard-rejected `step_created` leaves its (revoked) step message in + * flight, and the corrected replay may re-derive the same correlation id for + * a DIFFERENT step. Under a bare-correlationId key the corrected replay's + * dispatch would silently dedupe against the revoked in-flight message — + * which then resolves `skipped` against the re-created entity (the server's + * stepName fence rejects its bare start) — and the legitimate step would + * never be executed. Scoping by step name keeps every dedup property that + * matters (crash recovery re-dispatch, concurrent handlers, the delayed + * retry sharing the suspension re-dispatch's key — all name the same step) + * while letting the corrected schedule's dispatch through. + * + * Every producer of a step-dispatch (or step-retry) message must use this + * key. Cross-version mixing is not a concern: queue messages are pinned to + * the deployment that produced them, so one run never sees two key schemes. + */ +export function stepDispatchIdempotencyKey( + correlationId: string, + stepName: string +): string { + return `${correlationId}:${fnv1a32Hex(stepName)}`; +} + /** * Queues a message to the specified queue with tracing. */ diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index f725cf9ee3..ce88f86906 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -27,9 +27,10 @@ import { ROOT_RUN_ID_ATTRIBUTE, type RunInput, SPEC_VERSION_CURRENT, + SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT, type WorkflowRun, } from '@workflow/world'; -import { classifyRunError } from '../classify-error.js'; +import { classifyRunError, isRetryableWorldError } from '../classify-error.js'; import { runtimeLogger } from '../logger.js'; import { deriveRunPayloadKeys, @@ -47,9 +48,15 @@ import { serializeTraceCarrier } from '../telemetry.js'; import { getInlineOwnershipLeaseSeconds, getMaxInlineSteps, + isResilientStepDispatchEnabled, + MAX_RESILIENT_STEP_INPUT_BYTES, } from './constants.js'; import { getPortLazy } from './get-port-lazy.js'; -import { getWorkflowQueueName, queueMessage } from './helpers.js'; +import { + getWorkflowQueueName, + queueMessage, + stepDispatchIdempotencyKey, +} from './helpers.js'; import { BASELINE_BUNDLE_FILENAME, type PendingAttribute, @@ -121,7 +128,8 @@ async function queueStepMessage(params: { * 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 + * recovery) and uses the step-identity-scoped dispatch key + * (stepDispatchIdempotencyKey) 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 @@ -130,6 +138,14 @@ async function queueStepMessage(params: { * hop is enqueueable. */ purpose: 'dispatch' | `backstop:${string}` | `retry:${number}`; + /** + * Resilient step dispatch: the serialized (possibly encrypted) step input + * to carry on the message as `stepInput`, so the consumer can idempotently + * re-ensure the `step_created` event if the producer's parallel direct + * write failed transiently. Only set on `dispatch` publishes that + * dispatchPendingOps parallelizes with the step_created write. + */ + stepInput?: Uint8Array; wfdiag: (checkpoint: string, fields: Record) => void; }): Promise { const { @@ -141,6 +157,7 @@ async function queueStepMessage(params: { namespace, nextTraceCarrier, purpose, + stepInput, wfdiag, } = params; const traceCarrier = await nextTraceCarrier(); @@ -153,11 +170,17 @@ async function queueStepMessage(params: { stepName: step.stepId, traceCarrier, requestedAt: new Date(), + ...(stepInput !== undefined ? { stepInput: { input: stepInput } } : {}), }, { + // The 'dispatch' key is step-identity-scoped (correlationId + hashed + // step name) — shared with the node engine's dispatch of the same step + // so the two stay mutually exclusive, without a revoked resilient + // message absorbing a reassigned correlation id's legitimate dispatch. + // See stepDispatchIdempotencyKey. idempotencyKey: purpose === 'dispatch' - ? step.correlationId + ? stepDispatchIdempotencyKey(step.correlationId, step.stepId) : `${step.correlationId}:${purpose}`, ...(delaySeconds && delaySeconds > 0 ? { delaySeconds } : {}), } @@ -167,6 +190,7 @@ async function queueStepMessage(params: { correlationId: step.correlationId, purpose, delaySeconds: delaySeconds ?? 0, + ...(stepInput !== undefined ? { resilient: true } : {}), }); } @@ -175,11 +199,19 @@ async function queueStepMessage(params: { * step_created (+ optional queueing), hook_created / hook_received (aborts), * attr_set, hook_disposed, and wait_created events. * - * Steps are created but never queued here — queueing (or inline + * Steps are created but (usually) not 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). + * + * The one exception is resilient step dispatch: for step cids named in + * `queueStepCids` (the caller's overflow steps) that pass the eligibility + * gates, the step_created write is parallelized with the step's queue + * publish — the message carries the serialized input (`stepInput`) so the + * consumer can idempotently re-ensure the event if the direct write failed + * transiently. Steps queued this way are reported in `queuedStepCids`; the + * caller queues the rest itself. */ async function dispatchPendingOps(params: { world: Awaited>; @@ -194,6 +226,13 @@ async function dispatchPendingOps(params: { * instead of both invocations bare-starting the same step. */ skipStepCreation?: Set; + /** + * Step cids the caller intends to hand to the queue this turn (overflow + * steps beyond the inline cap). Eligible ones are published here, in + * parallel with their step_created write (resilient step dispatch), and + * reported back in `queuedStepCids`. + */ + queueStepCids?: Set; /** Queue namespace for all message publishes (see runtime.ts). */ namespace: string | undefined; /** @@ -209,6 +248,8 @@ async function dispatchPendingOps(params: { }): Promise<{ createdAttributeEvent: boolean; createdGetConflictHook: boolean; + /** Step cids already published via resilient dispatch — see above. */ + queuedStepCids: Set; }> { const { world, @@ -220,7 +261,29 @@ async function dispatchPendingOps(params: { nextTraceCarrier, } = params; const skipStepCreation = params.skipStepCreation; + const queueStepCids = params.queueStepCids; const wfdiag = params.wfdiag; + // Step cids published via resilient dispatch below (create + queue in + // parallel, message carrying `stepInput`). Reported to the caller so it + // skips them in its own queueing pass. + const queuedStepCids = new Set(); + // Resilient step dispatch eligibility, shared by every step op below (the + // per-step input-size check is applied inside the op): feature enabled and + // a binary-safe (CBOR) queue transport for the run. + // + // Unlike the node:vm suspension handler's gate (see + // SuspensionHandlerParams.stepDispatch), there is NO precondition-guard + // gate here: this engine's step_created writes are unguarded (no snapshot + // is attached), so a guard-enforcing World can never 412-reject them — + // the consumer's re-ensure therefore cannot materialize a step the guard + // rejected. If this engine ever adopts guarded suspension writes, the + // capability gate from the node:vm handler must be added here too. + const resilientDispatchEligible = + queueStepCids !== undefined && + queueStepCids.size > 0 && + isResilientStepDispatchEnabled() && + (workflowRun.specVersion ?? 0) >= + SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT; // Set when a hook with a parked getConflict() awaiter had its // hook_created written this invocation. The workflow must be re-invoked // so replay can confirm creation and resolve the awaiter. @@ -442,6 +505,93 @@ async function dispatchPendingOps(params: { // access to the CryptoKey, so encryption is applied here // on the host side — matching what // `dehydrateStepArguments` does in the node:vm engine. + const encryptedInput = await encryptSerializedData( + step.input, + encryptionKey + ); + + // Resilient step dispatch: fire the step_created write and the + // step's queue publish in parallel — the message carries the + // same serialized input (`stepInput`) so the consumer can + // idempotently re-ensure the event if the direct write failed + // transiently. Mirrors the node:vm suspension handler and the + // resilient start / resilient hook resume patterns. Only for + // caller-designated overflow steps with inputs the queue + // message can safely carry (binary, under the VQS size cap). + if ( + resilientDispatchEligible && + queueStepCids?.has(step.correlationId) && + encryptedInput instanceof Uint8Array && + encryptedInput.byteLength <= MAX_RESILIENT_STEP_INPUT_BYTES + ) { + const [createResult, queueResult] = await Promise.allSettled([ + world.events.create(runId, { + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: step.correlationId, + eventData: { + stepName: step.stepId, + input: encryptedInput, + }, + }), + queueStepMessage({ + world, + runId, + workflowRun, + step, + namespace, + nextTraceCarrier, + purpose: 'dispatch', + stepInput: encryptedInput, + wfdiag, + }), + ]); + // Queue failure is always fatal for this dispatch pass: + // without the message the step would rely on the create + // alone, and if the create ALSO failed there would be no + // durable record at all. Propagating redelivers the + // orchestrator message, which re-creates the (idempotent) + // step_created and re-dispatches. + if (queueResult.status === 'rejected') { + throw queueResult.reason; + } + queuedStepCids.add(step.correlationId); + if (createResult.status === 'rejected') { + const err = createResult.reason; + if (EntityConflictError.is(err)) { + // Concurrent invocation wrote it first — the message is + // already out; its duplicate publish dedupes on the + // shared step-identity-scoped idempotency key. + return; + } + if (isRetryableWorldError(err)) { + // Resilient: the write failed transiently (429 / 5xx / + // transport) but the step message — carrying the same + // serialized input — was published, so the consumer + // idempotently re-ensures the step_created before + // executing. + runtimeLogger.warn( + 'Step creation event write failed, but the step was ' + + 'dispatched via the queue. The step_created event ' + + 'will be ensured by the queue consumer.', + { + workflowRunId: runId, + correlationId: step.correlationId, + stepName: step.stepId, + error: err instanceof Error ? err.message : String(err), + } + ); + wfdiag('step_resilient_dispatch_recovered', { + stepId: step.stepId, + correlationId: step.correlationId, + }); + return; + } + throw err; + } + return; + } + try { await world.events.create(runId, { eventType: 'step_created', @@ -449,7 +599,7 @@ async function dispatchPendingOps(params: { correlationId: step.correlationId, eventData: { stepName: step.stepId, - input: await encryptSerializedData(step.input, encryptionKey), + input: encryptedInput, }, }); } catch (err) { @@ -457,9 +607,10 @@ async function dispatchPendingOps(params: { throw err; } - // 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). + // NOTE: step queueing is otherwise 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) { @@ -516,7 +667,7 @@ async function dispatchPendingOps(params: { // Per-op dispatch runs in parallel. await Promise.all(opsPromises); - return { createdAttributeEvent, createdGetConflictHook }; + return { createdAttributeEvent, createdGetConflictHook, queuedStepCids }; } /** @@ -1012,6 +1163,20 @@ export async function runWorkflowWithQuickJS(params: { recordedAbortIds.add(op.correlationId); } } + // Steps beyond the inline cap are handed to the queue in the same + // turn their step_created is written. Where eligible, the dispatch + // below parallelizes each overflow step's step_created write with + // its queue publish (resilient step dispatch — the message carries + // `stepInput` so the consumer can re-ensure the event); the rest + // are queued right after, in parallel. This must all 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 + // step-identity-scoped idempotency key makes repeats harmless. + const overflowSteps = freshSteps.slice(inlineCandidates.length); const dispatched = await dispatchPendingOps({ world, runId, @@ -1021,6 +1186,7 @@ export async function runWorkflowWithQuickJS(params: { nextTraceCarrier, pendingOperations: opsToDispatch, skipStepCreation: inlineClaimCids, + queueStepCids: new Set(overflowSteps.map((s) => s.correlationId)), wfdiag, }); if ( @@ -1030,29 +1196,26 @@ export async function runWorkflowWithQuickJS(params: { 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, - }); + for (const cid of dispatched.queuedStepCids) { + queuedStepIds.add(cid); } + await Promise.all( + overflowSteps + .filter((step) => !dispatched.queuedStepCids.has(step.correlationId)) + .map((step) => { + queuedStepIds.add(step.correlationId); + return 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). @@ -1130,7 +1293,7 @@ export async function runWorkflowWithQuickJS(params: { // 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 + // step-identity-scoped idempotency key dedupes against the // original handoff. const nowMs = Date.now(); for (const step of stepOps) { @@ -1322,7 +1485,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, // Suffixed key: this step was inline-claimed, so no dispatch - // publish exists under the bare correlationId — but suffixing + // publish exists under the dispatch key — 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', diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 6a6e883f7c..1d6bd7327e 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -4,11 +4,17 @@ import { PreconditionFailedError, WorkflowWorldError, } from '@workflow/errors'; -import type { Event, WorkflowRun, World } from '@workflow/world'; -import { slotToEventId } from '@workflow/world'; +import type { Event } from '@workflow/world'; +import { + SPEC_VERSION_CURRENT, + slotToEventId, + type ValidQueueName, + type WorkflowRun, + type World, +} from '@workflow/world'; import { describe, expect, it, vi } from 'vitest'; import { WorkflowSuspension } from '../global.js'; -import { maxEventSlot } from './helpers.js'; +import { maxEventSlot, stepDispatchIdempotencyKey } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; import { handleSuspension } from './suspension-handler.js'; @@ -622,6 +628,256 @@ describe('handleSuspension', () => { }); }); +describe('resilient step dispatch', () => { + const queueName = '__wkf_workflow_test-workflow' as ValidQueueName; + + /** A run whose queue transport supports binary payloads (CBOR). */ + const cborRun: WorkflowRun = { ...run, specVersion: SPEC_VERSION_CURRENT }; + + function createQueueWorld(overrides?: { + eventsCreate?: ReturnType; + queue?: ReturnType; + capabilities?: World['capabilities']; + }): { + world: World; + eventsCreate: ReturnType; + queue: ReturnType; + } { + const eventsCreate = + overrides?.eventsCreate ?? + vi.fn().mockImplementation(async (_runId, event) => ({ event })); + const queue = + overrides?.queue ?? vi.fn().mockResolvedValue({ messageId: 'msg_1' }); + const world = { + events: { create: eventsCreate }, + queue, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + ...(overrides?.capabilities + ? { capabilities: overrides.capabilities } + : {}), + } as unknown as World; + return { world, eventsCreate, queue }; + } + + /** Four parallel steps: s1-s3 are lazy-inline (default cap 3), s4 overflows. */ + function fourStepsPending() { + return new Map( + ['s1', 's2', 's3', 's4'].map((id) => [ + id, + { type: 'step' as const, correlationId: id, stepName: id, args: [] }, + ]) + ); + } + + const stepDispatch = () => ({ + queueName, + getTraceCarrier: vi.fn().mockResolvedValue({ traceparent: '00-abc' }), + }); + + it('publishes the overflow step alongside its step_created, carrying stepInput', async () => { + const { world, eventsCreate, queue } = createQueueWorld(); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }); + + // The overflow step is created AND queued by the suspension handler. + expect(eventsCreate).toHaveBeenCalledWith( + run.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's4', + }), + expect.anything() + ); + expect(queue).toHaveBeenCalledTimes(1); + const [calledQueueName, payload, opts] = queue.mock.calls[0]; + expect(calledQueueName).toBe(queueName); + expect(payload).toMatchObject({ + runId: run.runId, + stepId: 's4', + stepName: 's4', + traceCarrier: { traceparent: '00-abc' }, + }); + // The message carries the same serialized input as the direct write. + expect(payload.stepInput.input).toBeInstanceOf(Uint8Array); + const createdInput = eventsCreate.mock.calls.find( + ([, event]) => event.correlationId === 's4' + )?.[1].eventData.input; + expect(payload.stepInput.input).toBe(createdInput); + // Step-identity-scoped key — matches the dispatch key runtime.ts uses for + // the same step, so redundant publishes dedupe. + expect(opts).toMatchObject({ + idempotencyKey: stepDispatchIdempotencyKey('s4', 's4'), + }); + // Reported so the caller skips its own dispatch for this step. + expect([...result.queuedStepCorrelationIds]).toEqual(['s4']); + expect(result.createdStepCorrelationIds).toContain('s4'); + }); + + it('swallows a transient step_created failure once the message is out (resilient)', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => { + if (event.eventType === 'step_created') { + throw new WorkflowWorldError('backend blip', { status: 503 }); + } + return { event }; + }); + const { world, queue } = createQueueWorld({ eventsCreate }); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }); + + // The publish carried the payload, so the consumer re-ensures the event. + expect(queue).toHaveBeenCalledTimes(1); + expect([...result.queuedStepCorrelationIds]).toEqual(['s4']); + // The write did NOT land, so this handler does not claim creation. + expect(result.createdStepCorrelationIds.has('s4')).toBe(false); + }); + + it('propagates a queue publish failure (the message is the durability bar)', async () => { + const queue = vi.fn().mockRejectedValue(new Error('queue down')); + const { world } = createQueueWorld({ queue }); + + await expect( + handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }) + ).rejects.toThrow('queue down'); + }); + + it('propagates a non-retryable step_created failure even when the publish succeeded', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => { + if (event.eventType === 'step_created') { + throw new WorkflowWorldError('bad request', { status: 400 }); + } + return { event }; + }); + const { world } = createQueueWorld({ eventsCreate }); + + await expect( + handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }) + ).rejects.toThrow('bad request'); + }); + + it('falls back to create-only when the world enforces the precondition guard', async () => { + const { world, eventsCreate, queue } = createQueueWorld({ + capabilities: { preconditionGuard: true }, + }); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }); + + // The guarded create can be 412-rejected; a payload-carrying message + // would let the consumer materialize the rejected step. Sequential path: + // create here, caller dispatches. + expect(queue).not.toHaveBeenCalled(); + expect(eventsCreate).toHaveBeenCalledWith( + run.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's4', + }), + expect.anything() + ); + expect(result.queuedStepCorrelationIds.size).toBe(0); + expect(result.createdStepCorrelationIds).toContain('s4'); + }); + + it('stays sequential under an enforced guard regardless of other capabilities', async () => { + // The guard gate is deliberately not liftable by backend-side revocation + // bookkeeping: nothing orders a slow guarded create's eventual 412 before + // the consumer's redelivery re-ensure, so no capability may re-enable the + // payload-carrying publish while creates are guarded. + const { world, queue } = createQueueWorld({ + capabilities: { + preconditionGuard: true, + // Unknown/extra capability flags must not lift the gate. + ...({ resilientStepDispatch: true } as Record), + } as World['capabilities'], + }); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }); + + expect(queue).not.toHaveBeenCalled(); + expect(result.queuedStepCorrelationIds.size).toBe(0); + }); + + it('falls back to create-only when the run predates the CBOR queue transport', async () => { + const { world, queue } = createQueueWorld(); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: { ...run, specVersion: 2 }, + stepDispatch: stepDispatch(), + }); + + expect(queue).not.toHaveBeenCalled(); + expect(result.queuedStepCorrelationIds.size).toBe(0); + }); + + it('falls back to create-only when WORKFLOW_RESILIENT_STEP_DISPATCH=0', async () => { + const prev = process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; + process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = '0'; + try { + const { world, queue } = createQueueWorld(); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + stepDispatch: stepDispatch(), + }); + + expect(queue).not.toHaveBeenCalled(); + expect(result.queuedStepCorrelationIds.size).toBe(0); + } finally { + if (prev === undefined) { + delete process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; + } else { + process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = prev; + } + } + }); + + it('never queues from here when no stepDispatch is provided (terminal drain)', async () => { + const { world, queue } = createQueueWorld(); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(fourStepsPending(), globalThis), + world, + run: cborRun, + }); + + expect(queue).not.toHaveBeenCalled(); + expect(result.queuedStepCorrelationIds.size).toBe(0); + }); +}); + describe('retainedStepInputsSafe (serialization passivity gate)', () => { function stepPending(args: unknown[]) { return new Map([ diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index b02826df4c..13363a196c 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -14,10 +14,14 @@ import { type EventResult, type SerializedData, SPEC_VERSION_CURRENT, + SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT, SPEC_VERSION_SUPPORTS_COMPRESSION, + type TraceCarrier, + type ValidQueueName, type WorkflowRun, type World, } from '@workflow/world'; +import { isRetryableWorldError } from '../classify-error.js'; import { importKey } from '../encryption.js'; import type { AttributeInvocationQueueItem, @@ -31,12 +35,19 @@ import type { GuestCodeStats } from '../serialization/hardened.js'; import { dehydrateStepArguments } from '../serialization.js'; import * as Attribute from '../telemetry/semantic-conventions.js'; import { getAbortStreamIdFromToken } from '../util.js'; -import { getMaxInlineSteps } from './constants.js'; +import { + getMaxInlineSteps, + isResilientStepDispatchEnabled, + MAX_RESILIENT_STEP_INPUT_BYTES, +} from './constants.js'; import { type EventCreator, + isPreconditionGuardEnabled, type LoadedEventLog, mergeReportedEvents, preconditionSnapshotParams, + queueMessage, + stepDispatchIdempotencyKey, } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; @@ -68,6 +79,26 @@ export interface SuspensionHandlerParams { runReadyBarrier?: Promise; /** One-shot telemetry reporter, activated only after replay has recovered. */ replayRecoveryReporter?: ReplayRecoveryReporter; + /** + * Resilient step dispatch: when provided (and the per-step eligibility gates + * pass — see the step ops below), each newly created non-inline step's + * `step_created` write is parallelized with its step-execution queue + * publish, and the queue message carries the serialized step input + * (`stepInput`) so the consumer can idempotently re-ensure the event if the + * direct write failed transiently. Steps queued this way are reported in + * {@link SuspensionHandlerResult.queuedStepCorrelationIds} so the caller + * skips them in its own dispatch pass. Omitted by callers that must not + * queue (terminal drain, tests) — creates then behave exactly as before. + */ + stepDispatch?: { + /** The unified workflow queue this run's step messages are published to. */ + queueName: ValidQueueName; + /** + * Lazily resolves the trace carrier to stamp on the step messages. + * Called at most once per suspension (memoized here). + */ + getTraceCarrier: () => Promise; + }; } /** @@ -86,6 +117,16 @@ export interface SuspensionHandlerResult { * into the same batch boundary. */ createdStepCorrelationIds: Set; + /** + * Correlation IDs of steps this suspension call already published + * step-execution queue messages for, via resilient step dispatch (the + * `step_created` write parallelized with a `stepInput`-carrying queue + * publish). The caller MUST NOT dispatch these again — the message is + * already out (a duplicate would be deduped by its idempotency key, but + * costs a wasted round-trip). Empty when {@link SuspensionHandlerParams.stepDispatch} + * was not provided or no step was eligible. + */ + queuedStepCorrelationIds: Set; /** * How many events this phase's writes reported back as occupying slots they * skipped over, already merged into the caller's `eventLog.events`. Nonzero @@ -242,6 +283,7 @@ export async function handleSuspension({ eventLog, runReadyBarrier, replayRecoveryReporter, + stepDispatch, }: SuspensionHandlerParams): Promise { const runId = run.runId; @@ -638,7 +680,56 @@ export async function handleSuspension({ const ops: Promise[] = []; - // Steps: create step_created events (no queuing — V2 returns pending steps to caller) + // Correlation IDs of steps whose step-execution queue message was already + // published by the resilient-dispatch ops below (alongside the step_created + // write). Reported to the caller so its dispatch pass skips them. + const queuedStepCorrelationIds = new Set(); + + // Resilient step dispatch eligibility, shared by every step op below (the + // per-step input-size check is applied inside the op). All must hold: + // + // - The caller provided a dispatch target (`stepDispatch`) — terminal + // drains and other create-only callers never queue. + // - The feature is enabled (`WORKFLOW_RESILIENT_STEP_DISPATCH` opt-out). + // - The optimistic-concurrency guard is not in effect. A guard-enforcing + // backend can reject the step_created as stale (412) and the caller then + // restarts the replay — but a queue message carrying the payload would + // already be out, letting the consumer materialize a step the guard + // rejected. This gate is deliberately NOT liftable by backend-side + // revocation bookkeeping: nothing orders a slow create's eventual 412 + // (which is when the backend learns the dispatch is poisoned) before the + // consumer's redelivery re-ensure, and a best-effort marker that fails + // open cannot carry a correctness property. The sequential path is the + // only thing that gives the message a happens-after edge over its + // create's guard verdict. + // - The run's queue transport preserves binary payloads (CBOR, + // specVersion >= 3): `stepInput.input` is the serialized (possibly + // encrypted) input bytes, which the JSON transport would mangle. + const resilientDispatchEligible = + stepDispatch !== undefined && + isResilientStepDispatchEnabled() && + !( + isPreconditionGuardEnabled() && + world.capabilities?.preconditionGuard === true + ) && + (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT; + + // The trace carrier for resilient step dispatches, resolved at most once per + // suspension (the per-step ops run concurrently and share it). + let stepDispatchTraceCarrier: Promise | undefined; + const getStepDispatchTraceCarrier = (): Promise => { + stepDispatchTraceCarrier ??= + stepDispatch?.getTraceCarrier() ?? Promise.resolve({}); + return stepDispatchTraceCarrier; + }; + + // Producer-side resilient recovery count for the suspension span attribute. + let resilientDispatchRecovered = 0; + + // Steps: create step_created events (no queuing — V2 returns pending steps + // to caller — EXCEPT on the resilient dispatch path, which parallelizes the + // create with the step's queue publish and reports it in + // `queuedStepCorrelationIds`). for (const queueItem of stepItems) { if (stepsNeedingCreation.has(queueItem.correlationId)) { ops.push( @@ -685,6 +776,95 @@ export async function handleSuspension({ input: dehydratedInput as SerializedData, }, }; + + // Resilient step dispatch: fire the step_created write and the + // step-execution queue publish in parallel — the message carries the + // same serialized input (`stepInput`) so the consumer can + // idempotently re-ensure the event if the direct write failed + // transiently. Mirrors the resilient start (`runInput`) and + // resilient hook resume (`hookInput`) patterns. Only for inputs the + // queue message can safely carry (binary, under the VQS size cap). + if ( + resilientDispatchEligible && + dehydratedInput instanceof Uint8Array && + dehydratedInput.byteLength <= MAX_RESILIENT_STEP_INPUT_BYTES + ) { + await ensureRunReady(); + const traceCarrier = await getStepDispatchTraceCarrier(); + const [createResult, queueResult] = await Promise.allSettled([ + createGuarded(stepEvent, { requestId }), + queueMessage( + world, + // biome-ignore lint/style/noNonNullAssertion: implied by resilientDispatchEligible + stepDispatch!.queueName, + { + runId, + stepId: queueItem.correlationId, + stepName: queueItem.stepName, + traceCarrier, + requestedAt: new Date(), + stepInput: { input: dehydratedInput }, + }, + // Same key as the caller's dispatch pass and any concurrent + // handler's — redundant publishes for this step dedupe. The + // key is step-identity-scoped so a revoked message for a + // reassigned correlation id cannot absorb the corrected + // schedule's dispatch — see stepDispatchIdempotencyKey. + { + idempotencyKey: stepDispatchIdempotencyKey( + queueItem.correlationId, + queueItem.stepName + ), + } + ), + ]); + // Queue failure is always fatal for this suspension pass: without + // the message the step would rely on the create alone, and if the + // create ALSO failed there would be no durable record at all. + // Propagating redelivers the orchestrator message, which + // re-creates the (idempotent) step_created and re-dispatches — + // the same recovery as the sequential path. + if (queueResult.status === 'rejected') { + throw queueResult.reason; + } + queuedStepCorrelationIds.add(queueItem.correlationId); + if (createResult.status === 'rejected') { + const err = createResult.reason; + if (EntityConflictError.is(err)) { + // Concurrent handler wrote it first — same as the sequential + // path. The step message is already out; a duplicate publish + // by that handler dedupes on the shared idempotency key. + runtimeLogger.info('Step already exists, continuing', { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: err.message, + }); + } else if (isRetryableWorldError(err)) { + // Resilient: the write failed transiently (429 / 5xx / + // transport) but the step message — carrying the same + // serialized input — was published, so the consumer + // idempotently re-ensures the step_created before executing. + resilientDispatchRecovered++; + runtimeLogger.warn( + 'Step creation event write failed, but the step was ' + + 'dispatched via the queue. The step_created event will ' + + 'be ensured by the queue consumer.', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + stepName: queueItem.stepName, + error: err instanceof Error ? err.message : String(err), + } + ); + } else { + throw err; + } + } else { + createdStepCorrelationIds.add(queueItem.correlationId); + } + return; + } + try { await ensureRunReady(); await createGuarded(stepEvent, { requestId }); @@ -849,11 +1029,15 @@ export async function handleSuspension({ ...Attribute.WorkflowStepsCreated(stepItems.length), ...Attribute.WorkflowHooksCreated(hooksNeedingCreation.length), ...Attribute.WorkflowWaitsCreated(waitItems.length), + ...(resilientDispatchRecovered > 0 + ? Attribute.StepResilientDispatchRecovered(resilientDispatchRecovered) + : {}), }); return { pendingSteps: stepItems, createdStepCorrelationIds, + queuedStepCorrelationIds, lazyInlineSteps, // On hook conflict the caller re-invokes immediately and never reads // the wait timeout, so don't report one. diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index c77ec954a2..a6ef81c3ed 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -1,6 +1,7 @@ import { RuntimeDecryptionError, SerializationError, + StreamExpiredError, WorkflowRuntimeError, } from '@workflow/errors'; import { once } from '@workflow/utils'; @@ -945,7 +946,11 @@ export function createReconnectingFramedStream( try { await connect(); return; - } catch { + } catch (error) { + // Retention expiry is terminal and retrying cannot restore the stream. + // Preserve the typed error immediately instead of turning one 410 into + // 50 reconnect attempts and a generic budget-exhaustion error. + if (StreamExpiredError.is(error)) throw error; // Reopen failed transiently; loop to retry, counting against the // budget so a server that never recovers still terminates the stream. } diff --git a/packages/core/src/serialization/codec-devalue.ts b/packages/core/src/serialization/codec-devalue.ts index a7ac8e0037..983574f6e5 100644 --- a/packages/core/src/serialization/codec-devalue.ts +++ b/packages/core/src/serialization/codec-devalue.ts @@ -10,7 +10,13 @@ * typed arrays, Map, Set natively). */ -import { parse, stringify, unflatten } from 'devalue'; +import { + defaultParseOperations, + type ParseOptions, + parse, + stringify, + unflatten, +} from 'devalue'; import type { Codec, CodecOptions, SerializationMode } from './codec.js'; import { hardenedStringifyOperations, withGuestCodeStats } from './hardened.js'; import { getClassReducers, getClassRevivers } from './reducers/class.js'; @@ -23,6 +29,20 @@ import { type Reducers, type Revivers, SerializationFormat } from './types.js'; const encoder = new TextEncoder(); const decoder = new TextDecoder(); +const MAX_SPARSE_ARRAY_LENGTH = 100_000; + +const parseOptions = { + operations: { + createSparseArray(length) { + if (length > MAX_SPARSE_ARRAY_LENGTH) { + throw new RangeError( + `Sparse array length ${length} exceeds the maximum of ${MAX_SPARSE_ARRAY_LENGTH}` + ); + } + return defaultParseOperations.createSparseArray(length); + }, + }, +} satisfies ParseOptions; // ---- Reducer/Reviver composition per mode ---- @@ -136,7 +156,7 @@ export const devalueCodec: Codec = { options?.extraRevivers ); const str = decoder.decode(data); - return parse(str, revivers); + return parse(str, revivers, parseOptions); }, deserializeLegacy( @@ -149,6 +169,6 @@ export const devalueCodec: Codec = { options?.global, options?.extraRevivers ); - return unflatten(data as any[], revivers); + return unflatten(data as any[], revivers, parseOptions); }, }; diff --git a/packages/core/src/serialization/serialization.test.ts b/packages/core/src/serialization/serialization.test.ts index d711dec538..4a9a07afbb 100644 --- a/packages/core/src/serialization/serialization.test.ts +++ b/packages/core/src/serialization/serialization.test.ts @@ -1151,6 +1151,17 @@ describe('devalue codec', () => { expect(result).toEqual(value); }); + it('should reject oversized sparse arrays while decoding', () => { + const payload = new TextEncoder().encode('[[-7,4294967295]]'); + + expect(() => devalueCodec.deserialize(payload, 'workflow')).toThrow( + /exceeds the maximum/ + ); + expect(() => + devalueCodec.deserializeLegacy!([[-7, 4_294_967_295]], 'workflow') + ).toThrow(/exceeds the maximum/); + }); + it('should produce Uint8Array output from serialize', () => { const serialized = devalueCodec.serialize(42, 'workflow'); expect(serialized).toBeInstanceOf(Uint8Array); diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index f4fc7cbf4f..63bc0ebc79 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -471,6 +471,26 @@ export const HookResumeSetupSource = SemanticConvention( 'workflow.resume_setup_source' ); +/** + * Producer-side signal (on the suspension span) counting steps whose direct + * `step_created` write failed transiently while their `stepInput`-carrying + * queue publish succeeded, so step creation is recovered via the consumer's + * re-ensure. Mirrors {@link HookResilientResume}. + */ +export const StepResilientDispatchRecovered = SemanticConvention( + 'workflow.step.resilient_dispatch_recovered' +); + +/** + * Consumer-side signal (on the workflow execution span) that this delivery + * materialized the `step_created` event from the queue message's `stepInput` + * because the producer's direct write had not landed — the completion of the + * recovery path {@link StepResilientDispatchRecovered} began. + */ +export const StepResilientDispatchMaterialized = SemanticConvention( + 'workflow.step.resilient_dispatch_materialized' +); + // Webhook attributes /** Number of webhook handlers triggered */ diff --git a/packages/errors/src/index.ts b/packages/errors/src/index.ts index cc3157a2d6..fb53982131 100644 --- a/packages/errors/src/index.ts +++ b/packages/errors/src/index.ts @@ -789,6 +789,29 @@ export class RunExpiredError extends WorkflowWorldError { } } +/** + * Thrown when a stream is no longer readable because its owning run passed its + * storage-retention boundary. This is terminal: retrying cannot restore data. + * + * Unlike {@link RunExpiredError}, this identifies the failed stream read and + * exposes the server's authoritative expiry timestamp for user-facing errors. + */ +export class StreamExpiredError extends WorkflowWorldError { + constructor( + message: string, + readonly runId?: string, + readonly streamId?: string, + readonly expiredAt?: Date + ) { + super(message, { status: 410, code: 'stream-expired' }); + this.name = 'StreamExpiredError'; + } + + static is(value: unknown): value is StreamExpiredError { + return isError(value) && value.name === 'StreamExpiredError'; + } +} + /** * Thrown when an operation cannot proceed because a required timestamp * (e.g. retryAfter) has not been reached yet. diff --git a/packages/nest/CHANGELOG.md b/packages/nest/CHANGELOG.md index 876956d27e..04f3419fe7 100644 --- a/packages/nest/CHANGELOG.md +++ b/packages/nest/CHANGELOG.md @@ -1,5 +1,12 @@ # @workflow/nest +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/builders@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/nest/package.json b/packages/nest/package.json index 66cbb45073..ae17723391 100644 --- a/packages/nest/package.json +++ b/packages/nest/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/nest", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "NestJS integration for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/next/CHANGELOG.md b/packages/next/CHANGELOG.md index 2217e761ee..d0722b3258 100644 --- a/packages/next/CHANGELOG.md +++ b/packages/next/CHANGELOG.md @@ -1,5 +1,15 @@ # @workflow/next +## 5.0.0-beta.41 + +### Patch Changes + +- [#3084](https://github.com/vercel/workflow/pull/3084) [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727) Thanks [@shalabhc](https://github.com/shalabhc)! - Add an opt-in WebSocket transport for event writes during step execution, enabled with `WORKFLOW_EVENTS_TRANSPORT=ws`. Defaults to HTTP; no behavior change unless explicitly enabled. + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/builders@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/next/package.json b/packages/next/package.json index 5a472301cb..1872e1961b 100644 --- a/packages/next/package.json +++ b/packages/next/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/next", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Next.js integration for Workflow SDK", "type": "commonjs", "main": "dist/index.js", diff --git a/packages/nitro/CHANGELOG.md b/packages/nitro/CHANGELOG.md index 4cccd616a1..2071f33d46 100644 --- a/packages/nitro/CHANGELOG.md +++ b/packages/nitro/CHANGELOG.md @@ -1,5 +1,16 @@ # @workflow/nitro +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/builders@5.0.0-beta.41 + - @workflow/rollup@5.0.0-beta.41 + - @workflow/web@5.0.0-beta.41 + - @workflow/vite@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/nitro/package.json b/packages/nitro/package.json index 75651798fd..21d2311ab9 100644 --- a/packages/nitro/package.json +++ b/packages/nitro/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/nitro", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Nitro integration for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/nuxt/CHANGELOG.md b/packages/nuxt/CHANGELOG.md index bf3f4c438a..46e99faae6 100644 --- a/packages/nuxt/CHANGELOG.md +++ b/packages/nuxt/CHANGELOG.md @@ -1,5 +1,12 @@ # @workflow/nuxt +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies []: + - @workflow/nitro@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/nuxt/package.json b/packages/nuxt/package.json index 6b342ba87f..7971994180 100644 --- a/packages/nuxt/package.json +++ b/packages/nuxt/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/nuxt", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Nuxt integration for Workflow SDK", "license": "Apache-2.0", "type": "module", diff --git a/packages/rollup/CHANGELOG.md b/packages/rollup/CHANGELOG.md index bf0b728ee8..8204beb962 100644 --- a/packages/rollup/CHANGELOG.md +++ b/packages/rollup/CHANGELOG.md @@ -1,5 +1,14 @@ # @workflow/rollup +## 5.0.0-beta.41 + +### Patch Changes + +- [#3084](https://github.com/vercel/workflow/pull/3084) [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727) Thanks [@shalabhc](https://github.com/shalabhc)! - Add an opt-in WebSocket transport for event writes during step execution, enabled with `WORKFLOW_EVENTS_TRANSPORT=ws`. Defaults to HTTP; no behavior change unless explicitly enabled. + +- Updated dependencies [[`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/builders@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/rollup/package.json b/packages/rollup/package.json index eb7aaeaa3b..844a197cb6 100644 --- a/packages/rollup/package.json +++ b/packages/rollup/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/rollup", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Rollup plugin for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/sveltekit/CHANGELOG.md b/packages/sveltekit/CHANGELOG.md index 21ea2aa7db..058f85bf62 100644 --- a/packages/sveltekit/CHANGELOG.md +++ b/packages/sveltekit/CHANGELOG.md @@ -1,5 +1,14 @@ # @workflow/sveltekit +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/builders@5.0.0-beta.41 + - @workflow/rollup@5.0.0-beta.41 + - @workflow/vite@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/sveltekit/package.json b/packages/sveltekit/package.json index c5d6527677..4c0c4bd817 100644 --- a/packages/sveltekit/package.json +++ b/packages/sveltekit/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/sveltekit", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "SvelteKit integration for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/vite/CHANGELOG.md b/packages/vite/CHANGELOG.md index 1093745e6c..b62cc17d69 100644 --- a/packages/vite/CHANGELOG.md +++ b/packages/vite/CHANGELOG.md @@ -1,5 +1,12 @@ # @workflow/vite +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/builders@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/vite/package.json b/packages/vite/package.json index 0f5347117d..45c9649257 100644 --- a/packages/vite/package.json +++ b/packages/vite/package.json @@ -1,7 +1,7 @@ { "name": "@workflow/vite", "description": "Vite plugin for Workflow SDK", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "type": "module", "main": "dist/index.js", "files": [ diff --git a/packages/vitest/CHANGELOG.md b/packages/vitest/CHANGELOG.md index 95d7ef307f..24dc373022 100644 --- a/packages/vitest/CHANGELOG.md +++ b/packages/vitest/CHANGELOG.md @@ -1,5 +1,16 @@ # @workflow/vitest +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`4ec7aca`](https://github.com/vercel/workflow/commit/4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/world@5.0.0-beta.26 + - @workflow/world-local@5.0.0-beta.35 + - @workflow/builders@5.0.0-beta.41 + - @workflow/rollup@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/vitest/package.json b/packages/vitest/package.json index 76349576cc..dda4582201 100644 --- a/packages/vitest/package.json +++ b/packages/vitest/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/vitest", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Vitest plugin for testing Workflow SDK workflows", "type": "module", "main": "./dist/index.js", diff --git a/packages/web-shared/CHANGELOG.md b/packages/web-shared/CHANGELOG.md index 28b920280c..44d24e69ef 100644 --- a/packages/web-shared/CHANGELOG.md +++ b/packages/web-shared/CHANGELOG.md @@ -1,5 +1,13 @@ # @workflow/web-shared +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/world@5.0.0-beta.26 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/web-shared/package.json b/packages/web-shared/package.json index 920b3d3b54..2ac0237e90 100644 --- a/packages/web-shared/package.json +++ b/packages/web-shared/package.json @@ -1,7 +1,7 @@ { "name": "@workflow/web-shared", "description": "Shared components for Workflow Observability UI", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "private": false, "files": [ "dist", diff --git a/packages/web/CHANGELOG.md b/packages/web/CHANGELOG.md index 6d2b698fb6..6324319612 100644 --- a/packages/web/CHANGELOG.md +++ b/packages/web/CHANGELOG.md @@ -1,5 +1,12 @@ # @workflow/web +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/world-local@5.0.0-beta.35 + ## 5.0.0-beta.40 ### Minor Changes diff --git a/packages/web/package.json b/packages/web/package.json index 652dd15a11..b25c89b751 100644 --- a/packages/web/package.json +++ b/packages/web/package.json @@ -1,7 +1,7 @@ { "name": "@workflow/web", "description": "Workflow Observability UI", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "type": "module", "private": false, "files": [ diff --git a/packages/workflow/CHANGELOG.md b/packages/workflow/CHANGELOG.md index a66e1fa8e2..6ae9483730 100644 --- a/packages/workflow/CHANGELOG.md +++ b/packages/workflow/CHANGELOG.md @@ -1,5 +1,22 @@ # workflow +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/rollup@5.0.0-beta.41 + - @workflow/next@5.0.0-beta.41 + - @workflow/cli@5.0.0-beta.41 + - @workflow/nitro@5.0.0-beta.41 + - @workflow/typescript-plugin@5.0.0-beta.5 + - @workflow/errors@5.0.0-beta.16 + - @workflow/astro@5.0.0-beta.41 + - @workflow/nest@5.0.0-beta.41 + - @workflow/sveltekit@5.0.0-beta.41 + - @workflow/nuxt@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Minor Changes diff --git a/packages/workflow/package.json b/packages/workflow/package.json index 2b8d2423bc..ed72924759 100644 --- a/packages/workflow/package.json +++ b/packages/workflow/package.json @@ -1,6 +1,6 @@ { "name": "workflow", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Workflow SDK - Build durable, resilient, and observable workflows", "main": "dist/typescript-plugin.cjs", "type": "module", diff --git a/packages/world-local/CHANGELOG.md b/packages/world-local/CHANGELOG.md index d445cfae7c..251a003103 100644 --- a/packages/world-local/CHANGELOG.md +++ b/packages/world-local/CHANGELOG.md @@ -1,5 +1,21 @@ # @workflow/world-local +## 5.0.0-beta.35 + +### Patch Changes + +- [#3382](https://github.com/vercel/workflow/pull/3382) [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip + +- [#3389](https://github.com/vercel/workflow/pull/3389) [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - **Breaking**: SpecVersion 6: Event IDs are now a dense per-run slot number, allocated by the world at publish time so a rejected write leaves no gap in the event log. A replay tells the world how many events it had read and gets back the ones it did not see, so an event that arrives from outside the replay and lands ahead of an event the replay wrote no longer fails the run with `CORRUPTED_EVENT_LOG`: it is held for whichever part of the workflow awaits it. A gap in the numbering fails the run instead of being replayed over. + +- [#3205](https://github.com/vercel/workflow/pull/3205) [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load replay-event suffixes through one World request. + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Infer required run and step entities, including their start times, from + `events.create` request types. +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/world@5.0.0-beta.26 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.34 ### Major Changes diff --git a/packages/world-local/package.json b/packages/world-local/package.json index 9d4e65bd1f..0e40a11f0f 100644 --- a/packages/world-local/package.json +++ b/packages/world-local/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/world-local", - "version": "5.0.0-beta.34", + "version": "5.0.0-beta.35", "description": "Local development World implementation for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/world-postgres/CHANGELOG.md b/packages/world-postgres/CHANGELOG.md index 8cc59f599d..e08f0133b2 100644 --- a/packages/world-postgres/CHANGELOG.md +++ b/packages/world-postgres/CHANGELOG.md @@ -1,5 +1,24 @@ # @workflow/world-postgres +## 5.0.0-beta.33 + +### Patch Changes + +- [#3385](https://github.com/vercel/workflow/pull/3385) [`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Retry replay timeouts through normal queue redelivery instead of exiting the process, and keep Postgres jobs retryable through Core's terminal delivery limit. + +- [#3382](https://github.com/vercel/workflow/pull/3382) [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip + +- [#3389](https://github.com/vercel/workflow/pull/3389) [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - **Breaking**: SpecVersion 6: Event IDs are now a dense per-run slot number, allocated by the world at publish time so a rejected write leaves no gap in the event log. A replay tells the world how many events it had read and gets back the ones it did not see, so an event that arrives from outside the replay and lands ahead of an event the replay wrote no longer fails the run with `CORRUPTED_EVENT_LOG`: it is held for whichever part of the workflow awaits it. A gap in the numbering fails the run instead of being replayed over. + +- [#3205](https://github.com/vercel/workflow/pull/3205) [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load replay-event suffixes through one World request. + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Infer required run and step entities, including their start times, from + `events.create` request types. +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/world@5.0.0-beta.26 + - @workflow/world-local@5.0.0-beta.35 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.32 ### Major Changes diff --git a/packages/world-postgres/package.json b/packages/world-postgres/package.json index ff01c66a7a..3da50d6555 100644 --- a/packages/world-postgres/package.json +++ b/packages/world-postgres/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/world-postgres", - "version": "5.0.0-beta.32", + "version": "5.0.0-beta.33", "description": "A reference World implementation based on PostgreSQL", "type": "module", "main": "dist/index.js", diff --git a/packages/world-testing/CHANGELOG.md b/packages/world-testing/CHANGELOG.md index 7e96429ea2..be7810c28d 100644 --- a/packages/world-testing/CHANGELOG.md +++ b/packages/world-testing/CHANGELOG.md @@ -1,5 +1,15 @@ # @workflow/world-testing +## 5.0.0-beta.41 + +### Patch Changes + +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f), [`1a64f68`](https://github.com/vercel/workflow/commit/1a64f684723757c5a839abb94189b953dd3ac536), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`eb9e13f`](https://github.com/vercel/workflow/commit/eb9e13fd23eb12e353cd8f53ed4357da06f8e5ac), [`19b5b85`](https://github.com/vercel/workflow/commit/19b5b85c8b78cfd6c8ebbdf62ae7b8241a109595), [`fbebf71`](https://github.com/vercel/workflow/commit/fbebf7104d97219b73b6a51b0e77e42c45cdd99c), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`69c30ff`](https://github.com/vercel/workflow/commit/69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/core@5.0.0-beta.41 + - @workflow/world@5.0.0-beta.26 + - @workflow/cli@5.0.0-beta.41 + - workflow@5.0.0-beta.41 + ## 5.0.0-beta.40 ### Patch Changes diff --git a/packages/world-testing/package.json b/packages/world-testing/package.json index 03fddebff9..cf9c1f0e32 100644 --- a/packages/world-testing/package.json +++ b/packages/world-testing/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/world-testing", - "version": "5.0.0-beta.40", + "version": "5.0.0-beta.41", "description": "Testing utilities and World implementation for Workflow SDK", "main": "dist/src/index.mjs", "files": [ diff --git a/packages/world-vercel/CHANGELOG.md b/packages/world-vercel/CHANGELOG.md index b88ebc75b2..ee05db99a9 100644 --- a/packages/world-vercel/CHANGELOG.md +++ b/packages/world-vercel/CHANGELOG.md @@ -1,5 +1,33 @@ # @workflow/world-vercel +## 5.0.0-beta.37 + +### Minor Changes + +- [#3286](https://github.com/vercel/workflow/pull/3286) [`4bb86d3`](https://github.com/vercel/workflow/commit/4bb86d305423a2da813cbef8b7d6fe4421288bba) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Support Hook minimum retention on Vercel. + +- [#3084](https://github.com/vercel/workflow/pull/3084) [`264ddff`](https://github.com/vercel/workflow/commit/264ddff67b3cfceea24235ee6d865c48e7982727) Thanks [@shalabhc](https://github.com/shalabhc)! - Add an opt-in WebSocket transport for event writes during step execution, enabled with `WORKFLOW_EVENTS_TRANSPORT=ws`. Defaults to HTTP; no behavior change unless explicitly enabled. + +### Patch Changes + +- [#3385](https://github.com/vercel/workflow/pull/3385) [`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Retry replay timeouts through normal queue redelivery instead of exiting the process, and keep Postgres jobs retryable through Core's terminal delivery limit. + +- [#3382](https://github.com/vercel/workflow/pull/3382) [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip + +- [#3370](https://github.com/vercel/workflow/pull/3370) [`bf4dda6`](https://github.com/vercel/workflow/commit/bf4dda64780f5f8e13840871960cf2a647cdd329) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Recover from HTTP/2 connections that stop delivering event requests, and make `WORKFLOW_H2_MULTIPLEX=0` disable HTTP/2 on the events path entirely + +- [#3389](https://github.com/vercel/workflow/pull/3389) [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - **Breaking**: SpecVersion 6: Event IDs are now a dense per-run slot number, allocated by the world at publish time so a rejected write leaves no gap in the event log. A replay tells the world how many events it had read and gets back the ones it did not see, so an event that arrives from outside the replay and lands ahead of an event the replay wrote no longer fails the run with `CORRUPTED_EVENT_LOG`: it is held for whichever part of the workflow awaits it. A gap in the numbering fails the run instead of being replayed over. + +- [#3205](https://github.com/vercel/workflow/pull/3205) [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load replay-event suffixes through one World request. + +- [#3191](https://github.com/vercel/workflow/pull/3191) [`665110b`](https://github.com/vercel/workflow/commit/665110b3a255c920a3612dec521234efcf34d4f8) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load the current replay log from non-Turbo `run_started` responses while preserving Turbo's preload opt-out. + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Infer required run and step entities, including their start times, from + `events.create` request types. +- Updated dependencies [[`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef), [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea), [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040), [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d), [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f), [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f)]: + - @workflow/world@5.0.0-beta.26 + - @workflow/errors@5.0.0-beta.16 + ## 5.0.0-beta.36 ### Major Changes diff --git a/packages/world-vercel/package.json b/packages/world-vercel/package.json index 425642aa46..bc0ee97fb6 100644 --- a/packages/world-vercel/package.json +++ b/packages/world-vercel/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/world-vercel", - "version": "5.0.0-beta.36", + "version": "5.0.0-beta.37", "description": "Vercel platform World implementation for Workflow SDK", "type": "module", "main": "dist/index.js", diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index f68f35c880..0b6eb2b867 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -264,6 +264,13 @@ interface CreateEventV4InputBase { * digest on the server's `(runId, resumeId)` constraint (the v4 payload ref * is not content-stable server-side). Older servers ignore it. */ resumePayloadDigest?: string; + /** Marks a `step_created` as the queue consumer's re-ensure of a resilient + * step dispatch (`stepInput`-carrying step message). Advisory — see + * CreateEventParams.viaStepDispatch in @workflow/world: the server MAY + * refuse it with 410 (`step-dispatch-revoked` → RunExpiredError) as + * defense-in-depth when it recorded a 412 rejection for this correlation + * id and no step entity exists. Older servers ignore it. */ + viaStepDispatch?: boolean; } export type CreateEventV4Input = CreateEventV4InputBase & @@ -470,6 +477,9 @@ function buildPostFrameMeta( if (input.resumePayloadDigest !== undefined) { meta.resumePayloadDigest = input.resumePayloadDigest; } + if (input.viaStepDispatch !== undefined) { + meta.viaStepDispatch = input.viaStepDispatch; + } return meta; } diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index d5b981cef1..1e8c482844 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -625,6 +625,11 @@ async function createWorkflowRunEventInner( ...(params?.resumePayloadDigest ? { resumePayloadDigest: params.resumePayloadDigest } : {}), + // Resilient step dispatch re-ensure marker (step_created only). Advisory + // — the server MAY refuse it with 410 → RunExpiredError as + // defense-in-depth when it recorded a 412 rejection for this correlation + // id and no step entity exists. + ...(params?.viaStepDispatch ? { viaStepDispatch: true } : {}), remoteRefBehavior, payload, ...meta, diff --git a/packages/world-vercel/src/http-core.test.ts b/packages/world-vercel/src/http-core.test.ts index ed137db12d..f38cd90d6d 100644 --- a/packages/world-vercel/src/http-core.test.ts +++ b/packages/world-vercel/src/http-core.test.ts @@ -1,6 +1,7 @@ import { EntityConflictError, RunExpiredError, + StreamExpiredError, ThrottleError, TooEarlyError, WorkflowWorldError, @@ -26,10 +27,29 @@ describe('errorForResponse', () => { expect(errorForResponse(409, 'boom')).toBeInstanceOf(EntityConflictError); }); - it('maps 410 to RunExpiredError', () => { + it('maps an unstructured 410 to RunExpiredError', () => { expect(errorForResponse(410, 'boom')).toBeInstanceOf(RunExpiredError); }); + it('maps a stream-expired 410 to StreamExpiredError with its details', () => { + const err = errorForResponse(410, 'stream expired', { + code: 'stream-expired', + details: { + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt: '2026-08-10T14:40:00.000Z', + }, + }); + + expect(err).toMatchObject({ + name: 'StreamExpiredError', + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt: new Date('2026-08-10T14:40:00.000Z'), + }); + expect(err).toBeInstanceOf(StreamExpiredError); + }); + it('maps 425 to TooEarlyError carrying retryAfter', () => { const err = errorForResponse(425, 'too early', { retryAfter: 7 }); expect(err).toBeInstanceOf(TooEarlyError); diff --git a/packages/world-vercel/src/http-core.ts b/packages/world-vercel/src/http-core.ts index 4e0fe0a07a..7acbf1f3c6 100644 --- a/packages/world-vercel/src/http-core.ts +++ b/packages/world-vercel/src/http-core.ts @@ -21,6 +21,7 @@ import { EntityConflictError, PreconditionFailedError, RunExpiredError, + StreamExpiredError, ThrottleError, TooEarlyError, WorkflowWorldError, @@ -172,7 +173,8 @@ export function headersToRecord(headers: Headers): Record { * truth for the status → error-type contract the runtime branches on: * * - 409 → EntityConflictError (start() dedupe, terminal-state transitions) - * - 410 → RunExpiredError (runtime exits without retrying) + * - 410 → StreamExpiredError when the response code is `stream-expired`, + * otherwise RunExpiredError (both terminal) * - 412 → PreconditionFailedError + retryAfter + details (stale precondition * snapshot — the optimistic-concurrency guard on event creation; `details` * carries the events the backend returned inline, when it did) @@ -195,15 +197,41 @@ export function errorForResponse( code?: string; url?: string; mitigated?: string | null; - /** Rejection detail for a 412 — the events the backend says the client's - * snapshot was missing, when it returned them inline. Ignored for every - * other status. */ + /** Rejection detail returned by the backend. A stream-expired 410 carries + * its run, stream, and authoritative retention timestamp here; 412 carries + * events the backend says the client's snapshot was missing. */ details?: unknown; } = {} ): Error { const { retryAfter, code, url, mitigated, details } = opts; if (status === 409) return new EntityConflictError(message); - if (status === 410) return new RunExpiredError(message); + if (status === 410) { + if (code === 'stream-expired') { + const streamDetails = + details && typeof details === 'object' + ? (details as { + runId?: unknown; + streamId?: unknown; + expiredAt?: unknown; + }) + : undefined; + const expiredAt = + typeof streamDetails?.expiredAt === 'string' + ? new Date(streamDetails.expiredAt) + : undefined; + return new StreamExpiredError( + message, + typeof streamDetails?.runId === 'string' + ? streamDetails.runId + : undefined, + typeof streamDetails?.streamId === 'string' + ? streamDetails.streamId + : undefined, + expiredAt && !Number.isNaN(expiredAt.getTime()) ? expiredAt : undefined + ); + } + return new RunExpiredError(message); + } if (status === 412) return new PreconditionFailedError(message, { retryAfter, details }); if (status === 425) return new TooEarlyError(message, { retryAfter }); diff --git a/packages/world-vercel/src/streamer.test.ts b/packages/world-vercel/src/streamer.test.ts index 515981b784..f7ddac8d77 100644 --- a/packages/world-vercel/src/streamer.test.ts +++ b/packages/world-vercel/src/streamer.test.ts @@ -1,3 +1,4 @@ +import { StreamExpiredError } from '@workflow/errors'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { encodeMultiChunks, MAX_CHUNKS_PER_REQUEST } from './streamer.js'; @@ -207,6 +208,60 @@ describe('streams.get', () => { expect(url.pathname).toBe('/v3/runs/run-123/stream/my-stream'); }); + it('throws a typed terminal error with the retention details on 410', async () => { + const expiredAt = '2026-08-10T14:40:00.000Z'; + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + Response.json( + { + success: false, + error: 'stream-expired', + message: 'The stream reached its storage retention limit', + details: { + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt, + }, + }, + { status: 410 } + ) + ); + + const streamer = await getStreamer(); + const error = await streamer.streams + .get('wrun_test', 'stream-test') + .catch((cause: unknown) => cause); + + expect(error).toBeInstanceOf(StreamExpiredError); + expect(error).toMatchObject({ + message: 'The stream reached its storage retention limit', + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt: new Date(expiredAt), + status: 410, + code: 'stream-expired', + }); + const request = vi.mocked(globalThis.fetch).mock.calls[0]; + const headers = (request[1] as RequestInit).headers as Headers; + expect(headers.get('Accept')).toBe('application/json'); + }); + + it('falls back safely for non-stream-expired errors', async () => { + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + Response.json( + { error: 'run-expired', message: 'Run is unavailable' }, + { status: 410 } + ) + ); + + const streamer = await getStreamer(); + await expect( + streamer.streams.get('wrun_test', 'stream-test') + ).rejects.toThrow('Run is unavailable'); + await expect( + streamer.streams.get('wrun_test', 'stream-test') + ).rejects.not.toBeInstanceOf(StreamExpiredError); + }); + it('passes startIndex as a query parameter on the v3 read', async () => { const fetchSpy = vi .spyOn(globalThis, 'fetch') diff --git a/packages/world-vercel/src/streamer.ts b/packages/world-vercel/src/streamer.ts index 9d021e2317..59538ea6d2 100644 --- a/packages/world-vercel/src/streamer.ts +++ b/packages/world-vercel/src/streamer.ts @@ -10,7 +10,11 @@ import { getStreamCloseDispatcher, getStreamDispatcher, } from './http-client.js'; -import { getVercelDiagnostics, instrumentedFetch } from './http-core.js'; +import { + errorForResponse, + getVercelDiagnostics, + instrumentedFetch, +} from './http-core.js'; import { WorkflowRunId, WorkflowStreamName, @@ -101,6 +105,26 @@ function streamSpanAttributes(args: { }; } +async function createStreamReadError(response: Response): Promise { + const fallback = `Failed to fetch stream: ${response.status}`; + if (response.status !== 410) return new Error(fallback); + + try { + const body = (await response.json()) as { + error?: string; + message?: string; + details?: unknown; + }; + return errorForResponse( + response.status, + typeof body.message === 'string' ? body.message : fallback, + { code: body.error, details: body.details } + ); + } catch { + return new Error(fallback); + } +} + function createStreamRequestError( operation: 'write' | 'close', url: URL, @@ -297,6 +321,11 @@ export function createStreamer(config?: APIConfig): Streamer { async get(runId: string, name: string, startIndex?: number) { const httpConfig = await getHttpConfig(config); + // Stream bytes themselves are untyped binary, but any pre-header error + // is a JSON envelope. Asking explicitly avoids a CBOR 410 that this + // binary response path cannot decode while leaving successful stream + // bodies unchanged. + httpConfig.headers.set('Accept', 'application/json'); const url = getStreamReadUrl(name, runId, httpConfig); if (typeof startIndex === 'number') { url.searchParams.set('startIndex', String(startIndex)); @@ -322,8 +351,7 @@ export function createStreamer(config?: APIConfig): Streamer { operation: 'read', startIndex, }), - buildError: (res) => - new Error(`Failed to fetch stream: ${res.status}`), + buildError: createStreamReadError, }); if (!response.body) { throw new Error('No response body for stream'); diff --git a/packages/world-vercel/src/utils.test.ts b/packages/world-vercel/src/utils.test.ts index ffc7913ffe..002c1e1a0f 100644 --- a/packages/world-vercel/src/utils.test.ts +++ b/packages/world-vercel/src/utils.test.ts @@ -1,4 +1,4 @@ -import { PreconditionFailedError } from '@workflow/errors'; +import { PreconditionFailedError, StreamExpiredError } from '@workflow/errors'; import { encode } from 'cbor-x'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; @@ -289,6 +289,40 @@ describe('getHttpConfig (proxied path)', () => { }); }); +describe('makeRequest stream expiry errors', () => { + it.each([ + ['/v2/runs/wrun_test/streams/stream-test/chunks'], + ['/v2/runs/wrun_test/streams/stream-test/info'], + ])('preserves stream-expired details from %s', async (endpoint) => { + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => + Response.json( + { + error: 'stream-expired', + message: 'The stream reached its storage retention limit', + details: { + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt: '2026-08-10T14:40:00.000Z', + }, + }, + { status: 410 } + ) + ); + + const error = await makeRequest({ + endpoint, + schema: z.never(), + }).catch((cause: unknown) => cause); + + expect(error).toBeInstanceOf(StreamExpiredError); + expect(error).toMatchObject({ + runId: 'wrun_test', + streamId: 'stream-test', + expiredAt: new Date('2026-08-10T14:40:00.000Z'), + }); + }); +}); + describe('makeRequest body-parse retry', () => { const schema = z.object({ value: z.string() }); const originalEnv = process.env; diff --git a/packages/world-vercel/src/utils.ts b/packages/world-vercel/src/utils.ts index 94687e8f20..b9d46fae48 100644 --- a/packages/world-vercel/src/utils.ts +++ b/packages/world-vercel/src/utils.ts @@ -515,13 +515,22 @@ export async function makeRequest({ }); if (!response.ok) { - const errorData: { message?: string; code?: string; error?: string } = - await parseResponseBody(response) - .then( - (r) => - r.data as { message?: string; code?: string; error?: string } - ) - .catch(() => ({})); + const errorData: { + message?: string; + code?: string; + error?: string; + details?: unknown; + } = await parseResponseBody(response) + .then( + (r) => + r.data as { + message?: string; + code?: string; + error?: string; + details?: unknown; + } + ) + .catch(() => ({})); const errorCode = errorData.code ?? errorData.error; logCurlRepro(request.method, url, headers); @@ -544,6 +553,7 @@ export async function makeRequest({ code: errorCode, retryAfter, mitigated: response.headers.get('x-vercel-mitigated'), + details: errorData.details, }); span?.setAttributes({ ...ErrorType(errorCode || `HTTP ${response.status}`), diff --git a/packages/world/CHANGELOG.md b/packages/world/CHANGELOG.md index cd4b307ec8..97489d111d 100644 --- a/packages/world/CHANGELOG.md +++ b/packages/world/CHANGELOG.md @@ -1,5 +1,24 @@ # @workflow/world +## 5.0.0-beta.26 + +### Minor Changes + +- [#3374](https://github.com/vercel/workflow/pull/3374) [`439a495`](https://github.com/vercel/workflow/commit/439a495a715b9426ef4dbcf8d928a8fc50ffb040) Thanks [@karthikscale3](https://github.com/karthikscale3)! - Carry the run's pinned deployment on lazy resume messages so misrouted deliveries re-route before the `hook_received` write. + +### Patch Changes + +- [#3385](https://github.com/vercel/workflow/pull/3385) [`74dbf81`](https://github.com/vercel/workflow/commit/74dbf81d327b8574cca429b56757c7322a26b4ef) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Retry replay timeouts through normal queue redelivery instead of exiting the process, and keep Postgres jobs retryable through Core's terminal delivery limit. + +- [#3382](https://github.com/vercel/workflow/pull/3382) [`a8db185`](https://github.com/vercel/workflow/commit/a8db185c3b19b3dab971f51aa076aead81ed26ea) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip + +- [#3389](https://github.com/vercel/workflow/pull/3389) [`6786db9`](https://github.com/vercel/workflow/commit/6786db99538ef57c872d861ecfb28d99ae857d6d) Thanks [@VaguelySerious](https://github.com/VaguelySerious)! - **Breaking**: SpecVersion 6: Event IDs are now a dense per-run slot number, allocated by the world at publish time so a rejected write leaves no gap in the event log. A replay tells the world how many events it had read and gets back the ones it did not see, so an event that arrives from outside the replay and lands ahead of an event the replay wrote no longer fails the run with `CORRUPTED_EVENT_LOG`: it is held for whichever part of the workflow awaits it. A gap in the numbering fails the run instead of being replayed over. + +- [#3205](https://github.com/vercel/workflow/pull/3205) [`22349e9`](https://github.com/vercel/workflow/commit/22349e95fd85a112cbec3f425900b74bf5ccc77f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Load replay-event suffixes through one World request. + +- [#3124](https://github.com/vercel/workflow/pull/3124) [`65139ac`](https://github.com/vercel/workflow/commit/65139acfd7118d3b73672435a6e1c47115f6e23f) Thanks [@NathanColosimo](https://github.com/NathanColosimo)! - Infer required run and step entities, including their start times, from + `events.create` request types. + ## 5.0.0-beta.25 ### Major Changes diff --git a/packages/world/package.json b/packages/world/package.json index 2110621a40..a5be3cd257 100644 --- a/packages/world/package.json +++ b/packages/world/package.json @@ -1,6 +1,6 @@ { "name": "@workflow/world", - "version": "5.0.0-beta.25", + "version": "5.0.0-beta.26", "description": "The Workflows World interface", "type": "module", "main": "dist/index.js", diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index ba37c947ee..7eb2d0d603 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -732,6 +732,25 @@ export interface CreateEventParams { * alongside {@link resumeId}. */ resumePayloadDigest?: string; + /** + * Marks a `step_created` create as the queue consumer's re-ensure of a + * resilient step dispatch (a step message carrying `stepInput` — see + * `WorkflowInvokePayload.stepInput`): the producer's direct write was + * parallelized with the queue publish and may have failed. Only meaningful + * for `step_created`. + * + * Advisory. The runtime never parallelizes a *guarded* `step_created` with + * its publish (see the eligibility gate in the suspension handler), so in + * correct operation a re-ensure can only correspond to an unguarded create + * — there is no guard verdict for it to bypass. A guard-enforcing backend + * MAY nevertheless use this flag as defense-in-depth: refuse the re-ensure + * (world-vercel surfaces the backend's 410 as `RunExpiredError`, which the + * consumer treats as "nothing left to execute" and acks the message) when + * it has recorded a 412 rejection for this correlation id and no step + * entity exists — hardening against a misbehaving or future client. Worlds + * may ignore this flag entirely. + */ + viaStepDispatch?: boolean; /** Request ID (x-vercel-id when on Vercel) for correlating request logs with workflow events. */ requestId?: string; /** diff --git a/packages/world/src/queue.test.ts b/packages/world/src/queue.test.ts index f434d48155..3997465561 100644 --- a/packages/world/src/queue.test.ts +++ b/packages/world/src/queue.test.ts @@ -161,6 +161,43 @@ describe('QueuePayloadSchema', () => { QueuePayloadSchema.parse({ runId: 'wrun_01ABC', stepId: 'step_1' }) ).toEqual({ runId: 'wrun_01ABC', stepId: 'step_1' }); }); + + // Resilient step dispatch: a step message may carry `stepInput` with the + // serialized (binary) step input, which the consumer re-ensures the + // step_created event from. The bytes must survive parsing untouched. + it('round-trips a step message carrying stepInput', () => { + const input = new Uint8Array([1, 2, 3]); + const parsed = QueuePayloadSchema.parse({ + runId: 'wrun_01ABC', + stepId: 'step_1', + stepName: 'myStep', + stepInput: { input }, + }); + expect(parsed).toEqual({ + runId: 'wrun_01ABC', + stepId: 'step_1', + stepName: 'myStep', + stepInput: { input }, + }); + expect((parsed as { stepInput: { input: unknown } }).stepInput.input).toBe( + input + ); + }); + + // Producers only attach stepInput when the dehydrated input is binary and + // the queue transport preserves bytes (CBOR). A non-binary value means the + // payload was mangled in transit — fail the parse rather than letting it be + // written into a step_created as non-binary data. + it('rejects a stepInput whose input is not a Uint8Array', () => { + expect( + QueuePayloadSchema.safeParse({ + runId: 'wrun_01ABC', + stepId: 'step_1', + stepName: 'myStep', + stepInput: { input: 'mangled-to-string' }, + }).success + ).toBe(false); + }); }); describe('RunInputSchema environment', () => { diff --git a/packages/world/src/queue.ts b/packages/world/src/queue.ts index f32d8cf906..c756a99ed9 100644 --- a/packages/world/src/queue.ts +++ b/packages/world/src/queue.ts @@ -147,6 +147,40 @@ export type RunInput = z.infer; * `events.create`, so both server receipts hash to the same digest under the * `(runId, resumeId)` constraint. */ +/** + * Resilient step dispatch data carried through the queue alongside a + * step-execution message ({@link WorkflowInvokePayload.stepId}). Present when + * the producer (the suspension handler dispatching a newly created step) + * parallelized the `step_created` event write with the queue publish — the + * same shape as resilient start (`runInput`) and the resilient hook resume + * (`hookInput`). + * + * When the producer's `step_created` write fails transiently (429 / 5xx / + * transport), the step entity may not exist when this message is consumed. A + * consumer that understands `stepInput` idempotently re-ensures the + * `step_created` event — keyed by the message's `stepId` (the step's + * correlation id, unique per `(runId, correlationId)`) — before executing, so + * the producer's write and the consumer's re-ensure converge on exactly one + * event. + * + * The `input` is the already-serialized (and possibly encrypted) step input — + * the identical bytes the producer also sent on the direct `events.create`. + */ +export const StepDispatchInputSchema = z.object({ + /** + * The serialized step input, reused verbatim from the direct write. Always + * binary: producers only attach `stepInput` when the dehydrated input is a + * `Uint8Array` and the run's queue transport preserves bytes (CBOR). + * Validated here so a malformed or transport-mangled payload fails the + * message parse instead of being silently written into a `step_created` as + * non-binary data. `Buffer` is a `Uint8Array` subclass and passes. + */ + input: z.custom((value) => value instanceof Uint8Array, { + message: 'stepInput.input must be a Uint8Array', + }), +}); +export type StepDispatchInput = z.infer; + export const HookResumeInputSchema = z.object({ /** Stable idempotency key minted once per `resumeHook()` call. */ resumeId: z.string(), @@ -218,6 +252,13 @@ export const WorkflowInvokePayloadSchema = z.object({ * `hook_received` event exists (keyed by `resumeId`) before replaying. */ hookInput: HookResumeInputSchema.optional(), + /** + * Resilient step dispatch data, only present alongside `stepId` when the + * producer parallelized the `step_created` write with this queue publish. A + * consumer that understands this field idempotently ensures the + * `step_created` event exists (keyed by `stepId`) before executing the step. + */ + stepInput: StepDispatchInputSchema.optional(), }); export type WorkflowInvokePayload = z.infer;