diff --git a/.changeset/events-create-cursor-delta.md b/.changeset/events-create-cursor-delta.md new file mode 100644 index 0000000000..a443839007 --- /dev/null +++ b/.changeset/events-create-cursor-delta.md @@ -0,0 +1,9 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +'@workflow/world-local': patch +'@workflow/world-postgres': patch +'@workflow/world-vercel': patch +--- + +Fold new events returned by `events.create` into the replay log so a completed wait no longer needs a follow-up `events.list` round trip diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 322b4000d8..fdaa4ccb39 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -2000,6 +2000,36 @@ describe('workflowEntrypoint turbo mode', () => { expect((redeliverRunStarted?.[2] as any)?.skipPreload).toBeUndefined(); }); + it('never asks for an inline delta on a run-terminal write, or anywhere under turbo', async () => { + const turbo = await driveTurbo({ + runId: 'wrun_turbo_no_delta', + attempt: 1, + source: oneStepWorkflow, + }); + expect((await turbo.handlerPromise).status).toBe(204); + // Turbo exists to keep the first invocation's writes as cheap as + // possible and starts with no loaded log to extend, so nothing it writes + // asks the World to compute a delta. + expect( + turbo.eventsCreate.mock.calls.map((c) => (c[2] as any)?.sinceCursor) + ).toEqual(turbo.eventsCreate.mock.calls.map(() => undefined)); + + // A redelivery is not turbo and has a cursor by the time the run + // finishes, but nothing reads the log after a run-terminal write, so the + // delta would be work the World does for no one. + const redeliver = await driveTurbo({ + runId: 'wrun_turbo_no_delta_redeliver', + attempt: 2, + source: oneStepWorkflow, + }); + expect((await redeliver.handlerPromise).status).toBe(204); + const runCompleted = redeliver.eventsCreate.mock.calls.find( + (c) => (c[1] as any).eventType === 'run_completed' + ); + expect(runCompleted).toBeDefined(); + expect((runCompleted?.[2] as any)?.sinceCursor).toBeUndefined(); + }); + it('exits turbo (no forced optimistic) when the suspension creates a wait', async () => { const { handlerPromise, order } = await driveTurbo({ runId: 'wrun_turbo_wait', diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 885081b048..b829b4257a 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -18,9 +18,13 @@ import { workflowDisplayName, } from '@workflow/utils/parse-name'; import { + type CreateEventParams, + type CreateEventRequest, type Event, + type EventResult, getQueueTopicPrefix, isLegacySpecVersion, + isTerminalRunEventType, ROOT_RUN_ID_ATTRIBUTE, type RunInput, resolveQueueNamespace, @@ -799,10 +803,95 @@ export function workflowEntrypoint( const replayRecoveryReporter = replayDivergence ? new ReplayRecoveryReporter(replayDivergence.count) : ReplayRecoveryReporter.inert(); - const createEvent: EventCreator = (data, params) => - replayRecoveryReporter.withEventCreate(params, (p) => - world.events.create(runId, data, p) + // Every write this loop makes carries the cursor of the log + // it was computed against, and folds whatever the World + // hands back into that log. See `absorbCreateDelta` for the + // guards; `sinceCursor` in @workflow/world for the contract. + const createEvent: EventCreator = async (data, params) => { + const sinceCursor = deltaRequestCursor(data, params); + const result = await replayRecoveryReporter.withEventCreate( + sinceCursor === undefined + ? params + : { ...params, sinceCursor }, + (p) => world.events.create(runId, data, p) ); + if (sinceCursor !== undefined) { + absorbCreateDelta(sinceCursor, result); + } + return result; + }; + + /** + * The cursor to ask for an inline delta against, or + * undefined to not ask. + * + * Turbo is excluded on purpose: it exists to make the first + * invocation's writes as cheap as possible, and it has no + * loaded log to extend. Run-terminal writes are excluded + * because nothing reads the log afterwards, so the delta + * would be work the World does for no one. A caller that + * set its own `sinceCursor` (or asked for the `run_started` + * / `hook_received` preload, which owns the same response + * fields) keeps what it asked for. + */ + const deltaRequestCursor = ( + data: CreateEventRequest, + params: CreateEventParams | undefined + ): string | undefined => { + const skip = + turbo || + eventsCursor === null || + params?.sinceCursor !== undefined || + params?.preloadEvents === true || + isTerminalRunEventType(data.eventType); + return skip ? undefined : (eventsCursor ?? undefined); + }; + + /** + * Fold an inline delta returned by `events.create` into the + * in-memory event log. + * + * The delta is what `events.list({ cursor: sentCursor })` + * would have returned right after the write, so it carries + * back both our own event and anything another writer + * appended in band. Absorbing it here means the next reader + * already has the log and the loop's incremental + * `events.list` finds nothing left to fetch. + * + * Declining is always safe — an unabsorbed delta is a delta + * the next `events.list` returns — so the guards are free to + * be strict: + * + * - `hasMore` means the World truncated the page. Taking it + * while advancing the cursor to the page end would be + * fine, but taking it and NOT advancing would duplicate, + * and the truncated case is rare enough not to special-case. + * - The cursor must be exactly where it was when the request + * went out. Concurrent writes each diff against the cursor + * they saw, and `appendUniqueEvents` deliberately does not + * re-sort, so absorbing a second delta computed from an + * older cursor could append events behind ones already + * taken. First response back wins; the rest are dropped. + * - A pending inline delta is an unconsumed cursor move of + * its own. Rather than reason about which of the two is + * further ahead, leave the log alone until it is consumed. + */ + const absorbCreateDelta = ( + sentCursor: string, + result: EventResult + ): void => { + if ( + cachedEvents === null || + pendingInlineDelta !== null || + eventsCursor !== sentCursor || + result.events === undefined || + result.hasMore === true + ) { + return; + } + appendUniqueEvents(cachedEvents, result.events); + eventsCursor = result.cursor ?? eventsCursor; + }; // Event cache: keep loaded events in memory across loop iterations. // On the first iteration we do a full load; on subsequent iterations @@ -2537,15 +2626,34 @@ export function workflowEntrypoint( } } - if (waitsToComplete.length > 0) { - // The event list above may be stale by the time an - // elapsed wait is committed. Load only events after - // the original snapshot cursor so concurrent durable - // events, such as hook_received, keep their ordering - // relative to wait_completed. Fall back to a full - // reload for older worlds that cannot give us a stable - // cursor, or if the cursor delta does not include the - // wait completion this handler just attempted. + // The event list above may be stale by the time an + // elapsed wait is committed, and this replay has to see + // its own wait_completed before it can advance past the + // sleep. Each write asked the World for the delta since + // our cursor and folded it in (see createEvent), so a + // supporting World has already handed those events back + // and there is nothing left to do. Only fetch for the + // completions still missing locally: a World that + // ignores `sinceCursor`, a truncated delta, a concurrent + // absorb that lost the cursor race, or an + // EntityConflictError, whose rejection carries no delta. + const missingWaitCompletions = waitsToComplete.filter( + (waitEvent) => + !events.some( + (e) => + e.eventType === 'wait_completed' && + e.correlationId === waitEvent.correlationId + ) + ); + + if (missingWaitCompletions.length > 0) { + // Load only events after the original snapshot cursor + // so concurrent durable events, such as hook_received, + // keep their ordering relative to wait_completed. Fall + // back to a full reload for older worlds that cannot + // give us a stable cursor, or if the cursor delta does + // not include the wait completion this handler just + // attempted. if (eventsCursor) { const loaded = await loadWorkflowRunEvents( runId, @@ -2556,12 +2664,12 @@ export function workflowEntrypoint( .filter((e) => e.eventType === 'wait_completed') .map((e) => e.correlationId) ); - const sawAllWaitCompletions = waitsToComplete.every( - (waitEvent) => + const sawAllWaitCompletions = + missingWaitCompletions.every((waitEvent) => completedWaitIdsAfterCursor.has( waitEvent.correlationId ) - ); + ); if (sawAllWaitCompletions) { appendUniqueEvents(events, loaded.events); diff --git a/packages/core/src/runtime/wait-completion-replay.test.ts b/packages/core/src/runtime/wait-completion-replay.test.ts index 27ca033d37..e772761ff5 100644 --- a/packages/core/src/runtime/wait-completion-replay.test.ts +++ b/packages/core/src/runtime/wait-completion-replay.test.ts @@ -72,6 +72,14 @@ async function runStaleWaitReplayScenario(options: { preloadedHasMore?: boolean; omitWaitCompletionFromDelta?: boolean; terminalFailureAfterWaitCompletion?: boolean; + /** + * Model a World that honors `sinceCursor` on `events.create`: the write + * response carries the same page `events.list` would have returned, so the + * handler should absorb it and skip the follow-up fetch entirely. + */ + returnInlineDelta?: boolean; + /** Truncate that inline delta (hasMore: true), which must not be absorbed. */ + inlineDeltaHasMore?: boolean; }) { vi.spyOn(Date, 'now').mockReturnValue(+fixedNow); @@ -206,6 +214,22 @@ async function runStaleWaitReplayScenario(options: { ) => Promise) | undefined; + /** + * `events.list` semantics for this fake log: everything strictly after + * `cursor`. `staleEventsCursor` is the opaque cursor the run_started + * preload hands out; every other cursor is an event id. + */ + const eventsAfterCursor = (cursor?: string): Event[] => { + if (!cursor) { + return [...durableEvents]; + } + if (cursor === staleEventsCursor) { + return durableEvents.slice(staleEvents.length); + } + const index = durableEvents.findIndex((e) => e.eventId === cursor); + return index === -1 ? [...durableEvents] : durableEvents.slice(index + 1); + }; + const listEvents = vi.fn( async (params: { runId: string; @@ -213,10 +237,7 @@ async function runStaleWaitReplayScenario(options: { }) => { // Cursor reads simulate the optimized delta fetch. Without a cursor, the // runtime has fallen back to a full reload from the beginning. - let data = - params.pagination?.cursor === staleEventsCursor - ? durableEvents.slice(staleEvents.length) - : [...durableEvents]; + let data = eventsAfterCursor(params.pagination?.cursor); if ( params.pagination?.cursor === staleEventsCursor && options.omitWaitCompletionFromDelta @@ -251,7 +272,11 @@ async function runStaleWaitReplayScenario(options: { }; const createEvent = vi.fn( - async (_runId: string, request: CreateEventRequest) => { + async ( + _runId: string, + request: CreateEventRequest, + params?: { sinceCursor?: string } + ) => { if (request.eventType === 'run_started') { return runStartedResponse; } @@ -298,6 +323,20 @@ async function runStaleWaitReplayScenario(options: { const created = event(effectiveRequest); durableEvents.push(created); createdEvents.push(created); + // A World that honors sinceCursor answers the write with the delta the + // caller would otherwise fetch. Computed after the write is durable, so + // it includes the event just created. + const inlineDelta = (() => { + if (!(options.returnInlineDelta && params?.sinceCursor)) { + return {}; + } + const data = eventsAfterCursor(params.sinceCursor); + return { + events: data, + cursor: data.at(-1)?.eventId ?? null, + hasMore: options.inlineDeltaHasMore ?? false, + }; + })(); if (effectiveRequest.eventType === 'step_started') { return { event: created, @@ -307,6 +346,7 @@ async function runStaleWaitReplayScenario(options: { effectiveRequest.correlationId ), ...(lazyStepStart ? { stepCreated: true } : {}), + ...inlineDelta, }; } if ( @@ -323,7 +363,7 @@ async function runStaleWaitReplayScenario(options: { }) ); } - return { event: created }; + return { event: created, ...inlineDelta }; } ); @@ -397,6 +437,7 @@ async function runStaleWaitReplayScenario(options: { ); return { + createEvent, createdEvents, listEvents, listedPages, @@ -580,6 +621,58 @@ describe('workflow handler wait completion replay', () => { expectHookBranchQueued(result); }); + it('skips the follow-up fetch when the wait_completed write returns the delta', async () => { + // The write carries the cursor the handler's snapshot was taken at, so a + // supporting World answers it with exactly the page the follow-up + // events.list would have returned. Absorbing that page leaves nothing to + // fetch: the hook_received that raced the completion is already in the + // local log. + const result = await runStaleWaitReplayScenario({ + includePreloadedCursor: true, + returnInlineDelta: true, + }); + + const waitWrite = result.createEvent.mock.calls.find( + (call) => (call[1] as CreateEventRequest).eventType === 'wait_completed' + ); + expect(waitWrite?.[2]).toEqual( + expect.objectContaining({ sinceCursor: result.staleEventsCursor }) + ); + + // Only the next loop iteration's incremental fetch remains, and it reads + // from the cursor the absorbed delta advanced to — never from the + // pre-write cursor, which is what the deleted follow-up fetch used. + expect(result.listEvents).toHaveBeenCalledTimes(1); + expect(result.listEvents.mock.calls[0]?.[0].pagination?.cursor).not.toBe( + result.staleEventsCursor + ); + expectHookBranchQueued(result); + }); + + it('falls back to the follow-up fetch when the returned delta is truncated', async () => { + // hasMore means the page is not the whole delta. Absorbing it would leave + // a hole between the events taken and the cursor reported, so the handler + // must decline it and fetch as before. + const result = await runStaleWaitReplayScenario({ + includePreloadedCursor: true, + returnInlineDelta: true, + inlineDeltaHasMore: true, + }); + + expect(result.listEvents).toHaveBeenCalledTimes(2); + expect(result.listEvents.mock.calls[0]?.[0].pagination).toEqual( + expect.objectContaining({ + sortOrder: 'asc', + cursor: result.staleEventsCursor, + }) + ); + expect(result.listedPages[0]?.map((event) => event.eventType)).toEqual([ + 'hook_received', + 'wait_completed', + ]); + expectHookBranchQueued(result); + }); + it('stops after wait refresh when the event log contains a terminal run event', async () => { const result = await runStaleWaitReplayScenario({ includePreloadedCursor: true, diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index ff24fad4de..7f35f707ff 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -1477,7 +1477,7 @@ describe('Storage', () => { expect(result.cursor).toBeUndefined(); }); - it('does not return a delta for non-terminal step events', async () => { + it('returns a delta for non-terminal event types too', async () => { await updateRun(storage, testRunId, 'run_started'); const sinceCursor = await currentCursor(); await createStep(storage, testRunId, { @@ -1485,8 +1485,9 @@ describe('Storage', () => { stepName: 'seq-step', input: new Uint8Array(), }); - // step_started carries sinceCursor but is not a loop boundary, so the - // World should not compute a delta for it. + // Whether the delta is worth asking for is the caller's decision, not + // the World's: outside turbo the runtime sends `sinceCursor` on every + // write so each response carries its log forward. const result = await storage.events.create( testRunId, { @@ -1496,7 +1497,46 @@ describe('Storage', () => { }, { sinceCursor } ); - expect(result.events).toBeUndefined(); + const expected = await storage.events.list({ + runId: testRunId, + pagination: { sortOrder: 'asc', cursor: sinceCursor }, + }); + expect(result.events?.map((e) => e.eventId)).toEqual( + expected.data.map((e) => e.eventId) + ); + expect(result.events?.at(-1)?.eventType).toBe('step_started'); + expect(result.cursor).toBe(expected.cursor); + expect(result.hasMore).toBe(expected.hasMore); + }); + + it('returns a delta for a wait_completed write', async () => { + await updateRun(storage, testRunId, 'run_started'); + await storage.events.create(testRunId, { + eventType: 'wait_created' as const, + correlationId: 'corr_wait1', + eventData: { resumeAt: new Date(Date.now() - 1000) }, + }); + const sinceCursor = await currentCursor(); + // The runtime's elapsed-wait pass relies on this delta instead of a + // follow-up events.list, so the response must carry the completion it + // just wrote. + const result = await storage.events.create( + testRunId, + { + eventType: 'wait_completed' as const, + correlationId: 'corr_wait1', + eventData: { resumeAt: new Date(Date.now() - 1000) }, + }, + { sinceCursor } + ); + expect( + result.events?.some( + (e) => + e.eventType === 'wait_completed' && + e.correlationId === 'corr_wait1' + ) + ).toBe(true); + expect(result.hasMore).toBe(false); }); }); diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index 3873677420..77b73a03d1 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -31,7 +31,6 @@ import { isLegacySpecVersion, isStepEventType, isTerminalRunEventType, - isTerminalStepEventType, isTerminalStepStatus, isTerminalWorkflowRunStatus, requiresNewerWorld, @@ -2415,30 +2414,26 @@ export function createEventsStorage( hasMore = allEvents.hasMore; } - // Inline-delta optimization: on a step-terminal write the inline - // runtime loop can pass `sinceCursor` (the cursor from before it - // began writing this step's events). We return the delta of + // Inline-delta optimization: a writer can pass `sinceCursor` (the + // cursor of the event log as it last saw it). We return the delta of // events written strictly after that cursor — exactly what an // `events.list({ cursor: sinceCursor, sortOrder: 'asc' })` would - // return right now — so the loop can skip a redundant round-trip. + // return right now — so the caller can skip a redundant round-trip. // // This is computed against the same on-disk log the list path - // reads, so it captures everything the fetch would: this step's - // step_created/step_started/step_completed, any attr_set the step - // body wrote, and any in-band events (e.g. hook_received, - // wait_completed) another writer appended since the cursor. That - // equivalence is what makes skipping the fetch safe — a missed - // in-band event cannot diverge replay because the delta is the - // fetch. + // reads, so it captures everything the fetch would: the event just + // written, any attr_set a step body wrote, and any in-band events + // (e.g. hook_received, wait_completed) another writer appended since + // the cursor. That equivalence is what makes skipping the fetch safe + // — a missed in-band event cannot diverge replay because the delta + // is the fetch. // - // Only step-terminal events qualify: step_created/step_started are - // not loop boundaries (the loop fetches after step_completed / - // step_failed), and run-terminal events end the loop. `resolveData` - // matches the list path so eventData refs are handled identically. - if ( - isTerminalStepEventType(data.eventType) && - typeof params?.sinceCursor === 'string' - ) { + // Any event type qualifies. The write itself decides nothing here; + // whether the delta is worth requesting is the caller's call, and + // the runtime asks on every non-turbo write so each response carries + // the log forward. `resolveData` matches the list path so eventData + // refs are handled identically. + if (typeof params?.sinceCursor === 'string') { // Intentionally no `limit`: this returns a single default-size page, // unlike the `events.list` path which loops `while (hasMore)` to // exhaustion. That is safe — and must NOT be "fixed" by paginating diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index 7c97084b59..caeb325dbd 100644 --- a/packages/world-postgres/src/storage.ts +++ b/packages/world-postgres/src/storage.ts @@ -1879,6 +1879,35 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] { hasMore = false; } + // Inline delta: the caller told us the cursor of the log it holds, so + // return the page `events.list({ cursor: sinceCursor, sortOrder: 'asc' })` + // would return right now and save it the round-trip. Same query, same + // page size, same cursor semantics as `list` below — deliberately not + // paginated to exhaustion, since the contract is + // single-page-or-fall-back and the caller ignores a delta with + // `hasMore: true`. + if (typeof params?.sinceCursor === 'string') { + const limit = 100; + const deltaRows = await drizzle + .select() + .from(Schema.events) + .where( + and( + eq(Schema.events.runId, effectiveRunId), + gt(Schema.events.eventId, params.sinceCursor) + ) + ) + .orderBy(Schema.events.eventId) + .limit(limit + 1); + const page = deltaRows.slice(0, limit); + allEvents = page.map((e) => { + e.eventData ||= e.eventDataJson; + return stripEventDataRefs(EventSchema.parse(compact(e)), resolveData); + }); + cursor = allEvents.at(-1)?.eventId ?? null; + hasMore = deltaRows.length > limit; + } + return { event: stripEventDataRefs(parsed, resolveData), run, diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 1942517ee7..9222509c7c 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -730,10 +730,11 @@ async function createWorkflowRunEventInner( occurredAt: params?.occurredAt ?? new Date(), // Opt-in inline-delta: forward the cursor the runtime held before // this write so the server can return the authoritative event-log - // delta on the response (events/cursor/hasMore), letting the inline - // loop skip a follow-up events.list. The server only acts on it for - // step_completed/step_failed; older servers ignore it and the runtime - // falls back to events.list. + // delta on the response (events/cursor/hasMore), letting the caller + // skip a follow-up events.list. Outside turbo the runtime sends this on + // every write, but a server may act on only some event types (or none); + // a response without a delta just means the runtime keeps its cursor + // and fetches when it next needs to. ...(params?.sinceCursor ? { sinceCursor: params.sinceCursor } : {}), // Run-started preload opt-out: turbo backgrounds run_started as a write // barrier only and never reads the preloaded log, so tell the server to diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 6a092d863b..fa1c16e7cb 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -829,11 +829,15 @@ export interface CreateEventParams { * on the resulting {@link EventResult}, the first page of events written * strictly after this cursor (via `events`/`cursor`/`hasMore`) — the * same page an `events.list({ cursor: sinceCursor, sortOrder: 'asc' })` - * call would return immediately after this write. The inline runtime - * loop uses this to skip a redundant `events.list` round-trip between - * sequential steps: instead of re-reading its own just-written events - * (and any events interleaved in-band, such as `hook_received`), it - * consumes the authoritative delta the write already had to compute. + * call would return immediately after this write. Outside turbo mode the + * runtime sets this on every write it makes from the orchestrator loop + * and folds any returned delta into its in-memory log, so each write + * carries the log forward and the loop reads it back for free: instead of + * re-reading its own just-written events (and any events interleaved + * in-band, such as `hook_received`), it consumes the authoritative delta + * the write already had to compute. Turbo mode does not set it — the + * point there is to keep the first invocation's writes as cheap as + * possible, and it has no loaded log to extend. * * The cursor MUST share `events.list` semantics: the returned `events` * are everything sorted strictly after `sinceCursor`, `cursor` is the