diff --git a/.changeset/shared-hook-history.md b/.changeset/shared-hook-history.md new file mode 100644 index 0000000000..e6dec76422 --- /dev/null +++ b/.changeset/shared-hook-history.md @@ -0,0 +1,7 @@ +--- +"@workflow/core": patch +"@workflow/world": patch +"@workflow/world-vercel": patch +--- + +Avoid redundant history reads when independent hooks register concurrently. Keep existing single-event requests parallel and reconcile their committed events together before continuing, with the normal history reload when the responses do not cover a complete consecutive suffix. diff --git a/docs/content/docs/v5/how-it-works/event-sourcing.mdx b/docs/content/docs/v5/how-it-works/event-sourcing.mdx index 00c35fe031..9121e29fd7 100644 --- a/docs/content/docs/v5/how-it-works/event-sourcing.mdx +++ b/docs/content/docs/v5/how-it-works/event-sourcing.mdx @@ -287,7 +287,7 @@ Both save a reload, so the rule for which writes send them is not about the even | `run_created` | `start()` | Nothing | No log exists yet. | | `run_started` | Replay, first write of a delivery | `eventCount` when a log is loaded; no cursor | The `run_started` response already carries the full log as a preload. | | `step_created`, `wait_created`, `hook_disposed`, `attr_set` | Suspension handler | `eventCount` | The handler keeps writing from this log and the replay resumes from it. | -| `hook_created` | Suspension handler | `eventCount` and `sinceCursor` (only when the suspension creates exactly one hook) | The delta is how a hook whose payload already arrived resolves without another invocation. Two creates diffing against one cursor would give two deltas of which only the first could be taken. | +| `hook_created` | Suspension handler | `eventCount` and `sinceCursor` for a lone hook; independent concurrent Node.js registrations in a hooks-only suspension defer both | A lone create returns a delta. Concurrent registrations reconcile their committed events together and reload when the responses do not cover a consecutive history suffix. | | `hook_received` | Replay resuming a hook | `eventCount` | The replay continues from the log right after. | | `hook_received` | Webhook or `resumeHook()` | Nothing | Out-of-band. It enqueues a delivery that loads the log. | | `wait_completed` | Replay completing an elapsed wait | `eventCount` | The replay continues from the log right after. | diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index d902fc32c1..e8f009a6b9 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -235,6 +235,16 @@ The version a World stamps comes from `mintedSpecVersion()`: 8 by default or the Not every write carries `eventCount`, and a World is free to skip the report read for event types whose writer never holds a log to merge it into (the step executor's `step_started`, `step_retrying`, `step_completed`, and `step_failed`, and the run-terminal `run_completed`, `run_failed`, and `run_cancelled`). Which writes send `eventCount` or `sinceCursor`, and why, is tabulated in [Events returned on a write](/docs/how-it-works/event-sourcing#events-returned-on-a-write). When a write carries both, the `sinceCursor` delta is a superset of the report, so return the delta and skip the report. +Independent concurrent hook registrations in a hooks-only suspension deliberately omit `eventCount` and `sinceCursor`, even though the worker holds a loaded log. The Node.js runtime waits for every registration to settle and verifies that their returned events fill every position immediately after that log. A gap, incomplete response, or missing committed-event cursor keeps the normal history reload. Mixed suspensions, forced claims, and same-token lifecycle operations keep their usual catch-up parameters. + +### Optional: returning a committed-event cursor + +An `events.create()` result may include `eventCursor`: a World-owned cursor that its own ascending `events.list()` accepts as an exclusive position immediately after the returned `event`. It is distinct from a history-page `cursor`, and does not assert that the caller has loaded earlier events. + +Returning `eventCursor` opts into using the returned `event` as replay input. That event **must equal the event returned by `events.list()` for the same position and `resolveData` option**, including its stored `createdAt`, normalized `tokenRetentionUntil`, resolved metadata bytes, and any `hook_conflict` details such as `conflictingRunId` or `forceRefusedReason`. Return the persisted event rather than reconstructing it from the creation request. + +The runtime uses the cursor only after proving that the committed responses extend its loaded prefix without a hole. This is optional: a World that omits `eventCursor` remains supported and retains the reload path. The runtime never interprets a World's cursor encoding. + ### Optional: rejecting a stale write A replay writes events derived from the event log it loaded, so a write made from an event log that no longer matches the store can commit events no correct replay would produce. Rejecting one means answering `events.create()` with a `PreconditionFailedError` when the run's log already holds more events than the caller's `eventCount` says it had loaded. @@ -246,7 +256,7 @@ Implement rejection when your store allocates positions outside the commit and c Two rules make this safe: - **Only reject on evidence.** If your record of a run's events is incomplete, expired, or cannot answer the question, accept the creation. A rejection must always indicate a real discrepancy because the runtime responds by discarding a replay. -- **Accept a creation that carries no `eventCount`.** It came from a caller with no loaded log that could be stale, such as a queued step body or out-of-band writer, rather than a caller claiming the log is empty. +- **Accept a creation that carries no `eventCount`.** The caller is not claiming the log is empty. It may have no loaded log, as with a queued step body or out-of-band writer, or may deliberately defer catch-up until concurrent hook registrations settle. Without a count there is no evidence for count-based rejection; independent ownership checks and write fences still apply. A rejection may include the events the caller was missing as `{ events, cursor }` on the error's `details`. Include them only when you can prove that the set fully accounts for the discrepancy, isn't truncated, and contains only events from the run being written. The runtime merges them directly into the replay's event log, so incorrect data is worse than no delta. Otherwise, omit them, and the runtime performs a full reload. diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index c631f4c306..af1147573e 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -3915,6 +3915,123 @@ describe('workflowEntrypoint hook conflict beside a pre-claimed inline step', () }); }); +describe('workflowEntrypoint concurrent hook continuation', () => { + afterEach(() => { + setWorld(undefined); + vi.clearAllMocks(); + vi.unstubAllEnvs(); + }); + + it.each([ + false, + true, + ])('continues after concurrent registrations, interleaved writer=%s', async (interleaved) => { + vi.stubEnv('WORKFLOW_RETAINED_VM', '1'); + const runId = `wrun_hook_reconciliation_${interleaved}`; + const workflowRun = { + runId, + workflowName: 'workflow', + status: 'running', + specVersion: 6, + deploymentId: 'test-deployment', + input: await dehydrateWorkflowArguments([], runId, undefined, []), + createdAt: new Date(), + updatedAt: new Date(), + startedAt: new Date(), + } satisfies WorkflowRun; + const durableEvents: Event[] = []; + const record = (data: any): Event => { + const event = { + ...data, + runId, + eventId: slotToEventId(durableEvents.length + 1), + createdAt: new Date(), + } as Event; + durableEvents.push(event); + return event; + }; + record({ + eventType: 'run_created', + specVersion: 6, + eventData: { + workflowName: 'workflow', + deploymentId: 'test-deployment', + input: workflowRun.input, + }, + }); + let hooks = 0; + const create = vi.fn(async (_runId: string, data: any, _params?: any) => { + if (data.eventType === 'hook_created' && hooks++ === 1 && interleaved) { + // A concurrent delivery started after this invocation read its prefix. + record({ eventType: 'run_started', specVersion: 6 }); + } + const event = record(data); + if (data.eventType === 'run_started') + return { + event, + run: workflowRun, + events: [...durableEvents], + cursor: `eid:${event.eventId}`, + hasMore: false, + }; + if (data.eventType === 'run_completed') + workflowRun.status = 'completed' as any; + return { event, eventCursor: `eid:${event.eventId}` }; + }); + const list = vi.fn(async ({ pagination }: any) => ({ + data: durableEvents.filter( + (event) => + !pagination?.cursor || event.eventId > pagination.cursor.slice(4) + ), + cursor: `eid:${durableEvents.at(-1)!.eventId}`, + hasMore: false, + })); + const queue = vi.fn(async () => ({ messageId: null })); + let handler!: (message: unknown, metadata: unknown) => Promise; + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + createQueueHandler: vi.fn((_prefix: string, callback: typeof handler) => { + handler = callback; + return async () => new Response(null, { status: 204 }); + }), + events: { create, list }, + runs: { get: vi.fn(async () => workflowRun) }, + queue, + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as any); + const entry = workflowEntrypoint(` + const createHook = globalThis[Symbol.for('WORKFLOW_CREATE_HOOK')]; + async function workflow() { + const first = createHook({ token: 'customer:first' }); + const second = createHook({ token: 'customer:second' }); + const owners = await Promise.all([first.getConflict(), second.getConflict()]); + return owners.every(owner => owner === null); + } + globalThis.__private_workflows = new Map([['workflow', workflow]]);`); + await entry(new Request('https://example.test')); + await handler( + { runId, requestedAt: new Date() }, + { + requestId: 'request', + attempt: 1, + queueName: '__wkf_workflow_workflow', + messageId: 'msg_hooks', + } + ); + expect(workflowRun.status).toBe('completed'); + expect(queue).not.toHaveBeenCalled(); + expect(list).toHaveBeenCalledTimes(interleaved ? 1 : 0); + const registrations = create.mock.calls.filter( + (call) => call[1].eventType === 'hook_created' + ); + expect(registrations).toHaveLength(2); + for (const call of registrations) { + expect(call[2]?.eventCount).toBeUndefined(); + expect(call[2]?.sinceCursor).toBeUndefined(); + } + }); +}); + describe('workflowEntrypoint max-deliveries terminal write', () => { afterEach(() => { setWorld(undefined); diff --git a/packages/core/src/runtime/concurrent-hook-registration.test.ts b/packages/core/src/runtime/concurrent-hook-registration.test.ts new file mode 100644 index 0000000000..95bb6d4022 --- /dev/null +++ b/packages/core/src/runtime/concurrent-hook-registration.test.ts @@ -0,0 +1,476 @@ +import type { Span } from '@opentelemetry/api'; +import { + EntityConflictError, + PreconditionFailedError, + WorkflowWorldError, +} from '@workflow/errors'; +import { + type CreateEventParams, + type CreateEventRequest, + type Event, + type EventResult, + slotToEventId, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { describe, expect, it, vi } from 'vitest'; +import { type QueueItem, WorkflowSuspension } from '../global.js'; +import { hydrateStepArguments } from '../serialization.js'; +import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; +import { handleSuspension } from './suspension-handler.js'; + +vi.mock('../version.js', () => ({ version: '0.0.0-test' })); + +const run = { + runId: 'wrun_concurrent_hooks', + workflowName: 'test-workflow', + status: 'running', + specVersion: 6, + input: [], + createdAt: new Date(), + updatedAt: new Date(), + startedAt: new Date(), + deploymentId: 'test-deployment', +} satisfies WorkflowRun; + +function setup(count = 2) { + const seed = [1, 2, 3].map( + (slot) => + ({ + eventId: slotToEventId(slot), + runId: run.runId, + eventType: 'run_started', + createdAt: new Date(), + }) as Event + ); + const eventLog = { events: [...seed], cursor: `eid:${seed[2].eventId}` }; + const pending = new Map( + Array.from({ length: count }, (_, i) => [ + `hook_${i}`, + { + type: 'hook', + correlationId: `hook_${i}`, + token: `customer:${i}`, + hasConflictAwaiter: true, + }, + ]) + ); + let slot = 3; + const answer = (data: CreateEventRequest): EventResult => { + const event = { + ...data, + eventId: slotToEventId(++slot), + runId: run.runId, + createdAt: new Date(), + } as Event; + return { event, eventCursor: `eid:${event.eventId}` }; + }; + const create = vi.fn( + async ( + _runId: string, + data: CreateEventRequest, + _params?: CreateEventParams + ) => answer(data) + ); + const world = { + events: { create, list: vi.fn() }, + getEncryptionKeyForRun: vi.fn(async () => undefined), + } as unknown as World; + const span = { setAttributes: vi.fn() }; + const invoke = ( + options: Partial[0]> = {} + ) => + handleSuspension({ + world, + run, + eventLog, + span: span as unknown as Span, + requestId: 'request', + suspension: new WorkflowSuspension(pending, globalThis), + ...options, + }); + return { create, answer, world, span, seed, eventLog, pending, invoke }; +} + +describe('concurrent hook history reconciliation', () => { + it.each([ + 6, 7, + ])('uses existing parallel writes and no catch-up hints on spec %s', async (specVersion) => { + const t = setup(); + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + let active = 0; + t.create.mockImplementation(async (_id, data) => { + const result = t.answer(data); + if (++active === 2) started.resolve(); + await release.promise; + return result; + }); + const invocation = t.invoke({ run: { ...run, specVersion } }); + await started.promise; + expect(t.create).toHaveBeenCalledTimes(2); + expect(t.eventLog.events).toEqual(t.seed); + release.resolve(); + const result = await invocation; + + for (const call of t.create.mock.calls) { + expect(call[2]).toEqual({ requestId: 'request', resolveData: 'all' }); + } + expect(t.world.events.list).not.toHaveBeenCalled(); + expect(t.eventLog.events.map((event) => event.eventId)).toEqual( + [1, 2, 3, 4, 5].map(slotToEventId) + ); + expect(t.eventLog.cursor).toBe(`eid:${slotToEventId(5)}`); + expect(result.eventLogCarriedForward).toBe(true); + expect(t.span.setAttributes).toHaveBeenCalledWith({ + 'workflow.concurrent_hooks.count': 2, + 'workflow.concurrent_hooks.carried_forward': true, + 'workflow.concurrent_hooks.fallback_reason': 'none', + }); + expect(result.awaitedHookCorrelationIds.sort()).toEqual([ + 'hook_0', + 'hook_1', + ]); + }); + + it('orders committed responses by history position, not response arrival', async () => { + const t = setup(); + const secondReturned = Promise.withResolvers(); + t.create.mockImplementation(async (_id, data) => { + const result = t.answer(data); + if (data.correlationId === 'hook_0') await secondReturned.promise; + else secondReturned.resolve(); + return result; + }); + expect((await t.invoke()).eventLogCarriedForward).toBe(true); + expect( + t.eventLog.events.slice(3).map((event) => event.correlationId) + ).toEqual(['hook_0', 'hook_1']); + }); + + it('keeps tokens, metadata, retention, webhooks and internal flags generic', async () => { + const t = setup(3); + const retention = new Date(Date.now() + 60_000); + Object.assign(t.pending.get('hook_0')!, { + token: '', + metadata: { order: 123 }, + tokenRetentionUntil: retention, + }); + Object.assign(t.pending.get('hook_1')!, { + isWebhook: true, + metadata: ['webhook'], + }); + Object.assign(t.pending.get('hook_2')!, { + isSystem: true, + token: 'internal:custom', + }); + expect((await t.invoke()).eventLogCarriedForward).toBe(true); + const events = ['hook_0', 'hook_1', 'hook_2'].map( + (id) => t.eventLog.events.find((event) => event.correlationId === id)! + ); + expect(events[0].eventData).toMatchObject({ + token: '', + tokenRetentionUntil: retention, + }); + expect( + await hydrateStepArguments( + (events[0] as any).eventData.metadata, + run.runId, + undefined, + [] + ) + ).toEqual({ order: 123 }); + expect(events[1].eventData).toMatchObject({ isWebhook: true }); + expect( + await hydrateStepArguments( + (events[1] as any).eventData.metadata, + run.runId, + undefined, + [] + ) + ).toEqual(['webhook']); + expect(events[2].eventData).toMatchObject({ + token: 'internal:custom', + isSystem: true, + }); + }); + + it('preserves a committed token conflict as part of the complete suffix', async () => { + const t = setup(); + t.create.mockImplementation(async (_id, data) => { + const result = t.answer(data); + if (data.correlationId === 'hook_1') { + result.event = { + ...result.event!, + eventType: 'hook_conflict', + eventData: { token: 'customer:1', conflictingRunId: 'wrun_owner' }, + } as Event; + } + return result; + }); + const result = await t.invoke(); + expect(result.eventLogCarriedForward).toBe(true); + expect(result.hookConflictCorrelationIds).toEqual(['hook_1']); + expect(t.eventLog.events.at(-1)?.eventType).toBe('hook_conflict'); + }); + + it.each([ + ['gap before', 'nonconsecutive_slots'], + ['gap between', 'nonconsecutive_slots'], + ['missing event', 'missing_event'], + ['missing cursor', 'missing_event_cursor'], + ['duplicate slot', 'nonconsecutive_slots'], + ['old event', 'nonconsecutive_slots'], + ['wrong run', 'event_mismatch'], + ['wrong correlation', 'event_mismatch'], + ['wrong token', 'event_mismatch'], + ['wrong type', 'event_mismatch'], + ['non-slot ID', 'non_slot_event'], + ['truncated page', 'incomplete_response'], + ['lazy metadata', 'unresolved_metadata'], + ['omitted metadata', 'unresolved_metadata'], + ])('retains the reload path without changing the log: %s', async (mode, reason) => { + const t = setup(); + if (mode.includes('metadata')) + Object.assign(t.pending.get('hook_1')!, { metadata: 'payload' }); + t.create.mockImplementation(async (_id, data) => { + const result = t.answer(data); + const event = result.event!; + if (mode === 'gap before') + event.eventId = slotToEventId(data.correlationId === 'hook_0' ? 5 : 6); + if (data.correlationId === 'hook_1') { + if (mode === 'gap between') event.eventId = slotToEventId(6); + if (mode === 'missing event') delete result.event; + if (mode === 'missing cursor') delete result.eventCursor; + if (mode === 'duplicate slot') event.eventId = slotToEventId(4); + if (mode === 'old event') event.eventId = slotToEventId(2); + if (mode === 'wrong run') event.runId = 'wrun_other'; + if (mode === 'wrong correlation') event.correlationId = 'hook_other'; + if (mode === 'wrong token') (event as any).eventData.token = 'other'; + if (mode === 'wrong type') (event as any).eventType = 'hook_disposed'; + if (mode === 'non-slot ID') + event.eventId = 'evnt_01JXYZ12345678901234567890'; + if (mode === 'truncated page') + Object.assign(result, { + events: [], + cursor: 'truncated', + hasMore: true, + }); + if (mode === 'lazy metadata') + (event as any).eventData.metadata = { ref: 'unresolved' }; + if (mode === 'omitted metadata') + delete (event as any).eventData.metadata; + } + return result; + }); + expect((await t.invoke()).eventLogCarriedForward).toBe(false); + expect(t.eventLog.events).toEqual(t.seed); + expect(t.eventLog.cursor).toBe(`eid:${slotToEventId(3)}`); + expect(t.create).toHaveBeenCalledTimes(2); + expect(t.span.setAttributes).toHaveBeenCalledWith({ + 'workflow.concurrent_hooks.count': 2, + 'workflow.concurrent_hooks.carried_forward': false, + 'workflow.concurrent_hooks.fallback_reason': reason, + }); + }); + + it('does not hide an existing hole in the loaded prefix', async () => { + const t = setup(); + t.eventLog.events.splice(1, 1); + const before = [...t.eventLog.events]; + expect((await t.invoke()).eventLogCarriedForward).toBe(false); + expect(t.eventLog.events).toEqual(before); + expect(t.span.setAttributes).toHaveBeenCalledWith( + expect.objectContaining({ + 'workflow.concurrent_hooks.fallback_reason': 'prefix_gap', + }) + ); + }); + + it('uses the World cursor without assuming its encoding', async () => { + const t = setup(); + t.create.mockImplementation(async (_id, data) => ({ + ...t.answer(data), + eventCursor: `opaque:${data.correlationId}`, + })); + expect((await t.invoke()).eventLogCarriedForward).toBe(true); + expect(t.eventLog.cursor).toBe('opaque:hook_1'); + }); + + it('keeps a reload after a duplicate registration with no returned event', async () => { + const t = setup(); + t.create.mockImplementation(async (_id, data) => { + if (data.correlationId === 'hook_1') + throw new EntityConflictError('exists'); + return t.answer(data); + }); + expect((await t.invoke()).eventLogCarriedForward).toBe(false); + expect(t.eventLog.events).toEqual(t.seed); + expect(t.span.setAttributes).toHaveBeenCalledWith( + expect.objectContaining({ + 'workflow.concurrent_hooks.fallback_reason': 'missing_response', + }) + ); + }); + + it('settles siblings before propagating a stale-write failure, without fallback writes', async () => { + const t = setup(); + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + const stale = new PreconditionFailedError('stale'); + t.create.mockImplementation(async (_id, data) => { + if (data.correlationId === 'hook_0') throw stale; + started.resolve(); + await release.promise; + return t.answer(data); + }); + let settled = false; + const invocation = t.invoke().finally(() => { + settled = true; + }); + const rejection = expect(invocation).rejects.toBe(stale); + await started.promise; + expect(settled).toBe(false); + release.resolve(); + await rejection; + expect(t.create).toHaveBeenCalledTimes(2); + expect(t.eventLog.events).toEqual(t.seed); + }); + + it('does not retry an ambiguous transport failure alongside a committed sibling', async () => { + const t = setup(); + const error = new WorkflowWorldError('connection lost', { + code: 'TRANSPORT', + }); + t.create.mockImplementation(async (_id, data) => { + if (data.correlationId === 'hook_1') throw error; + return t.answer(data); + }); + await expect(t.invoke()).rejects.toBe(error); + expect(t.create).toHaveBeenCalledTimes(2); + expect(t.eventLog.events).toEqual(t.seed); + }); + + it('preserves one-shot replay recovery telemetry on individual requests', async () => { + const t = setup(); + const reporter = new ReplayRecoveryReporter(2); + reporter.activate(); + await t.invoke({ replayRecoveryReporter: reporter }); + expect( + t.create.mock.calls.filter((call) => call[2]?.replayDivergenceCount === 2) + ).toHaveLength(1); + }); + + it('keeps a reload when another suspension operation writes', async () => { + const t = setup(); + t.pending.set('wait_1', { + type: 'wait', + correlationId: 'wait_1', + resumeAt: new Date(Date.now() + 60_000), + }); + expect((await t.invoke()).eventLogCarriedForward).toBe(false); + const waitCall = t.create.mock.calls.find( + (call) => call[1].eventType === 'wait_created' + ); + expect(waitCall?.[2]?.eventCount).toBe(3); + for (const call of t.create.mock.calls.filter( + (call) => call[1].eventType === 'hook_created' + )) { + expect(call[2]?.eventCount).toBe(3); + expect(call[2]?.resolveData).toBeUndefined(); + } + expect(t.eventLog.events).toEqual(t.seed); + }); + + it.each([ + 'step', + 'attribute', + 'disposal', + 'abort', + 'force', + 'same token', + ])('keeps catch-up hints and avoids metadata echo when a sibling adds a %s', async (mode) => { + // Two independent hooks must not opt in when a third item makes the + // suspension ineligible, even though both would qualify individually. + const t = setup(); + Object.assign(t.pending.get('hook_0')!, { metadata: 'x'.repeat(10_000) }); + if (mode === 'step') { + t.pending.set('step_1', { + type: 'step', + correlationId: 'step_1', + stepName: 'test', + args: [], + }); + } else if (mode === 'attribute') { + t.pending.set('attr_1', { + type: 'attribute', + correlationId: 'attr_1', + changes: { status: 'idle' }, + }); + } else { + t.pending.set('hook_extra', { + type: 'hook', + correlationId: 'hook_extra', + token: mode === 'same token' ? 'customer:1' : 'customer:extra', + ...(mode === 'force' ? { force: true } : {}), + ...(mode === 'disposal' + ? { hasCreatedEvent: true, disposed: true } + : {}), + ...(mode === 'abort' + ? { hasCreatedEvent: true, abortRequested: true } + : {}), + }); + } + if (mode === 'abort') { + t.world.streams = { write: vi.fn(), close: vi.fn() } as any; + } + expect((await t.invoke()).eventLogCarriedForward).toBe(false); + for (const call of t.create.mock.calls.filter( + (call) => call[1].eventType === 'hook_created' + )) { + expect(call[2]?.eventCount).toBe(3); + expect(call[2]?.resolveData).toBeUndefined(); + } + expect(t.span.setAttributes).toHaveBeenCalledWith( + expect.objectContaining({ + 'workflow.concurrent_hooks.carried_forward': false, + 'workflow.concurrent_hooks.fallback_reason': + mode === 'force' || mode === 'same token' + ? 'dependent_hooks' + : 'other_suspension_writes', + }) + ); + }); + + it.each([ + 'legacy run', + 'no log', + 'empty log', + 'same token', + 'force', + 'disposed', + 'aborted', + ])('preserves the existing registration path for %s', async (mode) => { + const t = setup(); + const options: Partial[0]> = {}; + if (mode === 'legacy run') options.run = { ...run, specVersion: 5 }; + if (mode === 'no log') options.eventLog = undefined; + if (mode === 'empty log') options.eventLog = { events: [], cursor: null }; + if (mode === 'same token') + Object.assign(t.pending.get('hook_1')!, { token: 'customer:0' }); + if (mode === 'force') + Object.assign(t.pending.get('hook_1')!, { force: true }); + if (mode === 'disposed') + Object.assign(t.pending.get('hook_1')!, { disposed: true }); + if (mode === 'aborted') { + Object.assign(t.pending.get('hook_1')!, { abortRequested: true }); + t.world.streams = { write: vi.fn(), close: vi.fn() } as any; + } + expect((await t.invoke(options)).eventLogCarriedForward).toBe(false); + for (const call of t.create.mock.calls.filter( + (call) => call[1].eventType === 'hook_created' + )) { + expect(call[2]?.resolveData).toBeUndefined(); + } + }); +}); diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 9d81c2f0e5..3a7aef71ba 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -1119,7 +1119,7 @@ export interface SlotSnapshotParams { * | `run_started` | replay loop, first write of a delivery | `eventCount` when a log is loaded; no cursor, the `run_started` preload owns the response fields | the preload already returns the full log | * | `step_created`, `wait_created`, `hook_disposed`, `attr_set` | suspension handler (`createGuarded`) | `eventCount` | the handler keeps writing from, and resumes replay from, this log | * | `step_created` | queue-consumer re-ensure (raw) | nothing | runs from a queue message, no log | - * | `hook_created` | suspension handler (node), `dispatchPendingOps` (QuickJS) | `eventCount` + `sinceCursor` (only when the suspension creates exactly one hook) | the delta is how an already-received hook resolves without a re-invocation; two creates against one cursor would give two deltas of which only the first could be taken | + * | `hook_created` | suspension handler (node), `dispatchPendingOps` (QuickJS) | `eventCount` + `sinceCursor` (only when the suspension creates exactly one hook); independent concurrent node registrations in a hooks-only suspension defer both | a lone create returns a delta; concurrent registrations reconcile committed responses together, reloading when they do not prove a consecutive suffix | * | `hook_received` | replay loop lazy resume; handler in-suspension resume | `eventCount` (preload owns the cursor fields) | both replay from the log right after | * | `hook_received` | webhook / `resumeHook` (raw) | nothing | out-of-band, enqueues a delivery that loads the log | * | `wait_completed` | replay loop timer path | `eventCount` | the loop keeps replaying from the log | diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index f3bcf5beb4..24893dfe58 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -12,7 +12,9 @@ import { AttributeValidationError, type CreateEventParams, type CreateEventRequest, + type Event, type EventResult, + eventIdToSlot, type SerializedData, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_COMPRESSION, @@ -52,6 +54,7 @@ import { import { absorbSkippedSlotReport, type EventCreator, + findEventSlotGap, type LoadedEventLog, maxEventSlot, mergeReportedEvents, @@ -75,14 +78,17 @@ export interface SuspensionHandlerParams { span?: Span; requestId?: string; /** - * The runtime's loaded event log. Every event creation this suspension makes - * names the position it was derived from, so a backend that has recorded + * The runtime's loaded event log. Event creations normally name the position + * they were derived from, so a backend that has recorded * events the replay did not see can report them back on the write, or, if * it would rather refuse than report, reject it with a 412. A rejection is * not retried here: the event's correlation id was minted by *this* replay's * seeded sequence, so re-committing it against a corrected log would persist * an event no correct replay produces. The caller restarts the replay * instead. + * Independent concurrent hook registrations defer advisory catch-up until + * all of their existing single-event requests settle; a gap in their returned + * committed positions retains the caller's normal reload path. * * Extended in place by what those writes report back — the events on slots * they skipped over, and (on the hook create) the inline delta since its @@ -308,12 +314,10 @@ export interface SuspensionHandlerResult { * committed, so a caller that continues in this process can replay — or * resume a retained VM — straight off it with no read. * - * True only when the hook create's inline delta came back complete and was - * folded in (see `hookDeltaCursor` below), and nothing else in this - * suspension wrote an event. False whenever a read is needed first: no - * delta was asked for or returned, it was truncated, or a step / wait / - * attribute / abort write (single or batched) also committed and may not be - * in it. + * True when a lone hook create returns a complete inline delta, or when + * concurrent hook responses prove a complete consecutive suffix. In either + * case every write in this suspension must be accounted for. Missing events, + * intervening writers, or other suspension writes retain the reload path. * * Indifferent to which event the create committed: the delta is the slice * of the log after the caller's cursor either way, so it carries a @@ -799,6 +803,52 @@ export async function handleSuspension({ ? eventLog.cursor : undefined; + // Independent registrations share history reconciliation, not a new write + // API. Keep their existing requests parallel, but omit each request's + // advisory catch-up position: a sibling's commit is not unknown history. + // Actual write fencing, validation, and token ownership remain the World's. + // Same-token lifecycle operations and forced claims keep their original path. + // Mixed suspensions cannot carry this suffix forward, so retain their normal + // catch-up hints and avoid resolving metadata that replay would discard. + const hasOtherSuspensionWrites = + stepItems.some((item) => !item.hasCreatedEvent) || + waitItems.some((item) => !item.hasCreatedEvent) || + attributeItems.length > 0 || + allHookItems.some((item) => item.disposed || item.abortRequested); + const concurrentHookIneligibility = !eventLog + ? 'no_event_log' + : (run.specVersion ?? 0) < SPEC_VERSION_SUPPORTS_SLOT_IDENTITY + ? 'unsupported_spec' + : hasOtherSuspensionWrites + ? 'other_suspension_writes' + : hooksNeedingCreation.some( + (item) => + item.force || hookItemsByToken.get(item.token)?.length !== 1 + ) + ? 'dependent_hooks' + : undefined; + const concurrentHooks = concurrentHookIneligibility + ? [] + : hooksNeedingCreation; + const hookFrontier = + concurrentHooks.length > 1 && eventLog + ? maxEventSlot(eventLog.events) + : undefined; + const concurrentHookIds = new Set( + hookFrontier === undefined + ? [] + : concurrentHooks.map((item) => item.correlationId) + ); + const hookResults = new Map(); + const createConcurrentHook: EventCreator = async (data, params) => { + eventWrites++; + // Replay needs actual metadata bytes, not a lazy payload reference. This + // uses the existing single-event API's resolution option. + const result = await createEvent(data, { ...params, resolveData: 'all' }); + if (data.correlationId) hookResults.set(data.correlationId, result); + return result; + }; + const processHookGroup = async ( items: HookInvocationQueueItem[] ): Promise => { @@ -832,7 +882,9 @@ export async function handleSuspension({ queueItem, requestId, sinceCursor: hookDeltaCursor, - createEvent: createGuarded, + createEvent: concurrentHookIds.has(queueItem.correlationId) + ? createConcurrentHook + : createGuarded, world, forceClaimVictimWakes, }); @@ -1953,6 +2005,78 @@ export async function handleSuspension({ await owedVictimWakes; } + // Reconcile only after every write has settled. All responses must match + // their requests and fill exactly frontier+1..frontier+N. A missing response, + // duplicate old event, or a foreign event between these commits requires the + // caller's existing incremental reload; never advance a cursor across a gap. + const reconcileConcurrentHooks = (): string | undefined => { + if (concurrentHookIneligibility) return concurrentHookIneligibility; + if (!eventLog || hookFrontier === undefined) return 'missing_frontier'; + if (hookResults.size !== concurrentHookIds.size) return 'missing_response'; + if (eventWrites !== hookResults.size) return 'other_suspension_writes'; + if (maxEventSlot(eventLog.events) !== hookFrontier) return 'prefix_changed'; + if (findEventSlotGap(eventLog.events) !== undefined) return 'prefix_gap'; + + const committed: { event: Event; slot: number; cursor?: string }[] = []; + for (const item of concurrentHooks) { + const result = hookResults.get(item.correlationId)!; + const event = result.event; + if (!event) return 'missing_event'; + if (result.hasMore === true) return 'incomplete_response'; + if ( + event.runId !== runId || + event.correlationId !== item.correlationId || + (event.eventType !== 'hook_created' && + event.eventType !== 'hook_conflict') || + event.eventData.token !== item.token + ) { + return 'event_mismatch'; + } + if ( + event.eventType === 'hook_created' && + ((event.eventData.metadata !== undefined && + !(event.eventData.metadata instanceof Uint8Array)) || + (item.metadata !== undefined && + event.eventData.metadata === undefined)) + ) { + return 'unresolved_metadata'; + } + const slot = eventIdToSlot(event.eventId); + if (slot === null) return 'non_slot_event'; + committed.push({ event, slot, cursor: result.eventCursor }); + } + committed.sort((a, b) => a.slot - b.slot); + if ( + !committed.every(({ slot }, index) => slot === hookFrontier + index + 1) + ) { + return 'nonconsecutive_slots'; + } + const last = committed.at(-1); + if (!last?.cursor) return 'missing_event_cursor'; + eventLog.events.push(...committed.map(({ event }) => event)); + eventLog.cursor = last.cursor; + return undefined; + }; + let concurrentHooksCarriedForward = false; + if (hooksNeedingCreation.length > 1) { + const fallbackReason = reconcileConcurrentHooks(); + concurrentHooksCarriedForward = fallbackReason === undefined; + span?.setAttributes({ + 'workflow.concurrent_hooks.count': hooksNeedingCreation.length, + 'workflow.concurrent_hooks.carried_forward': + concurrentHooksCarriedForward, + // A span can cover several suspensions; clear a previous fallback when + // this registration set succeeds instead of leaving a stale reason. + 'workflow.concurrent_hooks.fallback_reason': fallbackReason ?? 'none', + }); + runtimeLogger.debug('Reconciled concurrent hook registrations', { + workflowRunId: runId, + hookCount: hooksNeedingCreation.length, + carriedForward: concurrentHooksCarriedForward, + ...(fallbackReason ? { fallbackReason } : {}), + }); + } + // The hook writes' share of this suspension's wall time: only the stretch // they held it after every other write had settled, since until then the // suspension was waiting on those writes too. @@ -2020,10 +2144,10 @@ export async function handleSuspension({ hookConflictCorrelationIds, hasAwaitedHookCreation: awaitedHookCorrelationIds.length > 0, awaitedHookCorrelationIds, - // The delta accounts for the whole log only if the write that returned it - // was this suspension's only one — any other write may have landed above - // the delta and be missing from the caller's log. - eventLogCarriedForward: deltaAbsorbed && eventWrites === 1, + // A lone delta or a proven consecutive suffix must account for every + // suspension write; otherwise the caller reloads before continuing. + eventLogCarriedForward: + (deltaAbsorbed && eventWrites === 1) || concurrentHooksCarriedForward, hasAttributeEvents: attributeItems.length > 0, hasHookEvents: hooksNeedingCreation.length > 0, hookCreationMs, diff --git a/packages/world-vercel/README.md b/packages/world-vercel/README.md index ec80ca049e..2eea915c2c 100644 --- a/packages/world-vercel/README.md +++ b/packages/world-vercel/README.md @@ -6,6 +6,23 @@ Integrates with Vercel's infrastructure for storage, queuing, and authentication Used by default for deployments on Vercel. Authentication and API endpoints are configured automatically in Vercel deployments. +## Concurrent hook registrations + +Concurrent independent hook registrations use the existing single-event +requests. The Node.js runtime defers per-request history catch-up and reconciles +the committed events after all requests settle. The adapter supplies an optional +`eventCursor` in Vercel's cursor format and honors resolved hook metadata for this +path. A consecutive suffix needs no history read; missing or intervening events +retain the normal reload before continuation. No server upgrade or batch +registration endpoint is required. + +The runtime applies this optimization only to hooks-only suspensions whose +registrations are all independent. Mixed suspensions, forced claims, and +same-token lifecycle operations keep the normal catch-up path without resolving +metadata for unused mutation responses. Runtime spans record +`workflow.concurrent_hooks.carried_forward` and, on fallback, +`workflow.concurrent_hooks.fallback_reason`; debug logs include the same outcome. + ## Connection failures Backend connection failures and interrupted event streams follow existing retry policies, including failures with unrecognized error codes. Repeated HTTP/2 session failures rebuild the shared events connection pool. Invalid backend URL protocols, embedded credentials, Fetch-blocked ports, and unsupported request headers fail immediately. Interrupted event writes retain their existing in-process retries; caller cancellations are excluded. diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index 56199ef9b3..1b960146d3 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -9,6 +9,7 @@ import { SEALED_LOG_ENV_VAR, SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT, SPEC_VERSION_SUPPORTS_SLOT_IDENTITY, + slotToEventId, } from '@workflow/world'; import { decode, encode } from 'cbor-x'; import { ulid } from 'ulid'; @@ -35,6 +36,217 @@ import { const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; const STARTED_AT = new Date('2026-06-10T00:00:00.000Z'); +describe('committed event cursors on existing single-event requests', () => { + it.each([ + 'hook_created', + 'hook_conflict', + ] as const)('preserves the listed replay representation of a persisted %s response', async (eventType) => { + const agent = mockAgent(); + const eventId = slotToEventId(4); + const requestedRetention = new Date('2026-06-12T12:00:00.000Z'); + const storedRetention = new Date('2026-06-11T00:00:00.000Z'); + const metadata = new Uint8Array(2_048).fill(42); + // These are persisted values, deliberately different from the request: + // registration time, normalized retention and token-conflict details. + const stored = { + eventId, + runId: 'wrun_1', + correlationId: 'hook_0', + specVersion: 6, + createdAt: STARTED_AT, + eventType, + eventData: + eventType === 'hook_created' + ? { + token: 'customer:0', + tokenRetentionUntil: storedRetention, + metadata, + isWebhook: true, + } + : { + token: 'customer:0', + conflictingRunId: 'wrun_owner', + forceRefusedReason: 'victim-spec-version', + }, + }; + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_created', + method: 'POST', + }) + .reply(200, encode({ event: stored }), { + headers: { 'content-type': 'application/cbor' }, + }); + const body = eventType === 'hook_created' ? metadata : new Uint8Array(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events', + method: 'GET', + query: { returnAll: 'true', remoteRefBehavior: 'resolve' }, + }) + .reply( + 200, + Buffer.concat([ + encodeFrame( + { + ...stored, + createdAt: STARTED_AT.toISOString(), + eventData: + eventType === 'hook_created' + ? { + ...stored.eventData, + metadata: { + _type: 'RemoteRef', + _ref: 's3rf:hook_0/metadata', + }, + } + : stored.eventData, + }, + body + ), + encodeFrame( + { _end: 1, next: `eid:${eventId}`, hasMore: false }, + new Uint8Array() + ), + ]), + { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } + ); + + const config = { token: 'test-token', dispatcher: agent }; + const result = await createWorkflowRunEvent( + 'wrun_1', + { + eventType: 'hook_created', + correlationId: 'hook_0', + specVersion: 6, + eventData: { + token: 'customer:0', + metadata, + isWebhook: true, + tokenRetentionUntil: requestedRetention, + }, + }, + { resolveData: 'all' }, + config + ); + const listed = await getWorkflowRunEvents( + { + runId: 'wrun_1', + resolveData: 'all', + }, + config + ); + expect(result.event).toEqual(listed.data[0]); + expect(result.event).toEqual(stored); + expect(result.eventCursor).toBe(listed.cursor); + agent.assertNoPendingInterceptors(); + }); + + it('returns World-owned cursors and resolved metadata without catch-up directives or a new endpoint', async () => { + const agent = mockAgent(); + const posted: Record[] = []; + const events = [0, 1].map((i) => ({ + eventId: slotToEventId(4 + i), + runId: 'wrun_1', + eventType: 'hook_created' as const, + correlationId: `hook_${i}`, + specVersion: 6, + createdAt: STARTED_AT, + eventData: { + token: `customer:${i}`, + metadata: new Uint8Array([i, 42]), + isWebhook: i === 1, + }, + })); + for (const event of events) { + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_created', + method: 'POST', + }) + .reply( + 200, + (opts: { body?: unknown }) => { + posted.push(decodePostedMeta(opts.body)); + return encode({ event }); + }, + { headers: { 'content-type': 'application/cbor' } } + ); + } + const results = await Promise.all( + events.map((event) => + createWorkflowRunEvent( + 'wrun_1', + event, + { resolveData: 'all' }, + { token: 'test-token', dispatcher: agent } + ) + ) + ); + expect(results.map((result) => result.eventCursor)).toEqual( + events.map((event) => `eid:${event.eventId}`) + ); + expect(results.map((result) => result.event)).toEqual(events); + for (const meta of posted) { + expect(meta.maxSlot).toBeUndefined(); + expect(meta.sinceCursor).toBeUndefined(); + expect(meta.remoteRefBehavior).toBe('resolve'); + } + expect(posted).toHaveLength(2); + agent.assertNoPendingInterceptors(); + }); + + it('preserves a server history page separately from the committed event cursor', async () => { + const agent = mockAgent(); + const event = { + eventId: slotToEventId(5), + runId: 'wrun_1', + eventType: 'hook_conflict', + correlationId: 'hook_1', + specVersion: 6, + createdAt: STARTED_AT, + eventData: { token: 'customer:1', conflictingRunId: 'wrun_owner' }, + }; + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hook_created', + method: 'POST', + }) + .reply( + 200, + encode({ + event, + events: [event], + cursor: 'page-cursor', + hasMore: true, + }), + { + headers: { 'content-type': 'application/cbor' }, + } + ); + const result = await createWorkflowRunEvent( + 'wrun_1', + { + eventType: 'hook_created', + correlationId: 'hook_1', + specVersion: 6, + eventData: { token: 'customer:1' }, + }, + { eventCount: 3 }, + { token: 'test-token', dispatcher: agent } + ); + expect(result.eventCursor).toBe(`eid:${event.eventId}`); + expect(result.events).toEqual([event]); + expect(result.cursor).toBe('page-cursor'); + expect(result.hasMore).toBe(true); + agent.assertNoPendingInterceptors(); + }); +}); + function createEventBody( event: AnyEventRequest, entities: Record = {} diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index a6c0948bdd..659702cf4c 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -803,11 +803,14 @@ async function createWorkflowRunEventInner( } } - const remoteRefBehavior: 'resolve' | 'lazy' = eventsNeedingResolve.has( - data.eventType - ) - ? 'resolve' - : 'lazy'; + const remoteRefBehavior: 'resolve' | 'lazy' = + eventsNeedingResolve.has(data.eventType) || + // Concurrent hook reconciliation consumes the committed event directly. + // Honor its explicit request for replay-ready metadata; other mutation + // responses keep their existing lightweight entity-only behavior. + (data.eventType === 'hook_created' && params?.resolveData === 'all') + ? 'resolve' + : 'lazy'; const { payload, meta } = splitEventDataForV4(data); @@ -932,12 +935,18 @@ async function createWorkflowRunEventInner( }; } - return createWorkflowRunEventV4( + const result = await createWorkflowRunEventV4( data.eventType === 'run_started' ? { ...input, eventType: 'run_started', skipPreload: true } : { ...input, eventType: data.eventType }, config ); + // This cursor addresses the committed event, not a complete history page. + // Construct it here, where the Vercel cursor format is owned, so the runtime + // can reconcile concurrent responses without interpreting opaque cursors. + return result.event + ? { ...result, eventCursor: `eid:${result.event.eventId}` } + : result; } /** diff --git a/packages/world/README.md b/packages/world/README.md index 50d68b6ac3..7bf345ce53 100644 --- a/packages/world/README.md +++ b/packages/world/README.md @@ -53,6 +53,22 @@ implementations must reject `step_started` on terminal runs, including redeliver for a step that still has `running` status. In-flight steps may still record `step_completed` or `step_failed` after the run ends. +## Committed event cursors + +`EventResult.eventCursor` optionally identifies the ascending `events.list` +position immediately after the returned `event`. The World owns its encoding; +the runtime must not derive a cursor from an event ID. A committed-event cursor +does not prove that earlier history is loaded. The Node.js runtime uses it only +after all concurrent hook registrations settle and their returned events prove +a complete consecutive suffix of its loaded log. Missing events or cursors keep +the normal reload path. Registrations use the existing single-event API. + +Returning `eventCursor` opts into replaying the returned `event` directly. It +must equal the event returned by `events.list` for that slot with the same +`resolveData` option, including stored timestamps, normalized retention, +resolved metadata, and hook-conflict details. Return persisted state rather than +echoing the creation request. + ## Implementation constraint: no mutable module state A World implementation must not keep mutable state at module scope. Hold it on diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 40ee6e0900..ea89e7e933 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -915,7 +915,11 @@ export interface CreateEventParams { * the replay loop and the suspension handler, which merge the report below * back into their loaded log. Omitted by callers with no loaded log to be * stale against, the step executor included: for those the report would be - * a read the World does for no one. + * a read the World does for no one. Independent concurrent hook registrations + * in a hooks-only suspension also omit it deliberately: the runtime verifies + * their committed events together after every write settles, and reloads if + * they do not form a complete suffix. Absence does not imply no loaded log; + * without a count there is no evidence for count-based stale rejection. * * A World's slots are dense and 1-based (see `Storage.events`), so a count * and a position are the same number. An id that is not a position does not @@ -1075,6 +1079,18 @@ export interface CreateEventParams { export type EventResult = { /** The created event (optional for legacy compatibility) */ event?: Event; + /** + * Optional ascending events.list cursor immediately after `event`, minted + * by the World that owns the cursor format. This does not attest that the + * caller holds any earlier events. The runtime may advance to it only after + * proving that its loaded prefix plus committed responses covers every slot + * through this event. Worlds that omit it retain the normal reload path. + * When supplied, `event` MUST match what ascending events.list returns for + * that slot with the same resolveData option, including stored timestamps, + * normalized retention, metadata bytes, and hook-conflict details. The + * runtime uses this stored event as replay input, not an echo of the request. + */ + eventCursor?: string; /** The workflow run entity (for run_* events) */ run?: WorkflowRun; /** The step entity (for step_* events) */