diff --git a/.changeset/lazy-hook-replay-preload.md b/.changeset/lazy-hook-replay-preload.md new file mode 100644 index 0000000000..d9ae64fe8d --- /dev/null +++ b/.changeset/lazy-hook-replay-preload.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': minor +'@workflow/world-vercel': patch +'@workflow/core': patch +--- + +Initialize lazy hook resume replay from the `hook_received` write via the new advisory `preloadEvents` param, skipping `run_started` and the initial `events.list`; safe fallback otherwise. diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 605021f660..87d11c4556 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -787,6 +787,12 @@ export function workflowEntrypoint( let workflowStartedAt = -1; let preloadedEvents: Event[] | undefined; let preloadedEventsCursor: string | null | undefined; + // True only when `preloadedEvents` is known to be the + // COMPLETE event log (the lazy hook fast path's validated + // hasMore-false preload). Lets QuickJS trust it as-is — + // its own heuristic only recognizes first-invocation + // (run_created/run_started-only) preloads. + let preloadedEventsComplete = false; // Latency telemetry (TTFS) state — see runtime/step-latency.ts. // Whether this invocation's FIRST event snapshot contained @@ -1099,6 +1105,7 @@ export function workflowEntrypoint( eventsCursor = null; preloadedEvents = undefined; preloadedEventsCursor = undefined; + preloadedEventsComplete = false; pendingInlineDelta = null; // The corrected log inserts the missing events BELOW the // length already scanned for payload prewarming, shifting @@ -1442,9 +1449,213 @@ export function workflowEntrypoint( } } + // --- Lazy hook resume fast path --- + // A lazy hook delivery (resumeId + digest on the queue + // message) hoists the consumer's idempotent hook_received + // re-ensure above run_started and asks the World to return + // the current replay log with the write (preloadEvents). A + // supporting World answers with the reconstructed run, the + // complete replay log, and the run's event ceiling — + // everything the generic setup below would spend a + // run_started POST and an events.list on. Any other result + // (older server, a World that ignores the param, or a + // preload that fails validation, including a bounded + // hasMore page) falls back to that generic setup; the + // hook_received write itself has still succeeded either + // way, so the re-ensure block below is skipped via + // `hookEnsured`. Never taken on turbo (turbo deliveries + // carry runInput, not hookInput) or after the background + // step path already loaded the run. + let hookEnsured = false; + if ( + !workflowRun && + hookInput && + hookInput.resumeId !== undefined && + hookInput.payloadDigest !== undefined + ) { + const hookResumeInput = hookInput; + // Date the materialized event to when the resume actually + // occurred — same derivation as the re-ensure below (the + // resumeId is a ULID minted by resumeHook() at resume + // time). + let occurredAt: Date | undefined; + try { + occurredAt = new Date( + decodeTime(hookResumeInput.resumeId) + ); + } catch { + occurredAt = undefined; + } + try { + span?.addEvent('workflow.hook_received.create.start', { + 'workflow.hook_received.preload_events': true, + }); + const result = await createEvent( + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hookResumeInput.hookId, + eventData: { + token: hookResumeInput.token, + payload: hookResumeInput.payload, + }, + }, + { + requestId, + occurredAt, + resumeId: hookResumeInput.resumeId, + resumePayloadDigest: hookResumeInput.payloadDigest, + preloadEvents: true, + } + ); + hookEnsured = true; + // Note: unlike the re-ensure below, this hoisted write + // does NOT set HookResilientResumeMaterialized — it + // runs on every fast-path resume, including the common + // case where the producer's direct write already landed + // and this call merely converged on it, so it carries + // no recovery signal. workflow.resume_setup_source + // (below) describes this path instead. + + // The preload is usable as replay input only when it is + // demonstrably the complete picture: a reconstructed + // run with a start time, a non-empty COMPLETE log + // (hasMore false — this path has no cursor-continuation + // machinery, so a bounded page must not be trusted), + // the server's event ceiling (this response plays + // run_started's role, so a missing ceiling would leave + // event-limit enforcement disabled for the run), + // both run lifecycle events, the canonical event this + // write converged on, and the hook_received matching + // THIS resume (so we never replay against a log that is + // missing the very event that triggered this delivery). + const usableReplayPreload = + result.run !== undefined && + result.run.startedAt !== undefined && + result.event !== undefined && + result.events !== undefined && + result.events.length > 0 && + result.cursor != null && + result.hasMore === false && + typeof result.maxEvents === 'number' && + result.events.some( + (e) => e.eventType === 'run_created' + ) && + result.events.some( + (e) => e.eventType === 'run_started' + ) && + result.events.some( + (e) => + e.eventType === 'hook_received' && + e.resumeId === hookResumeInput.resumeId + ); + + if ( + usableReplayPreload && + result.run?.startedAt && + result.events + ) { + // The reconstructed run always reads 'running', but a + // terminal event committed concurrently rides in the + // log itself. The node replay loop would catch it, but + // QuickJS dispatches before that check — so consume + // the delivery here, before any engine runs. Same + // outcome as the run_started path's non-running + // status check. + const terminalEvent = result.events.find( + (e) => + e.runId === runId && + (e.eventType === 'run_completed' || + e.eventType === 'run_failed' || + e.eventType === 'run_cancelled') + ); + if (terminalEvent) { + // The preload still initialized this delivery (it + // is how the terminal state was observed), so the + // setup-source and the run's actual terminal + // status are recorded before consuming. + span?.setAttributes({ + ...Attribute.WorkflowRunStatus( + terminalEvent.eventType === 'run_completed' + ? 'completed' + : terminalEvent.eventType === 'run_failed' + ? 'failed' + : 'cancelled' + ), + ...Attribute.HookResumeSetupSource( + 'hook_received_stream' + ), + }); + runtimeLogger.info( + 'Run already finished during lazy hook setup, skipping', + { + workflowRunId: runId, + eventType: terminalEvent.eventType, + } + ); + return; + } + workflowRun = result.run; + maxEventsLimit = clampMaxEvents(result.maxEvents); + // Anchors RSFS — see the declaration above. This + // response plays run_started's role on this path. + runStartedReceivedAtMs = Date.now(); + preloadedEvents = result.events; + preloadedEventsCursor = result.cursor; + // The validated preload is the COMPLETE log (hasMore + // false), so QuickJS may trust it as-is instead of + // refetching — its own first-invocation heuristic + // only recognizes run_created/run_started preloads. + preloadedEventsComplete = true; + workflowStartedAt = +result.run.startedAt; + span?.setAttributes({ + ...Attribute.WorkflowRunStatus(result.run.status), + ...Attribute.WorkflowStartedAt(workflowStartedAt), + ...Attribute.HookResumeSetupSource( + 'hook_received_stream' + ), + }); + } else { + // Successful write, no usable preload (CBOR response + // from an older server, a World that ignored the + // opt-in, a bounded hasMore page, or a preload that + // failed validation): take the generic run_started + // setup below. Its preload is loaded after this write + // committed, so the canonical hook_received is part + // of whatever log that setup reads — no splice + // needed. + span?.setAttributes( + Attribute.HookResumeSetupSource( + 'hook_received_fallback' + ) + ); + } + } catch (err) { + // Same classification as the re-ensure below: + // - HookNotFound / RunExpired: the run went terminal; + // nothing left to resume, consume the message. + // - anything else (EntityConflict, a truncated preload + // stream, transport failures): transient — rethrow so + // the queue redelivers and the idempotent + // (runId, resumeId) claim converges on retry. + if ( + HookNotFoundError.is(err) || + RunExpiredError.is(err) + ) { + runtimeLogger.info( + 'Run already finished during lazy hook setup, skipping', + { workflowRunId: runId, message: err.message } + ); + return; + } + throw err; + } + } + // --- Infrastructure: prepare the run state --- // Skip if workflowRun was already set by the background - // step path (inline replay after all parallel steps done). + // step path (inline replay after all parallel steps done) + // or by the lazy hook fast path above. if (!workflowRun) { // Always call run_started directly — this both transitions // the run to 'running' AND returns the run entity, saving @@ -1671,8 +1882,12 @@ export function workflowEntrypoint( // (the server resolves a matching claim as success, not an // error). `hookInput` never rides a turbo first-delivery // (that path carries `runInput`, not `hookInput`), so this - // only runs on the normal load-and-replay path. - if (hookInput) { + // only runs on the normal load-and-replay path. Skipped + // entirely when the fast path above already ensured the + // event (`hookEnsured`) — successfully or via its fallback, + // whose run_started preload was loaded after the write and + // therefore already contains the canonical event. + if (hookInput && !hookEnsured) { // Perf (Option A): if the producer's concurrent direct // write already landed in the run_started preload, the // canonical event is in the log and the re-ensure round trip @@ -1682,6 +1897,13 @@ export function workflowEntrypoint( // hook_received). Best-effort: the win lands only when the // producer's write beat this consumer's load; otherwise we // fall through to the idempotent re-ensure below. + // + // Intentionally unreachable for atomic lazy resumes + // (resumeId + digest): those set `hookEnsured` in the + // hoisted fast path above — on success AND on its + // fallback — so this block (and the skip check) now only + // serves hookInput shapes without the full idempotency + // pair. const alreadyPreloaded = hookInput.resumeId !== undefined && preloadedEvents?.some( @@ -1807,6 +2029,7 @@ export function workflowEntrypoint( } else { preloadedEvents = undefined; preloadedEventsCursor = undefined; + preloadedEventsComplete = false; } } // end else (re-ensure needed) } @@ -1940,6 +2163,7 @@ export function workflowEntrypoint( workflowName, workflowRun, preloadedEvents, + preloadedEventsComplete, runInput, parentSpan: span, maxEventsLimit, diff --git a/packages/core/src/runtime/quickjs-entrypoint.preload.test.ts b/packages/core/src/runtime/quickjs-entrypoint.preload.test.ts new file mode 100644 index 0000000000..bb4e3cb8fa --- /dev/null +++ b/packages/core/src/runtime/quickjs-entrypoint.preload.test.ts @@ -0,0 +1,197 @@ +/** + * Pins the QuickJS engine's event-log sourcing for the lazy hook resume + * fast path: a caller-attested complete preload (`preloadedEventsComplete`) + * is trusted as the full log — no `events.list` — while a non-attested + * hook-containing preload is NOT trusted (the first-invocation heuristic + * only recognizes run_created/run_started-only preloads) and the engine + * fetches the log itself. Also pins the non-empty guard on the attestation. + * + * The QuickJS VM itself is mocked (its WASM import chain is irrelevant to + * the sourcing decision): `startQuickJSWorkflow` records which events the + * engine handed it and completes immediately. + */ +import { + type CreateEventRequest, + type Event, + SPEC_VERSION_CURRENT, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { monotonicFactory } from 'ulid'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { dehydrateStepReturnValue } from '../serialization.js'; +import { setWorld } from './world.js'; + +vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn() })); +vi.mock('./get-port-lazy.js', () => ({ + getPortLazy: vi.fn().mockResolvedValue(3000), +})); + +const startQuickJSWorkflow = vi.fn(); +vi.mock('./quickjs-runtime.js', () => ({ + startQuickJSWorkflow: (...args: unknown[]) => startQuickJSWorkflow(...args), +})); + +async function runQuickJSScenario(options: { + preloadedEvents?: Event[]; + preloadedEventsComplete?: boolean; +}) { + const runId = 'wrun_quickjs_preload'; + const workflowName = 'workflow'; + const startedAt = new Date('2026-05-19T12:00:00.000Z'); + + const workflowRun: WorkflowRun = { + runId, + workflowName, + status: 'running', + input: [], + deploymentId: 'dpl_quickjs_preload', + specVersion: SPEC_VERSION_CURRENT, + startedAt, + createdAt: startedAt, + updatedAt: startedAt, + }; + + const durableEvents = options.preloadedEvents ?? []; + + const createdEvents: CreateEventRequest[] = []; + let listCallCount = 0; + const listEvents = vi.fn(async () => { + // Model real pagination: the full log on the first page, then an empty + // terminal page (a fake that always returns rows would loop the + // engine's fetch-all forever). + listCallCount++; + if (listCallCount === 1) { + return { + data: [...durableEvents], + cursor: durableEvents.at(-1)?.eventId ?? null, + hasMore: false, + }; + } + return { data: [], cursor: null, hasMore: false }; + }); + const createEvent = vi.fn( + async (_runId: string, request: CreateEventRequest) => { + createdEvents.push(request); + return { event: { ...request, runId, eventId: 'evnt_created' } }; + } + ); + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: {}, + events: { list: listEvents, create: createEvent }, + runs: { get: vi.fn(async () => workflowRun) }, + queue: vi.fn().mockResolvedValue({ messageId: 'msg_quickjs' }), + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World); + + const completedResult = await dehydrateStepReturnValue( + 'done', + runId, + undefined + ); + startQuickJSWorkflow.mockResolvedValue({ + result: { completed: { result: completedResult } }, + continueWithEvents: vi.fn(), + dispose: vi.fn(), + }); + + const { runWorkflowWithQuickJS } = await import('./quickjs-entrypoint.js'); + await runWorkflowWithQuickJS({ + workflowCode: '// not evaluated: the VM is mocked', + workflowName, + workflowRun, + preloadedEvents: options.preloadedEvents, + preloadedEventsComplete: options.preloadedEventsComplete, + }); + + expect(startQuickJSWorkflow).toHaveBeenCalledTimes(1); + const vmEvents = ( + startQuickJSWorkflow.mock.calls[0][0] as { events: Event[] } + ).events; + + return { listEvents, createdEvents, vmEvents }; +} + +function makeHookResumeLog(runId: string): Event[] { + const hostUlid = monotonicFactory(); + const startedAt = new Date('2026-05-19T12:00:00.000Z'); + let eventIndex = 0; + const event = (data: Record): Event => { + const t = +startedAt + ++eventIndex * 100; + return { + specVersion: SPEC_VERSION_CURRENT, + ...data, + runId, + eventId: `evnt_${hostUlid(t)}`, + createdAt: new Date(t), + } as Event; + }; + return [ + event({ + eventType: 'run_created', + eventData: { + deploymentId: 'dpl_quickjs_preload', + workflowName: 'workflow', + input: [], + }, + }), + event({ eventType: 'run_started' }), + event({ + eventType: 'hook_created', + correlationId: 'hook_1', + eventData: { token: 'tok-quickjs' }, + }), + event({ + eventType: 'hook_received', + correlationId: 'hook_1', + resumeId: 'resume-quickjs-1', + eventData: { token: 'tok-quickjs', payload: new Uint8Array() }, + }), + ]; +} + +describe('QuickJS lazy hook preload sourcing', () => { + afterEach(() => { + setWorld(undefined); + vi.clearAllMocks(); + }); + + it('trusts an attested complete preload: no events.list, VM gets the provided log', async () => { + const log = makeHookResumeLog('wrun_quickjs_preload'); + const { listEvents, createdEvents, vmEvents } = await runQuickJSScenario({ + preloadedEvents: log, + preloadedEventsComplete: true, + }); + + expect(listEvents).not.toHaveBeenCalled(); + expect(vmEvents.map((e) => e.eventId)).toEqual(log.map((e) => e.eventId)); + // The engine never posts run_started itself; the only write on this + // invocation is the completion. + expect(createdEvents.map((e) => e.eventType)).toEqual(['run_completed']); + }); + + it('does not trust a hook-containing preload without the attestation: fetches via events.list', async () => { + const log = makeHookResumeLog('wrun_quickjs_preload'); + const { listEvents, vmEvents } = await runQuickJSScenario({ + preloadedEvents: log, + preloadedEventsComplete: false, + }); + + // The first-invocation heuristic rejects a log with hook events, so the + // engine fetched the authoritative log itself... + expect(listEvents).toHaveBeenCalled(); + // ...and replayed what the fetch returned. + expect(vmEvents.map((e) => e.eventId)).toEqual(log.map((e) => e.eventId)); + }); + + it('does not trust an attested but empty preload: fetches via events.list', async () => { + const { listEvents } = await runQuickJSScenario({ + preloadedEvents: [], + preloadedEventsComplete: true, + }); + + expect(listEvents).toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index 57682050d1..1cf72cd399 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -539,12 +539,21 @@ export async function runWorkflowWithQuickJS(params: { workflowName: string; workflowRun: WorkflowRun; /** - * Events returned inline by `events.create('run_started', ...)`. When - * they indicate a first invocation, they are used as the event log - * instead of fetching via `events.list`, matching the node:vm engine's - * fast path. + * Events returned inline by `events.create('run_started', ...)` or by + * the lazy hook fast path's `hook_received` preload. When they indicate + * a first invocation — or when `preloadedEventsComplete` attests they + * are the complete log — they are used as the event log instead of + * fetching via `events.list`, matching the node:vm engine's fast path. */ preloadedEvents?: Event[]; + /** + * True when the caller has validated that `preloadedEvents` is the run's + * COMPLETE event log (e.g. the lazy hook fast path's hasMore-false + * replay preload). The first-invocation heuristic below only recognizes + * run_created/run_started-only preloads, so without this attestation a + * hook-resume preload would be discarded and refetched. + */ + preloadedEventsComplete?: boolean; /** * Run input carried through the queue message on first delivery. Used * as a last-resort fallback for `run_created.eventData.input` when @@ -601,6 +610,7 @@ export async function runWorkflowWithQuickJS(params: { workflowName, workflowRun, preloadedEvents, + preloadedEventsComplete, runInput, parentSpan, maxEventsLimit, @@ -672,10 +682,15 @@ export async function runWorkflowWithQuickJS(params: { // Load the FULL event log for the run. On first invocation the // preloaded events from the run_started response are the complete log - // and save the events.list round-trips. + // and save the events.list round-trips; a caller-attested complete + // preload (lazy hook fast path) is trusted the same way. let events: Event[]; let eventsFetchedPages = 0; - const usePreloaded = isFirstInvocation(preloadedEvents); + const usePreloaded = + (preloadedEventsComplete === true && + Array.isArray(preloadedEvents) && + preloadedEvents.length > 0) || + isFirstInvocation(preloadedEvents); if (usePreloaded && preloadedEvents) { events = preloadedEvents; } else { diff --git a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts index 3e37a92f2e..7c569f5514 100644 --- a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts +++ b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts @@ -1,19 +1,28 @@ /** - * Consumer-side coverage for lazy hook resume Perf (Option A): the queue - * consumer that receives a `hookInput` must idempotently ensure the - * `hook_received` event before replay — EXCEPT when the producer's concurrent - * direct write already landed in the `run_started` preload, in which case the - * re-ensure round trip is pure overhead and is skipped. + * Consumer-side coverage for lazy hook resume: the queue consumer that + * receives a `hookInput` hoists its idempotent `hook_received` re-ensure + * above `run_started` and asks the World to return the current replay log + * with the write (`preloadEvents`). A usable preload initializes the whole + * invocation from that one call (no `run_started` write, no `events.list`); + * anything else — including a bounded `hasMore` page, which this path has no + * continuation machinery for — falls back to the generic `run_started` setup + * without posting the hook a second time. * * Drives the real `workflowEntrypoint` replay loop (not just the helpers) so - * the skip / re-ensure decision, the `eventData` reconstruction from - * `hookInput`, and the in-order splice into the preloaded log are all exercised - * end to end. Uses real ULID event IDs and a seeded VM context so the derived - * hook correlation id matches what replay computes — modeled on - * precondition-guard-replay.test.ts. + * the fast path, the fallback, the preload validation, and the error + * classification are all exercised end to end. Uses real ULID event IDs and a + * seeded VM context so the derived hook correlation id matches what replay + * computes — modeled on precondition-guard-replay.test.ts. */ +import { trace as otelTrace } from '@opentelemetry/api'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; import { EntityConflictError, HookNotFoundError } from '@workflow/errors'; import { + type CreateEventParams, type CreateEventRequest, type Event, SPEC_VERSION_CURRENT, @@ -21,7 +30,15 @@ import { type World, } from '@workflow/world'; import { monotonicFactory } from 'ulid'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + afterAll, + afterEach, + beforeAll, + describe, + expect, + it, + vi, +} from 'vitest'; import { workflowEntrypoint } from '../runtime.js'; import { dehydrateStepReturnValue, @@ -36,6 +53,30 @@ vi.mock('@workflow/utils/get-port', () => ({ getPort: vi.fn().mockResolvedValue(3000), })); +// In-memory span capture so scenarios can assert the execution span's +// attributes (setup source, run status) — the runtime writes them via the +// global tracer provider. +const spanExporter = new InMemorySpanExporter(); +const tracerProvider = new BasicTracerProvider(); + +beforeAll(() => { + tracerProvider.addSpanProcessor(new SimpleSpanProcessor(spanExporter)); + otelTrace.setGlobalTracerProvider(tracerProvider); +}); + +afterAll(async () => { + await tracerProvider.shutdown(); + otelTrace.disable(); +}); + +/** Merged attributes of every span finished during the scenario. */ +function finishedSpanAttributes(): Record { + return Object.assign( + {}, + ...spanExporter.getFinishedSpans().map((s) => s.attributes) + ); +} + function getWorkflowTransformCode(workflowName: string) { return `;globalThis.__private_workflows = new Map([[${JSON.stringify(workflowName)}, ${workflowName}]]);`; } @@ -53,21 +94,52 @@ const HOOK_WORKFLOW = ` ${getWorkflowTransformCode('workflow')} `; +/** + * What the fake World returns for the consumer's hoisted `hook_received` + * create (the one carrying `preloadEvents: true`): + * + * - 'event-only' — the plain materialized result. Models an older server + * (CBOR response) or a World that ignores `preloadEvents`. + * - 'complete' — run + the complete replay log + cursor, hasMore: false. + * - 'partial' — same but hasMore: true (bounded stream); the runtime must + * NOT trust it and must fall back to the run_started setup. + * - 'missing-run' — the log without a reconstructed run entity. + * - 'missing-resume' — run + log whose hook_received lacks the resumeId. + * - 'missing-max-events' — complete preload without the event ceiling; the + * runtime must not run with limit enforcement disabled. + */ +type HookPreloadMode = + | 'event-only' + | 'complete' + | 'partial' + | 'missing-run' + | 'missing-resume' + | 'missing-max-events'; + async function runResumeConsumerScenario(options: { /** - * When true, seed the run_started preload with the producer's concurrent - * hook_received write (carrying the resumeId) so the consumer can skip its - * re-ensure. When false, the preload lacks it and the consumer must - * re-ensure + splice. + * When true, seed the durable log with the producer's concurrent + * hook_received write (carrying the resumeId) before the delivery runs — + * the producer won the resume claim, so the consumer's hoisted write + * converges on that canonical event instead of creating one. */ preloadHasHookReceived: boolean; + hookPreload?: HookPreloadMode; + /** + * When true, a `run_failed` committed concurrently (after the hook claim + * won its TOCTOU race) rides in the durable log — and therefore in the + * preload. The fast path must consume the delivery before any engine + * dispatch instead of replaying against a terminal run. + */ + logHasTerminalEvent?: boolean; /** - * When set, the consumer's `hook_received` re-ensure `events.create` rejects + * When set, the consumer's hoisted `hook_received` `events.create` rejects * with this error, exercising the consumer's terminal-vs-transient error * classification (consume the message vs rethrow for redelivery). */ reEnsureRejection?: Error; }) { + const hookPreload = options.hookPreload ?? 'event-only'; const runId = 'wrun_resume_consumer_preload'; const workflowName = 'workflow'; const deploymentId = 'dpl_resume_consumer_preload'; @@ -117,8 +189,8 @@ async function runResumeConsumerScenario(options: { } as Event; }; - // Bytes the queue message carries in `hookInput.payload` (and, on the skip - // path, what the preloaded hook_received also carries). + // Bytes the queue message carries in `hookInput.payload` (and what the + // canonical hook_received event carries in its eventData). const payloadBytes = await dehydrateStepReturnValue( { value: 'hook-wins' }, runId, @@ -127,7 +199,7 @@ async function runResumeConsumerScenario(options: { // The event log as it exists on this resume delivery: the hook was created // on a prior delivery, so hook_created is always present. - const preloadEvents: Event[] = [ + const durableEvents: Event[] = [ event({ eventType: 'run_created', specVersion: SPEC_VERSION_CURRENT, @@ -143,8 +215,8 @@ async function runResumeConsumerScenario(options: { ]; if (options.preloadHasHookReceived) { // The producer's concurrent direct write already landed — it carries the - // persisted resumeId that the consumer's skip check keys on. - preloadEvents.push({ + // persisted resumeId, so the consumer's hoisted write converges on it. + durableEvents.push({ ...event({ eventType: 'hook_received', specVersion: SPEC_VERSION_CURRENT, @@ -154,9 +226,18 @@ async function runResumeConsumerScenario(options: { resumeId, } as Event); } + if (options.logHasTerminalEvent) { + durableEvents.push( + event({ + eventType: 'run_failed', + specVersion: SPEC_VERSION_CURRENT, + eventData: { error: 'concurrent failure' }, + } as CreateEventRequest) + ); + } - const durableEvents: Event[] = [...preloadEvents]; const createdEvents: CreateEventRequest[] = []; + const createdParams: Array = []; const listEvents = vi.fn(async () => ({ data: [...durableEvents], @@ -165,24 +246,74 @@ async function runResumeConsumerScenario(options: { })); const createEvent = vi.fn( - async (_runId: string, request: CreateEventRequest) => { + async ( + _runId: string, + request: CreateEventRequest, + params?: CreateEventParams + ) => { createdEvents.push(request); + createdParams.push(params); if (request.eventType === 'run_started') { + // Like the real server, the duplicate run_started preload reads the + // CURRENT log — including a hook_received the consumer's hoisted + // write committed moments earlier. return { run: workflowRun, - events: [...preloadEvents], - cursor: preloadEvents.at(-1)?.eventId ?? null, + events: [...durableEvents], + cursor: durableEvents.at(-1)?.eventId ?? null, hasMore: false, }; } - // Simulate the re-ensure failing (terminal or transient) so the - // consumer's error classification runs. Recorded in `createdEvents` - // above, so the attempt is still observable to assertions. - if ( - request.eventType === 'hook_received' && - options.reEnsureRejection !== undefined - ) { - throw options.reEnsureRejection; + if (request.eventType === 'hook_received') { + // Simulate the write failing (terminal or transient) so the + // consumer's error classification runs. Recorded in `createdEvents` + // above, so the attempt is still observable to assertions. + if (options.reEnsureRejection !== undefined) { + throw options.reEnsureRejection; + } + // Converge on the producer's canonical event when it exists + // (the (runId, resumeId) claim), otherwise persist ours with the + // resumeId stamped on it — like the real server does. + let canonical = durableEvents.find( + (e) => e.eventType === 'hook_received' && e.resumeId === resumeId + ); + if (!canonical) { + canonical = { + ...event(request), + resumeId: params?.resumeId, + } as Event; + durableEvents.push(canonical); + } + if (params?.preloadEvents !== true || hookPreload === 'event-only') { + return { event: canonical }; + } + const page = { + events: [...durableEvents], + cursor: durableEvents.at(-1)?.eventId ?? null, + hasMore: hookPreload === 'partial', + ...(hookPreload === 'missing-max-events' + ? {} + : { maxEvents: 25_000 }), + }; + if (hookPreload === 'missing-run') { + return { event: canonical, ...page }; + } + if (hookPreload === 'missing-resume') { + return { + event: canonical, + run: workflowRun, + ...page, + // The streamed log's hook_received lost its resumeId (a server + // that doesn't emit it on the wire): the consumer must not trust + // the preload as replay input. + events: page.events.map((e) => + e.eventType === 'hook_received' + ? ({ ...e, resumeId: undefined } as Event) + : e + ), + }; + } + return { event: canonical, run: workflowRun, ...page }; } const created = event(request); durableEvents.push(created); @@ -211,9 +342,9 @@ async function runResumeConsumerScenario(options: { expect(capturedHandler).toBeDefined(); // A continuation delivery carrying the resume's hookInput (no runInput, so - // turbo is off and the hookInput re-ensure branch runs). Capture whether the - // handler rethrew: on the transient re-ensure failure it must reject so the - // queue redelivers; on a terminal one it resolves (consumes the message). + // turbo is off and the lazy hook fast path runs). Capture whether the + // handler rethrew: on a transient failure it must reject so the queue + // redelivers; on a terminal one it resolves (consumes the message). let handlerError: unknown; try { await capturedHandler?.( @@ -240,12 +371,20 @@ async function runResumeConsumerScenario(options: { const hookReceivedCreates = createdEvents.filter( (e) => e.eventType === 'hook_received' ); + const hookReceivedParams = createdParams.filter( + (_, i) => createdEvents[i]?.eventType === 'hook_received' + ); + const runStartedCreates = createdEvents.filter( + (e) => e.eventType === 'run_started' + ); const runCompletedCreates = createdEvents.filter( (e) => e.eventType === 'run_completed' ); return { hookReceivedCreates, + hookReceivedParams, + runStartedCreates, runCompletedCreates, listEvents, createEvent, @@ -255,42 +394,214 @@ async function runResumeConsumerScenario(options: { pinSharedCorrelationIds(); -describe('lazy hook resume consumer preload (Perf Option A)', () => { +describe('lazy hook resume consumer preload', () => { afterEach(() => { setWorld(undefined); vi.clearAllMocks(); + spanExporter.reset(); + }); + + it('initializes the invocation from a complete hook_received preload: no run_started, no events.list', async () => { + const { + hookReceivedCreates, + hookReceivedParams, + runStartedCreates, + runCompletedCreates, + listEvents, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'complete', + }); + + expect(handlerError).toBeUndefined(); + // Exactly one consumer-side HTTP setup request: the hoisted hook_received + // with the replay preload opt-in. + expect(hookReceivedCreates).toHaveLength(1); + expect(hookReceivedParams[0]?.preloadEvents).toBe(true); + expect(hookReceivedParams[0]?.resumeId).toBe('resume-consumer-1'); + expect(hookReceivedParams[0]?.resumePayloadDigest).toBe('c'.repeat(64)); + // The preload replaced the entire generic setup. + expect(runStartedCreates).toHaveLength(0); + expect(listEvents).not.toHaveBeenCalled(); + // Replay completed off the preloaded log. + expect(runCompletedCreates).toHaveLength(1); + }); + + it('initializes replay from the producer-won canonical event in the preload', async () => { + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + listEvents, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: true, + hookPreload: 'complete', + }); + + expect(handlerError).toBeUndefined(); + // The hoisted write converged on the producer's event (no second event + // was persisted — the fake's durable log kept a single hook_received) and + // its preload initialized replay. + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(0); + expect(listEvents).not.toHaveBeenCalled(); + expect(runCompletedCreates).toHaveLength(1); }); - it('skips the re-ensure when the run_started preload already carries the matching resumeId', async () => { - const { hookReceivedCreates, runCompletedCreates, listEvents } = - await runResumeConsumerScenario({ preloadHasHookReceived: true }); + it('falls back to run_started when the preload is bounded (hasMore: true)', async () => { + // This path deliberately has no cursor-continuation machinery: a bounded + // page must not be replayed (it could be missing the log's tail), so the + // runtime takes the generic setup instead. + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'partial', + }); - // The producer's concurrent write is already in the preloaded log, so the - // consumer must NOT issue its own hook_received create. - expect(hookReceivedCreates).toHaveLength(0); - // Replay still observed the hook and completed the run. + expect(handlerError).toBeUndefined(); + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(1); expect(runCompletedCreates).toHaveLength(1); - // The skip path consumes the preload as-is: no fresh events.list. + }); + + it('falls back to run_started without re-posting the hook when the World returns no preload', async () => { + const { + hookReceivedCreates, + hookReceivedParams, + runStartedCreates, + runCompletedCreates, + listEvents, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'event-only', + }); + + expect(handlerError).toBeUndefined(); + // The write succeeded (older server / CBOR response) — never posted twice. + expect(hookReceivedCreates).toHaveLength(1); + expect(hookReceivedParams[0]?.preloadEvents).toBe(true); + // Generic setup ran; its preload was read after the hook write committed, + // so the canonical event is already in the log — no events.list, no splice. + expect(runStartedCreates).toHaveLength(1); expect(listEvents).not.toHaveBeenCalled(); + expect(runCompletedCreates).toHaveLength(1); + }); + + it('falls back to run_started when the producer won and the World returns no preload', async () => { + const { hookReceivedCreates, runStartedCreates, runCompletedCreates } = + await runResumeConsumerScenario({ + preloadHasHookReceived: true, + hookPreload: 'event-only', + }); + + // The hoisted write converges on the producer's canonical event — exactly + // one write despite both sides attempting it. + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(1); + expect(runCompletedCreates).toHaveLength(1); }); - it('re-ensures and splices the canonical hook_received without a fresh events.list when the preload lacks it', async () => { - const { hookReceivedCreates, runCompletedCreates, listEvents } = - await runResumeConsumerScenario({ preloadHasHookReceived: false }); + it('uses the safe fallback when the preload lacks a reconstructed run', async () => { + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'missing-run', + }); - // The producer's write had not landed, so the consumer re-ensures exactly - // one hook_received event... + expect(handlerError).toBeUndefined(); expect(hookReceivedCreates).toHaveLength(1); - // ...and replay completes off the spliced-in event. + // Unusable preload → generic run_started setup, no second hook post. + expect(runStartedCreates).toHaveLength(1); expect(runCompletedCreates).toHaveLength(1); - // The canonical event was spliced into the preloaded log in-order, so no - // fresh events.list round trip was needed to observe it. + }); + + it('consumes the delivery when the complete preload contains a terminal run event', async () => { + // The hook claim won its race, but a run_failed committed concurrently + // and rides in the streamed log. The reconstructed run always reads + // 'running', and QuickJS dispatches before the node replay loop's + // terminal check — so the fast path itself must consume the delivery + // before any engine runs. + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + listEvents, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'complete', + logHasTerminalEvent: true, + }); + + // The write converged and the delivery was consumed (no rethrow)... + expect(handlerError).toBeUndefined(); + expect(hookReceivedCreates).toHaveLength(1); + // ...without falling back to run_started, replaying, or completing. + expect(runStartedCreates).toHaveLength(0); expect(listEvents).not.toHaveBeenCalled(); + expect(runCompletedCreates).toHaveLength(0); + + // The preload still initialized (and ended) this delivery, so the span + // records the setup source and the run's ACTUAL terminal status — not + // the reconstructed run's synthetic 'running'. + const attributes = finishedSpanAttributes(); + expect(attributes['workflow.resume_setup_source']).toBe( + 'hook_received_stream' + ); + expect(attributes['workflow.run.status']).toBe('failed'); }); - it('consumes the message (no rethrow, no replay) when the re-ensure hits a terminal run', async () => { + it('falls back to run_started when the preload lacks the event ceiling (maxEvents)', async () => { + // The preload response plays run_started's role, so a missing/invalid + // x-wf-max-events would leave event-limit enforcement disabled for the + // whole run. Require it, and take the generic setup when absent. + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'missing-max-events', + }); + + expect(handlerError).toBeUndefined(); + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(1); + expect(runCompletedCreates).toHaveLength(1); + }); + + it('uses the safe fallback when the preload lacks the matching resumeId', async () => { + const { + hookReceivedCreates, + runStartedCreates, + runCompletedCreates, + handlerError, + } = await runResumeConsumerScenario({ + preloadHasHookReceived: false, + hookPreload: 'missing-resume', + }); + + expect(handlerError).toBeUndefined(); + expect(hookReceivedCreates).toHaveLength(1); + expect(runStartedCreates).toHaveLength(1); + expect(runCompletedCreates).toHaveLength(1); + }); + + it('consumes the message (no rethrow, no replay) when the hoisted write hits a terminal run', async () => { // The run went terminal between the producer's dispatch and this delivery, - // so the re-ensure rejects with HookNotFoundError. There is nothing left to + // so the write rejects with HookNotFoundError. There is nothing left to // resume: the consumer must consume the message (resolve) and NOT replay — // acking a terminal delivery is safe because the run is already ended. const { hookReceivedCreates, runCompletedCreates, handlerError } = @@ -299,7 +610,7 @@ describe('lazy hook resume consumer preload (Perf Option A)', () => { reEnsureRejection: new HookNotFoundError('resume-consumer-token'), }); - // The re-ensure was attempted... + // The write was attempted... expect(hookReceivedCreates).toHaveLength(1); // ...it rejected terminally, so the handler resolved without rethrowing... expect(handlerError).toBeUndefined(); @@ -307,7 +618,7 @@ describe('lazy hook resume consumer preload (Perf Option A)', () => { expect(runCompletedCreates).toHaveLength(0); }); - it('rethrows for queue redelivery when the re-ensure hits a transient conflict', async () => { + it('rethrows for queue redelivery when the hoisted write hits a transient conflict', async () => { // The (runId, resumeId) constraint exists but the matching event is not yet // observable — the producer's parallel write is still in flight, or a // redrive raced the claim. This is transient: the consumer must rethrow so @@ -319,11 +630,29 @@ describe('lazy hook resume consumer preload (Perf Option A)', () => { reEnsureRejection: new EntityConflictError('resumeId claim in flight'), }); - // The re-ensure was attempted... + // The write was attempted... expect(hookReceivedCreates).toHaveLength(1); // ...and rethrew so VQS redelivers the message. expect(handlerError).toBeInstanceOf(EntityConflictError); // Replay never proceeded to complete the run on this failed delivery. expect(runCompletedCreates).toHaveLength(0); }); + + it('rethrows for queue redelivery when the preload stream is interrupted', async () => { + // world-vercel surfaces a frame stream that ends without the _end + // sentinel as a plain error. The write may have committed, but the + // (runId, resumeId) claim makes the retry idempotent — rethrow and let + // the queue redeliver. + const truncated = new Error( + 'v4 createEvent: frame stream ended without the end-of-stream sentinel' + ); + const { runCompletedCreates, handlerError } = + await runResumeConsumerScenario({ + preloadHasHookReceived: false, + reEnsureRejection: truncated, + }); + + expect(handlerError).toBe(truncated); + expect(runCompletedCreates).toHaveLength(0); + }); }); diff --git a/packages/core/src/runtime/resume-hook.parallel.test.ts b/packages/core/src/runtime/resume-hook.parallel.test.ts index e3a6959f8a..d0c1237d97 100644 --- a/packages/core/src/runtime/resume-hook.parallel.test.ts +++ b/packages/core/src/runtime/resume-hook.parallel.test.ts @@ -102,6 +102,10 @@ describe('resumeHook (parallel fast path)', () => { 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(); expect(queue).toHaveBeenCalledTimes(1); const [, payloadArg] = queue.mock.calls[0]; diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index f830197d8a..bf891851ee 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -390,11 +390,46 @@ export const HookResilientResume = SemanticConvention( * materialized the `hook_received` event from the queue message's `hookInput` * because the producer's direct write had not landed — the completion of 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 + * response cannot tell whether the producer or the consumer won the + * `(runId, resumeId)` claim — so this attribute is deliberately NOT emitted + * for them (emitting `true` unconditionally would count every producer-won + * resume as a recovery). The producer-begin ({@link HookResilientResume}) / + * consumer-materialized pairing is therefore no longer complete for atomic + * lazy resumptions; use {@link HookResumeSetupSource} to observe that path. */ export const HookResilientResumeMaterialized = SemanticConvention( 'workflow.hook.resilient_resume_materialized' ); +/** + * Consumer-side signal (on the workflow execution span) of how a lazy hook + * resume initialized its replay state: + * + * - `hook_received_stream` — the hoisted `hook_received` write returned a + * usable replay preload (run + complete event log), so the invocation + * skipped both the `run_started` write and the initial `events.list`. + * - `hook_received_fallback` — the hoisted write succeeded but returned no + * usable preload (a CBOR response from an older server, a World that + * ignored the opt-in, a bounded `hasMore` page, or a preload that failed + * validation); the invocation fell back to the `run_started` setup without + * re-posting the hook. + * + * Absent on legacy hook deliveries (no resumeId/digest) and on every other + * delivery kind, which take the `run_started` setup unconditionally. + * + * This is a latency/setup-path signal: it says which requests initialized + * the invocation, NOT that this consumer created the `hook_received` event + * (the hoisted write may equally have converged on the producer's — claim + * ownership is not observable client-side; cf. + * {@link HookResilientResumeMaterialized}). + */ +export const HookResumeSetupSource = SemanticConvention( + 'workflow.resume_setup_source' +); + // Webhook attributes /** Number of webhook handlers triggered */ diff --git a/packages/world-vercel/src/event-retry.test.ts b/packages/world-vercel/src/event-retry.test.ts index d514576c7b..15c16abc45 100644 --- a/packages/world-vercel/src/event-retry.test.ts +++ b/packages/world-vercel/src/event-retry.test.ts @@ -225,4 +225,77 @@ describe('withEventPostRetry', () => { await expect(withEventPostRetry(fn, 'hook_received')).rejects.toThrow(); expect(fn).toHaveBeenCalledTimes(1); }); + + describe('idempotentHookResume opt-in', () => { + it('retries an atomic hook resume (resumeId + digest) past an ECONNRESET', async () => { + // The (runId, resumeId) claim makes the write idempotent-on-retry: a + // retry whose original landed converges on the same canonical event. + let calls = 0; + const fn = vi.fn(async () => { + calls++; + if (calls === 1) throw transportErr('ECONNRESET'); + return 'ok'; + }); + + const p = withEventPostRetry(fn, 'hook_received', { + idempotentHookResume: true, + }); + await vi.runAllTimersAsync(); + + await expect(p).resolves.toBe('ok'); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it('keeps plain hook_received single-attempt when the opt-in is absent', async () => { + const fn = vi.fn(async () => { + throw transportErr('ECONNRESET'); + }); + + await expect( + withEventPostRetry(fn, 'hook_received', {}) + ).rejects.toThrow(); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('keeps an incomplete idempotency shape single-attempt (opt-in false)', async () => { + // The caller computes the opt-in from resumeId AND digest presence — + // resumeId-only / digest-only writes arrive here with false. + const fn = vi.fn(async () => { + throw transportErr('ECONNRESET'); + }); + + await expect( + withEventPostRetry(fn, 'hook_received', { + idempotentHookResume: false, + }) + ).rejects.toThrow(); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('never retries a definitive response, even with the opt-in', async () => { + const fn = vi.fn(async () => { + throw new WorkflowWorldError('digest reuse', { status: 422 }); + }); + + await expect( + withEventPostRetry(fn, 'hook_received', { + idempotentHookResume: true, + }) + ).rejects.toThrow(); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it('does not widen retries for other event types', async () => { + const fn = vi.fn(async () => { + throw transportErr('ECONNRESET'); + }); + + await expect( + withEventPostRetry(fn, 'step_started', { + idempotentHookResume: true, + }) + ).rejects.toThrow(); + expect(fn).toHaveBeenCalledTimes(1); + }); + }); }); diff --git a/packages/world-vercel/src/event-retry.ts b/packages/world-vercel/src/event-retry.ts index fc0d3bd87f..103e798b02 100644 --- a/packages/world-vercel/src/event-retry.ts +++ b/packages/world-vercel/src/event-retry.ts @@ -30,8 +30,14 @@ * - `step_retrying` — re-applies `pending` (idempotent state) but its handler * does NOT throw on a duplicate, so a retry appends a * second event-log row. - * - `hook_received` — has no server-side guard at all; a retry appends a - * duplicate row and can re-deliver the payload. + * - `hook_received` — an ORDINARY hook write has no server-side guard, so a + * retry appends a duplicate row and can re-deliver the + * payload. The atomic lazy-resume shape (resumeId + + * resumePayloadDigest) IS guarded — the server's + * (runId, resumeId) claim converges a retry on the same + * canonical event — so those writes opt back into the + * standard policy via + * {@link EventPostRetryOptions.idempotentHookResume}. * * Only transient/ambiguous transport failures are retried; definitive responses * (409/410/425/429 and any other 4xx) surface immediately, exactly as before. @@ -137,7 +143,10 @@ export const EVENT_RETRY_ELIGIBILITY = { hook_received: { retryable: false, reason: - 'no server guard → a retry duplicates the row / re-delivers payload', + 'no server guard → a retry duplicates the row / re-delivers payload. ' + + 'EXCEPTION: the atomic lazy-resume shape (resumeId + digest) is ' + + 'deduplicated server-side by the (runId, resumeId) claim, so those ' + + 'writes opt in via withEventPostRetry({ idempotentHookResume })', }, // Server-originated; the SDK never POSTs it. hook_conflict: { @@ -257,6 +266,20 @@ function errorMarker(err: unknown): string { ); } +export interface EventPostRetryOptions { + /** + * Narrow opt-in for `hook_received` writes carrying the atomic lazy-resume + * idempotency pair (`resumeId` + `resumePayloadDigest`). Those writes are + * deduplicated server-side by the `(runId, resumeId)` claim — a retry whose + * original landed converges on the same canonical event instead of + * appending a duplicate row — so they get the standard transient retry + * policy. Legacy `hook_received` (no pair, or an incomplete one) stays + * single-attempt per {@link EVENT_RETRY_ELIGIBILITY}; definitive 4xx + * responses stay non-retryable regardless. + */ + idempotentHookResume?: boolean; +} + /** * Run an event POST, retrying transient transport failures in-process when the * event type is idempotent-on-retry. Non-retryable event types and definitive @@ -264,9 +287,12 @@ function errorMarker(err: unknown): string { */ export async function withEventPostRetry( fn: () => Promise, - eventType: WorkflowEventType + eventType: WorkflowEventType, + options?: EventPostRetryOptions ): Promise { - const retryable = EVENT_RETRY_ELIGIBILITY[eventType]?.retryable ?? false; + const retryable = + (eventType === 'hook_received' && options?.idempotentHookResume === true) || + (EVENT_RETRY_ELIGIBILITY[eventType]?.retryable ?? false); for (let attempt = 0; ; attempt++) { try { return await fn(); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index c3a6b47bc2..1d2b6dfddc 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -99,6 +99,11 @@ export const V4_RESPONSE_HEADERS = { eventId: 'x-wf-event-id', runId: 'x-wf-run-id', createdAt: 'x-wf-created-at', + /** + * Server-owned per-run event ceiling. Set on streamed replay-log + * responses, where there is no CBOR body to carry `maxEvents`. + */ + maxEvents: 'x-wf-max-events', } as const; export interface CreateEventV4Input { @@ -539,6 +544,13 @@ export async function createWorkflowRunEventV4( 'createEvent' ); + return decodeMaterializedCreateEventResponse(response); +} + +/** Decode the ids + materialized-entity bag of a v4 POST CBOR response. */ +async function decodeMaterializedCreateEventResponse( + response: Response +): Promise { const eventId = response.headers.get(V4_RESPONSE_HEADERS.eventId); const runId = response.headers.get(V4_RESPONSE_HEADERS.runId); const createdAt = response.headers.get(V4_RESPONSE_HEADERS.createdAt); @@ -560,6 +572,89 @@ export async function createWorkflowRunEventV4( return { eventId, runId, createdAt, body }; } +/** + * Result of a `hook_received` POST that opted into the replay-log preload, + * discriminated on `kind` (keyed on the response content type). + */ +export type HookReceivedPreloadV4Result = + /** The server streamed the replay log back as v4 frames. */ + | (ListEventsV4Result & { + kind: 'stream'; + /** + * The canonical event this write created or converged on (the resume + * claim winner's — ours or the producer's), named by the + * {@link V4_RESPONSE_HEADERS.eventId} response header. Undefined when + * the server did not send the header. + */ + canonicalEventId: string | undefined; + /** Per-run event ceiling from the response header, when present. */ + maxEvents: number | undefined; + }) + /** + * The server answered with the normal materialized CBOR body instead — + * an older server, or one that declined the optimization. The + * hook_received write itself has still succeeded; callers must not + * re-post it. + */ + | { kind: 'materialized'; result: CreateEventV4Result }; + +/** + * POST /api/v4/runs/:runId/events/hook_received with the v4-frame `Accept`, + * consuming either response mode. + * + * A server that supports the lazy-hook replay stream answers the consumer's + * idempotent re-ensure with the run's complete replay log as v4 frames — + * the same event-frame sequence LIST uses, ending with the `_end` sentinel. + * A truncated stream (EOF without the sentinel) throws; the write is + * deduplicated by the server's `(runId, resumeId)` constraint, so retrying + * the whole request is safe and converges on the same canonical event. + */ +export async function createHookReceivedPreloadEventV4( + input: CreateEventV4Input, + config?: APIConfig +): Promise { + const { baseUrl, headers: baseHeaders } = await getHttpConfig(config); + const headers = new Headers(baseHeaders); + headers.set('Content-Type', 'application/octet-stream'); + headers.set('Accept', V4_FRAME_CONTENT_TYPE); + + const frame = encodeFrame( + buildPostFrameMeta(input), + input.payload ?? new Uint8Array(0) + ); + + const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/${encodeURIComponent(input.eventType)}`; + const response = await fetchV4( + url, + { method: 'POST', headers, body: frame }, + config, + 'createEvent' + ); + + const contentType = response.headers.get('content-type'); + if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) { + return { + kind: 'materialized', + result: await decodeMaterializedCreateEventResponse(response), + }; + } + + const page = await decodeListFrameResponse(response, 'createEvent'); + const maxEventsRaw = response.headers.get(V4_RESPONSE_HEADERS.maxEvents); + const maxEventsParsed = + maxEventsRaw === null ? Number.NaN : Number(maxEventsRaw); + return { + kind: 'stream', + ...page, + canonicalEventId: + response.headers.get(V4_RESPONSE_HEADERS.eventId) ?? undefined, + maxEvents: + Number.isInteger(maxEventsParsed) && maxEventsParsed > 0 + ? maxEventsParsed + : undefined, + }; +} + /** * Decoded event entity returned by GET /api/v4/runs/:runId/events/:eventId. * The server CBOR-encodes the full entity with refs resolved server-side, @@ -576,6 +671,13 @@ export interface DecodedV4Event { occurredAt?: Date | string; specVersion?: number; eventData?: Record; + /** + * Lazy hook resume idempotency key, persisted on `hook_received` events + * created through the `(runId, resumeId)` claim and emitted back in the + * frame meta. The runtime matches it against the queue message's + * `hookInput.resumeId` to recognize its own resume in a preloaded log. + */ + resumeId?: string; } function readHeader( @@ -701,6 +803,18 @@ async function consumeListFrameStream( config, opName ); + return decodeListFrameResponse(response, opName); +} + +/** + * Decode a v4 event-frame response body into an in-memory page. Shared by + * the GET LIST paths and the streamed `hook_received` replay preload — + * the wire shape is identical regardless of the verb that produced it. + */ +async function decodeListFrameResponse( + response: Response, + opName: string +): Promise { const contentType = response.headers.get('content-type'); if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) { throw new Error( diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index fff1b7874e..874f6d9c90 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -1286,3 +1286,408 @@ describe('getWorkflowRunEvents by correlation id is scoped to the run', () => { expect(result.cursor).toBe('eid:evnt_2'); }); }); + +describe('createWorkflowRunEvent hook_received replay preload', () => { + const RESUME_ID = 'resume-preload-1'; + const DIGEST = 'e'.repeat(64); + const PAYLOAD = new TextEncoder().encode('"resume payload"'); + + const preloadParams: CreateEventParams = { + resumeId: RESUME_ID, + resumePayloadDigest: DIGEST, + preloadEvents: true, + }; + + function hookReceivedRequest() { + return { + eventType: 'hook_received', + specVersion: 2, + correlationId: 'hook_1', + eventData: { token: 'tok-preload', payload: PAYLOAD }, + } as AnyEventRequest; + } + + function concatFrames(frames: Uint8Array[]): Uint8Array { + const total = frames.reduce((n, f) => n + f.byteLength, 0); + const out = new Uint8Array(total); + let off = 0; + for (const f of frames) { + out.set(f, off); + off += f.byteLength; + } + return out; + } + + function hookReplayStreamResponse(): Uint8Array { + return concatFrames([ + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_created', + createdAt: new Date('2026-06-10T00:00:00.000Z'), + specVersion: 2, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'wf', + executionContext: { region: 'iad1' }, + }, + }, + new Uint8Array() + ), + encodeFrame( + { + eventId: 'evnt_2', + runId: 'wrun_1', + eventType: 'run_started', + createdAt: new Date('2026-06-10T00:00:01.000Z'), + specVersion: 2, + eventData: {}, + }, + new Uint8Array() + ), + encodeFrame( + { + eventId: 'evnt_3', + runId: 'wrun_1', + eventType: 'hook_created', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:02.000Z'), + specVersion: 2, + eventData: { token: 'tok-preload' }, + }, + new Uint8Array() + ), + encodeFrame( + { + eventId: 'evnt_4', + runId: 'wrun_1', + eventType: 'hook_received', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:03.000Z'), + specVersion: 2, + resumeId: RESUME_ID, + eventData: { token: 'tok-preload' }, + }, + PAYLOAD + ), + encodeFrame( + { _end: 1, next: 'eid:evnt_4', hasMore: false }, + new Uint8Array() + ), + ]); + } + + it('decodes a streamed replay log into event + reconstructed run + page', async () => { + const agent = mockAgent(); + let capturedMeta: Record | undefined; + agent + .get(ORIGIN) + .intercept({ + // The headers matcher proves the frame Accept was sent — an + // unmatched request would leave the interceptor pending. + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply( + 200, + (opts: { body?: unknown }) => { + capturedMeta = decodePostedMeta(opts.body); + return hookReplayStreamResponse(); + }, + { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_4', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:03.000Z', + 'x-wf-max-events': '10000', + }, + } + ); + + const result = await createWorkflowRunEvent( + 'wrun_1', + hookReceivedRequest(), + preloadParams, + { token: 'test-token', dispatcher: agent } + ); + + // The idempotency key + digest rode the frame meta. The request keeps + // hook_received's lazy default: a supporting server owns frame-body + // resolution regardless, and an older server then answers the CBOR + // fallback without resolving a payload the runtime would discard. + expect(capturedMeta?.resumeId).toBe(RESUME_ID); + expect(capturedMeta?.resumePayloadDigest).toBe(DIGEST); + expect(capturedMeta?.remoteRefBehavior).toBe('lazy'); + + // The canonical event is the one the x-wf-event-id header names. + expect(result.event?.eventId).toBe('evnt_4'); + expect(result.event?.eventType).toBe('hook_received'); + expect(result.event?.resumeId).toBe(RESUME_ID); + expect(result.event?.eventData?.payload).toEqual(PAYLOAD); + + // The run is reconstructed from the streamed lifecycle events. + expect(result.run).toMatchObject({ + runId: 'wrun_1', + status: 'running', + deploymentId: 'dpl_1', + workflowName: 'wf', + executionContext: { region: 'iad1' }, + }); + expect(result.run?.startedAt?.getTime()).toBe( + new Date('2026-06-10T00:00:01.000Z').getTime() + ); + + expect(result.events?.map((event) => event.eventType)).toEqual([ + 'run_created', + 'run_started', + 'hook_created', + 'hook_received', + ]); + expect(result.cursor).toBe('eid:evnt_4'); + expect(result.hasMore).toBe(false); + expect(result.maxEvents).toBe(10000); + agent.assertNoPendingInterceptors(); + }); + + it('returns the page without a run when the stream lacks the lifecycle events', async () => { + // A hook_received preload missing run_created / run_started is not + // fatal: the write converged, so return the page without a run and let + // the runtime take its safe fallback. + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply( + 200, + concatFrames([ + encodeFrame( + { + eventId: 'evnt_4', + runId: 'wrun_1', + eventType: 'hook_received', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:03.000Z'), + specVersion: 2, + resumeId: RESUME_ID, + eventData: { token: 'tok-preload' }, + }, + PAYLOAD + ), + encodeFrame( + { _end: 1, next: 'eid:evnt_4', hasMore: false }, + new Uint8Array() + ), + ]), + { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_4', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:03.000Z', + 'x-wf-max-events': '10000', + }, + } + ); + + const result = await createWorkflowRunEvent( + 'wrun_1', + hookReceivedRequest(), + preloadParams, + { token: 'test-token', dispatcher: agent } + ); + + expect(result.event?.eventId).toBe('evnt_4'); + expect(result.run).toBeUndefined(); + expect(result.events).toHaveLength(1); + expect(result.cursor).toBe('eid:evnt_4'); + agent.assertNoPendingInterceptors(); + }); + + it('retries the atomic preload write past a transient transport failure', async () => { + // The (runId, resumeId) claim makes the write idempotent-on-retry, so + // createWorkflowRunEvent opts this shape into withEventPostRetry — an + // ECONNRESET on the first attempt rides out in-process instead of + // failing the delivery back to the queue. + const agent = mockAgent(); + const reset = Object.assign(new Error('socket hang up'), { + code: 'ECONNRESET', + }); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .replyWithError(reset); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply(200, hookReplayStreamResponse(), { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_4', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:03.000Z', + 'x-wf-max-events': '10000', + }, + }); + + const result = await createWorkflowRunEvent( + 'wrun_1', + hookReceivedRequest(), + preloadParams, + { token: 'test-token', dispatcher: agent } + ); + + expect(result.event?.eventId).toBe('evnt_4'); + expect(result.events).toHaveLength(4); + agent.assertNoPendingInterceptors(); + }); + + it('keeps the CBOR result when the server does not stream (older server)', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply( + 200, + encode({ + event: { + eventId: 'evnt_4', + runId: 'wrun_1', + eventType: 'hook_received', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:03.000Z'), + specVersion: 2, + eventData: { token: 'tok-preload' }, + }, + }), + { + headers: { + 'content-type': 'application/cbor', + 'x-wf-event-id': 'evnt_4', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:03.000Z', + }, + } + ); + + const result = await createWorkflowRunEvent( + 'wrun_1', + hookReceivedRequest(), + preloadParams, + { token: 'test-token', dispatcher: agent } + ); + + // A successful write with no replay preload — the runtime falls back to + // the run_started setup without posting the hook again. + expect(result.event?.eventType).toBe('hook_received'); + expect(result.events).toBeUndefined(); + expect(result.run).toBeUndefined(); + agent.assertNoPendingInterceptors(); + }); + + it('rejects a truncated preload stream (no end sentinel)', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply( + 200, + encodeFrame( + { + eventId: 'evnt_4', + runId: 'wrun_1', + eventType: 'hook_received', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:03.000Z'), + specVersion: 2, + resumeId: RESUME_ID, + eventData: { token: 'tok-preload' }, + }, + PAYLOAD + ), + { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } + ); + + await expect( + createWorkflowRunEvent('wrun_1', hookReceivedRequest(), preloadParams, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toThrow(/end-of-stream sentinel/); + agent.assertNoPendingInterceptors(); + }); + + it('does not request the frame response without preloadEvents (producer write)', async () => { + const agent = mockAgent(); + let capturedAccept: string | undefined; + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + }) + .reply( + 200, + (opts: { headers?: unknown }) => { + const headers = opts.headers as + | Record + | undefined; + capturedAccept = headers?.accept ?? headers?.Accept; + return encode({ + event: { + eventId: 'evnt_4', + runId: 'wrun_1', + eventType: 'hook_received', + correlationId: 'hook_1', + createdAt: new Date('2026-06-10T00:00:03.000Z'), + specVersion: 2, + eventData: { token: 'tok-preload' }, + }, + }); + }, + { + headers: { + 'content-type': 'application/cbor', + 'x-wf-event-id': 'evnt_4', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:03.000Z', + }, + } + ); + + const result = await createWorkflowRunEvent( + 'wrun_1', + hookReceivedRequest(), + { resumeId: RESUME_ID, resumePayloadDigest: DIGEST }, + { token: 'test-token', dispatcher: agent } + ); + + // fetch fills a default `accept: */*`; what matters is the producer + // never opts into the frame response. + expect(capturedAccept ?? '').not.toContain(V4_FRAME_CONTENT_TYPE); + expect(result.event?.eventType).toBe('hook_received'); + agent.assertNoPendingInterceptors(); + }); +}); diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 679d278410..1942517ee7 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -35,6 +35,7 @@ import { HookNotFoundError, WorkflowWorldError } from '@workflow/errors'; import { type AnyEventRequest, + applyAttributeChanges, type CreateEventParams, type Event, type EventDataPayloadField, @@ -55,6 +56,8 @@ import { decode } from 'cbor-x'; import { coerceEventDates } from './event-coerce.js'; import { withEventPostRetry } from './event-retry.js'; import { + type CreateEventV4Result, + createHookReceivedPreloadEventV4, createWorkflowRunEventV4, type DecodedV4Event, getEventsByCorrelationIdV4, @@ -503,6 +506,11 @@ function buildEventFromV4( ...(decoded.specVersion !== undefined ? { specVersion: decoded.specVersion } : {}), + // The persisted lazy-resume idempotency key. The runtime matches it + // against the queue message's hookInput.resumeId to recognize its own + // resume in a preloaded log — dropping it here would silently disable + // that check for frame-decoded events. + ...(decoded.resumeId ? { resumeId: decoded.resumeId } : {}), }; const event = coerceNormalizedEvent(raw); @@ -606,7 +614,16 @@ export async function createWorkflowRunEvent( // ./event-retry for the validated per-event classification. return await withEventPostRetry( () => createWorkflowRunEventInner(id, data, params, config), - data.eventType + data.eventType, + { + // The atomic lazy-resume shape is deduplicated server-side by the + // (runId, resumeId) claim, so its POST is idempotent-on-retry even + // though plain hook_received is not — see EVENT_RETRY_ELIGIBILITY. + idempotentHookResume: + data.eventType === 'hook_received' && + params?.resumeId !== undefined && + params?.resumePayloadDigest !== undefined, + } ); } catch (err) { // 404 on hook_disposed / hook_received → already-disposed hook. @@ -678,84 +695,145 @@ async function createWorkflowRunEventInner( } } - const remoteRefBehavior = eventsNeedingResolve.has(data.eventType) + const remoteRefBehavior: 'resolve' | 'lazy' = eventsNeedingResolve.has( + data.eventType + ) ? 'resolve' : 'lazy'; const { payload, meta } = splitEventDataForV4(data); + const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; - const result = await createWorkflowRunEventV4( - { - runId: id, - eventType: data.eventType, - specVersion: data.specVersion ?? 2, - ...(data.correlationId ? { correlationId: data.correlationId } : {}), - ...(params?.requestId ? { vercelId: params.requestId } : {}), - ...(params?.computeInstanceId - ? { computeInstanceId: params.computeInstanceId } - : {}), - // Precondition snapshot. The three fields describe one snapshot and the - // runtime always sends them together (or not at all); each is spread - // independently only so an older server that knows one but not the - // others still gets what it understands. - ...(params?.stateUpdatedAt !== undefined - ? { stateUpdatedAt: params.stateUpdatedAt } - : {}), - ...(params?.stateEventCount !== undefined - ? { stateEventCount: params.stateEventCount } - : {}), - ...(params?.stateCursor ? { stateCursor: params.stateCursor } : {}), - ...(params?.replayDivergenceCount !== undefined - ? { replayDivergenceCount: params.replayDivergenceCount } - : {}), - occurredAt: params?.occurredAt ?? new Date(), - // Opt-in inline-delta: forward the cursor the runtime held before - // this write so the server can return the authoritative event-log - // delta on the response (events/cursor/hasMore), letting the inline - // loop skip a follow-up events.list. The server only acts on it for - // step_completed/step_failed; older servers ignore it and the runtime - // falls back to events.list. - ...(params?.sinceCursor ? { sinceCursor: params.sinceCursor } : {}), - // Run-started preload opt-out: turbo backgrounds run_started as a write - // barrier only and never reads the preloaded log, so tell the server to - // skip the list+resolve. The server only acts on it for run_started; - // older servers ignore it and simply preload as before. - ...(params?.skipPreload ? { skipPreload: true } : {}), - // Lazy hook resume idempotency key (hook_received only). Routes the - // write through the server's (runId, resumeId) constraint so a - // concurrent queue-consumer re-ensure deduplicates to one event. - ...(params?.resumeId ? { resumeId: params.resumeId } : {}), - // Content digest forwarded alongside resumeId so the direct write and the - // queue re-ensure record an identical digest on the server constraint. - ...(params?.resumePayloadDigest - ? { resumePayloadDigest: params.resumePayloadDigest } - : {}), - remoteRefBehavior, - payload, - ...meta, - }, - config - ); + const v4Input = { + runId: id, + eventType: data.eventType, + specVersion: data.specVersion ?? 2, + ...(data.correlationId ? { correlationId: data.correlationId } : {}), + ...(params?.requestId ? { vercelId: params.requestId } : {}), + ...(params?.computeInstanceId + ? { computeInstanceId: params.computeInstanceId } + : {}), + // Precondition snapshot. The three fields describe one snapshot and the + // runtime always sends them together (or not at all); each is spread + // independently only so an older server that knows one but not the + // others still gets what it understands. + ...(params?.stateUpdatedAt !== undefined + ? { stateUpdatedAt: params.stateUpdatedAt } + : {}), + ...(params?.stateEventCount !== undefined + ? { stateEventCount: params.stateEventCount } + : {}), + ...(params?.stateCursor ? { stateCursor: params.stateCursor } : {}), + ...(params?.replayDivergenceCount !== undefined + ? { replayDivergenceCount: params.replayDivergenceCount } + : {}), + occurredAt: params?.occurredAt ?? new Date(), + // Opt-in inline-delta: forward the cursor the runtime held before + // this write so the server can return the authoritative event-log + // delta on the response (events/cursor/hasMore), letting the inline + // loop skip a follow-up events.list. The server only acts on it for + // step_completed/step_failed; older servers ignore it and the runtime + // falls back to events.list. + ...(params?.sinceCursor ? { sinceCursor: params.sinceCursor } : {}), + // Run-started preload opt-out: turbo backgrounds run_started as a write + // barrier only and never reads the preloaded log, so tell the server to + // skip the list+resolve. The server only acts on it for run_started; + // older servers ignore it and simply preload as before. + ...(params?.skipPreload ? { skipPreload: true } : {}), + // Lazy hook resume idempotency key (hook_received only). Routes the + // write through the server's (runId, resumeId) constraint so a + // concurrent queue-consumer re-ensure deduplicates to one event. + ...(params?.resumeId ? { resumeId: params.resumeId } : {}), + // Content digest forwarded alongside resumeId so the direct write and the + // queue re-ensure record an identical digest on the server constraint. + ...(params?.resumePayloadDigest + ? { resumePayloadDigest: params.resumePayloadDigest } + : {}), + remoteRefBehavior, + payload, + ...meta, + }; - // The server already CBOR-decoded into result.body — just thread the - // fields through. This is the runtime's event-append path (world.events - // .create is only ever called from the workflow runtime, never from - // o11y), and the runtime re-hydrates every payload it consumes through - // the decompress-aware helpers (hydrateStepReturnValue, hydrateRunError, - // …). So we deliberately do NOT decompress here: doing so would be - // redundant work on the TTFB-sensitive run_started/inline-delta path and - // would make the runtime's deserialize compression telemetry report - // `codec: none` for payloads that were compressed at rest. gzip/zstd - // normalization for o11y/display lives on the read paths (getEvent, - // getWorkflowRunEvents, getStep, getRun, getHook). - // - // `event`/`events` go through coerceEventDates only: they can be read - // back from the backing store server-side (e.g. the run_started TTFB - // preload queries the event log), where nested eventData dates are ISO - // strings — same coercion the GET/LIST path applies. The returned event - // honors the caller's resolveData: 'none' strips payload fields, - // matching the v3 path's stripEventAndLegacyRefs behavior. - const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; + if ( + data.eventType === 'hook_received' && + params?.preloadEvents === true && + params.resumeId !== undefined && + params.resumePayloadDigest !== undefined + ) { + // Lazy hook resume: the queue consumer's idempotent re-ensure doubles + // as the invocation's setup request. A supporting server streams the + // complete replay log back in this response with resolved frame bodies + // — the SERVER owns that resolution (the preload contract requires + // replay-ready bytes; v4 has no /refs endpoint to hydrate a lazy + // descriptor during replay), so the request keeps hook_received's lazy + // default. Against an older server this makes the CBOR fallback + // lightweight: it answers the mutation without resolving and echoing + // an S3-backed hook payload the runtime would discard anyway. + const outcome = await createHookReceivedPreloadEventV4( + { ...v4Input, remoteRefBehavior: 'lazy' }, + config + ); + if (outcome.kind === 'materialized') { + // Older server (or optimization declined): the write still succeeded + // and this is its normal materialized result. The runtime sees no + // replay preload on it and falls back to the run_started setup. + return materializedV4ToEventResult(outcome.result, resolveData); + } + const { canonicalEventId, maxEvents, next, hasMore } = outcome; + const events = outcome.events.map((listed) => + buildEventFromV4(listed.event, listed.body, 'all') + ); + const canonicalEvent = events.find( + (event) => event.eventId === canonicalEventId + ); + // Unlike lifecycle streams, a preload missing run_created/run_started is + // not fatal here: the write has already converged, so return the page + // without a run and let the runtime take its safe fallback. + const run = reconstructRunFromReplayEvents(events); + return { + ...(canonicalEvent ? { event: canonicalEvent } : {}), + ...(run ? { run } : {}), + events, + cursor: next ?? null, + // Our streaming server always stamps hasMore on the sentinel; treat a + // missing flag as "not the complete log" so the runtime falls back + // rather than replaying a possibly-truncated prefix. + hasMore: hasMore ?? true, + ...(maxEvents !== undefined ? { maxEvents } : {}), + }; + } + + const result = await createWorkflowRunEventV4(v4Input, config); + return materializedV4ToEventResult(result, resolveData); +} + +/** + * Map a materialized v4 POST response onto the EventResult the runtime + * consumes. + * + * The server already CBOR-decoded into result.body — just thread the + * fields through. This is the runtime's event-append path (world.events + * .create is only ever called from the workflow runtime, never from + * o11y), and the runtime re-hydrates every payload it consumes through + * the decompress-aware helpers (hydrateStepReturnValue, hydrateRunError, + * …). So we deliberately do NOT decompress here: doing so would be + * redundant work on the TTFB-sensitive run_started/inline-delta path and + * would make the runtime's deserialize compression telemetry report + * `codec: none` for payloads that were compressed at rest. gzip/zstd + * normalization for o11y/display lives on the read paths (getEvent, + * getWorkflowRunEvents, getStep, getRun, getHook). + * + * `event`/`events` go through coerceEventDates only: they can be read + * back from the backing store server-side (e.g. the run_started TTFB + * preload queries the event log), where nested eventData dates are ISO + * strings — same coercion the GET/LIST path applies. The returned event + * honors the caller's resolveData: 'none' strips payload fields, + * matching the v3 path's stripEventAndLegacyRefs behavior. + */ +function materializedV4ToEventResult( + result: CreateEventV4Result, + resolveData: NonNullable +): EventResult { const body = result.body; return { event: body.event @@ -787,3 +865,46 @@ async function createWorkflowRunEventInner( : {}), }; } + +/** + * Reconstruct the run entity from a streamed replay log: identity and input + * from `run_created`, start time from `run_started`, later `attr_set` events + * folded into `attributes`/`updatedAt`. Returns undefined when the log does + * not contain both lifecycle events (the caller decides whether that is + * fatal). The reconstructed status is always `running` — a terminal event + * committed concurrently still rides in the log itself, and the runtime's + * replay-time terminal detection handles it. + */ +function reconstructRunFromReplayEvents( + events: Event[] +): (WorkflowRun & { startedAt: Date }) | undefined { + const runCreated = events.find((event) => event.eventType === 'run_created'); + const runStarted = events.find((event) => event.eventType === 'run_started'); + if (!runCreated || !runStarted) { + return undefined; + } + + let attributes = runCreated.eventData.attributes ?? {}; + let updatedAt = runStarted.createdAt; + for (const event of events) { + if (event.eventType === 'attr_set') { + attributes = applyAttributeChanges(attributes, event.eventData.changes); + updatedAt = event.createdAt; + } + } + + return { + runId: runCreated.runId, + status: 'running', + deploymentId: runCreated.eventData.deploymentId, + workflowName: runCreated.eventData.workflowName, + specVersion: runCreated.specVersion, + executionContext: runCreated.eventData.executionContext, + input: runCreated.eventData.input, + attributes, + encryptionPublicKey: runCreated.eventData.encryptionPublicKey, + startedAt: runStarted.createdAt, + createdAt: runCreated.createdAt, + updatedAt, + }; +} diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts index eca41bbe8b..9eff51cb8e 100644 --- a/packages/world-vercel/src/trace-propagation.test.ts +++ b/packages/world-vercel/src/trace-propagation.test.ts @@ -18,7 +18,10 @@ import { vi, } from 'vitest'; import { z } from 'zod'; -import { getWorkflowRunEventsV4 } from './events-v4.js'; +import { + createHookReceivedPreloadEventV4, + getWorkflowRunEventsV4, +} from './events-v4.js'; import { encodeFrame, V4_FRAME_CONTENT_TYPE } from './frames.js'; import { injectTraceContextIntoHeaders } from './telemetry.js'; import { makeRequest, WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; @@ -172,6 +175,69 @@ describe('v4 event requests (fetchV4) trace propagation', () => { agent.assertNoPendingInterceptors(); fetchSpy.mockRestore(); }); + + it('sends traceparent and the frame Accept on the hook_received preload POST', async () => { + const origin = + WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(origin) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_received', + method: 'POST', + }) + .reply(200, encodeFrame({ _end: 1, hasMore: false }, new Uint8Array(0)), { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_1', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:00.000Z', + 'x-wf-max-events': '10000', + }, + }); + + const fetchSpy = vi.spyOn(globalThis, 'fetch'); + + const tracer = otelTrace.getTracer('test'); + let traceId = ''; + await tracer.startActiveSpan('flow-invocation', async (span) => { + traceId = span.spanContext().traceId; + await createHookReceivedPreloadEventV4( + { + runId: 'wrun_1', + eventType: 'hook_received', + specVersion: 2, + correlationId: 'hook_1', + resumeId: 'resume-trace-1', + resumePayloadDigest: 'e'.repeat(64), + hookToken: 'tok-trace', + }, + { token: 'test-token', dispatcher: agent } + ); + span.end(); + }); + + expect(fetchSpy).toHaveBeenCalledTimes(1); + const calledInit = fetchSpy.mock.calls[0][1]; + const sent = new Headers(calledInit?.headers as HeadersInit); + // The preload POST rides the same fetchV4 envelope as every other v4 + // event request: trace context injected inside the client span... + const traceparent = sent.get('traceparent'); + expect(traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/); + const clientSpan = exporter + .getFinishedSpans() + .find((s) => s.name === 'http POST'); + expect(clientSpan).toBeDefined(); + expect(clientSpan?.spanContext().traceId).toBe(traceId); + expect(traceparent).toBe( + `00-${traceId}-${clientSpan?.spanContext().spanId}-01` + ); + // ...while still negotiating the streamed replay-log response. + expect(sent.get('accept')).toBe(V4_FRAME_CONTENT_TYPE); + agent.assertNoPendingInterceptors(); + fetchSpy.mockRestore(); + }); }); describe('streamer write trace propagation', () => { diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 9294bc992e..6a092d863b 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -867,6 +867,42 @@ export interface CreateEventParams { * across the SDK and the backend. */ skipPreload?: boolean; + /** + * Replay-log preload opt-in (advisory) — the `hook_received` dual of + * {@link skipPreload}. Set only by the queue consumer's idempotent + * `hook_received` re-ensure on a lazy hook resume (alongside + * {@link resumeId} + {@link resumePayloadDigest}). A World MAY return the + * run's current replay event log with the event creation + * (`events`/`cursor`/`hasMore`, plus `run` and `maxEvents`) so the runtime + * can initialize replay from this one request and skip both the + * `run_started` write and the initial `events.list`. + * + * The runtime trusts a returned preload as replay input ONLY when all of + * the following hold — a World that cannot guarantee them should return + * its normal {@link EventResult} instead: + * + * - `events` is the COMPLETE log with `hasMore: false` (the runtime has no + * cursor-continuation machinery on this path; a bounded page is + * rejected). + * - `cursor` is a valid non-null resume point matching `events.list` + * semantics (present even on the final page). + * - `run` (with `run.startedAt`) and `maxEvents` are present — this + * response plays `run_started`'s role, including the event-ceiling + * handshake. + * - The log contains `run_created`, `run_started`, and the canonical + * `hook_received` carrying the requested {@link resumeId}. + * - `events` uses the same ascending ordering semantics as `events.list`. + * - The log is read atomically/consistently WITH (i.e. no earlier than) + * the `hook_received` write, so no concurrently committed event can be + * omitted from the replay input. + * + * Anything less and the runtime observes that no usable replay preload + * came back and falls back to the existing `run_started` setup — a World + * that ignores the param entirely remains fully correct. Only meaningful + * for `hook_received`; ignored for other event types. Producer-side + * `resumeHook()` must not set it. + */ + preloadEvents?: true; } /** @@ -887,7 +923,7 @@ export interface EventResult { /** The wait entity (for wait_created/wait_completed events) */ wait?: import('./waits.js').Wait; /** - * Events with data resolved. Two producers populate this: + * Events with data resolved. Three producers populate this: * * - On a `run_started` response: all events up to this point, so the * runtime can skip the initial `events.list` call and reduce TTFB. @@ -895,6 +931,11 @@ export interface EventResult { * the caller passed {@link CreateEventParams.sinceCursor}: the delta * of events written strictly after that cursor, so the inline loop * can skip the per-step incremental `events.list` round-trip. + * - On a `hook_received` response when the caller passed + * {@link CreateEventParams.preloadEvents}: the run's current replay + * log through the canonical `hook_received`, so the lazy hook queue + * consumer can skip both the `run_started` write and the initial + * `events.list`. */ events?: Event[]; /** Pagination cursor for `events`, matching events.list semantics. */