diff --git a/.changeset/prepare-streamed-replay-payloads.md b/.changeset/prepare-streamed-replay-payloads.md new file mode 100644 index 0000000000..20facbfc94 --- /dev/null +++ b/.changeset/prepare-streamed-replay-payloads.md @@ -0,0 +1,7 @@ +--- +"@workflow/core": patch +"@workflow/world": patch +"@workflow/world-vercel": patch +--- + +Prepare replay payloads as validated event frames arrive and reuse immutable primitive values across fresh workflow VMs. diff --git a/packages/core/src/abort-consistency.test.ts b/packages/core/src/abort-consistency.test.ts index ec93e93227..c1f2f5a4b9 100644 --- a/packages/core/src/abort-consistency.test.ts +++ b/packages/core/src/abort-consistency.test.ts @@ -37,7 +37,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { return { runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/abort-controller.test.ts b/packages/core/src/abort-controller.test.ts index 3adf469366..96c4f8bc04 100644 --- a/packages/core/src/abort-controller.test.ts +++ b/packages/core/src/abort-controller.test.ts @@ -35,7 +35,7 @@ function setupWorkflowContext( return { runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/abort-replay-ordering.test.ts b/packages/core/src/abort-replay-ordering.test.ts index 8961a64925..06296ee362 100644 --- a/packages/core/src/abort-replay-ordering.test.ts +++ b/packages/core/src/abort-replay-ordering.test.ts @@ -70,7 +70,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { return { runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/async-deserialization-ordering.test.ts b/packages/core/src/async-deserialization-ordering.test.ts index 41d4684563..b23346e3e7 100644 --- a/packages/core/src/async-deserialization-ordering.test.ts +++ b/packages/core/src/async-deserialization-ordering.test.ts @@ -50,7 +50,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { return { runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index 25a2486a2b..a8c836b30b 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -90,7 +90,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/replay-payload-cache.test.ts b/packages/core/src/replay-payload-cache.test.ts index 7f89f43cd4..f868b981ba 100644 --- a/packages/core/src/replay-payload-cache.test.ts +++ b/packages/core/src/replay-payload-cache.test.ts @@ -1,7 +1,11 @@ +import { Buffer } from 'node:buffer'; import type { Event, WorkflowRun } from '@workflow/world'; import { assert, describe, expect, it, vi } from 'vitest'; import { importKey } from './encryption.js'; -import { ReplayPayloadCache } from './replay-payload-cache.js'; +import { + ReplayPayloadCache, + type ReplayPayloadPreparer, +} from './replay-payload-cache.js'; import { decodePayload } from './serialization/payload.js'; import { dehydrateStepReturnValue, @@ -56,15 +60,17 @@ function makeEvents(payloads: unknown[]): Event[] { describe('ReplayPayloadCache', () => { it('deduplicates preparation', async () => { const payload = new Uint8Array([1]); - const preparer = vi.fn(async (value) => value); + const preparer = vi.fn(async (value) => value); + const hydrate = vi.fn((prepared: unknown) => prepared); const cache = new ReplayPayloadCache(undefined, preparer); - const first = cache.prepareEventPayload('evnt_one', 'result', payload); - const second = cache.prepareEventPayload('evnt_one', 'result', payload); + const first = cache.getEventValue('evnt_one', payload, hydrate); + const second = cache.getEventValue('evnt_one', payload, hydrate); - expect(first).toBe(second); await expect(first).resolves.toEqual({ data: payload }); + await expect(second).resolves.toEqual({ data: payload }); expect(preparer).toHaveBeenCalledOnce(); + expect(hydrate).toHaveBeenCalledTimes(2); }); it('compacts prepared bytes before retaining them', async () => { @@ -72,37 +78,53 @@ describe('ReplayPayloadCache', () => { const prepared = backing.subarray(1024, 2048); const cache = new ReplayPayloadCache( undefined, - vi.fn(async () => prepared) + vi.fn(async () => prepared) ); - const retained = await cache.prepareEventPayload( + const retained = await cache.getEventValue( 'evnt_compact', - 'result', - new Uint8Array([1]) + new Uint8Array([1]), + (value) => value ); - assert(retained.data instanceof Uint8Array); + assert( + typeof retained === 'object' && + retained !== null && + 'data' in retained && + retained.data instanceof Uint8Array + ); expect(Array.from(retained.data)).toEqual(Array.from(prepared)); expect(retained.data).not.toBeInstanceOf(Buffer); expect(retained.data.buffer.byteLength).toBe(retained.data.byteLength); }); + it('hydrates and memoizes a primitive', async () => { + const payload = new Uint8Array([1]); + const hydrate = vi.fn(() => 42); + const cache = new ReplayPayloadCache(undefined, async (value) => value); + + await expect( + cache.getEventValue('evnt_one', payload, hydrate) + ).resolves.toBe(42); + expect(cache.getEventValue('evnt_one', payload, hydrate)).toBe(42); + expect(hydrate).toHaveBeenCalledOnce(); + }); + it('keeps a failed prewarm until its consumer observes it, then retries', async () => { const payload = new Uint8Array([1]); const run = makeRun(payload); const preparer = vi - .fn() + .fn() .mockRejectedValueOnce(new Error('decrypt failed')) .mockResolvedValueOnce(payload); const cache = new ReplayPayloadCache(undefined, preparer); - await cache.prewarm(run, []); - await expect(cache.prepareWorkflowInput(run)).rejects.toThrow( - 'decrypt failed' - ); + cache.prepareAll(run, []); + await Promise.resolve(); + await expect(cache.getWorkflowInput(run)).rejects.toThrow('decrypt failed'); expect(preparer).toHaveBeenCalledOnce(); - await expect(cache.prepareWorkflowInput(run)).resolves.toEqual({ + await expect(cache.getWorkflowInput(run)).resolves.toEqual({ data: payload, }); expect(preparer).toHaveBeenCalledTimes(2); @@ -111,7 +133,7 @@ describe('ReplayPayloadCache', () => { it('prewarms workflow, step, error, and hook payloads concurrently', async () => { const payloads = [0, 1, 2, 3].map((value) => new Uint8Array([value])); const resolvers: Array<() => void> = []; - const preparer = vi.fn( + const preparer = vi.fn( (value) => new Promise((resolve) => { resolvers.push(() => resolve(value)); @@ -121,16 +143,61 @@ describe('ReplayPayloadCache', () => { const run = makeRun(payloads[0]); const events = makeEvents(payloads.slice(1)); - const warming = cache.prewarm(run, events); + cache.prepareAll(run, events); expect(preparer).toHaveBeenCalledTimes(4); for (const resolve of resolvers.reverse()) resolve(); - await warming; + await Promise.all([ + cache.getWorkflowInput(run), + ...events.map((event) => { + switch (event.eventType) { + case 'step_completed': + return cache.getEventValue( + event.eventId, + event.eventData?.result, + (prepared) => prepared + ); + case 'step_failed': + return cache.getEventValue( + event.eventId, + event.eventData?.error, + (prepared) => prepared + ); + case 'hook_received': + return cache.getEventValue( + event.eventId, + event.eventData?.payload, + (prepared) => prepared + ); + default: + throw new Error(`Unexpected event: ${event.eventType}`); + } + }), + ]); - const allSettled = vi.spyOn(Promise, 'allSettled'); - await cache.prewarm(run, events); + cache.prepareAll(run, events); expect(preparer).toHaveBeenCalledTimes(4); - expect(allSettled).toHaveBeenLastCalledWith([]); - allSettled.mockRestore(); + }); + + it('starts streamed preparation inside the decoder callback', async () => { + const payload = new Uint8Array([1]); + const order: string[] = []; + const preparer = vi.fn(async (value) => { + order.push('prepare'); + return value; + }); + const cache = new ReplayPayloadCache(undefined, preparer); + const [event] = makeEvents([payload]); + + cache.prepareEvent(event); + expect(preparer).toHaveBeenCalledOnce(); + expect(order).toEqual(['prepare']); + + cache.prepareEvent(event); + expect(order).toEqual(['prepare']); + + await expect( + cache.getEventValue(event.eventId, payload, (prepared) => prepared) + ).resolves.toEqual({ data: payload }); }); it('caches real decrypt/decompress output but revives fresh objects', async () => { @@ -145,18 +212,22 @@ describe('ReplayPayloadCache', () => { false, true ); - const preparer = vi.fn(decodePayload); + const preparer = vi.fn(decodePayload); const cache = new ReplayPayloadCache(key, preparer); - const prepared = await cache.prepareEventPayload( + const directPreparation = decodePayload(serialized, key); + expect(directPreparation).toBeInstanceOf(Promise); + await directPreparation; + + const prepared = await cache.getEventValue( 'evnt_encrypted', - 'result', - serialized + serialized, + (value) => value ); - const samePrepared = await cache.prepareEventPayload( + const samePrepared = await cache.getEventValue( 'evnt_encrypted', - 'result', - serialized + serialized, + (value) => value ); const first = deserializePreparedReplayPayload(prepared) as { count: number; @@ -171,93 +242,136 @@ describe('ReplayPayloadCache', () => { expect(second.count).toBe(0); }); - it('rescans a log whose missing events were filled in below the scanned prefix', async () => { - // A stale-snapshot (412) restart replaces the log with a corrected one, so - // the events it was missing appear BELOW the length already scanned and - // shift every later position. Resuming from that length skips exactly the - // events the reload was for, which is what `resetScan` exists to prevent. + it('prepares only events appended after the scanned prefix', () => { const payloads = [0, 1, 2].map((value) => new Uint8Array([value])); - const preparer = vi.fn(async (value) => value); + const preparer = vi.fn(async (value) => value); const cache = new ReplayPayloadCache(undefined, preparer); const run = makeRun(undefined); - const [first, missing, second] = makeEvents(payloads); + const [first, second, third] = makeEvents(payloads); + const prepareEvent = vi.spyOn(cache, 'prepareEvent'); - await cache.prewarm(run, [first, second]); + cache.prepareAll(run, [first, second]); expect(preparer).toHaveBeenCalledTimes(2); - // Positional resume: `missing` sits inside the scanned prefix, so it is - // skipped and its payload is only prepared on demand. - await cache.prewarm(run, [first, missing, second]); + prepareEvent.mockClear(); + cache.prepareAll(run, [first, second, third]); + expect(prepareEvent).toHaveBeenCalledOnce(); + expect(prepareEvent).toHaveBeenCalledWith(third); + expect(preparer).toHaveBeenCalledTimes(3); + }); + + it('rescans a corrected event log after reset', () => { + const payloads = [0, 1, 2].map((value) => new Uint8Array([value])); + const preparer = vi.fn(async (value) => value); + const cache = new ReplayPayloadCache(undefined, preparer); + const run = makeRun(undefined); + const [first, missing, second] = makeEvents(payloads); + + cache.prepareAll(run, [first, second]); + cache.prepareAll(run, [first, missing, second]); expect(preparer).toHaveBeenCalledTimes(2); cache.resetScan(); - await cache.prewarm(run, [first, missing, second]); - // Only the inserted event is new: the other two are keyed by event id and - // stay prepared across the rescan. + cache.prepareAll(run, [first, missing, second]); expect(preparer).toHaveBeenCalledTimes(3); - expect(preparer).toHaveBeenLastCalledWith( - payloads[1], - undefined, - expect.any(Object) - ); + expect(preparer).toHaveBeenLastCalledWith(payloads[1], undefined); }); - it('bypasses legacy values and ignores missing event data during prewarm', async () => { + it('bypasses legacy values and ignores missing event data during preparation', async () => { const legacy = [0, { value: 1 }]; - const preparer = vi.fn(async (value) => value); + const preparer = vi.fn(async (value) => value); const cache = new ReplayPayloadCache(undefined, preparer); - await cache.prepareEventPayload('evnt_legacy', 'result', legacy); - await cache.prepareEventPayload('evnt_legacy', 'result', legacy); + await cache.getEventValue('evnt_legacy', legacy, (prepared) => prepared); + await cache.getEventValue('evnt_legacy', legacy, (prepared) => prepared); expect(preparer).not.toHaveBeenCalled(); const events = makeEvents([legacy, legacy, legacy]); events[2] = { ...events[2], eventData: undefined } as unknown as Event; - await cache.prewarm(makeRun(legacy), events); + cache.prepareAll(makeRun(legacy), events); expect(preparer).not.toHaveBeenCalled(); }); it('memoizes primitive step results, including undefined', async () => { for (const value of [0, false, '', null, undefined]) { - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const hydrate = vi.fn().mockResolvedValue(value); - expect(await cache.getStepResult('evnt_result', hydrate)).toBe(value); - expect(await cache.getStepResult('evnt_result', hydrate)).toBe(value); + expect(await cache.getEventValue('evnt_result', undefined, hydrate)).toBe( + value + ); + expect(await cache.getEventValue('evnt_result', undefined, hydrate)).toBe( + value + ); expect(hydrate).toHaveBeenCalledOnce(); } }); - it('rehydrates mutable and oversized step results', async () => { - const oversized = 'x'.repeat(4097); - for (const value of [{ count: 0 }, oversized]) { - const cache = new ReplayPayloadCache(undefined); + it('isolates primitive values by event id', async () => { + const cache = new ReplayPayloadCache(); + const result = vi.fn().mockResolvedValue('result'); + const error = vi.fn().mockResolvedValue('error'); + + await expect( + cache.getEventValue('evnt_result', undefined, result) + ).resolves.toBe('result'); + await expect( + cache.getEventValue('evnt_error', undefined, error) + ).resolves.toBe('error'); + expect(cache.getEventValue('evnt_result', undefined, result)).toBe( + 'result' + ); + expect(result).toHaveBeenCalledOnce(); + expect(error).toHaveBeenCalledOnce(); + }); + + it('memoizes primitives of any size and rehydrates object results', async () => { + const commonText = 'x'.repeat(256 * 1024); + const oversizedText = 'x'.repeat(16 * 1024 * 1024 + 1); + for (const [value, expectedHydrations] of [ + [{ count: 0 }, 2], + [commonText, 1], + [oversizedText, 1], + ] as const) { + const cache = new ReplayPayloadCache(); const hydrate = vi .fn() .mockImplementation(async () => typeof value === 'object' ? { ...value } : value ); - const first = await cache.getStepResult('evnt_result', hydrate); - const second = await cache.getStepResult('evnt_result', hydrate); - expect(hydrate).toHaveBeenCalledTimes(2); - if (typeof value === 'object') expect(second).not.toBe(first); + const first = await cache.getEventValue( + 'evnt_result', + undefined, + hydrate + ); + const second = await cache.getEventValue( + 'evnt_result', + undefined, + hydrate + ); + expect(hydrate).toHaveBeenCalledTimes(expectedHydrations); + if (typeof value === 'object') { + expect(second).not.toBe(first); + } else if (expectedHydrations === 1) { + expect(second).toBe(first); + } } }); it('does not memoize failed step hydration', async () => { - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const hydrate = vi .fn() .mockRejectedValueOnce(new Error('boom')) .mockResolvedValueOnce('ok'); - await expect(cache.getStepResult('evnt_result', hydrate)).rejects.toThrow( - 'boom' - ); - await expect(cache.getStepResult('evnt_result', hydrate)).resolves.toBe( - 'ok' - ); + await expect( + cache.getEventValue('evnt_result', undefined, hydrate) + ).rejects.toThrow('boom'); + await expect( + cache.getEventValue('evnt_result', undefined, hydrate) + ).resolves.toBe('ok'); expect(hydrate).toHaveBeenCalledTimes(2); }); }); diff --git a/packages/core/src/replay-payload-cache.ts b/packages/core/src/replay-payload-cache.ts index 0922171662..283b55392b 100644 --- a/packages/core/src/replay-payload-cache.ts +++ b/packages/core/src/replay-payload-cache.ts @@ -7,8 +7,23 @@ import { } from './serialization/payload.js'; import { recordCompression } from './serialization/telemetry.js'; -const MAX_MEMOIZED_PRIMITIVE_LENGTH = 4096; -type ReplayPayloadField = 'result' | 'error' | 'payload'; +export type ReplayPayloadPreparer = ( + data: Uint8Array, + key: DecryptionKey | undefined +) => Promise; + +async function prepareReplayPayload( + data: Uint8Array, + key: DecryptionKey | undefined +): Promise { + const compressionStats: CompressionStats = {}; + const prepared = await decodePayload(data, key, compressionStats); + await recordCompression(compressionStats, 'deserialize'); + return prepared; +} + +const WORKFLOW_INPUT = Symbol('workflow-input'); +type ReplayPayloadKey = string | typeof WORKFLOW_INPUT; /** Copy a view only when retaining it would also retain unrelated bytes. */ function compactOwnedBytes(data: Uint8Array): Uint8Array { @@ -17,215 +32,141 @@ function compactOwnedBytes(data: Uint8Array): Uint8Array { : new Uint8Array(data); } -function isMemoizablePrimitive(value: unknown): boolean { - if (value === null) return true; +function isCacheablePrimitive(value: unknown): boolean { const type = typeof value; - if (type === 'object' || type === 'function') return false; - if (type === 'string') { - return (value as string).length <= MAX_MEMOIZED_PRIMITIVE_LENGTH; - } - if (type === 'bigint') { - return (value as bigint).toString().length <= MAX_MEMOIZED_PRIMITIVE_LENGTH; - } - return true; + return ( + value === null || + (type !== 'object' && type !== 'function' && type !== 'symbol') + ); } /** * Invocation-scoped cache for replay payload hydration. * - * A workflow invocation may replay the same event log through several fresh - * VMs. This cache keeps the VM-independent decrypt/decompress result across - * those replays. Deserialization still runs against each VM's globals so every - * replay receives fresh object graphs and correctly revived Workflow objects. + * The cache retains VM-independent decrypt/decompress output across fresh VMs. + * Deserialization still runs against each VM's globals so object graphs and + * Workflow objects remain realm-local. Primitive final values are safe to + * share and skip that repeated deserialization entirely. * - * Successful prepared plaintext remains resident for the invocation lifetime. - * Its memory cost is the sum of decrypted and decompressed payload sizes, but - * it never crosses workflow runs or queue deliveries. + * Key lookup is deliberately outside this class. The runtime creates the + * cache once the run's key has resolved, then feeds it decoded events. */ export class ReplayPayloadCache { - private readonly preparedPayloads = new Map< - string, - Promise + private readonly preparations = new Map< + ReplayPayloadKey, + Promise >(); - private readonly primitiveStepResults = new Map(); - private nextUnscannedEventIndex = 0; + private readonly primitiveValues = new Map(); + private nextUnpreparedEventIndex = 0; constructor( - private readonly encryptionKey: DecryptionKey | undefined, - private readonly preparer: typeof decodePayload = decodePayload + private readonly encryptionKey?: DecryptionKey, + private readonly preparer: ReplayPayloadPreparer = prepareReplayPayload ) {} - /** - * Start every missing binary preparation before workflow execution. Failures - * are intentionally retained: the ordered event consumer must observe the - * original rejection before that entry becomes retryable. - */ - async prewarm(workflowRun: WorkflowRun, events: Event[]): Promise { - const preparations: Promise[] = []; - const start = (cacheKey: string, value: unknown): void => { - // Legacy flattened values may be mutated by devalue's unflatten and are - // therefore prepared only by their eventual consumer, never cached. - if (!(value instanceof Uint8Array)) return; - - // Each replay scans the full event log, so awaiting cached promises here - // would add O(N^2) promise reactions over an N-step invocation. Only wait - // for preparations first discovered by this prewarm pass. - if (this.preparedPayloads.has(cacheKey)) return; - preparations.push(this.ensurePreparation(cacheKey, value)); - }; + /** Prepare a payload as soon as its event frame has been decoded. */ + prepareEvent(event: Event): void { + switch (event.eventType) { + case 'run_created': + this.cachePayload(WORKFLOW_INPUT, event.eventData.input); + break; + case 'run_started': + this.cachePayload(WORKFLOW_INPUT, event.eventData?.input); + break; + case 'step_completed': + this.cachePayload(event.eventId, event.eventData?.result); + break; + case 'step_failed': + this.cachePayload(event.eventId, event.eventData?.error); + break; + case 'hook_received': + this.cachePayload(event.eventId, event.eventData?.payload); + } + } - start(this.workflowInputKey(workflowRun.runId), workflowRun.input); - // This cache is scoped to one invocation. Incremental loads and write - // response deltas only ever append, so the scanned length locates the - // events added since the previous replay. A reload that can insert events - // BELOW that length — a stale-snapshot restart replacing the log with a - // corrected one — must call `resetScan()` first, or the inserted events are - // never scanned. Prepared entries stay valid across that: they are keyed by - // event id, not by position. + /** Prepare every payload not already seen through the event stream. */ + prepareAll(workflowRun: WorkflowRun, events: Event[]): void { + this.cachePayload(WORKFLOW_INPUT, workflowRun.input); for ( - let index = this.nextUnscannedEventIndex; + let index = this.nextUnpreparedEventIndex; index < events.length; index++ ) { - const event = events[index]; - switch (event.eventType) { - case 'step_completed': - start( - this.eventPayloadKey(event.eventId, 'result'), - event.eventData?.result - ); - break; - case 'step_failed': - start( - this.eventPayloadKey(event.eventId, 'error'), - event.eventData?.error - ); - break; - case 'hook_received': - start( - this.eventPayloadKey(event.eventId, 'payload'), - event.eventData?.payload - ); - break; - } + this.prepareEvent(events[index]); } - this.nextUnscannedEventIndex = events.length; - - // Prewarming is speculative and must not fail replay before the matching - // event is consumed. allSettled also attaches rejection handlers eagerly. - await Promise.allSettled(preparations); + this.nextUnpreparedEventIndex = events.length; } - /** - * Forget how much of the event log has been scanned, so the next - * {@link prewarm} walks it from the start again. - * - * Required before a replay whose event log was reloaded rather than extended: - * a corrected log inserts the events the previous load was missing, which - * shifts every later position, so a positional resume would skip exactly the - * events the reload was for. Already-prepared payloads are kept — they are - * keyed by event id, so re-scanning re-observes them for free. - */ + /** Rescan after an event log is replaced or reordered. */ resetScan(): void { - this.nextUnscannedEventIndex = 0; + this.nextUnpreparedEventIndex = 0; } - /** Return the workflow input after shared host-side preparation. */ - prepareWorkflowInput( + getWorkflowInput( workflowRun: WorkflowRun - ): Promise { - return this.consumePreparation( - this.workflowInputKey(workflowRun.runId), - workflowRun.input - ); + ): PreparedReplayPayload | Promise { + return this.getPayload(WORKFLOW_INPUT, workflowRun.input); } - /** - * Return an event payload after shared host-side preparation. A rejected - * preparation is evicted only after this ordered consumer requests it, so a - * later replay can retry without hiding the original failure. - */ - prepareEventPayload( + getEventValue( eventId: string, - field: ReplayPayloadField, - value: unknown - ): Promise { - return this.consumePreparation(this.eventPayloadKey(eventId, field), value); - } - - /** - * Reuse final step values only when sharing them across VMs is unobservable. - * Objects and large strings/bigints always run `hydrate` again, producing a - * fresh VM-specific value from the separately cached prepared payload. - */ - async getStepResult( - eventId: string, - hydrate: () => Promise - ): Promise { - if (this.primitiveStepResults.has(eventId)) { - return this.primitiveStepResults.get(eventId); + serializedValue: unknown, + hydrate: (prepared: PreparedReplayPayload) => unknown | Promise + ): unknown | Promise { + if (this.primitiveValues.has(eventId)) { + return this.primitiveValues.get(eventId); } - const value = await hydrate(); - if (isMemoizablePrimitive(value)) { - this.primitiveStepResults.set(eventId, value); - } - return value; + const prepared = this.getPayload(eventId, serializedValue); + const hydrateAndCache = (payload: PreparedReplayPayload) => { + const hydrated = hydrate(payload); + return hydrated instanceof Promise + ? hydrated.then((value) => this.cachePrimitive(eventId, value)) + : this.cachePrimitive(eventId, hydrated); + }; + return prepared instanceof Promise + ? prepared.then(hydrateAndCache) + : hydrateAndCache(prepared); } - /** - * Consumer-facing lookup. Binary payloads share preparation; legacy values - * bypass the cache because their flattened representation may be mutated. - */ - private consumePreparation( - cacheKey: string, - value: unknown - ): Promise { - if (!(value instanceof Uint8Array)) { - return Promise.resolve({ data: value }); - } - - const preparation = this.ensurePreparation(cacheKey, value); - void preparation.catch(() => { - if (this.preparedPayloads.get(cacheKey) === preparation) { - this.preparedPayloads.delete(cacheKey); - } - }); - return preparation; + private cachePrimitive(eventId: string, value: unknown): unknown { + if (!isCacheablePrimitive(value)) return value; + this.primitiveValues.set(eventId, value); + return value; } - /** Start preparation once and share the exact in-flight promise. */ - private ensurePreparation( - cacheKey: string, - value: Uint8Array - ): Promise { - const cached = this.preparedPayloads.get(cacheKey); - if (cached) return cached; - - const preparation = this.runPreparation(value); - this.preparedPayloads.set(cacheKey, preparation); - return preparation; - } + private cachePayload(cacheKey: ReplayPayloadKey, value: unknown): void { + if (!(value instanceof Uint8Array) || this.preparations.has(cacheKey)) { + return; + } - /** Compact prepared bytes before retaining them for the invocation. */ - private async runPreparation( - value: Uint8Array - ): Promise { - const compressionStats: CompressionStats = {}; - const prepared = await this.preparer( - value, - this.encryptionKey, - compressionStats + const preparation = this.preparer(value, this.encryptionKey).then( + compactOwnedBytes ); - await recordCompression(compressionStats, 'deserialize'); - return { data: compactOwnedBytes(prepared) }; + this.preparations.set(cacheKey, preparation); + void preparation.catch(() => {}); } - private workflowInputKey(runId: string): string { - return `run:${runId}:input`; - } + private getPayload( + cacheKey: ReplayPayloadKey, + value: unknown + ): PreparedReplayPayload | Promise { + if (!(value instanceof Uint8Array)) return { data: value }; - private eventPayloadKey(eventId: string, field: ReplayPayloadField): string { - return `event:${eventId}:${field}`; + this.cachePayload(cacheKey, value); + const prepared = this.preparations.get(cacheKey); + if (!prepared) { + throw new Error('Replay payload preparation was not cached'); + } + + return prepared.then( + (data) => ({ data }), + (error) => { + if (this.preparations.get(cacheKey) === prepared) { + this.preparations.delete(cacheKey); + } + throw error; + } + ); } } diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 5fa19d63ea..d25ebe0b4b 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -82,6 +82,7 @@ import { parseHealthCheckPayload, preconditionEventDelta, queueMessage, + resolveRunEncryptionKey, type SlotSnapshotParams, settleEventSlotGap, slotSnapshotParams, @@ -119,6 +120,7 @@ import { getWorldHandlers, type WorldHandlers, } from './runtime/world.js'; +import type { DecryptionKey } from './serialization/encryption.js'; import { dehydrateRunError } from './serialization.js'; import { remapErrorStack } from './source-map.js'; import * as Attribute from './telemetry/semantic-conventions.js'; @@ -526,6 +528,24 @@ type ReplayEventLog = | ({ type: 'ready' } & LoadedEventLog) | ({ type: 'loadAfter'; cursor: string } & LoadedEventLog); +type ReplayEncryptionKeySource = + | { type: 'run'; run: WorkflowRun } + | { type: 'deployment'; deploymentId: string }; + +interface ReplayPayloads { + encryptionKey: DecryptionKey | undefined; + cache: ReplayPayloadCache; +} + +type ReplayPayloadCacheState = + | { type: 'waitingForKey'; events: Event[] } + | { + type: 'resolvingKey'; + events: Event[]; + payloads: Promise; + } + | { type: 'ready'; payloads: ReplayPayloads }; + function nextEventLogLoad(log: LoadedEventLog): ReplayEventLog { if (log.cursor === null) { return { type: 'loadAll' }; @@ -542,6 +562,13 @@ function appendEventLog(log: LoadedEventLog, appended: LoadedEventLog): void { log.cursor = appended.cursor ?? log.cursor; } +function replayEventDeploymentId(event: Event): string | undefined { + if (event.eventType === 'run_created' || event.eventType === 'run_started') { + return event.eventData?.deploymentId; + } + return undefined; +} + /** * The whole retention predicate: keep the session only for a pure step * boundary (every suspension item is a step — any other item type, present @@ -1015,6 +1042,82 @@ export function workflowEntrypoint( // describe the next load exactly. let eventLog: ReplayEventLog = { type: 'loadAll' }; + // Resolve the run-scoped key as soon as the deployment id is + // known. On a normal replay that is the streamed run_created + // frame; resilient start and turbo already carry it in the + // queue payload. This starts key resolution and payload + // preparation before the remainder of the event log arrives, + // without guessing the key for a cross-deployment run. + let replayPayloadState: ReplayPayloadCacheState = { + type: 'waitingForKey', + events: [], + }; + const getReplayPayloads = ( + source: ReplayEncryptionKeySource + ): ReplayPayloads | Promise => { + switch (replayPayloadState.type) { + case 'waitingForKey': { + const events = replayPayloadState.events; + const encryptionKey = + source.type === 'run' + ? resolveRunEncryptionKey(world, source.run) + : resolveRunEncryptionKey(world, runId, { + deploymentId: source.deploymentId, + }); + const resolving = encryptionKey.then( + (encryptionKey) => { + const cache = new ReplayPayloadCache(encryptionKey); + for (const event of events) { + cache.prepareEvent(event); + } + const payloads = { encryptionKey, cache }; + replayPayloadState = { + type: 'ready', + payloads, + }; + return payloads; + } + ); + replayPayloadState = { + type: 'resolvingKey', + events, + payloads: resolving, + }; + void resolving.catch(() => {}); + return resolving; + } + case 'resolvingKey': + case 'ready': + return replayPayloadState.payloads; + } + replayPayloadState satisfies never; + }; + const onReplayEvent = (event: Event): void => { + const deploymentId = replayEventDeploymentId(event); + if (deploymentId) { + void getReplayPayloads({ + type: 'deployment', + deploymentId, + }); + } + switch (replayPayloadState.type) { + case 'waitingForKey': + case 'resolvingKey': + replayPayloadState.events.push(event); + return; + case 'ready': + replayPayloadState.payloads.cache.prepareEvent(event); + return; + } + replayPayloadState satisfies never; + }; + if (runInput?.deploymentId) { + void getReplayPayloads({ + type: 'deployment', + deploymentId: runInput.deploymentId, + }); + } + // Shared state: set by either the background step path // or the run_started setup below. let workflowRun: WorkflowRun | undefined; @@ -1335,9 +1438,6 @@ export function workflowEntrypoint( // incremental load starts above the hole and never // returns it. eventLog = { type: 'loadAll' }; - // The corrected log inserts the missing events BELOW the - // length already scanned for payload prewarming, shifting - // every later position. Only a full rescan sees them. replayPayloadCache.resetScan(); } runtimeLogger.warn( @@ -1636,7 +1736,7 @@ export function workflowEntrypoint( getStepFunction(incomingStepName)?.maxRetries ?? DEFAULT_STEP_MAX_RETRIES; if (metadata.attempt > bgMaxRetries + 1) { - const loaded = await loadWorkflowRunEvents(runId); + const loaded = await loadWorkflowRunEvents({ runId }); bgAuthoritativeAttempt = countStepStartedEvents( loaded.events, @@ -1751,7 +1851,7 @@ export function workflowEntrypoint( // Load events to check if all parallel steps are done. // Use cursor-based loading so the main loop can continue // incrementally from here. - const loaded = await loadWorkflowRunEvents(runId); + const loaded = await loadWorkflowRunEvents({ runId }); eventLog = nextEventLogLoad(loaded); // Check for pending steps: any step_created without @@ -1986,6 +2086,7 @@ export function workflowEntrypoint( resumeId: hookResumeInput.resumeId, resumePayloadDigest: hookResumeInput.payloadDigest, preloadEvents: true, + onEvent: onReplayEvent, } ); hookEnsured = true; @@ -2097,13 +2198,12 @@ export function workflowEntrypoint( ), }); } 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 + // Successful write, no usable preload (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( @@ -2260,6 +2360,7 @@ export function workflowEntrypoint( }); const result = await createEvent(runStartedEvent, { requestId, + onEvent: onReplayEvent, }); workflowRun = result.run; maxEventsLimit = clampMaxEvents(result.maxEvents); @@ -2544,31 +2645,21 @@ export function workflowEntrypoint( // do we fall back to reloading the complete log. if (eventLog.type !== 'loadAll' && ensuredEvent) { insertEventByEventId(eventLog.events, ensuredEvent); + onReplayEvent(ensuredEvent); } else { eventLog = { type: 'loadAll' }; } } // end else (re-ensure needed) } - // Resolve the encryption key for this run's deployment. - // Used eagerly here since both workflow execution (input - // hydration / hook payload decryption) and the run_failed - // dehydrate path below need it. Memoized accessor: first - // call triggers the actual fetch / HKDF derivation, - // subsequent calls await the cached promise. - const getEncryptionKey = memoizeEncryptionKey( - world, - workflowRun - ); - const encryptionKey = await getEncryptionKey(); - - // Invocation-scoped cache of VM-independent prepared payloads - // and immutable final values. It survives the fresh workflow - // VM created by each inline replay, but never crosses runs or - // queue deliveries. - const replayPayloadCache = new ReplayPayloadCache( - encryptionKey - ); + // Worlds that do not implement streamed observation still + // resolve from the materialized run. This is also the final + // cross-deployment-safe source of truth. + const { encryptionKey, cache: replayPayloadCache } = + await getReplayPayloads({ + type: 'run', + run: workflowRun, + }); // The live VM parked at the previous boundary, when the // retention decision kept it. null → this iteration cold- @@ -2663,7 +2754,11 @@ export function workflowEntrypoint( if (eventLog.type === 'loadAfter') { appendEventLog( eventLog, - await loadWorkflowRunEvents(runId, eventLog.cursor) + await loadWorkflowRunEvents({ + runId, + afterCursor: eventLog.cursor, + onEvent: onReplayEvent, + }) ); eventLog = { ...eventLog, type: 'ready' }; } @@ -2716,12 +2811,14 @@ export function workflowEntrypoint( } if (eventLog.type !== 'ready') { - const page = await loadWorkflowRunEvents( + const page = await loadWorkflowRunEvents({ runId, - eventLog.type === 'loadAfter' - ? eventLog.cursor - : undefined - ); + afterCursor: + eventLog.type === 'loadAfter' + ? eventLog.cursor + : undefined, + onEvent: onReplayEvent, + }); if (eventLog.type === 'loadAfter') { appendEventLog(eventLog, page); eventLog = { ...eventLog, type: 'ready' }; @@ -2838,10 +2935,11 @@ export function workflowEntrypoint( // not include the wait completion this handler just // attempted. if (eventLog.cursor) { - const page = await loadWorkflowRunEvents( + const page = await loadWorkflowRunEvents({ runId, - eventLog.cursor - ); + afterCursor: eventLog.cursor, + onEvent: onReplayEvent, + }); const completedWaitIdsAfterCursor = new Set( page.events .filter((e) => e.eventType === 'wait_completed') @@ -2858,13 +2956,19 @@ export function workflowEntrypoint( appendEventLog(eventLog, page); } else { eventLog = { - ...(await loadWorkflowRunEvents(runId)), + ...(await loadWorkflowRunEvents({ + runId, + onEvent: onReplayEvent, + })), type: 'ready', }; } } else { eventLog = { - ...(await loadWorkflowRunEvents(runId)), + ...(await loadWorkflowRunEvents({ + runId, + onEvent: onReplayEvent, + })), type: 'ready', }; } @@ -2964,15 +3068,13 @@ export function workflowEntrypoint( if (resumeTracking) { resumeTracking.replayStartedAtMs ??= replayStart; } - // Start every missing decrypt/decompress operation up - // front (already-prepared payloads are skipped). Web - // Crypto work overlaps VM setup on the replay path and - // the appended events' consumption on the resume path; - // consumers still deserialize and resolve in event order. - const payloadPrewarm = replayPayloadCache.prewarm( - workflowRun, - eventLog.events - ); + // Finish scheduling every missing decrypt/decompress + // operation (stream-observed payloads are already in + // flight). Preparation overlaps VM setup on replay and + // appended-event consumption on resume; consumers still + // deserialize and resolve in event order. + const replayEvents = eventLog.events; + replayPayloadCache.prepareAll(workflowRun, replayEvents); let workflowResult: WorkflowResumeResult = retainedSession ? await resumeWorkflow(retainedSession, eventLog.events) : { type: 'replay' }; @@ -2992,8 +3094,6 @@ export function workflowEntrypoint( worldCapabilities: world.capabilities, }); } - await payloadPrewarm; - if (workflowResult.type === 'suspended') { // Park the live session; the suspension catch below // makes the one retention decision — keep it for the @@ -3266,8 +3366,7 @@ export function workflowEntrypoint( } if (suspensionResult.reportedEventCount > 0) { // Bump-and-report merged events BELOW the tail and - // re-sorted the array to slot order, shifting every - // position the prewarm scan had already recorded. + // re-sorted the array to slot order. // The cursor is deliberately left alone: the report // is a lower bound on what was skipped, so the next // incremental read still has to cover the same range. diff --git a/packages/core/src/runtime/helpers.test.ts b/packages/core/src/runtime/helpers.test.ts index 0ce474ef07..cc397963bd 100644 --- a/packages/core/src/runtime/helpers.test.ts +++ b/packages/core/src/runtime/helpers.test.ts @@ -414,7 +414,7 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test'); + const result = await loadWorkflowRunEvents({ runId: 'wrun_test' }); expect(result.events).toHaveLength(2); expect(result.cursor).toBe('eid:evnt_b'); @@ -447,7 +447,7 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test'); + const result = await loadWorkflowRunEvents({ runId: 'wrun_test' }); expect(result.events).toHaveLength(2); expect(result.cursor).toBe('eid:evnt_b'); @@ -461,7 +461,7 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test'); + const result = await loadWorkflowRunEvents({ runId: 'wrun_test' }); expect(result.events).toHaveLength(0); expect(result.cursor).toBeNull(); @@ -484,7 +484,7 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test'); + const result = await loadWorkflowRunEvents({ runId: 'wrun_test' }); expect(result.events.map((e) => e.eventId)).toEqual([ 'evnt_a', @@ -501,7 +501,10 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test', 'eid:evnt_z'); + const result = await loadWorkflowRunEvents({ + runId: 'wrun_test', + afterCursor: 'eid:evnt_z', + }); expect(result.events).toHaveLength(0); // Preserving the input cursor avoids the runtime treating "no new events @@ -521,7 +524,7 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test'); + const result = await loadWorkflowRunEvents({ runId: 'wrun_test' }); expect(result.events.map((event) => event.eventId)).toEqual([ 'evnt_a', @@ -540,7 +543,10 @@ describe('loadWorkflowRunEvents', () => { hasMore: false, }); - const result = await loadWorkflowRunEvents('wrun_test', 'opaque-cursor'); + const result = await loadWorkflowRunEvents({ + runId: 'wrun_test', + afterCursor: 'opaque-cursor', + }); expect(result.events.map((event) => event.eventId)).toEqual([ 'evnt_a', @@ -568,7 +574,9 @@ describe('loadWorkflowRunEvents', () => { hasMore: true, }); - await expect(loadWorkflowRunEvents('wrun_test')).rejects.toMatchObject({ + await expect( + loadWorkflowRunEvents({ runId: 'wrun_test' }) + ).rejects.toMatchObject({ code: 'WORLD_CONTRACT_ERROR', }); expect(eventsListMock).toHaveBeenCalledTimes(2); @@ -581,7 +589,9 @@ describe('loadWorkflowRunEvents', () => { hasMore: true, }); - await expect(loadWorkflowRunEvents('wrun_test')).rejects.toMatchObject({ + await expect( + loadWorkflowRunEvents({ runId: 'wrun_test' }) + ).rejects.toMatchObject({ code: 'WORLD_CONTRACT_ERROR', }); expect(eventsListMock).toHaveBeenCalledTimes(1); @@ -971,6 +981,18 @@ describe('memoizeEncryptionKey', () => { expect(spy).toHaveBeenCalledTimes(1); }); + it('passes deployment context when resolving before the run is materialized', async () => { + const spy = vi.fn().mockResolvedValue(MATERIAL); + const getKey = memoizeEncryptionKey(worldWithKey(spy), 'wrun_1', { + deploymentId: 'dpl_streamed', + }); + + await getKey(); + expect(spy).toHaveBeenCalledWith('wrun_1', { + deploymentId: 'dpl_streamed', + }); + }); + it('resolves undefined when encryption is not configured', async () => { const getKey = memoizeEncryptionKey(worldWithKey(undefined), 'wrun_1'); await expect(getKey()).resolves.toBeUndefined(); diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index f24944478f..939369b329 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -592,10 +592,15 @@ function shouldRetryWithoutEventCursor( * The returned cursor can be passed back in on a subsequent call for * incremental loading. */ -export async function loadWorkflowRunEvents( - runId: string, - afterCursor?: string -): Promise { +export async function loadWorkflowRunEvents({ + runId, + afterCursor, + onEvent, +}: { + runId: string; + afterCursor?: string; + onEvent?: (event: Event) => void; +}): Promise { const incremental = afterCursor !== undefined; return trace( incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents', @@ -630,6 +635,7 @@ export async function loadWorkflowRunEvents( sortOrder: 'asc', cursor: requestedCursor ?? undefined, }, + onEvent, }); } catch (error) { if ( @@ -981,7 +987,7 @@ export async function settleEventSlotGap( await new Promise((resolve) => setTimeout(resolve, SLOT_GAP_RECHECK_BASE_DELAY_MS * 2 ** attempt) ); - log = await loadWorkflowRunEvents(runId); + log = await loadWorkflowRunEvents({ runId }); gap = findEventSlotGap(log.events); } return { log, gap }; @@ -1215,56 +1221,37 @@ export function getQueueOverhead(message: { requestedAt?: Date }) { } /** - * Returns a memoized accessor for a run's full encryption capability. - * - * The first call resolves the run's key material via - * `world.getEncryptionKeyForRun` (which may do HKDF derivation locally on - * Vercel, or a network fetch from external contexts) and derives a - * {@link PayloadKey} from it; subsequent calls await the same cached promise. - * If the world doesn't support encryption or the run has no key configured, - * the cached value is `undefined`. - * - * The resolved value is deliberately the *full* capability — the symmetric AES - * key plus the run's X25519 keypair — not just a `CryptoKey`. A run reading - * its own event log can encounter sealed (`encp`) payloads that another run - * wrote to it (a cross-deployment hook resumption, say), and opening those - * needs the keypair. Resolving only the symmetric key would leave those - * payloads unopenable and wedge the run. - * - * Used by step / workflow handlers to defer the (potentially expensive) - * key fetch until the first code path that actually needs it — typically - * input hydration on the success path, or error dehydration on a failure - * path. Both paths can race-call the accessor without triggering duplicate - * fetches. - * - * Errors thrown by `getEncryptionKeyForRun` propagate to every caller - * (the cached promise rejects). This is intentional: when encryption is - * configured, we never want to silently fall back to plaintext - * serialization. A propagated error in an event-emission path leaves the - * outer try/catch to log and surface the issue; the queue's redelivery - * semantics will retry the key fetch on the next attempt. + * Resolves a run's full encryption capability: its symmetric key and X25519 + * keypair. The keypair is required to open sealed cross-deployment payloads. */ +export async function resolveRunEncryptionKey( + world: World, + runOrId: WorkflowRun | string, + context?: Record +): Promise { + // The `getEncryptionKeyForRun` overload set takes either a `WorkflowRun` or + // a `runId: string` (with optional context). Branch here so TypeScript picks + // the right overload for each shape. + const rawKey = + typeof runOrId === 'string' + ? await world.getEncryptionKeyForRun?.(runOrId, context) + : await world.getEncryptionKeyForRun?.(runOrId); + // Resolve the *full* capability, not just the symmetric key: a run reading + // its own event log may encounter sealed (`encp`) payloads that another run + // wrote to it, and opening those needs the run's X25519 scalar as well. + return rawKey ? await deriveRunPayloadKeys(rawKey) : undefined; +} + +/** Returns a lazy accessor that shares the first key lookup and its outcome. */ export function memoizeEncryptionKey( world: World, - runOrId: WorkflowRun | string + runOrId: WorkflowRun | string, + context?: Record ): () => Promise { let cached: Promise | undefined; return () => { if (!cached) { - cached = (async () => { - // The `getEncryptionKeyForRun` overload set takes either a - // `WorkflowRun` or a `runId: string` (with optional context). Branch - // here so TypeScript picks the right overload for each shape. - const rawKey = - typeof runOrId === 'string' - ? await world.getEncryptionKeyForRun?.(runOrId) - : await world.getEncryptionKeyForRun?.(runOrId); - // Resolve the *full* capability, not just the symmetric key: a run - // reading its own event log may encounter sealed (`encp`) payloads - // that another run wrote to it, and opening those needs the run's - // X25519 scalar as well. - return rawKey ? await deriveRunPayloadKeys(rawKey) : undefined; - })(); + cached = resolveRunEncryptionKey(world, runOrId, context); } return cached; }; diff --git a/packages/core/src/runtime/quickjs-partial-preload.test.ts b/packages/core/src/runtime/quickjs-partial-preload.test.ts index d9d3edf528..33f7b1df80 100644 --- a/packages/core/src/runtime/quickjs-partial-preload.test.ts +++ b/packages/core/src/runtime/quickjs-partial-preload.test.ts @@ -116,6 +116,7 @@ describe('QuickJS partial run_started preload', () => { expect(listEvents).toHaveBeenCalledWith({ runId, pagination: { sortOrder: 'asc', cursor: preloadCursor }, + onEvent: expect.any(Function), }); expect(runWorkflowWithQuickJS).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/packages/core/src/serialization/compression-telemetry.test.ts b/packages/core/src/serialization/compression-telemetry.test.ts index 9ca73262fd..a3ce736689 100644 --- a/packages/core/src/serialization/compression-telemetry.test.ts +++ b/packages/core/src/serialization/compression-telemetry.test.ts @@ -117,10 +117,10 @@ describe('compression telemetry attributes', () => { ); recordedAttributes.length = 0; - await new ReplayPayloadCache(undefined).prepareEventPayload( + await new ReplayPayloadCache(undefined).getEventValue( 'evnt_test', - 'result', - data + data, + (prepared) => prepared ); const attrs = lastAttrs(); diff --git a/packages/core/src/step-delivery-hop-count.test.ts b/packages/core/src/step-delivery-hop-count.test.ts index 7653f85d4e..65f628206f 100644 --- a/packages/core/src/step-delivery-hop-count.test.ts +++ b/packages/core/src/step-delivery-hop-count.test.ts @@ -44,7 +44,7 @@ import { createSleep } from './workflow/sleep.js'; function setupWorkflowContext( events: Event[], - replayPayloadCache: ReplayPayloadCache = new ReplayPayloadCache(undefined) + replayPayloadCache: ReplayPayloadCache = new ReplayPayloadCache() ): WorkflowOrchestratorContext { const context = createContext({ seed: 'test', @@ -229,7 +229,7 @@ describe('step delivery ordering is independent of consumer hop count: hook payl const spy = await slowHydration(); try { const events = await buildEventLog(); - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const c1 = setupWorkflowContext(events, cache); const r1 = await runWithDiscontinuation(c1, body(c1, extraHops)); if (!WorkflowSuspension.is(r1.error)) { @@ -346,7 +346,7 @@ describe('step delivery ordering is independent of consumer hop count: wait comp const spy = await slowHydration(); try { const events = await buildWaitEventLog(); - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const c1 = setupWorkflowContext(events, cache); const r1 = await runWithDiscontinuation(c1, waitBody(c1, extraHops)); if (!WorkflowSuspension.is(r1.error)) @@ -472,7 +472,7 @@ describe('step delivery ordering is independent of consumer hop count: step fail const spy = await slowHydration(); try { const events = await buildFailedEventLog(); - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const c1 = setupWorkflowContext(events, cache); const r1 = await runWithDiscontinuation(c1, failedBody(c1, extraHops)); if (!WorkflowSuspension.is(r1.error)) { diff --git a/packages/core/src/step-delivery-ordering.test.ts b/packages/core/src/step-delivery-ordering.test.ts index 93a21be71e..bf5c35bd51 100644 --- a/packages/core/src/step-delivery-ordering.test.ts +++ b/packages/core/src/step-delivery-ordering.test.ts @@ -41,7 +41,7 @@ import { createSleep } from './workflow/sleep.js'; * - `wait_completed` resolves through a detached chain with a fixed, small * microtask-hop count (`workflow/sleep.ts`). A `step_completed` instead * resolves inside a serial `ctx.promiseQueue` slot that first hydrates the - * payload via `ReplayPayloadCache.getStepResult(...)`. That hop count is not + * payload via `ReplayPayloadCache.getEventValue(...)`. That hop count is not * fixed: the first hydration pays async decrypt/deserialize, while a later * replay sharing the same `ReplayPayloadCache` hits the * `primitiveStepResults` memo for small primitive results and resolves in @@ -91,7 +91,7 @@ import { createSleep } from './workflow/sleep.js'; */ function setupWorkflowContext( events: Event[], - replayPayloadCache: ReplayPayloadCache = new ReplayPayloadCache(undefined) + replayPayloadCache: ReplayPayloadCache = new ReplayPayloadCache() ): WorkflowOrchestratorContext { const context = createContext({ seed: 'test', @@ -340,7 +340,7 @@ describe('step result delivery ordering across replays', () => { // One cache for both replays: production shares a single // `ReplayPayloadCache` across every replay of one queue delivery. - const sharedCache = new ReplayPayloadCache(undefined); + const sharedCache = new ReplayPayloadCache(); const firstCtx = setupWorkflowContext(events, sharedCache); const first = await runWithDiscontinuation( @@ -532,7 +532,7 @@ describe('step result delivery ordering across replays', () => { const hydration = delayHydration(); spy = await hydration.install(); const events = await buildEventLog(); - const sharedCache = new ReplayPayloadCache(undefined); + const sharedCache = new ReplayPayloadCache(); const firstCtx = setupWorkflowContext(events, sharedCache); const first = await runWithDiscontinuation( @@ -578,7 +578,7 @@ describe('step result delivery ordering across replays', () => { const hydration = delayHydration(); spy = await hydration.install(); const events = await buildEventLog(); - const sharedCache = new ReplayPayloadCache(undefined); + const sharedCache = new ReplayPayloadCache(); for (const replay of [1, 2]) { const ctx = setupWorkflowContext(events, sharedCache); diff --git a/packages/core/src/step-hydration-memoization.test.ts b/packages/core/src/step-hydration-memoization.test.ts index e8aabc1d0d..ede78c99b2 100644 --- a/packages/core/src/step-hydration-memoization.test.ts +++ b/packages/core/src/step-hydration-memoization.test.ts @@ -25,7 +25,7 @@ import { createContext } from './vm/index.js'; // the inline loop threads one cache across replay iterations. function setupWorkflowContext( events: Event[], - replayPayloadCache = new ReplayPayloadCache(undefined) + replayPayloadCache = new ReplayPayloadCache() ): WorkflowOrchestratorContext { const context = createContext({ seed: 'test', @@ -90,7 +90,7 @@ describe('step hydration memoization through the step consumer', () => { it('skips re-hydration of primitive step results on a second replay sharing the cache', async () => { const events = await makeStepEvents(); - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); const serialization = await import('./serialization.js'); const hydrateSpy = vi.spyOn(serialization, 'hydrateStepReturnValue'); @@ -122,7 +122,7 @@ describe('step hydration memoization through the step consumer', () => { it('preserves event-log resolution order on cache hits even with variable timing', async () => { const events = await makeStepEvents(); - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); // Replay 1: populate the cache (no timing games needed). const ctx1 = setupWorkflowContext(events, cache); @@ -168,7 +168,7 @@ describe('step hydration memoization through the step consumer', () => { createdAt: new Date(), }, ]; - const cache = new ReplayPayloadCache(undefined); + const cache = new ReplayPayloadCache(); // Replay 1: hydrate the object, then mutate it (as workflow code might). const ctx1 = setupWorkflowContext(events, cache); diff --git a/packages/core/src/step.test.ts b/packages/core/src/step.test.ts index 574e441110..cd759f9004 100644 --- a/packages/core/src/step.test.ts +++ b/packages/core/src/step.test.ts @@ -54,7 +54,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { return { runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts index 1fcc4c11df..78a9379cee 100644 --- a/packages/core/src/step.ts +++ b/packages/core/src/step.ts @@ -190,18 +190,18 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) { ctx.pendingDeliveries++; ctx.promiseQueue = ctx.promiseQueue.then(async () => { try { - const prepared = await ctx.replayPayloadCache.prepareEventPayload( + rejection = await ctx.replayPayloadCache.getEventValue( event.eventId, - 'error', - event.eventData.error - ); - rejection = await hydrateStepError( event.eventData.error, - ctx.runId, - ctx.encryptionKey, - ctx.globalThis, - {}, - prepared + (prepared) => + hydrateStepError( + event.eventData.error, + ctx.runId, + ctx.encryptionKey, + ctx.globalThis, + {}, + prepared + ) ); } catch (hydrateErr) { // If hydration fails for any reason, fall back to a generic @@ -301,24 +301,18 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) { ctx.pendingDeliveries++; ctx.promiseQueue = ctx.promiseQueue.then(async () => { try { - const hydratedResult = await ctx.replayPayloadCache.getStepResult( + const hydratedResult = await ctx.replayPayloadCache.getEventValue( completedEventId, - async () => { - const prepared = - await ctx.replayPayloadCache.prepareEventPayload( - completedEventId, - 'result', - serializedResult - ); - return await hydrateStepReturnValue( + serializedResult, + (prepared) => + hydrateStepReturnValue( serializedResult, ctx.runId, ctx.encryptionKey, ctx.globalThis, {}, prepared - ); - } + ) ); outcome = { ok: true, value: hydratedResult as Result }; } catch (error) { diff --git a/packages/core/src/test-support/orchestrator-context.ts b/packages/core/src/test-support/orchestrator-context.ts index 0b68b46bb3..329ca5eac5 100644 --- a/packages/core/src/test-support/orchestrator-context.ts +++ b/packages/core/src/test-support/orchestrator-context.ts @@ -38,7 +38,7 @@ export function setupWorkflowContext( suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 644e60a9f4..1f02426fc7 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -1099,7 +1099,7 @@ async function createWorkflowSession({ // workflow function subscribing its first step callbacks. let args: unknown[] = []; workflowContext.promiseQueue = workflowContext.promiseQueue.then(async () => { - const prepared = await replayPayloadCache.prepareWorkflowInput(workflowRun); + const prepared = await replayPayloadCache.getWorkflowInput(workflowRun); args = await hydrateWorkflowArguments( workflowRun.input, workflowRun.runId, diff --git a/packages/core/src/workflow/abort-controller.ts b/packages/core/src/workflow/abort-controller.ts index 4d5d9a08af..6b7cacd6cb 100644 --- a/packages/core/src/workflow/abort-controller.ts +++ b/packages/core/src/workflow/abort-controller.ts @@ -230,19 +230,18 @@ export function createCreateAbortController(ctx: WorkflowOrchestratorContext) { try { if (rawPayload !== undefined) { try { - const prepared = - await ctx.replayPayloadCache.prepareEventPayload( - event.eventId, - 'payload', - rawPayload - ); - const hydrated = (await hydrateStepReturnValue( + const hydrated = (await ctx.replayPayloadCache.getEventValue( + event.eventId, rawPayload, - ctx.runId, - ctx.encryptionKey, - ctx.globalThis, - {}, - prepared + (prepared) => + hydrateStepReturnValue( + rawPayload, + ctx.runId, + ctx.encryptionKey, + ctx.globalThis, + {}, + prepared + ) )) as { reason?: unknown } | undefined; if ( hydrated && diff --git a/packages/core/src/workflow/hook.test.ts b/packages/core/src/workflow/hook.test.ts index c6c59727db..0ac3f008ae 100644 --- a/packages/core/src/workflow/hook.test.ts +++ b/packages/core/src/workflow/hook.test.ts @@ -39,7 +39,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { runId: 'wrun_test', encryptionKey: undefined, worldCapabilities: { hookRetention: { active: true } }, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, eventsConsumer: new EventsConsumer(events, { // Fake context: no deliveries are modeled, so the gate is a no-op here. diff --git a/packages/core/src/workflow/hook.ts b/packages/core/src/workflow/hook.ts index bcb714d89d..16e14ccedc 100644 --- a/packages/core/src/workflow/hook.ts +++ b/packages/core/src/workflow/hook.ts @@ -352,19 +352,18 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { | { ok: false; error: unknown }; ctx.promiseQueue = ctx.promiseQueue.then(async () => { try { - const prepared = - await ctx.replayPayloadCache.prepareEventPayload( - event.eventId, - 'payload', - event.eventData.payload - ); - const payload = await hydrateStepReturnValue( + const payload = await ctx.replayPayloadCache.getEventValue( + event.eventId, event.eventData.payload, - ctx.runId, - ctx.encryptionKey, - ctx.globalThis, - {}, - prepared + (prepared) => + hydrateStepReturnValue( + event.eventData.payload, + ctx.runId, + ctx.encryptionKey, + ctx.globalThis, + {}, + prepared + ) ); hydrateOutcome = { ok: true, value: payload as T }; } catch (error) { @@ -425,18 +424,18 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { ctx.pendingDeliveries++; ctx.promiseQueue = ctx.promiseQueue.then(async () => { try { - const prepared = await ctx.replayPayloadCache.prepareEventPayload( + const payload = await ctx.replayPayloadCache.getEventValue( event.eventId, - 'payload', - event.eventData.payload - ); - const payload = await hydrateStepReturnValue( event.eventData.payload, - ctx.runId, - ctx.encryptionKey, - ctx.globalThis, - {}, - prepared + (prepared) => + hydrateStepReturnValue( + event.eventData.payload, + ctx.runId, + ctx.encryptionKey, + ctx.globalThis, + {}, + prepared + ) ); outcome = { ok: true, value: payload as T }; } catch (error) { diff --git a/packages/core/src/workflow/sleep.test.ts b/packages/core/src/workflow/sleep.test.ts index 7780bc05f1..139b185653 100644 --- a/packages/core/src/workflow/sleep.test.ts +++ b/packages/core/src/workflow/sleep.test.ts @@ -23,7 +23,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, - replayPayloadCache: new ReplayPayloadCache(undefined), + replayPayloadCache: new ReplayPayloadCache(), globalThis: context.globalThis, // ctx.onWorkflowError is accessed via closure — it's defined below on the same object eventsConsumer: new EventsConsumer(events, { diff --git a/packages/world-vercel/src/event-retry.ts b/packages/world-vercel/src/event-retry.ts index b461c21c80..cff34a7d83 100644 --- a/packages/world-vercel/src/event-retry.ts +++ b/packages/world-vercel/src/event-retry.ts @@ -88,6 +88,14 @@ export class EventPostResponseError extends Error { } } +/** Preserve an observer's original thrown value without letting it retry I/O. */ +export class EventObserverError extends EventPostResponseError { + constructor(readonly error: unknown) { + super('event observer failed', { cause: error }); + this.name = 'EventObserverError'; + } +} + export interface EventRetryPolicy { /** Whether a failed POST of this event type may be retried in-process. */ retryable: boolean; diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts index 12314793b5..27c81e2e70 100644 --- a/packages/world-vercel/src/events-v4.test.ts +++ b/packages/world-vercel/src/events-v4.test.ts @@ -374,6 +374,51 @@ describe('getWorkflowRunEventsV4 over HTTP', () => { agent.assertNoPendingInterceptors(); }); + it('does not retry an observer failure that resembles transport failure', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true', + method: 'GET', + }) + .reply( + 200, + Buffer.concat([ + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_started', + createdAt: CREATED_AT, + }, + new Uint8Array() + ), + encodeFrame( + { _end: 1, next: 'eid:evnt_1', hasMore: false }, + new Uint8Array() + ), + ]), + { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } + ); + + const observerError = new WorkflowWorldError('observer failed', { + code: 'TRANSPORT', + }); + await expect( + getWorkflowRunEventsV4( + { + runId: 'wrun_1', + onEvent: () => { + throw observerError; + }, + }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toBe(observerError); + agent.assertNoPendingInterceptors(); + }); + it('captures an explicit hasMore from the sentinel, independent of next', async () => { const agent = mockAgent(); @@ -1022,6 +1067,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { it('continues a truncated run_started stream after its last event', async () => { const agent = mockAgent(); + const observed: string[] = []; agent .get(ORIGIN) @@ -1081,7 +1127,8 @@ describe('createWorkflowRunEventV4 over HTTP', () => { const result = await createWorkflowRunStartedEventV4( { runId: 'wrun_1', specVersion: 5 }, - { token: 'test-token', dispatcher: agent } + { token: 'test-token', dispatcher: agent }, + (event) => observed.push(event.eventId) ); expect(result.events.map((event) => event.eventId)).toEqual([ @@ -1089,6 +1136,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { 'evnt_2', ]); expect(result.hasMore).toBe(false); + expect(observed).toEqual(['evnt_1', 'evnt_2']); agent.assertNoPendingInterceptors(); }); @@ -1160,10 +1208,13 @@ describe('createWorkflowRunEventV4 over HTTP', () => { agent.assertNoPendingInterceptors(); }); - it('retries a truncated continuation that produced no complete event', async () => { + it('resumes after an event observed during a truncated continuation', async () => { const agent = mockAgent(); + const observed: string[] = []; const continuationPath = '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_1&remoteRefBehavior=resolve&returnAll=true'; + const resumedPath = + '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_2&remoteRefBehavior=resolve&returnAll=true'; agent .get(ORIGIN) @@ -1208,33 +1259,32 @@ describe('createWorkflowRunEventV4 over HTTP', () => { agent .get(ORIGIN) .intercept({ path: continuationPath, method: 'GET' }) - .reply(200, runStartedFrame.subarray(0, 4), { + .reply(200, runStartedFrame, { headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, }); agent .get(ORIGIN) - .intercept({ path: continuationPath, method: 'GET' }) + .intercept({ path: resumedPath, method: 'GET' }) .reply( 200, - Buffer.concat([ - runStartedFrame, - encodeFrame( - { _end: 1, next: 'eid:evnt_2', hasMore: false }, - new Uint8Array() - ), - ]), + encodeFrame( + { _end: 1, next: 'eid:evnt_2', hasMore: false }, + new Uint8Array() + ), { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } ); const result = await createWorkflowRunStartedEventV4( { runId: 'wrun_1', specVersion: 5 }, - { token: 'test-token', dispatcher: agent } + { token: 'test-token', dispatcher: agent }, + (event) => observed.push(event.eventId) ); expect(result.events.map((event) => event.eventId)).toEqual([ 'evnt_1', 'evnt_2', ]); + expect(observed).toEqual(['evnt_1', 'evnt_2']); expect(result.cursor).toBe('eid:evnt_2'); agent.assertNoPendingInterceptors(); }); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 8867db298a..51c90283e3 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -42,6 +42,7 @@ import { import { decode } from 'cbor-x'; import { z } from 'zod'; import { + EventObserverError, EventPostResponseError, isRetryableEventRequestError, } from './event-retry.js'; @@ -179,9 +180,10 @@ async function withV4ResponseBody( } catch (error) { if (!outcomeReported) { const incomplete = - error instanceof IncompleteFrameError || - error instanceof PartialEventStreamError || - isRecyclableTransportError(error); + !(error instanceof EventObserverError) && + (error instanceof IncompleteFrameError || + error instanceof PartialEventStreamError || + isRecyclableTransportError(error)); report(incomplete ? error : undefined); } throw error; @@ -856,11 +858,13 @@ async function decodeCreateEventResponse( export async function createWorkflowRunStartedEventV4( input: CreateEventV4InputBase, - config?: APIConfig + config?: APIConfig, + onEvent?: (event: Event) => void ) { const { responseHeaders, ...replay } = await postReplayLogEvent( { ...input, eventType: 'run_started' }, - config + config, + onEvent ); assert(replay.cursor, 'v4 createEvent: event stream missing cursor'); const maxEvents = MaxEventsHeaderSchema.safeParse( @@ -1283,7 +1287,8 @@ interface ReplayLog { */ export async function createHookReceivedPreloadEventV4( input: CreateEventV4InputBase, - config?: APIConfig + config?: APIConfig, + onEvent?: (event: Event) => void ): Promise< ReplayLog & { canonicalEventId: string | undefined; @@ -1292,7 +1297,8 @@ export async function createHookReceivedPreloadEventV4( > { const { responseHeaders, ...replay } = await postReplayLogEvent( { ...input, eventType: 'hook_received' }, - config + config, + onEvent ); const maxEvents = MaxEventsHeaderSchema.safeParse( responseHeaders.get(MAX_EVENTS_HEADER) @@ -1385,7 +1391,8 @@ const MAX_PARTIAL_EVENT_STREAM_RETRIES = 3; async function consumeEventFrameStream( response: Response, opName: string, - events: Event[] + events: Event[], + onEvent?: (event: Event) => void ): Promise<{ cursor: string | null; hasMore: boolean }> { try { for await (const frame of decodeFrames( @@ -1401,7 +1408,15 @@ async function consumeEventFrameStream( if (Object.keys(frame.meta).some((key) => key.startsWith('_'))) { throw new Error(`v4 ${opName}: unexpected control frame`); } - events.push(decodeEventFrame(frame)); + const event = decodeEventFrame(frame); + events.push(event); + if (onEvent) { + try { + onEvent(event); + } catch (error) { + throw new EventObserverError(error); + } + } } } catch (cause) { if (!(cause instanceof IncompleteFrameError)) throw cause; @@ -1422,7 +1437,8 @@ async function postReplayLogEvent( input: CreateEventV4InputBase & { eventType: 'run_started' | 'hook_received'; }, - config?: APIConfig + config?: APIConfig, + onEvent?: (event: Event) => void ): Promise { const events: Event[] = []; let responseHeaders: Headers | undefined; @@ -1432,7 +1448,12 @@ async function postReplayLogEvent( config, (response) => { responseHeaders = response.headers; - return consumeEventFrameStream(response, 'createEvent', events); + return consumeEventFrameStream( + response, + 'createEvent', + events, + onEvent + ); } ); assert(responseHeaders); @@ -1445,8 +1466,12 @@ async function postReplayLogEvent( const continuationCursor = `eid:${lastEvent.eventId}`; try { - const suffix = await getWorkflowRunEventsV4( - { runId: input.runId, pagination: { cursor: continuationCursor } }, + const suffix = await listWorkflowRunEventsV4( + { + runId: input.runId, + pagination: { cursor: continuationCursor }, + onEvent, + }, config ); events.push(...suffix.data); @@ -1457,6 +1482,7 @@ async function postReplayLogEvent( responseHeaders, }; } catch (cause) { + if (cause instanceof EventObserverError) throw cause; throw new EventPostResponseError( `v4 createEvent: replay continuation failed for run ${input.runId}`, { cause } @@ -1506,7 +1532,7 @@ function paginationToQuery( * `getWorkflowRunEvents` contract. A truncated full response resumes * after its last validated event instead of downloading accepted frames again. */ -export async function getWorkflowRunEventsV4( +async function listWorkflowRunEventsV4( params: ListEventsParams, config?: APIConfig ): Promise> { @@ -1527,10 +1553,17 @@ export async function getWorkflowRunEventsV4( opName: 'listEvents', streamResponse: true, }, - (response) => consumeEventFrameStream(response, 'listEvents', events) + (response) => + consumeEventFrameStream( + response, + 'listEvents', + events, + params.onEvent + ) ); return { data: events, cursor: result.cursor, hasMore: result.hasMore }; } catch (error) { + if (error instanceof EventObserverError) throw error; const lastEvent = events.at(-1); if ( retries === MAX_PARTIAL_EVENT_STREAM_RETRIES || @@ -1544,6 +1577,18 @@ export async function getWorkflowRunEventsV4( } } +export async function getWorkflowRunEventsV4( + params: ListEventsParams, + config?: APIConfig +): Promise> { + try { + return await listWorkflowRunEventsV4(params, config); + } catch (error) { + if (error instanceof EventObserverError) throw error.error; + throw error; + } +} + /** * GET /api/v4/events?correlationId=...&runId=... * diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index c396ad9724..b41280e17b 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -1,10 +1,11 @@ import { Buffer } from 'node:buffer'; import { gzipSync } from 'node:zlib'; +import { WorkflowWorldError } from '@workflow/errors'; import type { AnyEventRequest, CreateEventParams } from '@workflow/world'; import { decode, encode } from 'cbor-x'; import { ulid } from 'ulid'; import { MockAgent } from 'undici'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { EventPostResponseError } from './event-retry.js'; import { createWorkflowRunEvent, @@ -376,6 +377,117 @@ describe('createWorkflowRunEvent result contract', () => { agent.assertNoPendingInterceptors(); }); + it('does not repeat a POST when its event observer throws', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/run_started', + method: 'POST', + }) + .reply(200, runStartedResponse(), { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_1', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': STARTED_AT.toISOString(), + 'x-wf-max-events': '10000', + }, + }); + + const observerError = new WorkflowWorldError('observer failed', { + code: 'TRANSPORT', + }); + const onEvent = vi.fn(() => { + throw observerError; + }); + + await expect( + createWorkflowRunEvent( + 'wrun_1', + { eventType: 'run_started', specVersion: 2 } as AnyEventRequest, + { onEvent }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toBe(observerError); + expect(onEvent).toHaveBeenCalledOnce(); + agent.assertNoPendingInterceptors(); + }); + + it('preserves an observer failure from a replay continuation', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/run_started', + method: 'POST', + }) + .reply( + 200, + encodeFrame( + { + eventId: 'evnt_0', + runId: 'wrun_1', + eventType: 'run_created', + createdAt: new Date('2026-06-09T23:59:59.000Z'), + specVersion: 2, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'workflow', + }, + }, + new Uint8Array() + ), + { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-max-events': '10000', + }, + } + ); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_0&remoteRefBehavior=resolve&returnAll=true', + method: 'GET', + }) + .reply( + 200, + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_started', + createdAt: STARTED_AT, + specVersion: 2, + eventData: {}, + }, + new Uint8Array() + ), + { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } + ); + + const observerError = new WorkflowWorldError('observer failed', { + code: 'TRANSPORT', + }); + const observed: string[] = []; + const onEvent = vi.fn((event: { eventId: string }) => { + observed.push(event.eventId); + if (event.eventId === 'evnt_1') throw observerError; + }); + + await expect( + createWorkflowRunEvent( + 'wrun_1', + { eventType: 'run_started', specVersion: 2 } as AnyEventRequest, + { onEvent }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toBe(observerError); + expect(observed).toEqual(['evnt_0', 'evnt_1']); + agent.assertNoPendingInterceptors(); + }); + it('does not repeat a run_started POST when its continuation exhausts retries', async () => { const agent = mockAgent(); agent diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index d2903ff116..c7e6cb9d98 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -54,7 +54,7 @@ import { validateUlidTimestamp, type WorkflowRun, } from '@workflow/world'; -import { withEventPostRetry } from './event-retry.js'; +import { EventObserverError, withEventPostRetry } from './event-retry.js'; import { createHookReceivedPreloadEventV4, createWorkflowRunEventsBatchV4, @@ -622,6 +622,7 @@ export async function createWorkflowRunEvent( } return result as EventResult; } catch (err) { + if (err instanceof EventObserverError) throw err.error; // 404 on hook_disposed / hook_received → already-disposed hook. if ( isHookEventRequiringExistence(data.eventType) && @@ -736,7 +737,11 @@ async function createWorkflowRunEventInner( }; if (data.eventType === 'run_started' && !params?.skipPreload) { - const result = await createWorkflowRunStartedEventV4(input, config); + const result = await createWorkflowRunStartedEventV4( + input, + config, + params?.onEvent + ); const runCreated = result.events.find( (event) => event.eventType === 'run_created' ); @@ -781,7 +786,8 @@ async function createWorkflowRunEventInner( // default. const outcome = await createHookReceivedPreloadEventV4( { ...input, remoteRefBehavior: 'lazy' }, - config + config, + params.onEvent ); const { canonicalEventId, maxEvents, events, cursor, hasMore } = outcome; const canonicalEvent = events.find( diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index df89aaa07d..f12fe173c8 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -1021,6 +1021,14 @@ export interface CreateEventParams { * `resumeHook()` must not set it. */ preloadEvents?: true; + /** + * Observe replay-preload events immediately after validation, in stream + * order. The callback is synchronous: it delays the next frame and therefore + * applies response backpressure. Recovery continues after the last observed + * event instead of replaying the accepted prefix. Throwing aborts the + * operation and is never treated as a transport failure or retried. + */ + onEvent?: (event: Event) => void; } /** @@ -1208,6 +1216,14 @@ export interface ListEventsParams { /** Omit `limit` to return every remaining event. */ pagination?: PaginationOptions; resolveData?: ResolveData; + /** + * Observe events immediately after validation, in stream order. The callback + * is synchronous: it delays the next frame and therefore applies response + * backpressure. Recovery continues after the last observed event instead of + * replaying the accepted prefix. Throwing aborts the operation and is never + * treated as a transport failure or retried. + */ + onEvent?: (event: Event) => void; } export interface ListEventsByCorrelationIdParams {