diff --git a/.changeset/prepare-streamed-replay-payloads.md b/.changeset/prepare-streamed-replay-payloads.md new file mode 100644 index 0000000000..9678641541 --- /dev/null +++ b/.changeset/prepare-streamed-replay-payloads.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Prepare replay payloads as 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 ee113b8ff0..27b29d2b5d 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -86,7 +86,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 f09ec0f4ca..dc2f36eede 100644 --- a/packages/core/src/replay-payload-cache.test.ts +++ b/packages/core/src/replay-payload-cache.test.ts @@ -54,17 +54,29 @@ function makeEvents(payloads: unknown[]): Event[] { } describe('ReplayPayloadCache', () => { - it('deduplicates preparation and accepts a synchronous preparer', async () => { + it('deduplicates synchronous preparation without creating a promise', () => { const payload = new Uint8Array([1]); const preparer = vi.fn((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.toBe(payload); + expect(first).toEqual(payload); expect(preparer).toHaveBeenCalledOnce(); + expect(hydrate).toHaveBeenCalledTimes(2); + }); + + it('hydrates and memoizes a primitive without creating a promise', () => { + const payload = new Uint8Array([1]); + const hydrate = vi.fn(() => 42); + const cache = new ReplayPayloadCache(undefined, (value) => value); + + expect(cache.getEventValue('evnt_one', payload, hydrate)).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 () => { @@ -76,13 +88,12 @@ describe('ReplayPayloadCache', () => { .mockReturnValueOnce(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.toBe(payload); + expect(cache.getWorkflowInput(run)).toEqual(payload); expect(preparer).toHaveBeenCalledTimes(2); }); @@ -99,16 +110,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('prepares streamed events synchronously inside the decoder callback', async () => { + const payload = new Uint8Array([1]); + const order: string[] = []; + const preparer = vi.fn((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']); + + expect( + cache.getEventValue(event.eventId, payload, (prepared) => prepared) + ).toEqual(payload); }); it('caches real decrypt/decompress output but revives fresh objects', async () => { @@ -126,15 +182,19 @@ describe('ReplayPayloadCache', () => { const preparer = vi.fn(prepareReplayPayload); const cache = new ReplayPayloadCache(key, preparer); - const prepared = await cache.prepareEventPayload( + const directPreparation = prepareReplayPayload(serialized, key); + expect(directPreparation).not.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; @@ -149,89 +209,115 @@ 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('finds events inserted below a previously prepared prefix', () => { + // A stale-snapshot restart can replace the log with a corrected one whose + // missing events appear below the old tail. Full scans are cheap because + // event-id cache hits do no payload work. const payloads = [0, 1, 2].map((value) => new Uint8Array([value])); const preparer = vi.fn((value) => value); const cache = new ReplayPayloadCache(undefined, preparer); const run = makeRun(undefined); const [first, missing, second] = makeEvents(payloads); - 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]); - 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); }); - 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((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 () => { + 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('rehydrates mutable results and memoizes primitives of any size', async () => { const oversized = 'x'.repeat(4097); for (const value of [{ count: 0 }, oversized]) { - const cache = new ReplayPayloadCache(undefined); + 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 + ); + if (typeof value === 'object') { + expect(hydrate).toHaveBeenCalledTimes(2); + expect(second).not.toBe(first); + } else { + expect(hydrate).toHaveBeenCalledOnce(); + 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 084c6fd3c9..c5d91622b7 100644 --- a/packages/core/src/replay-payload-cache.ts +++ b/packages/core/src/replay-payload-cache.ts @@ -5,211 +5,149 @@ import { prepareReplayPayload, } from './serialization/replay.js'; -const MAX_MEMOIZED_PRIMITIVE_LENGTH = 4096; -type ReplayPayloadField = 'result' | 'error' | 'payload'; +const WORKFLOW_INPUT = Symbol('workflow-input'); +type ReplayPayloadKey = string | typeof WORKFLOW_INPUT; +type CachedPreparation = Uint8Array | Promise; -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. Most + * Node preparation is synchronous; only codecs that are inherently async + * leave a Promise in the cache. */ export class ReplayPayloadCache { - private readonly preparedPayloads = new Map< - string, - Promise + private readonly preparations = new Map< + ReplayPayloadKey, + CachedPreparation >(); - private readonly primitiveStepResults = new Map(); - private nextUnscannedEventIndex = 0; + private readonly primitiveValues = new Map(); constructor( - private readonly encryptionKey: DecryptionKey | undefined, + private readonly encryptionKey?: DecryptionKey, private readonly preparer: typeof prepareReplayPayload = 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)); - }; - - 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. - for ( - let index = this.nextUnscannedEventIndex; - 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; - } + /** 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); } - 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); } - /** - * 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. - */ - resetScan(): void { - this.nextUnscannedEventIndex = 0; + /** Prepare every payload not already seen through the event stream. */ + prepareAll(workflowRun: WorkflowRun, events: Event[]): void { + this.cachePayload(WORKFLOW_INPUT, workflowRun.input); + for (const event of events) this.prepareEvent(event); } - /** 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); + 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); + } + + private cachePrimitive(eventId: string, value: unknown): unknown { + if (isCacheablePrimitive(value)) { + this.primitiveValues.set(eventId, value); } return value; } - /** - * 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({ legacy: value }); + private cachePayload(cacheKey: ReplayPayloadKey, value: unknown): void { + if (!(value instanceof Uint8Array) || this.preparations.has(cacheKey)) { + return; } - const preparation = this.ensurePreparation(cacheKey, value); - void preparation.catch(() => { - if (this.preparedPayloads.get(cacheKey) === preparation) { - this.preparedPayloads.delete(cacheKey); - } - }); - return preparation; - } - - /** 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; + let preparation: CachedPreparation; + try { + preparation = this.preparer(value, this.encryptionKey); + } catch (error) { + // Preparation is speculative. Preserve a synchronous failure for the + // ordered consumer without failing event loading or creating an + // unhandled rejection. + preparation = Promise.reject(error); + } + this.preparations.set(cacheKey, preparation); + + if (preparation instanceof Promise) { + void preparation.then( + (prepared) => { + if (this.preparations.get(cacheKey) === preparation) { + this.preparations.set(cacheKey, prepared); + } + }, + () => {} + ); + } } - /** Normalize synchronous and asynchronous preparers to one promise contract. */ - private async runPreparation( - value: Uint8Array - ): Promise { - return this.preparer(value, this.encryptionKey); - } + private getPayload( + cacheKey: ReplayPayloadKey, + value: unknown + ): PreparedReplayPayload | Promise { + if (!(value instanceof Uint8Array)) return { legacy: value }; - private workflowInputKey(runId: string): string { - return `run:${runId}:input`; - } + this.cachePayload(cacheKey, value); + const prepared = this.preparations.get(cacheKey); + if (!prepared) { + throw new Error('Replay payload preparation was not cached'); + } - private eventPayloadKey(eventId: string, field: ReplayPayloadField): string { - return `event:${eventId}:${field}`; + if (!(prepared instanceof Promise)) return prepared; + return prepared.catch((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 0c9bc472d2..fe8e1e00e7 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -81,6 +81,7 @@ import { parseHealthCheckPayload, preconditionEventDelta, queueMessage, + resolveRunEncryptionKey, type SlotSnapshotParams, settleEventSlotGap, slotSnapshotParams, @@ -117,6 +118,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'; @@ -540,6 +542,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 @@ -1013,6 +1022,65 @@ 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 replayEncryptionKey: + | Promise + | undefined; + let streamedPayloadCache: ReplayPayloadCache | undefined; + const eventsWaitingForKey: Event[] = []; + const activatePayloadCache = ( + key: DecryptionKey | undefined + ): ReplayPayloadCache => { + if (!streamedPayloadCache) { + streamedPayloadCache = new ReplayPayloadCache(key); + for (const event of eventsWaitingForKey) { + streamedPayloadCache.prepareEvent(event); + } + eventsWaitingForKey.length = 0; + } + return streamedPayloadCache; + }; + const getReplayEncryptionKey = ( + runOrId: WorkflowRun | string, + context?: Record + ): Promise => { + if (!replayEncryptionKey) { + replayEncryptionKey = resolveRunEncryptionKey( + world, + runOrId, + context + ); + void replayEncryptionKey.then( + activatePayloadCache, + () => {} + ); + } + return replayEncryptionKey; + }; + const onReplayEvent = (event: Event): void => { + const deploymentId = replayEventDeploymentId(event); + if (deploymentId) { + void getReplayEncryptionKey(runId, { + deploymentId, + }); + } + if (streamedPayloadCache) { + streamedPayloadCache.prepareEvent(event); + } else { + eventsWaitingForKey.push(event); + } + }; + if (runInput?.deploymentId) { + void getReplayEncryptionKey(runId, { + deploymentId: runInput.deploymentId, + }); + } + // Shared state: set by either the background step path // or the run_started setup below. let workflowRun: WorkflowRun | undefined; @@ -1333,10 +1401,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( 'Event creation rejected as stale; restarting replay in-process', @@ -1634,7 +1698,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, @@ -1748,7 +1812,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 @@ -1983,6 +2047,7 @@ export function workflowEntrypoint( resumeId: hookResumeInput.resumeId, resumePayloadDigest: hookResumeInput.payloadDigest, preloadEvents: true, + onEvent: onReplayEvent, } ); hookEnsured = true; @@ -2257,6 +2322,7 @@ export function workflowEntrypoint( }); const result = await createEvent(runStartedEvent, { requestId, + onEvent: onReplayEvent, }); workflowRun = result.run; maxEventsLimit = clampMaxEvents(result.maxEvents); @@ -2541,31 +2607,20 @@ 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 = + await getReplayEncryptionKey(workflowRun); + const replayPayloadCache = + activatePayloadCache(encryptionKey); // The live VM parked at the previous boundary, when the // retention decision kept it. null → this iteration cold- @@ -2660,7 +2715,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' }; } @@ -2712,12 +2771,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' }; @@ -2834,10 +2895,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') @@ -2854,13 +2916,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', }; } @@ -2943,15 +3011,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' }; @@ -2971,8 +3037,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 @@ -3230,12 +3294,10 @@ 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. - replayPayloadCache.resetScan(); } // Open hooks/waits in the log as loaded for this 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..f774ed0c6b 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 }; @@ -1244,27 +1250,33 @@ export function getQueueOverhead(message: { requestedAt?: Date }) { * outer try/catch to log and surface the issue; the queue's redelivery * semantics will retry the key fetch on the next attempt. */ +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; +} + 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/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 300cb6dba4..18df1c7b77 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 1ae7a78204..a9aa11f3e9 100644 --- a/packages/core/src/test-support/orchestrator-context.ts +++ b/packages/core/src/test-support/orchestrator-context.ts @@ -34,7 +34,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 3bcd3f08fa..de9a248cf1 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -1085,7 +1085,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, {