diff --git a/.changeset/lazy-hook-resume-durable-abort.md b/.changeset/lazy-hook-resume-durable-abort.md new file mode 100644 index 0000000000..952f6cbda5 --- /dev/null +++ b/.changeset/lazy-hook-resume-durable-abort.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Keep the eager `hook_received` write for the internal resume that records a step-issued abort, so it stays committed before the step completes. diff --git a/.changeset/lazy-hook-resume-vitest.md b/.changeset/lazy-hook-resume-vitest.md new file mode 100644 index 0000000000..238b92ba5e --- /dev/null +++ b/.changeset/lazy-hook-resume-vitest.md @@ -0,0 +1,5 @@ +--- +'@workflow/vitest': patch +--- + +`waitForHook()` accepts `notHookId` to skip a hook the caller already resumed, whose `hook_received` may not be written yet. diff --git a/.changeset/lazy-only-hook-resume-world.md b/.changeset/lazy-only-hook-resume-world.md new file mode 100644 index 0000000000..e86dd935ed --- /dev/null +++ b/.changeset/lazy-only-hook-resume-world.md @@ -0,0 +1,5 @@ +--- +'@workflow/world': patch +--- + +Document that `hookInput` and the `(runId, resumeId)` dedup contract now cover repeated deliveries of one resume rather than a producer/consumer race, and accept `lazy` as a hook-resume strategy. diff --git a/.changeset/lazy-only-hook-resume.md b/.changeset/lazy-only-hook-resume.md new file mode 100644 index 0000000000..30a5a48ac7 --- /dev/null +++ b/.changeset/lazy-only-hook-resume.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +`resumeHook()` no longer writes the `hook_received` event itself on the lazy path: the queue consumer creates it from the queue message, so a resume costs one round trip. A resume against an already-ended run now resolves instead of throwing `HookNotFoundError`. diff --git a/docs/content/docs/v5/api-reference/workflow-api/resume-hook.mdx b/docs/content/docs/v5/api-reference/workflow-api/resume-hook.mdx index 06257a6584..8dfe6390e2 100644 --- a/docs/content/docs/v5/api-reference/workflow-api/resume-hook.mdx +++ b/docs/content/docs/v5/api-reference/workflow-api/resume-hook.mdx @@ -12,9 +12,9 @@ related: Resumes a workflow run by sending a payload to a hook identified by its token. -It creates a `hook_received` event and re-triggers the workflow to continue execution. +It publishes a workflow invocation carrying the payload; the runtime creates the `hook_received` event and continues execution from it. -A Hook kept by `experimental_minRetention` after its workflow ends cannot be resumed. `resumeHook()` throws `HookNotFoundError` in that case. +`resumeHook()` throws `HookNotFoundError` when no hook holds the token. A run that has already ended cannot be resumed, including one whose Hook is kept by `experimental_minRetention`, but whether the call reports that depends on the path it takes: a resume dispatched without reading the run resolves and the ended state is only detected once the payload arrives, while one that reads the run, or that falls back to writing the event up front, throws `HookNotFoundError`. See [lazy hook resume](/docs/changelog/lazy-hook-resume). `resumeHook` is a runtime function that must be called from outside a workflow function. @@ -50,7 +50,7 @@ showSections={["parameters"]} ### Returns -Returns a `Promise`, a `Hook` extended with an optional `resilientResume` flag. Resolving means the resume was accepted and the workflow will continue, whether the `hook_received` event was written directly or, on the parallel fast path, delivered through the workflow queue for the runtime to materialize (see the [lazy hook resume changelog](/docs/changelog/resilient-resume)). `resilientResume` is `true` only when the direct event write failed transiently and the resume was recovered through the queue; on the happy path it is absent. The resolved hook: +Returns a `Promise`, a `Hook` extended with an optional `resilientResume` flag. Resolving means the resume was accepted for delivery: the payload rides the workflow queue message and the runtime materializes the `hook_received` event from it before replaying (see the [lazy hook resume changelog](/docs/changelog/lazy-hook-resume)). `resilientResume` is retained for source compatibility and is no longer set by any path. The resolved hook: `resumeWebhook` is a runtime function that must be called from outside a workflow function. diff --git a/docs/content/docs/v5/changelog/index.mdx b/docs/content/docs/v5/changelog/index.mdx index 37a3b2901e..9742e0a08f 100644 --- a/docs/content/docs/v5/changelog/index.mdx +++ b/docs/content/docs/v5/changelog/index.mdx @@ -12,6 +12,7 @@ Stay up to date with the latest changes to Workflow SDK. ## 2026 +- [Lazy hook resume](/docs/changelog/lazy-hook-resume) (August 2026) - [Resilient hook resume](/docs/changelog/resilient-resume) (July 2026) - [Eager processing of steps and incremental event replay](/docs/changelog/eager-processing) (March 2026) - Serializable AbortController and AbortSignal (March 12, 2026) diff --git a/docs/content/docs/v5/changelog/lazy-hook-resume.mdx b/docs/content/docs/v5/changelog/lazy-hook-resume.mdx new file mode 100644 index 0000000000..ea812036d0 --- /dev/null +++ b/docs/content/docs/v5/changelog/lazy-hook-resume.mdx @@ -0,0 +1,51 @@ +--- +title: Lazy hook resume +description: resumeHook() no longer writes hook_received itself. The queue consumer materializes the event from the message, so a resume costs one round trip. +--- + +# Lazy hook resume + +## Motivation + +[Resilient hook resume](/docs/changelog/resilient-resume) made `resumeHook()` write the `hook_received` event and publish the workflow queue message concurrently, with the queue consumer re-ensuring the event from the message's `hookInput` before replay. Both sides then wrote the same event, and a `(runId, resumeId)` constraint collapsed them onto one. + +Running the two concurrently removed the second round trip from the critical path, but the write itself stayed: every resume still spent a request on an event the consumer was about to write anyway, and the producer still had to classify its outcome (conflict, throttle, terminal run) to decide whether the resume had survived. + +This change drops the producer's write entirely. On the lazy path `resumeHook()` publishes the queue message and nothing else. + +## Design + +- `resumeHook()` publishes one message carrying `hookInput`: the dehydrated payload, a client-minted `resumeId`, the hook token, and a payload digest. It writes no event. +- The queue consumer materializes `hook_received` from `hookInput` before replay, keyed by `resumeId`. This is the same write it already performed; it is now the only one. +- The `(runId, resumeId)` constraint still matters: a queue redelivery, or a delivery re-routed for deployment affinity, repeats the write with the same key, and the backend collapses those onto exactly one committed event. +- **A failed publish fails the resume.** The message carries both the trigger and the only copy of the payload, so `resumeHook()` throws and nothing is persisted for a later delivery to pick up. This replaces the previous rule where a failed event write could still be recovered through the queue. +- `ResumedHook.resilientResume` is retained on the type but is never set: with a single writer there is no partial outcome to report. The `workflow.hook.resilient_resume` span attribute is likewise no longer emitted. +- The resume span reports `workflow.hook.resume_strategy: lazy` (previously `parallel`). + +## Behavior change: the event is not visible when `resumeHook()` returns + +`resumeHook()` used to await its own `hook_received` write, so by the time it resolved the event was in the log. It no longer writes, so **resolving means the message was published, not that the event exists**. The event appears when the run picks the resume up. + +Code that reads the run back immediately after resuming now races. The pattern that breaks is a loop that resumes and then looks for the next thing to resume, keying off "this hook has no `hook_received` yet": it can be handed back the hook it just resumed and deliver a second payload to it. Wait for something that implies the run made progress instead. `waitForHook()` in `@workflow/vitest` takes a `notHookId` option for exactly this. + +The runtime has one caller that needs the old guarantee. A step that aborts a shared `AbortController` resumes a hook to record the abort in the event log, and that write is an ordering barrier: it must land before the step completes, or the continuation `step_completed` enqueues can dispatch the next step with a stale, non-aborted signal. That path uses an internal durable resume which keeps the eager write and reports `resume_fallback_reason: durable_required`. + +Nothing about delivery changes. The payload is on the queue message and reaches the workflow exactly once. + +## Behavior change: resumes against an ended run + +The hook lookup is unchanged: `resumeHook(token, ...)` still resolves the token through `hooks.getByToken()`, which throws `HookNotFoundError` when no hook holds it. Hook existence, and the token's binding to a run, are still validated before anything is published. + +What the lookup does not carry is the run's *mutable* status. `HookResumeContext` is deliberately an immutable slice of the run, so a resume that runs off it never learns whether the run is still live. That used to be caught by the `hook_received` write being rejected. With no write, **a resume against an ended run resolves instead of throwing `HookNotFoundError`**. + +This is only reachable when the hook record outlives its run, since otherwise the lookup itself fails: a hook kept by `experimental_minRetention`, or one whose token has not been released yet. Resumes that fall back to reading the run keep their terminal pre-check, as does the sequential path, so a resume on either of those still fails loudly. + +Nothing resumes either way. The consumer's write is rejected the same way and the delivery is consumed, so the ended run is untouched. Only the producer's report changes: an accepted publish means the resume was dispatched, not that the run was still live when it arrived. A [webhook](/docs/api-reference/workflow-api/resume-webhook) whose run has ended can answer `202` rather than surfacing an error. + +Callers that need the distinction have to read the run. + +## Compatibility + +The gating is unchanged: the lazy path activates only when the target run's queue consumer and the live backend both attest support, re-checked on every resume. Oversized payloads, legacy runs, non-CBOR transports, and `WORKFLOW_DISABLE_LAZY_HOOK_RESUME=1` fall back to the sequential write-then-publish path. + +Consumers still accept a message from an older producer that wrote the event itself: such a message reports `strategy: parallel`, and the consumer's write converges on the producer's committed event exactly as before. No coordinated deploy is needed in either direction. diff --git a/docs/content/docs/v5/changelog/meta.json b/docs/content/docs/v5/changelog/meta.json index 63e53a02e7..746696a0d3 100644 --- a/docs/content/docs/v5/changelog/meta.json +++ b/docs/content/docs/v5/changelog/meta.json @@ -2,6 +2,7 @@ "title": "Changelog", "pages": [ "index", + "lazy-hook-resume", "eager-processing", "resilient-resume", "resilient-start", diff --git a/docs/content/docs/v5/changelog/resilient-resume.mdx b/docs/content/docs/v5/changelog/resilient-resume.mdx index 6a61a5f85a..09e6bd8898 100644 --- a/docs/content/docs/v5/changelog/resilient-resume.mdx +++ b/docs/content/docs/v5/changelog/resilient-resume.mdx @@ -5,6 +5,14 @@ description: resumeHook() now tolerates transient event storage failures when th # Resilient `resumeHook()` + + Superseded by [lazy hook resume](/docs/changelog/lazy-hook-resume): + `resumeHook()` no longer writes the `hook_received` event at all on the fast + path, so the two-writer design and the `resilientResume` flag described below + are historical. The `(runId, resumeId)` constraint and the queue-carried + payload remain. + + ## Motivation `resumeHook()` used to write the `hook_received` event and dispatch the workflow queue message strictly one after the other, so every resume paid two sequential round trips and a transient event-storage failure failed the whole resume even when the queue was healthy. This change runs both writes **concurrently** (cutting a round trip off resume latency) and, on the same path, brings `resumeHook()` to parity with [resilient `start()`](/docs/changelog/resilient-start): a transient event-write failure no longer fails the resume when the payload can still be delivered through the queue. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index f07fd7ef34..45fd0c9502 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -84,8 +84,9 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_DISABLE_LAZY_HOOK_RESUME` - Default: enabled (lazy hook resume on) -- Resuming a hook persists the `hook_received` event and publishes the workflow invocation concurrently, cutting a round trip off resume latency. On this parallel path, the queue message also carries the payload, so a transient event-write failure still resumes the run. The queue consumer re-ensures the `hook_received` event before replay. A backend `(runId, resumeId)` constraint keeps the two writers converging on exactly one event. -- The runtime falls back to the sequential path when the consumer or backend does not attest dedup support (or the payload is too large to inline on the queue message). On the sequential path, the event is written *before* dispatch, and its failure fails the resume. The fallback trades away that resilience to stay safe when dedup is not enforced; it does not preserve it. +- Resuming a hook publishes the workflow invocation and writes no event, so the resume costs one round trip. The queue message carries the payload, and the consumer creates the `hook_received` event from it before replay. A backend `(runId, resumeId)` constraint keeps redeliveries of that message converging on exactly one event. Because the message is the only copy of the payload, a failed publish fails the resume. +- The runtime falls back to the sequential path when the consumer or backend does not attest dedup support (or the payload is too large to inline on the queue message). On the sequential path, the event is written *before* dispatch, and its failure fails the resume. +- The two paths differ on ended runs: the sequential write is rejected and surfaces as `HookNotFoundError`, while the lazy path resolves because it never writes. Neither resumes the run. - Set `1` to force the sequential path as a kill switch. The chosen strategy is reported on the resume span as `workflow.hook.resume_strategy`. ### `WORKFLOW_DEPLOYMENT_MISMATCH_MAX_RETRIES` diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index d8868d2c47..782186335e 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -2389,9 +2389,15 @@ describe.concurrent('e2e', () => { const hook = await getHookByToken(token); expect(hook.runId).toBe(owner.runId); - await expect(resumeHook(hook, { duplicate: true })).rejects.toSatisfy( - (error: unknown) => HookNotFoundError.is(error) - ); + // Whether the call itself rejects depends on which dispatch path + // `resumeHook` takes, so this only asserts the invariant both share: the + // ended run is never resumed. The sequential path writes `hook_received` + // and surfaces the server's rejection as HookNotFoundError; the lazy + // path writes nothing, so it resolves and the same rejection lands on + // the queue consumer, which consumes the delivery. + await resumeHook(hook, { duplicate: true }).catch((error: unknown) => { + if (!HookNotFoundError.is(error)) throw error; + }); const duplicate = await start(await e2e('hookMinRetentionWorkflow'), [ token, @@ -2402,6 +2408,18 @@ describe.concurrent('e2e', () => { conflictRunId: owner.runId, conflictStatus: 'completed', }); + + // Nothing was appended to the terminal run: no path may materialize a + // `hook_received` for it. Checked after the duplicate run so a lazy + // resume's consumer has had time to attempt (and be refused) its write. + const world = await getWorld(); + const { data: ownerEvents } = await world.events.list({ + runId: owner.runId, + }); + expect( + ownerEvents.some((e) => e.eventType === 'hook_received'), + 'a resume against an ended run must not append hook_received' + ).toBe(false); } ); diff --git a/packages/core/src/abort-controller-step.test.ts b/packages/core/src/abort-controller-step.test.ts index e96c379c79..187fae0e54 100644 --- a/packages/core/src/abort-controller-step.test.ts +++ b/packages/core/src/abort-controller-step.test.ts @@ -30,6 +30,14 @@ const mockStreamReads = vi.hoisted(() => ({ })); const mockResumeHook = vi.hoisted(() => vi.fn().mockResolvedValue(undefined)); +// The abort path must use the DURABLE entry point: a plain `resumeHook` only +// guarantees the resume was published (the lazy path defers the event write to +// the queue consumer), which would satisfy preCompletionOps while leaving the +// stale-signal race it exists to prevent. Mocked separately so a regression to +// `resumeHook` shows up as this staying uncalled. +const mockResumeHookDurable = vi.hoisted(() => + vi.fn().mockResolvedValue(undefined) +); // Mock version module vi.mock('./version.js', () => ({ version: '0.0.0-test' })); @@ -84,6 +92,7 @@ vi.mock('./runtime/get-world-lazy.js', () => ({ // Mock resume-hook vi.mock('./runtime/resume-hook.js', () => ({ resumeHook: mockResumeHook, + resumeHookDurable: mockResumeHookDurable, })); // ============================================================================ @@ -164,7 +173,7 @@ function reviveAbortController(opts: { if (opts.hookToken) { ctx.ops.push( (async () => { - await mockResumeHook(opts.hookToken, { + await mockResumeHookDurable(opts.hookToken, { aborted: true, reason, }); @@ -422,7 +431,7 @@ describe('AbortSignal deserialized in step context', () => { await Promise.allSettled(stepCtx.ops); - expect(mockResumeHook).toHaveBeenCalledWith('abrt_test9', { + expect(mockResumeHookDurable).toHaveBeenCalledWith('abrt_test9', { aborted: true, reason: 'hook-resume-test', }); @@ -615,10 +624,17 @@ describe('AbortSignal deserialized in step context', () => { * flake). The hook resume must land in `ctx.preCompletionOps`, which the step * handler awaits before writing `step_completed`. The real-time stream write * (which reaches an in-flight sibling) stays in the background `ctx.ops`. + * + * Routing alone stopped being sufficient once `resumeHook()` went lazy: it + * resolves when the resume is PUBLISHED, leaving the event write to the queue + * consumer, so awaiting it in `preCompletionOps` would re-open the same race. + * The abort path therefore calls `resumeHookDurable()`, which forces the eager + * write, and this suite pins both the routing and the entry point. */ describe('step-initiated abort: durable hook resume is committed before completion', () => { beforeEach(() => { mockResumeHook.mockClear(); + mockResumeHookDurable.mockClear(); mockStreamReads.readResults.clear(); mockStreamReads.writeLog = []; mockStreamReads.closeLog = []; @@ -708,10 +724,14 @@ describe('step-initiated abort: durable hook resume is committed before completi // Draining preCompletionOps (what the step executor awaits before // step_completed) actually fires the resume with the correct payload. await Promise.all(preCompletionOps); - expect(mockResumeHook).toHaveBeenCalledTimes(1); - expect(mockResumeHook).toHaveBeenCalledWith('abrt_pre_completion', { + expect(mockResumeHookDurable).toHaveBeenCalledTimes(1); + expect(mockResumeHookDurable).toHaveBeenCalledWith('abrt_pre_completion', { aborted: true, reason: 'aborted from step', }); + // Routing is only half of it: the plain entry point would resolve as soon + // as the resume was published, so draining preCompletionOps would prove + // nothing about the event existing. + expect(mockResumeHook).not.toHaveBeenCalled(); }); }); diff --git a/packages/core/src/capabilities.ts b/packages/core/src/capabilities.ts index f991e07dc9..ef7c7f74f8 100644 --- a/packages/core/src/capabilities.ts +++ b/packages/core/src/capabilities.ts @@ -40,8 +40,8 @@ * - Lazy hook resume ("consumer re-ensures `hook_received` from `hookInput`"): * deliberately NOT tracked here. Rather than predict a release cutoff, the * run's creating deployment stamps an explicit `hookResumeInputVersion` - * execution-context marker; `resumeHook()` gates the parallel fast path on - * that marker (mirrored onto the hook's resumeContext by the server). + * execution-context marker; `resumeHook()` gates the lazy path on that + * marker (mirrored onto the hook's resumeContext by the server). */ import semver from 'semver'; @@ -120,7 +120,7 @@ const CAPABILITY_VERSION_TABLE: ReadonlyArray<{ // version-compare against a predicted release cutoff is a guess; instead the // run's creating deployment stamps an explicit `hookResumeInputVersion` // marker into its execution context, which the server mirrors onto the hook's - // resumeContext. `resumeHook()` gates the parallel fast path on that marker. + // resumeContext. `resumeHook()` gates the lazy path on that marker. ]; /** diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 24c17fd430..8c7a4ce530 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -105,12 +105,12 @@ export interface HealthCheckResult { encryptionPublicKey?: string; /** * The responding deployment's `HOOK_RESUME_INPUT_VERSION`: the protocol - * version at which the *consumer* (queue-message target) re-ensures the + * version at which the *consumer* (queue-message target) materializes the * `hook_received` event from `hookInput` on replay. A cross-deployment * `start()` stamps the *target's* value (not the caller's) into the new * run's `executionContext.hookResumeInputVersion` so that `resumeHook()` - * only takes the parallel path when the deployment that will actually - * consume the queue message is known to honor `hookInput`. Omitted when the + * only takes the lazy path when the deployment that will actually consume + * the queue message is known to honor `hookInput`. Omitted when the * responding deployment predates this field (an older consumer that ignores * `hookInput`), which fails the gate closed. */ diff --git a/packages/core/src/runtime/resume-hook.parallel.test.ts b/packages/core/src/runtime/resume-hook.lazy.test.ts similarity index 60% rename from packages/core/src/runtime/resume-hook.parallel.test.ts rename to packages/core/src/runtime/resume-hook.lazy.test.ts index 488950f3f8..5dcfa88906 100644 --- a/packages/core/src/runtime/resume-hook.parallel.test.ts +++ b/packages/core/src/runtime/resume-hook.lazy.test.ts @@ -1,20 +1,16 @@ -import { - EntityConflictError, - HookNotFoundError, - RunExpiredError, - ThrottleError, -} from '@workflow/errors'; +import { HookNotFoundError, RunExpiredError } from '@workflow/errors'; import { HOOK_RESUME_DEDUP_VERSION, HOOK_RESUME_INPUT_VERSION, type Hook, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT, + type WorkflowRun, type World, } from '@workflow/world'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { dehydrateStepReturnValue } from '../serialization.js'; -import { resumeHook, resumeWebhook } from './resume-hook.js'; +import { resumeHook, resumeHookDurable, resumeWebhook } from './resume-hook.js'; import { setWorld } from './world.js'; vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn() })); @@ -23,9 +19,9 @@ vi.mock('../telemetry.js', () => ({ trace: vi.fn((_name, fn) => fn(undefined)), })); // Return raw bytes from dehydration so `dehydratedPayload instanceof Uint8Array` -// is true and the parallel resume strategy activates. The sibling +// is true and the lazy resume strategy activates. The sibling // `resume-hook.fast-path.test.ts` returns a string and thus stays sequential; -// this file exercises the complementary parallel branch. +// this file exercises the complementary lazy branch. const PAYLOAD_BYTES = new Uint8Array([1, 2, 3, 4]); vi.mock('../serialization.js', async (importActual) => { const actual = await importActual(); @@ -36,27 +32,27 @@ vi.mock('../serialization.js', async (importActual) => { }; }); -describe('resumeHook (parallel fast path)', () => { +describe('resumeHook (lazy path)', () => { afterEach(() => setWorld(undefined)); const baseHook = { - runId: 'wrun_par', - hookId: 'hook_par', - token: 'order:par', + runId: 'wrun_lazy', + hookId: 'hook_lazy', + token: 'order:lazy', ownerId: 'owner_1', projectId: 'project_1', environment: 'production', createdAt: new Date(), - // Non-legacy run: v1Compat is false, so the parallel path is eligible. + // Non-legacy run: v1Compat is false, so the lazy path is eligible. specVersion: SPEC_VERSION_CURRENT, } satisfies Hook; // The run carries an explicit `hookResumeInputVersion` marker (its creating - // deployment re-ensures from `hookInput`). Combined with a backend that - // declares `hookResumeDedup`, a CBOR-transport spec version, and a raw-byte - // payload, resumeHook takes the parallel path. - const parallelContext = { - deploymentId: 'deployment_par', + // deployment materializes the event from `hookInput`). Combined with a + // backend that declares `hookResumeDedup`, a CBOR-transport spec version, and + // a raw-byte payload, resumeHook takes the lazy path. + const lazyContext = { + deploymentId: 'deployment_lazy', workflowName: 'processOrder', runSpecVersion: SPEC_VERSION_CURRENT, workflowCoreVersion: '5.0.0', @@ -83,48 +79,53 @@ describe('resumeHook (parallel fast path)', () => { return { createEvent, queue, getByToken }; }; - it('dispatches the event write and queue publish concurrently with a shared resumeId + digest', async () => { - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + it('publishes the resume on the queue and writes no hook_received event', async () => { + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld(hook); const result = await resumeHook(hook.token, { foo: 'bar' }); - // Happy path (direct write landed): no resilience flag on the ResumedHook. + // The flag is retained on the type but no longer produced by any path. expect(result.resilientResume).toBeUndefined(); - expect(createEvent).toHaveBeenCalledTimes(1); - const [runIdArg, eventArg, optsArg] = createEvent.mock.calls[0]; - expect(runIdArg).toBe(hook.runId); - expect(eventArg).toMatchObject({ - eventType: 'hook_received', - correlationId: hook.hookId, - }); - // Both writers must carry the same idempotency key and content digest. - const resumeId = optsArg.resumeId as string; - const digest = optsArg.resumePayloadDigest as string; - expect(resumeId).toEqual(expect.any(String)); - expect(digest).toMatch(/^[0-9a-f]{64}$/); - // The replay-log preload is the consumer re-ensure's opt-in only: the - // producer never reads the log, so it must not ask the World (and, on - // world-vercel, the server) to assemble one. - expect(optsArg.preloadEvents).toBeUndefined(); + // The whole point of the lazy path: the producer performs no event write, + // so the resume costs one round trip. The consumer materializes + // `hook_received` from `hookInput` before it replays. + expect(createEvent).not.toHaveBeenCalled(); expect(queue).toHaveBeenCalledTimes(1); const [, payloadArg] = queue.mock.calls[0]; expect(payloadArg.runId).toBe(hook.runId); expect(payloadArg.hookInput).toEqual({ - resumeId, + // Idempotency key for the consumer's write: a redelivery of this message + // converges on the one committed event via (runId, resumeId). + resumeId: expect.any(String), hookId: hook.hookId, token: hook.token, payload: PAYLOAD_BYTES, - payloadDigest: digest, + payloadDigest: expect.stringMatching(/^[0-9a-f]{64}$/), // The run's pinned deployment from the resume context, for the // consumer's cheap pre-write affinity check. - deploymentId: 'deployment_par', + deploymentId: 'deployment_lazy', }); }); + it('mints a distinct resumeId per resume of the same hook', async () => { + // Two resumes of a reusable hook must not collide on the dedup constraint, + // or the second would be swallowed as a redelivery of the first and the + // run would only ever see one payload. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; + const { queue } = makeWorld(hook); + + await resumeHook(hook.token, { foo: 'one' }); + await resumeHook(hook.token, { foo: 'two' }); + + const [, first] = queue.mock.calls[0]; + const [, second] = queue.mock.calls[1]; + expect(first.hookInput.resumeId).not.toBe(second.hookInput.resumeId); + }); + it('stamps the resume TTR window on the queue message', async () => { - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { queue } = makeWorld(hook); const before = Date.now(); @@ -133,7 +134,7 @@ describe('resumeHook (parallel fast path)', () => { const [, payloadArg] = queue.mock.calls[0]; const timing = payloadArg.hookResumeTiming; - expect(timing.strategy).toBe('parallel'); + expect(timing.strategy).toBe('lazy'); // T0 is entry into resumeHook and T1 the publish request, so both fall // inside this call and in that order. expect(timing.resumeRequestedAtMs).toBeGreaterThanOrEqual(before); @@ -146,8 +147,10 @@ describe('resumeHook (parallel fast path)', () => { expect(timing.setupSource).toBeUndefined(); }); - it('always throws when the queue publish fails', async () => { - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + it('throws when the queue publish fails', async () => { + // The message carries both the trigger and the only copy of the payload, + // so a failed publish is a failed resume with nothing persisted behind it. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const queueErr = new Error('queue unavailable'); const { queue } = makeWorld(hook, { queue: vi.fn().mockRejectedValue(queueErr), @@ -157,139 +160,124 @@ describe('resumeHook (parallel fast path)', () => { expect(queue).toHaveBeenCalledTimes(1); }); - it('swallows a retryable event-write failure because the queue consumer re-ensures the event', async () => { - // 429/5xx/transport on the direct write is resilient: the run WAS - // re-triggered via the queue, whose consumer idempotently re-ensures the - // hook_received event before replay. resumeHook must not fail the caller. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + it('accepts a resume against an ended run instead of throwing HookNotFoundError', async () => { + // Contract change from the write-then-publish paths: this resume runs off + // a stored `resumeContext`, so it never reads the run, and it never + // writes, so it cannot observe the server's rejection of `hook_received` + // for a terminal run either. It resolves. Nothing resumes — the consumer's + // own write is rejected the same way and the delivery is consumed. A World + // whose events.create would reject is never consulted. The complementary + // run-fallback case is the test below. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const createEvent = vi .fn() - .mockRejectedValue(new ThrottleError('slow down')); - const queue = vi.fn().mockResolvedValue({ messageId: 'm_1' }); - makeWorld(hook, { createEvent, queue }); + .mockRejectedValue(new RunExpiredError('run has expired')); + const { queue } = makeWorld(hook, { createEvent }); - const result = await resumeHook(hook.token, { foo: 'bar' }); - // Recovered via the queue: the ResumedHook carries resilientResume=true so - // callers/telemetry can distinguish the fallback from the happy path. - expect(result).toMatchObject({ - hookId: hook.hookId, - resilientResume: true, - }); - expect(createEvent).toHaveBeenCalledTimes(1); + await expect(resumeHook(hook.token, { foo: 'bar' })).resolves.toMatchObject( + { hookId: hook.hookId } + ); + expect(createEvent).not.toHaveBeenCalled(); expect(queue).toHaveBeenCalledTimes(1); - // The payload rode the queue message so the consumer can materialize it. - const [, payloadArg] = queue.mock.calls[0]; - expect(payloadArg.hookInput).toMatchObject({ - hookId: hook.hookId, - token: hook.token, - payload: PAYLOAD_BYTES, - }); }); - it('re-keys a terminal-run rejection from the event write to HookNotFoundError(token)', async () => { - // The queue publish succeeds, but the run has genuinely ended: the direct - // write rejects with a terminal "hook gone" error and resumeHook surfaces - // the pre-fast-path contract (HookNotFoundError keyed on the token). The - // queue consumer's re-ensure will also no-op against the terminal run. - for (const err of [ - new HookNotFoundError(baseHook.hookId), - new RunExpiredError('run has expired'), - ]) { - const hook = { - ...baseHook, - resumeContext: parallelContext, - } satisfies Hook; - const createEvent = vi.fn().mockRejectedValue(err); - const queue = vi.fn().mockResolvedValue({ messageId: 'm_1' }); - makeWorld(hook, { createEvent, queue }); - - await expect(resumeHook(hook.token, { foo: 'bar' })).rejects.toSatisfy( - (e: unknown) => - HookNotFoundError.is(e) && - (e as HookNotFoundError).token === hook.token - ); - setWorld(undefined); - } - }); + it('still rejects an ended run when the hook carries no resumeContext', async () => { + // The lazy path removes the producer's write, not the run-fallback + // terminal pre-check. A World that serves no `resumeContext` on its hooks + // (world-local) makes every resume fetch the run, so an ended run is + // caught locally and throws before anything is published — even though + // that World statically attests dedup and would otherwise go lazy. + const hook = { ...baseHook } satisfies Hook; + const run = { + runId: hook.runId, + status: 'completed', + deploymentId: 'deployment_lazy', + workflowName: 'processOrder', + createdAt: new Date(), + updatedAt: new Date(), + attributes: {}, + specVersion: SPEC_VERSION_CURRENT, + } as unknown as WorkflowRun; + const createEvent = vi.fn(); + const queue = vi.fn(); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { hookResumeDedup: true }, + hooks: { getByToken: vi.fn().mockResolvedValue(hook) }, + runs: { get: vi.fn().mockResolvedValue(run) }, + events: { create: createEvent }, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + queue, + } as unknown as World); - it('rethrows a non-retryable, non-terminal event-write failure (e.g. a 400) even though the queue publish succeeded', async () => { - // Not every event-write rejection is recoverable: a genuine client error - // (400 / validation) is neither a terminal "hook gone" (re-keyed to - // HookNotFoundError) nor a transient failure (swallowed). It falls through - // to the default branch and surfaces to the caller unchanged — the queue - // message went out, but the caller must see the real error. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; - const badRequest = new Error('invalid event payload'); - const createEvent = vi.fn().mockRejectedValue(badRequest); - const queue = vi.fn().mockResolvedValue({ messageId: 'm_1' }); - makeWorld(hook, { createEvent, queue }); - - await expect(resumeHook(hook.token, { foo: 'bar' })).rejects.toBe( - badRequest + await expect(resumeHook(hook.token, { foo: 'bar' })).rejects.toSatisfy( + (e: unknown) => + HookNotFoundError.is(e) && (e as HookNotFoundError).token === hook.token ); - expect(createEvent).toHaveBeenCalledTimes(1); - expect(queue).toHaveBeenCalledTimes(1); + expect(createEvent).not.toHaveBeenCalled(); + expect(queue).not.toHaveBeenCalled(); }); - it('prioritizes the queue error when both the event write and the queue publish fail', async () => { - // Queue failure is always fatal (no consumer will re-ensure), and it is - // checked before the event-write result — so even a recoverable-looking - // event error is superseded by the queue rejection the caller must see. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; - const queueErr = new Error('queue unavailable'); - const createEvent = vi - .fn() - .mockRejectedValue(new ThrottleError('slow down')); - const queue = vi.fn().mockRejectedValue(queueErr); - makeWorld(hook, { createEvent, queue }); + it('resumeHookDurable writes the event before resolving, even when every lazy precondition passes', async () => { + // The runtime resumes a hook to record a step-issued abort in the event + // log, and that write is an ordering barrier: it must be committed before + // the aborting step completes, or the continuation `step_completed` + // enqueues can dispatch the next step with a stale, non-aborted signal. + // Publishing is not enough, so this entry point forces the eager write. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; + const { createEvent, queue } = makeWorld(hook); + + await resumeHookDurable(hook.token, { aborted: true }); - await expect(resumeHook(hook.token, { foo: 'bar' })).rejects.toBe(queueErr); expect(createEvent).toHaveBeenCalledTimes(1); - expect(queue).toHaveBeenCalledTimes(1); + const [, eventArg, optsArg] = createEvent.mock.calls[0]; + expect(eventArg).toMatchObject({ + eventType: 'hook_received', + correlationId: hook.hookId, + }); + // Sequential shape: no idempotency key on the write, no hookInput on the + // message. The payload rides the event log. + expect(optsArg.resumeId).toBeUndefined(); + const [, payloadArg] = queue.mock.calls[0]; + expect(payloadArg.hookInput).toBeUndefined(); + expect(payloadArg.hookResumeTiming.strategy).toBe('sequential'); }); - it('swallows an EntityConflict (409) from the event write on the parallel path', async () => { - // Unlike the sequential path, a 409 here is NOT "hook gone": the parallel - // write raced its own re-ensuring queue consumer (or a redrive) on the - // shared resumeId. The run was re-triggered via the queue, whose consumer - // converges on the single committed event, so resumeHook must resolve - // rather than re-key to HookNotFoundError. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + it('resumeHookDurable surfaces a terminal-run rejection as HookNotFoundError', async () => { + // The barrier path keeps the older loud contract precisely because it + // writes: a resume recording an abort against an ended run must not look + // like it landed. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const createEvent = vi .fn() - .mockRejectedValue(new EntityConflictError('resumeId already claimed')); - const queue = vi.fn().mockResolvedValue({ messageId: 'm_1' }); - makeWorld(hook, { createEvent, queue }); - - const result = await resumeHook(hook.token, { foo: 'bar' }); - // A 409 here is expected concurrency, recovered via the queue consumer, so - // it is surfaced as a resilient resume rather than an error. - expect(result).toMatchObject({ - hookId: hook.hookId, - resilientResume: true, - }); - expect(createEvent).toHaveBeenCalledTimes(1); - expect(queue).toHaveBeenCalledTimes(1); + .mockRejectedValue(new RunExpiredError('run has expired')); + const { queue } = makeWorld(hook, { createEvent }); + + await expect( + resumeHookDurable(hook.token, { aborted: true }) + ).rejects.toSatisfy( + (e: unknown) => + HookNotFoundError.is(e) && (e as HookNotFoundError).token === hook.token + ); + expect(queue).not.toHaveBeenCalled(); }); it('forces the sequential path when WORKFLOW_DISABLE_LAZY_HOOK_RESUME=1 despite every other precondition passing', async () => { - // The operational kill switch must win over an otherwise fully fast-path- + // The operational kill switch must win over an otherwise fully lazy- // eligible resume (marker present, dedup-capable backend, CBOR transport, // raw-byte payload). Follows the SDK convention of other disable flags // (e.g. WORKFLOW_DISABLE_COMPRESSION): enabled by default, strict '1'. const ORIG = process.env.WORKFLOW_DISABLE_LAZY_HOOK_RESUME; process.env.WORKFLOW_DISABLE_LAZY_HOOK_RESUME = '1'; try { - const hook = { - ...baseHook, - resumeContext: parallelContext, - } satisfies Hook; + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld(hook); await resumeHook(hook.token, { foo: 'bar' }); - // Sequential: no shared idempotency key on the write, no hookInput on the - // queue message — the payload rides the event log. + // Sequential: the event is written before the publish, with no + // idempotency key, and the queue message carries no hookInput — the + // payload rides the event log. expect(createEvent).toHaveBeenCalledTimes(1); const [, , optsArg] = createEvent.mock.calls[0]; expect(optsArg.resumeId).toBeUndefined(); @@ -309,20 +297,16 @@ describe('resumeHook (parallel fast path)', () => { it('does NOT force sequential for values other than the exact string "1"', async () => { // Strict comparison: only '1' disables. A stray 'true'/'0'/'' must leave - // the fast path enabled, matching the other WORKFLOW_DISABLE_* flags. + // the lazy path enabled, matching the other WORKFLOW_DISABLE_* flags. const ORIG = process.env.WORKFLOW_DISABLE_LAZY_HOOK_RESUME; process.env.WORKFLOW_DISABLE_LAZY_HOOK_RESUME = 'true'; try { - const hook = { - ...baseHook, - resumeContext: parallelContext, - } satisfies Hook; + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld(hook); await resumeHook(hook.token, { foo: 'bar' }); - const [, , optsArg] = createEvent.mock.calls[0]; - expect(optsArg.resumeId).toEqual(expect.any(String)); + expect(createEvent).not.toHaveBeenCalled(); const [, payloadArg] = queue.mock.calls[0]; expect(payloadArg.hookInput).toBeDefined(); } finally { @@ -335,14 +319,13 @@ describe('resumeHook (parallel fast path)', () => { }); it('falls back to the sequential path when the payload exceeds the inline queue bound', async () => { - // A payload larger than the queue's inline ceiling would fail the oversized - // publish on the parallel path, persisting hook_received but never - // re-triggering the run. The size gate must instead select the sequential - // path, whose queue message carries only the run ID (no resumeId / no - // hookInput) — the payload rides the event log. + // A payload larger than the queue's inline ceiling would fail the publish + // on the lazy path, and with no eager write there would be nothing left of + // the resume. The size gate must instead select the sequential path, whose + // queue message carries only the run ID — the payload rides the event log. const oversized = new Uint8Array(256 * 1024).fill(7); vi.mocked(dehydrateStepReturnValue).mockResolvedValueOnce(oversized); - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld(hook); await resumeHook(hook.token, { foo: 'bar' }); @@ -366,15 +349,15 @@ describe('resumeHook (parallel fast path)', () => { it('falls back to the sequential path when the run lacks the hookResumeInput marker', async () => { // The run's creating deployment did not stamp `hookResumeInputVersion`, so - // its queue consumer will NOT re-ensure hook_received from hookInput. Even - // with a dedup-capable backend, raw-byte payloads, and CBOR transport, - // resumeHook writes then publishes and carries neither resumeId nor - // hookInput — the payload rides the event log. + // its queue consumer will NOT materialize hook_received from hookInput. + // Without an eager write the resume would be lost outright, so even with a + // dedup-capable backend, raw-byte payloads, and CBOR transport, resumeHook + // writes then publishes and carries neither resumeId nor hookInput. expect(SPEC_VERSION_CURRENT).toBeGreaterThanOrEqual( SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT ); const { hookResumeInputVersion: _omit, ...contextWithoutMarker } = - parallelContext; + lazyContext; const hook = { ...baseHook, resumeContext: contextWithoutMarker, @@ -394,14 +377,14 @@ describe('resumeHook (parallel fast path)', () => { }); it('falls back to the sequential path for a legacy (v1Compat) run', async () => { - // A legacy run omits `token` from the producer's event body, but the queue - // consumer's re-ensure always includes it — the two writers would disagree - // on the event body. Legacy runs must stay sequential regardless of every - // other precondition. + // A legacy run omits `token` from the eagerly written event body, but the + // consumer's write always includes it — the same resume would produce a + // different event depending on which side wrote it. Legacy runs must stay + // sequential regardless of every other precondition. const hook = { ...baseHook, specVersion: 1, - resumeContext: parallelContext, + resumeContext: lazyContext, } satisfies Hook; const { createEvent, queue } = makeWorld(hook); @@ -421,8 +404,9 @@ describe('resumeHook (parallel fast path)', () => { // The target runtime supports lazy resume (marker present, CBOR transport, // raw bytes) but the World backend has not opted in — e.g. Postgres, which // has no (runId, resumeId) dedup. resumeHook must fail closed to the - // sequential path so the two writers can never diverge. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + // sequential path, or a queue redelivery would commit a second + // hook_received. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld( hook, {}, @@ -441,25 +425,21 @@ describe('resumeHook (parallel fast path)', () => { expect(payloadArg.hookInput).toBeUndefined(); }); - it('takes the parallel path on a dynamic backend attestation (resumeCapabilities) with no static capability', async () => { + it('takes the lazy path on a dynamic backend attestation (resumeCapabilities) with no static capability', async () => { // world-vercel no longer declares the static `hookResumeDedup`; it attests // dedup support FRESH per by-token lookup via the response-only - // `resumeCapabilities`. The parallel path must engage on that signal alone, + // `resumeCapabilities`. The lazy path must engage on that signal alone, // with an otherwise-empty World capability set. const hook = { ...baseHook, - resumeContext: parallelContext, + resumeContext: lazyContext, resumeCapabilities: { hookResumeDedupVersion: HOOK_RESUME_DEDUP_VERSION }, } satisfies Hook; const { createEvent, queue } = makeWorld(hook, {}, {}); await resumeHook(hook.token, { foo: 'bar' }); - expect(createEvent).toHaveBeenCalledTimes(1); - const [, , optsArg] = createEvent.mock.calls[0]; - expect(optsArg.resumeId).toEqual(expect.any(String)); - expect(optsArg.resumePayloadDigest).toMatch(/^[0-9a-f]{64}$/); - + expect(createEvent).not.toHaveBeenCalled(); expect(queue).toHaveBeenCalledTimes(1); const [, payloadArg] = queue.mock.calls[0]; expect(payloadArg.hookInput).toBeDefined(); @@ -469,8 +449,8 @@ describe('resumeHook (parallel fast path)', () => { // A rolled-back or kill-switched server returns a hook with no // `resumeCapabilities`, and world-vercel declares no static capability. // With both attestations absent, resumeHook must fail closed — every new - // resume degrades to the single-writer path with no stranded hooks. - const hook = { ...baseHook, resumeContext: parallelContext } satisfies Hook; + // resume degrades to the eager write with no stranded hooks. + const hook = { ...baseHook, resumeContext: lazyContext } satisfies Hook; const { createEvent, queue } = makeWorld(hook, {}, {}); await resumeHook(hook.token, { foo: 'bar' }); @@ -489,12 +469,12 @@ describe('resumeHook (parallel fast path)', () => { // The response-only `resumeCapabilities` is only trustworthy when fetched // during THIS resume. A public caller passing a pre-fetched Hook object — // e.g. one cached before a server rollback or kill switch, still carrying - // `hookResumeDedupVersion` — must NOT reactivate the parallel path against - // a backend that no longer dedups. Passing a Hook (not a token) skips the + // `hookResumeDedupVersion` — must NOT reactivate the lazy path against a + // backend that no longer dedups. Passing a Hook (not a token) skips the // by-token lookup, so its capability is stale by construction and ignored. const hook = { ...baseHook, - resumeContext: parallelContext, + resumeContext: lazyContext, resumeCapabilities: { hookResumeDedupVersion: HOOK_RESUME_DEDUP_VERSION }, } satisfies Hook; // Empty world capabilities (world-vercel: no static hookResumeDedup) and @@ -503,7 +483,8 @@ describe('resumeHook (parallel fast path)', () => { await resumeHook(hook, { foo: 'bar' }); - // Sequential: no shared idempotency key, no hookInput on the queue message. + // Sequential: eager write with no idempotency key, no hookInput on the + // queue message. expect(createEvent).toHaveBeenCalledTimes(1); const [, , optsArg] = createEvent.mock.calls[0]; expect(optsArg.resumeId).toBeUndefined(); @@ -512,19 +493,19 @@ describe('resumeHook (parallel fast path)', () => { expect(payloadArg.hookInput).toBeUndefined(); }); - it('resumeWebhook takes the parallel path via its internal fresh attestation on a dynamic-only backend', async () => { + it('resumeWebhook takes the lazy path via its internal fresh attestation on a dynamic-only backend', async () => { // The complement to the "caller supplies a stale Hook" fail-closed test: // `resumeWebhook` fetches the hook by token in-line (`getHookByTokenWithKey`) // during this resume, then calls the private `resumeHookImpl` with the // freshness attestation set. That is the ONLY path allowed to trust the // response-only `resumeCapabilities` on a Hook object, so with no static - // world capability the parallel path must still engage — proving the + // world capability the lazy path must still engage — proving the // attestation flows through the webhook entry point (which cannot be // exercised through the public three-arg `resumeHook`). const hook = { ...baseHook, isWebhook: true, - resumeContext: parallelContext, + resumeContext: lazyContext, resumeCapabilities: { hookResumeDedupVersion: HOOK_RESUME_DEDUP_VERSION }, } satisfies Hook; const { createEvent, queue } = makeWorld(hook, {}, {}); @@ -533,11 +514,8 @@ describe('resumeHook (parallel fast path)', () => { // Default webhook (no `respondWith`) resolves to a 202. expect(response.status).toBe(202); - // Parallel: shared idempotency key + hookInput on the queue message. - expect(createEvent).toHaveBeenCalledTimes(1); - const [, , optsArg] = createEvent.mock.calls[0]; - expect(optsArg.resumeId).toEqual(expect.any(String)); - expect(optsArg.resumePayloadDigest).toMatch(/^[0-9a-f]{64}$/); + // Lazy: no event write, hookInput on the queue message. + expect(createEvent).not.toHaveBeenCalled(); const [, payloadArg] = queue.mock.calls[0]; expect(payloadArg.hookInput).toBeDefined(); }); @@ -555,7 +533,7 @@ describe('resumeHook (parallel fast path)', () => { const hook = { ...baseHook, isWebhook: true, - resumeContext: parallelContext, + resumeContext: lazyContext, } satisfies Hook; const { queue } = makeWorld(hook, { // The lookup takes 500ms of wall clock. @@ -581,10 +559,10 @@ describe('resumeHook (parallel fast path)', () => { it('ignores a stale resumeCapabilities below the required dedup version', async () => { // Forward-compat: a future server that lowers its attested version (or a // corrupted/old field below HOOK_RESUME_DEDUP_VERSION) must not engage the - // parallel path — the version gate is a floor, not a mere presence check. + // lazy path — the version gate is a floor, not a mere presence check. const hook = { ...baseHook, - resumeContext: parallelContext, + resumeContext: lazyContext, resumeCapabilities: { hookResumeDedupVersion: HOOK_RESUME_DEDUP_VERSION - 1, }, @@ -593,6 +571,7 @@ describe('resumeHook (parallel fast path)', () => { await resumeHook(hook.token, { foo: 'bar' }); + expect(createEvent).toHaveBeenCalledTimes(1); const [, , optsArg] = createEvent.mock.calls[0]; expect(optsArg.resumeId).toBeUndefined(); const [, payloadArg] = queue.mock.calls[0]; diff --git a/packages/core/src/runtime/resume-hook.ts b/packages/core/src/runtime/resume-hook.ts index 1deb367fd7..39c572e973 100644 --- a/packages/core/src/runtime/resume-hook.ts +++ b/packages/core/src/runtime/resume-hook.ts @@ -21,7 +21,6 @@ import { } from '@workflow/world'; import { monotonicFactory } from 'ulid'; import { getRunCapabilities } from '../capabilities.js'; -import { isRetryableWorldError } from '../classify-error.js'; import { importKey } from '../encryption.js'; import { runtimeLogger } from '../logger.js'; import { decodeRunPublicKey } from '../sealed-box.js'; @@ -44,24 +43,26 @@ import { safeWaitUntil, waitedUntil } from './wait-until.js'; const generateResumeId = monotonicFactory(); /** - * Upper bound on the serialized hook payload that the parallel fast path will - * inline into the queue message's `hookInput`. Vercel Queues caps a single - * message at ~256 KiB, and the message also carries the runId, hookId, token, - * resumeId, digest, and trace carrier alongside CBOR framing overhead. Staying - * well under that ceiling keeps the queue publish from rejecting an oversized - * message, which, on the parallel path, would persist `hook_received` but never - * re-trigger the run. Above this size we fall back to the sequential path, whose - * queue message carries only the run ID (the payload lives in the event log). + * Upper bound on the serialized hook payload that the lazy path will inline + * into the queue message's `hookInput`. Vercel Queues caps a single message at + * ~256 KiB, and the message also carries the runId, hookId, token, resumeId, + * digest, and trace carrier alongside CBOR framing overhead. Staying well under + * that ceiling keeps the queue publish from rejecting an oversized message, + * which on the lazy path would drop the resume entirely: the message is the + * only copy of the payload. Above this size we fall back to the sequential + * path, whose queue message carries only the run ID (the payload lives in the + * event log). */ const MAX_INLINE_RESUME_PAYLOAD_BYTES = 128 * 1024; /** * Hex SHA-256 of the serialized resume payload bytes. Computed once by the - * producer and forwarded verbatim on both the direct `hook_received` write and - * the queue `hookInput`, so both writers of the same `resumeId` record an - * identical digest on the server's `(runId, resumeId)` constraint. Hashing the - * already-serialized bytes (not the raw value) keeps producer and consumer in - * lockstep: the consumer forwards this string without recomputing. + * producer and carried on the queue message's `hookInput`, so every delivery of + * that message records an identical digest against the server's + * `(runId, resumeId)` constraint and redeliveries converge on the one committed + * `hook_received`. Hashing the already-serialized bytes (not the raw value) + * keeps the digest stable across deliveries: the consumer forwards this string + * without recomputing. */ async function computeResumePayloadDigest(bytes: Uint8Array): Promise { const digest = await crypto.subtle.digest('SHA-256', bytes); @@ -203,18 +204,13 @@ export async function getHookByToken(token: string): Promise { * The result of {@link resumeHook}: a {@link Hook} augmented with an optional * resilience signal. * - * On the parallel fast path, `resumeHook()` writes the `hook_received` event and - * dispatches the workflow queue message concurrently. When the direct event - * write fails *transiently* (a 429/5xx, a transport error, or an expected - * `(runId, resumeId)` conflict with its own re-ensuring consumer) but the queue - * dispatch succeeds, the resume is still guaranteed: the queue consumer - * idempotently materializes the `hook_received` event from the payload carried - * on the message before replay. In that recovered case the returned hook carries - * `resilientResume: true`. - * - * On the happy path (the direct write landed) and on the sequential fallback - * path, the flag is absent (`undefined`). Callers that don't care about the - * distinction can treat the result as a plain {@link Hook}. + * `resilientResume` is retained for source compatibility and is never set. It + * signalled a resume whose direct `hook_received` write had failed while the + * queue dispatch succeeded, back when the lazy path raced the two. The lazy + * path no longer writes the event at all (the queue consumer materializes it + * from `hookInput`), so there is no longer a distinction to report, and the + * sequential path never set the flag either. Treat the result as a plain + * {@link Hook}. */ export type ResumedHook = Hook & { resilientResume?: boolean }; @@ -268,7 +264,42 @@ export async function resumeHook( payload, encryptionKeyOverride, false, - Date.now() + Date.now(), + false + ); +} + +/** + * {@link resumeHook} with the `hook_received` event written BEFORE this + * resolves, for the one caller that needs the resume to be durable at that + * instant rather than merely dispatched. + * + * The lazy path leaves the write to the queue consumer, so a normal + * `resumeHook()` resolves while the event is still in flight. That is fine for + * an external resume, whose caller has nothing racing it. It is NOT fine for a + * resume the runtime itself issues as a barrier: a step that aborts a shared + * `AbortController` resumes the hook that records the abort in the event log, + * and that write has to land before the step completes, or the workflow + * continuation `step_completed` enqueues can dispatch the next step with a + * stale, non-aborted signal (see `reviveAbortController` in serialization.ts). + * + * Forcing the eager write costs the round trip the lazy path removes, which is + * the right trade here: this is an internal ordering barrier, not the + * latency-sensitive external resume the optimization targets. The resume span + * reports `resume_fallback_reason: durable_required`. + */ +export async function resumeHookDurable( + tokenOrHook: string | Hook, + payload: T, + encryptionKeyOverride?: PayloadKey +): Promise { + return resumeHookImpl( + tokenOrHook, + payload, + encryptionKeyOverride, + false, + Date.now(), + true ); } @@ -288,13 +319,16 @@ export async function resumeHook( * resolution that hydrates hook metadata, and the `respondWith` setup) and * stamping locally would silently exclude all of it, so the two entry points * would report the same metric over different windows. + * @param requireDurableWrite - Force the sequential path so `hook_received` is + * committed before this resolves. See {@link resumeHookDurable}. */ async function resumeHookImpl( tokenOrHook: string | Hook, payload: T, encryptionKeyOverride: PayloadKey | undefined, hookFreshlyLookedUp: boolean, - resumeRequestedAtMs: number + resumeRequestedAtMs: number, + requireDurableWrite: boolean ): Promise { return await waitedUntil(() => { return trace('hook.resume', async (span) => { @@ -425,38 +459,34 @@ async function resumeHookImpl( // Link to the run-origin context from the stored trace carrier // (skipped when absent or invalid). Resolved before dispatch so both - // the sequential and parallel paths attach it. + // the sequential and lazy paths attach it. const originLink = await linkToTraceCarrier(resumeContext.traceCarrier); if (originLink) { span?.addLink?.(originLink); } - const eventData = { - ...(v1Compat ? {} : { token: hook.token }), - payload: dehydratedPayload, - }; - const queueName = getWorkflowQueueName(resumeContext.workflowName); const queueOptions = { deploymentId: resumeContext.deploymentId, specVersion: resumeContext.runSpecVersion ?? SPEC_VERSION_LEGACY, }; - // Decide whether to parallelize the `hook_received` write and the queue - // publish. The fast path is only safe when EVERY precondition holds; the - // first that fails names the fallback reason (emitted as a span - // attribute for observability, and to make "why did this run sequential" - // answerable in production). All conditions: + // Decide whether the `hook_received` event is written lazily by the + // queue consumer (from the message's `hookInput`) or eagerly here, + // before the publish. The lazy path is only safe when EVERY + // precondition holds; the first that fails names the fallback reason + // (emitted as a span attribute for observability, and to make "why did + // this run sequential" answerable in production). All conditions: // // - kill switch: `WORKFLOW_DISABLE_LAZY_HOOK_RESUME` forces sequential - // if the fast path ever misbehaves. It is an SDK-deployment env var, + // if the lazy path ever misbehaves. It is an SDK-deployment env var, // so changing it generally requires redeploying the workflow // deployment. (The backend can independently drop new resumes to the // sequential path fleet-wide by ceasing to attest dedup support on // the by-token lookup; see the backend-dedup condition below.) // - backend dedup: the live backend must enforce the - // `(runId, resumeId)` constraint, or the two writers would commit two - // `hook_received` events. Fail closed. Attested by EITHER a fresh, + // `(runId, resumeId)` constraint, or a queue redelivery would commit + // a second `hook_received`. Fail closed. Attested by EITHER a fresh, // response-only `hook.resumeCapabilities.hookResumeDedupVersion` from // the by-token lookup (world-vercel: recomputed every read, so a // server rollback or kill switch drops to sequential immediately) OR @@ -466,23 +496,26 @@ async function resumeHookImpl( // resume (`hookResumeCapabilitiesAreFresh`); a Hook object handed in // by a public caller may carry a capability cached before a rollback, // so it is ignored and the path falls back to sequential. - // - consumer support: the target run's deployment must re-ensure the + // - consumer support: the target run's deployment must materialize the // event from `hookInput` on replay. Attested by the run's explicit // `hookResumeInputVersion` execution-context marker (mirrored onto // resumeContext), NOT a version-compare against a predicted release - // cutoff. Absent → a lost producer write would be a lost resume. - // - not legacy: v1Compat runs omit `token` from the producer's event - // body but the consumer's re-ensure always includes it, so the two - // writers would disagree on the event body. Legacy stays sequential. + // cutoff. Absent → nothing would ever write the event, so the resume + // would be lost outright. + // - not legacy: v1Compat runs omit `token` from the eagerly written + // event body but the consumer always includes it, so a legacy run + // would get a different event depending on which path ran. Legacy + // stays sequential. // - CBOR transport: the run must use CBOR queue transport so the binary // payload survives the queue message. // - raw bytes: the dehydrated payload must be a `Uint8Array` (the // content digest that keys the dedup constraint is over these bytes). - // - size: a payload above the queue's message ceiling would fail the - // publish (which, on the parallel path, would persist the event but - // never re-trigger the run), so oversized payloads stay sequential - // (their queue message carries only the run ID). - const parallelResumeDisabled = + // - size: on the lazy path the queue message carries the only copy of + // the payload, so a payload above the message ceiling would fail the + // publish and lose the resume. Oversized payloads stay sequential + // (their queue message carries only the run ID; the payload lives in + // the event log). + const lazyResumeDisabled = process.env.WORKFLOW_DISABLE_LAZY_HOOK_RESUME === '1'; // Backend dedup is supported when EITHER the live server attests it // fresh on this by-token hook (world-vercel: response-only, recomputed @@ -495,60 +528,67 @@ async function resumeHookImpl( ? (hook.resumeCapabilities?.hookResumeDedupVersion ?? 0) : 0) >= HOOK_RESUME_DEDUP_VERSION || world.capabilities?.hookResumeDedup === true; - const fallbackReason: string | null = parallelResumeDisabled - ? 'disabled' - : !backendDedupSupported - ? 'backend_unsupported' - : (resumeContext.hookResumeInputVersion ?? 0) < - HOOK_RESUME_INPUT_VERSION - ? 'consumer_unsupported' - : v1Compat - ? 'legacy' - : (resumeContext.runSpecVersion ?? 0) < - SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT - ? 'non_cbor_transport' - : !(dehydratedPayload instanceof Uint8Array) - ? 'non_bytes' - : dehydratedPayload.byteLength > - MAX_INLINE_RESUME_PAYLOAD_BYTES - ? 'oversized' - : null; - const useParallelResume = fallbackReason === null; + const fallbackReason: string | null = requireDurableWrite + ? // An internal caller needs the event committed before this + // resolves (an ordering barrier), which only the eager write + // provides. Checked first so the span names the real reason + // rather than whichever gate happens to fail alongside it. + 'durable_required' + : lazyResumeDisabled + ? 'disabled' + : !backendDedupSupported + ? 'backend_unsupported' + : (resumeContext.hookResumeInputVersion ?? 0) < + HOOK_RESUME_INPUT_VERSION + ? 'consumer_unsupported' + : v1Compat + ? 'legacy' + : (resumeContext.runSpecVersion ?? 0) < + SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT + ? 'non_cbor_transport' + : !(dehydratedPayload instanceof Uint8Array) + ? 'non_bytes' + : dehydratedPayload.byteLength > + MAX_INLINE_RESUME_PAYLOAD_BYTES + ? 'oversized' + : null; + const useLazyResume = fallbackReason === null; span?.setAttributes({ - 'workflow.hook.resume_strategy': useParallelResume - ? 'parallel' + 'workflow.hook.resume_strategy': useLazyResume + ? 'lazy' : 'sequential', ...(fallbackReason ? { 'workflow.hook.resume_fallback_reason': fallbackReason } : {}), }); - // Re-key any "hook can no longer be received" rejection to - // HookNotFoundError(hook.token) so `.token` matches the pre-fast-path - // contract, where resumeHook threw `HookNotFoundError(hook.token)` - // after its own terminal check. The specific error depends on the - // World: - // - a genuinely missing hook maps to HookNotFoundError (keyed on the - // event correlationId / hook ID); - // - a terminal run on Vercel rejects hook_received with 404, which - // world-vercel maps to HookNotFoundError; - // - a terminal run on world-local / world-postgres rejects with - // RunExpiredError. - // - // On the SEQUENTIAL path an EntityConflictError (HTTP 409) is also - // treated as "hook gone" for historical / conflict-shaped-rejection - // compatibility: there is no queue message in flight, so a conflict has - // no re-ensuring consumer to converge on. The PARALLEL path handles 409 - // differently (see below): it published the queue message before the - // conflict, so the consumer will converge the event. - const isHookGoneError = (err: unknown): boolean => - HookNotFoundError.is(err) || - EntityConflictError.is(err) || - RunExpiredError.is(err); - - if (!useParallelResume) { + if (!useLazyResume) { // Sequential path: create a hook_received event, then re-trigger. + // + // Re-key any "hook can no longer be received" rejection to + // HookNotFoundError(hook.token) so `.token` matches the historical + // contract, where resumeHook threw `HookNotFoundError(hook.token)` + // after its own terminal check. The specific error depends on the + // World: + // - a genuinely missing hook maps to HookNotFoundError (keyed on + // the event correlationId / hook ID); + // - a terminal run on Vercel rejects hook_received with 404, which + // world-vercel maps to HookNotFoundError; + // - a terminal run on world-local / world-postgres rejects with + // RunExpiredError. + // + // An EntityConflictError (HTTP 409) is also treated as "hook gone" + // here for historical / conflict-shaped-rejection compatibility: + // this path holds no queue message in flight, so a conflict has no + // consumer to converge on. + // + // The lazy path performs no write, so it raises none of these: see + // the note on its terminal-run behavior below. + const isHookGoneError = (err: unknown): boolean => + HookNotFoundError.is(err) || + EntityConflictError.is(err) || + RunExpiredError.is(err); try { await world.events.create( hook.runId, @@ -556,7 +596,10 @@ async function resumeHookImpl( eventType: 'hook_received', specVersion: SPEC_VERSION_CURRENT, correlationId: hook.hookId, - eventData, + eventData: { + ...(v1Compat ? {} : { token: hook.token }), + payload: dehydratedPayload, + }, }, { v1Compat } ); @@ -589,124 +632,77 @@ async function resumeHookImpl( return hook; } - // Parallel fast path. A stable idempotency key ties the direct write to - // the queue consumer's re-ensure; the payload digest lets the server - // detect key reuse across byte-identical payloads. + // Lazy path: publish the queue message and let the consumer write the + // `hook_received` event from `hookInput` before it replays. The + // producer writes nothing, so the resume costs exactly one round trip + // (the publish) instead of two, and the event is created once, by the + // side that is about to replay it. + // + // `resumeId` is the idempotency key the consumer sends with that write; + // `payloadDigest` lets the server detect key reuse across + // byte-different payloads. Both ride the message, so every redelivery + // of it converges on the one committed event via the backend's + // `(runId, resumeId)` constraint (the precondition gated above). + // + // Two things a caller could previously infer from a resolved resume, + // and can no longer: + // + // - Visibility. This returns once the publish is accepted, not once + // `hook_received` exists, so a caller that reads the run back + // immediately can see a log without it. Delivery is unaffected: the + // payload is on the message. + // - The run still being live. The hook lookup above still validates + // that a hook holds the token (and which run it belongs to), but + // `resumeContext` is an immutable slice and carries no status, and + // with no write there is no server rejection to observe. A resume + // against an ended run therefore resolves rather than throwing + // HookNotFoundError. It is reachable only while the hook outlives + // its run (minimum retention, or before the token is released); + // otherwise the lookup itself fails. Nothing resumes either way: + // the consumer's write is rejected the same way and the delivery is + // consumed. The paths that do observe the status are unchanged: the + // `run_fallback` terminal pre-check above, and the sequential + // path's own write. const resumeId = generateResumeId(); const payloadDigest = await computeResumePayloadDigest( dehydratedPayload as Uint8Array ); span?.setAttributes({ 'workflow.hook.resume_id': resumeId }); - // Wrapped in a thunk purely so T1 of the TTR window is stamped at the - // exact instant the publish is requested, after the racing - // `hook_received` write has been kicked off, which is where the - // additive `producer_prep` phase must end. The two calls still start - // in the same turn and race exactly as before. - const publishInvocation = () => { - const queuePublishRequestedAtMs = Date.now(); - return world.queue( - queueName, - { - runId: hook.runId, - traceCarrier: resumeContext.traceCarrier ?? undefined, - hookInput: { - resumeId, - hookId: hook.hookId, - token: hook.token, - payload: dehydratedPayload, - payloadDigest, - // Deployment affinity for the consumer's cheap pre-write - // check: lets a misrouted delivery re-route before its - // hoisted hook_received write instead of after. - deploymentId: resumeContext.deploymentId, - }, - hookResumeTiming: { - resumeRequestedAtMs, - queuePublishRequestedAtMs, - strategy: 'parallel', - }, - } satisfies WorkflowInvokePayload, - queueOptions - ); - }; - const [eventResult, queueResult] = await Promise.allSettled([ - world.events.create( - hook.runId, - { - eventType: 'hook_received', - specVersion: SPEC_VERSION_CURRENT, - correlationId: hook.hookId, - eventData, + // T1 of the TTR window, stamped at the instant the publish is + // requested: `producer_prep` covers exactly the work above it (hook + // lookup, key resolution, serialization) and nothing else, since no + // event write remains on this path. + const queuePublishRequestedAtMs = Date.now(); + await world.queue( + queueName, + { + runId: hook.runId, + traceCarrier: resumeContext.traceCarrier ?? undefined, + hookInput: { + resumeId, + hookId: hook.hookId, + token: hook.token, + payload: dehydratedPayload, + payloadDigest, + // Deployment affinity for the consumer's cheap pre-write + // check: lets a misrouted delivery re-route before its + // hoisted hook_received write instead of after. + deploymentId: resumeContext.deploymentId, }, - { v1Compat, resumeId, resumePayloadDigest: payloadDigest } - ), - publishInvocation(), - ]); - - // Queue failure is always fatal: the run was not re-triggered, so no - // consumer will re-ensure the event. - if (queueResult.status === 'rejected') { - throw queueResult.reason; - } - - // Set when the direct write did not land but the queue dispatch did, so - // the consumer materializes the event before replay. Surfaced to the - // caller as `ResumedHook.resilientResume` and on the span. - let resilientResume = false; - if (eventResult.status === 'rejected') { - const err = eventResult.reason; - if (HookNotFoundError.is(err) || RunExpiredError.is(err)) { - // The hook's run has genuinely ended: surface the same contract as - // the sequential path. The queue message already went out, but the - // consumer's re-ensure will also reject against the terminal run - // and no-op, so the workflow does not resume. - throw new HookNotFoundError(hook.token); - } - if (EntityConflictError.is(err) || isRetryableWorldError(err)) { - // Resilient. Two shapes reach here, both non-terminal: - // - EntityConflict (409): this write raced its own re-ensuring - // consumer (or a redrive) on the shared `resumeId`; the run is - // NOT gone, and the consumer converges on the one committed - // event. Unlike the sequential path, a 409 here is expected - // concurrency, not a vanished hook. - // - 429 / 5xx / transport: a transient backend failure. - // In both cases the run WAS re-triggered via the queue and the - // consumer idempotently ensures the hook_received before replay, so - // swallow rather than failing the caller. - // - // Producer telemetry: record that the direct write did not land but - // the resume is still guaranteed via queue-delivered `hookInput`. - // This is the operational signal that the recovery path fired, - // surfaced both on the span and as the public `resilientResume` - // flag on the returned hook (the resume outcome is otherwise - // identical to the happy path from the caller's perspective). - resilientResume = true; - span?.setAttributes({ - ...Attribute.HookResilientResume(true), - 'workflow.hook.resume_event_write_recovered': true, - 'workflow.hook.resume_event_write_error': - err instanceof Error ? err.name : 'unknown', - }); - runtimeLogger.warn( - 'Hook resume event write failed, but the run was re-triggered via ' + - 'the queue. The hook_received event will be ensured by the ' + - 'queue consumer.', - { - workflowRunId: hook.runId, - hookId: hook.hookId, - resumeId, - error: err instanceof Error ? err.message : String(err), - } - ); - } else { - throw err; - } - } + hookResumeTiming: { + resumeRequestedAtMs, + queuePublishRequestedAtMs, + strategy: 'lazy', + }, + } satisfies WorkflowInvokePayload, + queueOptions + ); - return ( - resilientResume ? { ...hook, resilientResume: true } : hook - ) satisfies ResumedHook; + // A rejected publish propagates: the message is the only carrier of + // both the trigger and the payload, so a failed publish is a failed + // resume, with nothing persisted for a later delivery to pick up. + return hook satisfies ResumedHook; } catch (err) { span?.setAttributes({ ...Attribute.HookToken( @@ -805,9 +801,16 @@ export async function resumeWebhook( // `hook` was just fetched via `getHookByTokenWithKey` (a fresh by-token // lookup) above, so its response-only `resumeCapabilities` reflects the live // backend. Call the internal implementation with the fresh attestation so - // the parallel fast path stays available without a second GET. (The public + // the lazy path stays available without a second GET. (The public // `resumeHook` never sets this, so a caller cannot forge it.) - await resumeHookImpl(hook, request, encryptionKey, true, resumeRequestedAtMs); + await resumeHookImpl( + hook, + request, + encryptionKey, + true, + resumeRequestedAtMs, + false + ); if (responseReadable) { // Wait for the readable stream to emit one chunk, diff --git a/packages/core/src/runtime/resume-latency.ts b/packages/core/src/runtime/resume-latency.ts index 2d9c87e67e..0aac11ea65 100644 --- a/packages/core/src/runtime/resume-latency.ts +++ b/packages/core/src/runtime/resume-latency.ts @@ -24,10 +24,13 @@ import * as Attribute from '../telemetry/semantic-conventions.js'; * T7 immediately before stepFn.apply() * ``` * - * The producer's direct `hook_received` POST races the queue publish on the - * parallel fast path, so it deliberately has no phase of its own: the two - * overlap, and representing both as additive phases would double-count. It - * remains visible as a contextual span (`hook.resume`). + * On the lazy path the producer writes no `hook_received` at all (the + * consumer materializes it from `hookInput`), so the window has no producer + * write phase. On the sequential path that write is awaited inside + * `producer_prep`, and for messages from an older producer, which raced the + * write against the publish, it overlapped `producer_prep` rather than adding + * to it. Either way it has no phase of its own; it remains visible as a + * contextual span (`hook.resume`). * * T0/T1 are stamped on the producer's machine and T2..T7 on the consumer's, so * the measurement is subject to cross-machine clock skew. Rather than clamp @@ -39,8 +42,15 @@ import * as Attribute from '../telemetry/semantic-conventions.js'; /** What caused the resumption being measured. Only hooks are measured today. */ export type ResumeTrigger = 'hook'; -/** Which `resumeHook()` dispatch path produced this resume. */ -export type ResumeStrategy = 'parallel' | 'sequential'; +/** + * Which `resumeHook()` dispatch path produced this resume. + * + * `parallel` is only ever received from an older producer, which raced its own + * `hook_received` write against the publish. Current producers send `lazy` (no + * producer write: the consumer materializes the event from `hookInput`) or + * `sequential`. + */ +export type ResumeStrategy = 'lazy' | 'parallel' | 'sequential'; /** * How the consuming invocation initialized its replay state: @@ -125,7 +135,9 @@ export function resumeTrackingFromMessage( } return { trigger: 'hook', - ...(timing.strategy === 'parallel' || timing.strategy === 'sequential' + ...(timing.strategy === 'lazy' || + timing.strategy === 'parallel' || + timing.strategy === 'sequential' ? { strategy: timing.strategy } : {}), resumeRequestedAtMs: timing.resumeRequestedAtMs, diff --git a/packages/core/src/runtime/start.ts b/packages/core/src/runtime/start.ts index fab61ad943..410a399373 100644 --- a/packages/core/src/runtime/start.ts +++ b/packages/core/src/runtime/start.ts @@ -348,8 +348,8 @@ export async function start( let framedByteStreams: boolean; let targetSupportsCompression: boolean; // The consumer's hook-resume protocol version, stamped onto the new run - // so a later `resumeHook()` gates its parallel path on the deployment - // that will actually consume the queue message. `undefined` means "could + // so a later `resumeHook()` gates its lazy path on the deployment that + // will actually consume the queue message. `undefined` means "could // not attest" and fails the gate closed. let targetHookResumeInputVersion: number | undefined; // Public key of the target run, when the capability probe was able to diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index e32f393ea4..794b6f7aff 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -2605,15 +2605,22 @@ function reviveAbortController( // completion) rather than `ops` (best-effort, background). The stream // write above stays in `ops`: it must fire ASAP to reach an in-flight // sibling step and is not the durable record. + // + // `resumeHookDurable`, not `resumeHook`: awaiting the latter only + // guarantees the resume was published, since the lazy path leaves the + // event write to the queue consumer. That would satisfy + // `preCompletionOps` while leaving the very race this ordering exists + // to prevent. + // // Swallow errors here so the promise can only ever enforce ordering // when awaited (see the no-reject contract on // StepContext.preCompletionOps); a failed resume retries on next replay. const hookResume = (async () => { try { - const { resumeHook: resumeHookFn } = await import( + const { resumeHookDurable } = await import( './runtime/resume-hook.js' ); - await resumeHookFn(value.hookToken, { + await resumeHookDurable(value.hookToken, { aborted: true, reason, }); diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index 9695d1a952..fb6754017f 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -430,6 +430,11 @@ export const HookFound = SemanticConvention('workflow.hook.found'); * `hook_received` write failed transiently but the queue dispatch succeeded, so * the resume is recovered via the consumer's re-ensure. Corresponds to * `ResumedHook.resilientResume === true`. + * + * No longer emitted: the lazy path writes no event to fail, and the sequential + * path has no queue-delivered payload to recover from. Retained so dashboards + * and queries built on the attribute keep resolving while older producers are + * still deployed. */ export const HookResilientResume = SemanticConvention( 'workflow.hook.resilient_resume' @@ -438,8 +443,8 @@ export const HookResilientResume = SemanticConvention( /** * Consumer-side signal (on the workflow execution span) that this replay * materialized the `hook_received` event from the queue message's `hookInput` - * because the producer's direct write had not landed, which completes the - * recovery path {@link HookResilientResume} began. + * because no committed event was found, which completes the recovery path + * {@link HookResilientResume} began. * * Legacy / non-atomic re-ensure signal only. Atomic lazy resumes * (resumeId + digest) go through the hoisted preload write instead, whose @@ -568,10 +573,14 @@ export const ResumeTrigger = SemanticConvention<'hook'>( 'workflow.resume.trigger' ); -/** Which `resumeHook()` dispatch path produced this resume. */ -export const ResumeStrategy = SemanticConvention<'parallel' | 'sequential'>( - 'workflow.resume.strategy' -); +/** + * Which `resumeHook()` dispatch path produced this resume. `parallel` only + * appears for messages published by an older producer, which wrote + * `hook_received` itself in parallel with the publish. + */ +export const ResumeStrategy = SemanticConvention< + 'lazy' | 'parallel' | 'sequential' +>('workflow.resume.strategy'); /** * How the consuming invocation initialized replay state. Distinct from the diff --git a/packages/vitest/src/index.ts b/packages/vitest/src/index.ts index 7978889919..bcabf50a9a 100644 --- a/packages/vitest/src/index.ts +++ b/packages/vitest/src/index.ts @@ -320,6 +320,13 @@ export async function waitForSleep( * filter that hasn't had a `hook_received` event. Returns the matching hook, * which you can then resume with `resumeHook(hook.token, data)`. * + * `resumeHook()` resolving does NOT mean the `hook_received` event is visible: + * on the lazy resume path the consuming invocation writes it, so the event + * appears once the run picks the resume up. A loop that resumes and then + * immediately calls this again can therefore be handed back the hook it just + * resumed. Pass `notHookId` with the hook you resumed to wait for the NEXT + * one, or await something that implies the run made progress. + * * @example * ```ts * const run = await start(myWorkflow, ["doc-1"]); @@ -330,7 +337,7 @@ export async function waitForSleep( */ export async function waitForHook( run: Run, - options?: WaitOptions & { token?: string } + options?: WaitOptions & { token?: string; notHookId?: string } ): Promise { const w = getWorldOrThrow(); const timeout = options?.timeout ?? 30_000; @@ -352,7 +359,12 @@ export async function waitForHook( const pendingHook = hooks.find( (h) => !receivedCorrelationIds.has(h.hookId) && - (!options?.token || h.token === options.token) + (!options?.token || h.token === options.token) && + // Skip a hook the caller has already resumed. Its `hook_received` may + // not be written yet (the lazy resume path defers that to the + // consuming invocation), so "no hook_received" alone cannot tell a + // fresh hook from one whose payload is still in flight. + (!options?.notHookId || h.hookId !== options.notHookId) ); if (pendingHook) return pendingHook; @@ -361,6 +373,6 @@ export async function waitForHook( } throw new Error( - `waitForHook timed out after ${timeout}ms: no pending hook found for run ${run.runId}${options?.token ? ` with token "${options.token}"` : ''}` + `waitForHook timed out after ${timeout}ms: no pending hook found for run ${run.runId}${options?.token ? ` with token "${options.token}"` : ''}${options?.notHookId ? ` other than "${options.notHookId}"` : ''}` ); } diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index 91fb379de6..721f149103 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -1412,16 +1412,18 @@ export function createEventsStorage( throw new HookNotFoundError(data.correlationId); } - // Lazy hook resume idempotency: the parallel fast path writes this - // `hook_received` directly AND has the queue consumer re-ensure it, - // both carrying the same `resumeId`, so both may reach here under the - // per-hook lock. They must converge on ONE event. Keyed on - // `(runId, resumeId)` (NOT on the hookId) because a reusable hook + // Lazy hook resume idempotency: the queue consumer writes this + // `hook_received` from the message's `hookInput`, so every delivery + // of one resume's message repeats the write with the same + // `resumeId` and several may reach here under the per-hook lock (a + // redelivery, a deployment-affinity re-route, or an older producer + // that also wrote directly). They must converge on ONE event. Keyed + // on `(runId, resumeId)` (NOT on the hookId) because a reusable hook // receives many distinct resumes and each must record its own event; - // only the two writers of a single resume collapse. The claim pins - // the canonical eventId BEFORE the append so a cross-process writer - // converges too. Gated on `resumeId` so the historical single-write - // path is untouched. + // only the repeated writers of a single resume collapse. The claim + // pins the canonical eventId BEFORE the append so a cross-process + // writer converges too. Gated on `resumeId` so the historical + // single-write path is untouched. if (data.eventType === 'hook_received' && params?.resumeId) { const claimPath = hookResumeClaimPath( basedir, diff --git a/packages/world-local/src/storage/hook-resume-producer-consumer.test.ts b/packages/world-local/src/storage/hook-resume-producer-consumer.test.ts index c3eab3eec3..5807df4739 100644 --- a/packages/world-local/src/storage/hook-resume-producer-consumer.test.ts +++ b/packages/world-local/src/storage/hook-resume-producer-consumer.test.ts @@ -11,22 +11,24 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { createHook, createRun, updateRun } from '../test-helpers.js'; import { createStorage } from './index.js'; -// End-to-end coverage of the lazy hook resume's TWO writers against the +// End-to-end coverage of the lazy hook resume's repeated writers against the // world-local backend (the dev World that advertises `hookResumeDedup`). // -// On the parallel fast path, `resumeHook()` (the PRODUCER) writes the -// `hook_received` event directly AND publishes a queue message carrying the -// same `resumeId`; the queue consumer's re-ensure (the CONSUMER) writes the -// same `hook_received` again before replay. Whichever lands first, the backend -// must converge both onto exactly ONE committed event so replay never sees a -// duplicated resume — the same guarantee the Vercel server enforces with its -// `(runId, resumeId)` constraint. +// On the lazy path, `resumeHook()` publishes a queue message carrying the +// `resumeId` and writes nothing itself; the consumer materializes +// `hook_received` from that message before replay. Every delivery of the +// message repeats that write with the same `resumeId`, and a run's own +// deployment-affinity re-route or a queue redelivery can therefore issue it +// more than once. The backend must converge them onto exactly ONE committed +// event so replay never sees a duplicated resume — the same guarantee the +// Vercel server enforces with its `(runId, resumeId)` constraint. The same +// convergence still covers an older producer that raced its own direct write +// against the publish, which is why both orderings are exercised below. // // These tests drive `storage.events.create` with the exact arguments the real -// producer (packages/core/src/runtime/resume-hook.ts) and consumer -// (packages/core/src/runtime.ts) pass, so they exercise the convergence, -// ordering, concurrency, and terminal-run behavior that the mocked core unit -// tests (resume-hook.parallel.test.ts) cannot. +// consumer (packages/core/src/runtime.ts) passes, so they exercise the +// convergence, ordering, concurrency, and terminal-run behavior that the +// mocked core unit tests (resume-hook.lazy.test.ts) cannot. describe('world-local lazy hook resume: producer/consumer convergence', () => { let testDir: string; let storage: Storage; diff --git a/packages/world-sim/DESIGN.md b/packages/world-sim/DESIGN.md index 134b42df84..d861c6c452 100644 --- a/packages/world-sim/DESIGN.md +++ b/packages/world-sim/DESIGN.md @@ -816,11 +816,11 @@ Two step bodies inside one delivery are genuinely concurrent and separately steerable, which is enough to reach the interesting corruption without a second invocation. That is why the limit has been acceptable so far. -**The parallel hook-resume path is never exercised.** `resumeHook` picks -parallel ("lazy") vs sequential from `world.capabilities.hookResumeDedup` (or a -fresh server attestation). `world-local` declares it and `world-vercel` attests -it per lookup, so **every real world takes the parallel path**, where the queue -publish races the `hook_received` write and the consumer re-ensures the event +**The lazy hook-resume path is never exercised.** `resumeHook` picks lazy vs +sequential from `world.capabilities.hookResumeDedup` (or a fresh server +attestation). `world-local` declares it and `world-vercel` attests it per +lookup, so **every real world takes the lazy path**, where the producer writes +no event and the consumer materializes `hook_received` from the queue message through the durable `(runId, resumeId)` claim. The sim advertises neither the capability nor a `resumeId` dedupe, so every sim hook delivery takes the sequential path, meaning the hook-timing shapes in this book are the *legacy* diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 2d8bb2c4f9..85283f11aa 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -193,10 +193,10 @@ interface CreateEventV4InputBase { hookIsWebhook?: boolean; hookIsSystem?: boolean; /** Lazy hook resume idempotency key. Set only on a `hook_received` written - * by `resumeHook()`'s parallel fast path; routes the event through the - * server's `(runId, resumeId)` constraint so the direct write and the - * queue consumer's re-ensure converge on one event. Older servers ignore - * it (the deduplication then falls to the sequential path). */ + * from a queue message's `hookInput`; routes the event through the + * server's `(runId, resumeId)` constraint so repeated deliveries of one + * resume converge on one event. Older servers ignore it (the resume then + * falls to the sequential path, which writes the event eagerly). */ resumeId?: string; errorCode?: string; /** run_cancelled's optional free-text cancellation reason. Small plaintext diff --git a/packages/world-vercel/src/index.ts b/packages/world-vercel/src/index.ts index beb0dde9bd..2ab93b4f25 100644 --- a/packages/world-vercel/src/index.ts +++ b/packages/world-vercel/src/index.ts @@ -49,7 +49,7 @@ export function createWorld(config?: APIConfig): World { // Vercel deployments are atomic and immutable, so a deployment id names // one fixed build for its whole lifetime. deploymentAffinity: true, - // NOTE: the backend half of resumeHook()'s parallel fast path (that + // NOTE: the backend half of resumeHook()'s lazy path (that // the server enforces the `(runId, resumeId)` dedup constraint) is // NO LONGER a static world capability here. It is attested per-lookup by // the server via `Hook.resumeCapabilities.hookResumeDedupVersion` diff --git a/packages/world/src/hooks.ts b/packages/world/src/hooks.ts index 136330e793..cad3364357 100644 --- a/packages/world/src/hooks.ts +++ b/packages/world/src/hooks.ts @@ -30,8 +30,8 @@ export const HookResumeContextSchema = z.object({ // Feature marker: the version of the lazy-hook-resume consumer protocol the // run's creating deployment supports. Present (>= 1) means that deployment's // `@workflow/core` re-ensures the `hook_received` event from the queue - // message's `hookInput` on replay, so `resumeHook()`'s parallel fast path is - // safe to use. Because a run is pinned to its creating deployment, this + // message's `hookInput` on replay, so `resumeHook()`'s lazy path is safe to + // use. Because a run is pinned to its creating deployment, this // marker is a reliable per-run attestation, unlike inferring support from a // version compare against a predicted release cutoff. Absent on runs created // before the marker existed (fall back to the sequential path). @@ -45,18 +45,18 @@ export type HookResumeContext = z.infer; * deployment stamps this into its execution context (and the server mirrors it * onto `HookResumeContext.hookResumeInputVersion`) to attest that its * `@workflow/core` re-ensures the `hook_received` event from a queue message's - * `hookInput`. `resumeHook()`'s parallel fast path requires the target run's - * marker to be at least this value. Bump only on a breaking change to the + * `hookInput`. `resumeHook()`'s lazy path requires the target run's marker to + * be at least this value. Bump only on a breaking change to the * `hookInput` re-ensure contract. */ export const HOOK_RESUME_INPUT_VERSION = 1; /** * Current version of the backend lazy-hook-resume dedup contract: the live - * backend enforces a `(runId, resumeId)` constraint so the direct write and the - * queue consumer's re-ensure converge on exactly one `hook_received`. - * `resumeHook()`'s parallel fast path requires the backend to attest at least - * this version. Bump only on a breaking change to the constraint semantics. + * backend enforces a `(runId, resumeId)` constraint so repeated deliveries of + * one resume's queue message converge on exactly one `hook_received`. + * `resumeHook()`'s lazy path requires the backend to attest at least this + * version. Bump only on a breaking change to the constraint semantics. */ export const HOOK_RESUME_DEDUP_VERSION = 1; @@ -116,7 +116,7 @@ export const HookSchema = z.object({ // lookup: RESPONSE-ONLY and TRANSIENT. Never persisted on the hook entity // and never part of `resumeContext`, so a server rollback or kill switch // takes effect on the next lookup (the field stops appearing). - // `resumeHook()` gates its parallel fast path on this being present and + // `resumeHook()` gates its lazy path on this being present and // current. Absent against an older/rolled-back server or when the kill switch // is active. resumeCapabilities: HookResumeCapabilitiesSchema.optional(), diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index d2b0b396b9..0dc2705260 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -494,14 +494,15 @@ export interface WorldCapabilities { * The World's `events.create` deduplicates concurrent `hook_received` writes * that carry the same `(runId, resumeId)`, collapsing them onto a single * committed event and returning the canonical one to every caller. This is - * the backend half of `resumeHook()`'s parallel fast path: the producer's - * direct write and the queue consumer's re-ensure both write the same - * `resumeId`, and exactly one event must survive or the run replays a + * the backend half of `resumeHook()`'s lazy path: every delivery of one + * resume's queue message writes the same `resumeId` (a redelivery, a + * deployment-affinity re-route, or an older producer's direct write racing + * its own consumer), and exactly one event must survive or the run replays a * duplicated `hook_received`. * - * The core runtime fails closed on this: the parallel path is taken ONLY - * when the World declares `hookResumeDedup === true` AND the target run's - * deployment can re-ensure from `hookInput` (see the execution-context + * The core runtime fails closed on this: the lazy path is taken ONLY when + * the World declares `hookResumeDedup === true` AND the target run's + * deployment can materialize from `hookInput` (see the execution-context * marker `hookResumeInputVersion`). A World that accepts a `resumeId` but * does not enforce the `(runId, resumeId)` constraint must leave this unset * so the runtime keeps the sequential single-writer path. diff --git a/packages/world/src/queue.ts b/packages/world/src/queue.ts index ba4ccd6d8f..912c19f065 100644 --- a/packages/world/src/queue.ts +++ b/packages/world/src/queue.ts @@ -136,15 +136,15 @@ export type RunInput = z.infer; /** * Lazy hook resume data carried through the queue alongside a workflow - * invocation. Present only when `resumeHook()` takes the parallel fast path: - * the producer persists the `hook_received` event and publishes this invocation - * concurrently. On receipt, a consumer that understands `hookInput` idempotently - * ensures the `hook_received` event exists (keyed by `resumeId`) before - * replaying, so the two concurrent writes converge on exactly one event. + * invocation. Present only when `resumeHook()` takes the lazy path, where the + * producer publishes this invocation and writes no event of its own. On + * receipt, a consumer that understands `hookInput` idempotently ensures the + * `hook_received` event exists (keyed by `resumeId`) before replaying, so + * repeated deliveries of the same message converge on exactly one event. * * The `payload` is the already-serialized (and possibly encrypted) resume - * payload: the identical bytes the producer also sent on the direct - * `events.create`, so both server receipts hash to the same digest under the + * payload, and on this path the queue message is its only carrier. Every write + * derived from this message therefore hashes to the same digest under the * `(runId, resumeId)` constraint. */ /** @@ -250,7 +250,10 @@ export const HookResumeTimingSchema = z.object({ resumeRequestedAtMs: z.number(), /** Epoch ms immediately before the queue publish was requested. */ queuePublishRequestedAtMs: z.number(), - /** Which `resumeHook()` dispatch path ran: `parallel` or `sequential`. */ + /** + * Which `resumeHook()` dispatch path ran: `lazy` or `sequential` (`parallel` + * from producers predating lazy-only resume). + */ strategy: z.string().optional(), /** Epoch ms the final consumer's queue handler was entered. */ consumerStartedAtMs: z.number().optional(), @@ -342,7 +345,7 @@ export const WorkflowInvokePayloadSchema = z.object({ stepInput: StepDispatchInputSchema.optional(), /** * Hook-resume TTR timing. Present on both `resumeHook()` dispatch paths - * (unlike `hookInput`, which only rides the parallel fast path), and + * (unlike `hookInput`, which only rides the lazy path), and * forwarded onto a dispatched step message when the resuming invocation * hands the next durable step to another invocation. Purely observational. * See {@link HookResumeTimingSchema}. diff --git a/workbench/vitest/test/hook-token-reuse.test.ts b/workbench/vitest/test/hook-token-reuse.test.ts index ffb18f1501..937bca46f7 100644 --- a/workbench/vitest/test/hook-token-reuse.test.ts +++ b/workbench/vitest/test/hook-token-reuse.test.ts @@ -14,12 +14,24 @@ describe('hook token reuse after dispose', () => { const rounds = 3; const run = await start(reuseHookTokenWorkflow, [token, rounds]); + // Each round must wait for a hook OTHER than the one it just resumed. + // `resumeHook` returning does not mean the payload has been recorded: on + // the lazy path the consuming invocation writes `hook_received`, so the + // previous round's hook can still look un-received here and be resumed a + // second time, starving the round that was waiting for its own payload. + let resumedHookId: string | undefined; for (let round = 0; round < rounds; round++) { const settled = await Promise.race([ - waitForHook(run, { token }).then(() => 'hook' as const), + waitForHook(run, { token, notHookId: resumedHookId }).then( + (hook) => hook + ), run.returnValue.then((value) => ({ value })), ]); - expect(settled, `round ${round} should register a hook`).toBe('hook'); + expect( + settled, + `round ${round} should register a hook, got ${JSON.stringify(settled)}` + ).toHaveProperty('hookId'); + resumedHookId = (settled as { hookId: string }).hookId; await resumeHook(token, { n: round }); }